# -*- coding: utf-8 -*-
|
r"""匹配与校验.py —— Step1 的写入前一步：读「解析结果 + 目标明细表 + 投资系统查询结果」，产出变更清单与校验报告（只读）
|
|
用法：
|
python docs\投资\工具\匹配与校验.py --month 2026-07
|
|
输入：
|
1) docs\投资\输出\解析结果_<YYYY-MM>.tsv 由 市州月报解析.py 生成（市州报上来的项目行）
|
2) docs\投资\输出\X月客运站投资明细表.xlsx 目标表（客运，sheet「明细」）
|
docs\投资\输出\X月物流站场投资明细.xls 目标表（物流，sheet「 分项目投资完成情况」）
|
3) docs\投资\输入\模板_查询结果(投资系统-客运+物流 YYYY.M).xls
|
—— 本期那份（权威对照）；另按用户口径 (a) 找**上月**那份做校验基数
|
|
输出：
|
docs\投资\输出\变更清单_<YYYY-MM>.tsv 给「写入侧」用的逐行变更（动作/目标行号/6 列新值）
|
docs\投资\输出\匹配校验报告_<YYYY-MM>.md 人读：匹配统计、名称对不上清单、三方数值差异、月度校验
|
"""
|
import sys, os, io, re, argparse, importlib.util, difflib
|
|
_here = os.path.dirname(os.path.abspath(__file__))
|
_root = os.path.abspath(os.path.join(_here, '..', '..', '..'))
|
for _b in (_here, _root, os.getcwd()):
|
_lib = os.path.abspath(os.path.join(_b, '_pylibs'))
|
if os.path.isdir(_lib) and _lib not in sys.path:
|
sys.path.insert(0, _lib)
|
_spec = importlib.util.spec_from_file_location('_fmt', os.path.join(_here, '格式探测.py'))
|
F = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(F)
|
|
TARGETS = {
|
'客运': dict(path=os.path.join('docs', '投资', '输出', 'X月客运站投资明细表.xlsx'),
|
sheet='明细', kind='xlsx', name_col='B', city_col='A',
|
cols=dict(项目名称='B', 建设单位='C', 建设性质='D', 开工时间='E', 竣工时间='F',
|
总投资='G', 自开始建设累计='H', 本年计划投资='I', 自年初累计='J',
|
本月完成='K', 建设阶段='L', 形象进度='M')),
|
'物流': dict(path=os.path.join('docs', '投资', '输出', 'X月物流站场投资明细.xls'),
|
sheet=' 分项目投资完成情况', kind='xls', name_col='B', city_col='A',
|
cols=dict(项目名称='B', 建设单位='C', 建设性质='D', 开工时间='E', 竣工时间='F',
|
总投资='G', 自开始建设累计='H', 本年计划投资='I', 自年初累计='J',
|
本月完成='K', 建设阶段='L', 形象进度='M')),
|
}
|
WANT = ['自开始建设累计', '本年计划投资', '自年初累计', '本月完成', '建设阶段', '形象进度']
|
NUMF = ['总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成']
|
GROUP_PAT = re.compile(r'^(一|二|三|四|五|六|七|八)、')
|
SKIP_PAT = re.compile(r'^(小\s*计|合\s*计|总\s*计)$')
|
|
|
def normname(s, loose=False):
|
"""规范化项目名称用于匹配。
|
|
loose=False：只统一全角/半角、去空白（保留括号内信息）——精确匹配用；
|
loose=True ：再去掉所有括号字符与结尾的「项目」二字——判断「疑似同一项目」用
|
（例：鄂湘赣商贸物流中心一期 ⇄ 鄂湘赣商贸物流中心（一期）、
|
荆门智慧冷链物流园 ⇄ 荆门智慧冷链物流园项目）
|
"""
|
s = str(s or '')
|
for a, b in (('（', '('), ('）', ')'), ('【', '['), ('】', ']'), ('，', ','), ('、', ','),
|
('：', ':'), (' ', ''), (' ', ''), ('\u3000', '')):
|
s = s.replace(a, b)
|
s = s.strip().upper()
|
if loose:
|
for ch in '()[]':
|
s = s.replace(ch, '')
|
while s.endswith('项目'):
|
s = s[:-2]
|
s = s.rstrip('.。-')
|
return s
|
|
|
def to_num(v):
|
if isinstance(v, (int, float)) and not isinstance(v, bool):
|
return float(v)
|
s = str(v or '').replace(',', '').strip()
|
try:
|
return float(s)
|
except ValueError:
|
return None
|
|
|
def pretty(v):
|
if v is None:
|
return ''
|
if isinstance(v, float) and v.is_integer():
|
return str(int(v))
|
return ('%.4f' % v).rstrip('0').rstrip('.') if isinstance(v, float) else str(v)
|
|
|
def read_target(spec):
|
"""读目标明细表 → (行列表, 表头行号, 说明)"""
|
from openpyxl import load_workbook
|
import xlrd
|
rows = []
|
if spec['kind'] == 'xlsx':
|
wb = load_workbook(spec['path'], data_only=True)
|
ws = wb[spec['sheet']]
|
def cells(r, c):
|
return ws.cell(r, c).value
|
nrows, ncols = ws.max_row, ws.max_column
|
else:
|
bk = xlrd.open_workbook(spec['path'])
|
sh = bk.sheet_by_name(spec['sheet'])
|
def cells(r, c):
|
return sh.cell_value(r - 1, c - 1) if (r - 1 < sh.nrows and c - 1 < sh.ncols) else ''
|
nrows, ncols = sh.nrows, sh.ncols
|
namec = F.colindex(spec['name_col']) + 1
|
anchors = [F.colindex(spec['cols'][k]) + 1 for k in ('自开始建设累计', '自年初累计', '总投资')]
|
for r in range(1, nrows + 1):
|
name = str(cells(r, namec) or '').strip()
|
a = str(cells(r, 1) or '').strip()
|
if not name or len(name) < 3:
|
continue
|
if GROUP_PAT.match(F.norm(a) + F.norm(name)) or SKIP_PAT.match(F.norm(name)):
|
continue
|
if any(name.startswith(x) for x in F.BAD_NAME):
|
continue
|
if not any(to_num(cells(r, c)) is not None for c in anchors):
|
continue
|
rec = {'目标行号': r, '市州或行首': a}
|
for k, col in spec['cols'].items():
|
v = cells(r, F.colindex(col) + 1)
|
rec[k] = pretty(to_num(v)) if k in NUMF else str(v or '').strip()
|
rows.append(rec)
|
return rows
|
|
|
def read_system(path):
|
"""读投资系统查询结果 → {normname: {项目名称, 计划总投资, 自开始建设累计, 本年计划投资, 自年初累计, 本月完成, 时期}}"""
|
import xlrd
|
bk = xlrd.open_workbook(path)
|
sh = bk.sheet_by_index(0)
|
out, period = {}, ''
|
for r in range(1, sh.nrows):
|
name = str(sh.cell_value(r, 1)).strip()
|
if not name or name in ('单位',):
|
continue
|
period = period or str(sh.cell_value(r, 2)).strip()
|
out[normname(name)] = {
|
'项目名称': name, '时期': str(sh.cell_value(r, 2)).strip(),
|
'总投资': to_num(sh.cell_value(r, 3)),
|
'自开始建设累计': to_num(sh.cell_value(r, 4)),
|
'本年计划投资': to_num(sh.cell_value(r, 5)),
|
'自年初累计': to_num(sh.cell_value(r, 6)),
|
'本月完成': to_num(sh.cell_value(r, 7)),
|
}
|
return out, period
|
|
|
def find_system_files(indir, ym):
|
"""在输入目录里找「本月/上月」投资系统查询结果。文件名或『时期』列里的年月都认。"""
|
found = {}
|
if not os.path.isdir(indir):
|
return found
|
for f in os.listdir(indir):
|
if not f.lower().endswith(('.xls', '.xlsx')) or '查询结果' not in f:
|
continue
|
m = re.search(r'(20\d\d)\s*[-.年]\s*(\d{1,2})', f)
|
key = '%s%02d' % (m.group(1), int(m.group(2))) if m else None
|
if key is None:
|
try:
|
_d, period = read_system(os.path.join(indir, f))
|
m2 = re.search(r'(20\d\d)\D+(\d{1,2})', period or '')
|
key = '%s%02d' % (m2.group(1), int(m2.group(2))) if m2 else None
|
except Exception:
|
key = None
|
if key:
|
found[key] = os.path.join(indir, f)
|
return found
|
|
|
def prev_ym(ym):
|
y, m = int(ym[:4]), int(ym[4:])
|
return '%d%02d' % (y - 1, 12) if m == 1 else '%d%02d' % (y, m - 1)
|
|
|
def main():
|
ap = argparse.ArgumentParser()
|
ap.add_argument('--month', required=True)
|
ap.add_argument('--root', default=os.path.join('docs', '投资'))
|
a = ap.parse_args()
|
ym = a.month.replace('-', '')
|
out = os.path.join(a.root, '输出')
|
tsv_in = os.path.join(out, '解析结果_%s.tsv' % a.month)
|
if not os.path.exists(tsv_in):
|
print('缺少解析结果：%s（先跑 市州月报解析.py）' % tsv_in); return
|
|
parsed = []
|
for line in io.open(tsv_in, encoding='utf-8-sig').read().splitlines()[1:]:
|
p = line.split('\t')
|
if len(p) < 8:
|
continue
|
parsed.append(dict(zip(['品类', '市州', '文件', '工作表', '行号', '序号', '分类'] +
|
['项目名称', '建设单位', '建设性质', '开工时间', '竣工时间', '总投资',
|
'自开始建设累计', '本年计划投资', '自年初累计', '本月完成',
|
'建设阶段', '形象进度'], p)))
|
rep = ['# 匹配校验报告（%s）' % a.month, '']
|
rep.append('> 只读：本步骤不改任何文件；输出的《变更清单》供「写入侧」执行。')
|
rep.append('')
|
changes = []
|
|
# 投资系统文件（本期 / 上月）
|
indir = os.path.join(a.root, '输入')
|
sysfiles = find_system_files(indir, ym)
|
cur_path = sysfiles.get(ym)
|
pre_path = sysfiles.get(prev_ym(ym))
|
cur_sys = cur_pre = {}
|
if cur_path:
|
cur_sys, _ = read_system(cur_path)
|
if pre_path:
|
cur_pre, _ = read_system(pre_path)
|
rep.append('## 一、投资系统数据（本期 / 上月）')
|
rep.append('')
|
rep.append('| 用途 | 文件 | 项目数 |')
|
rep.append('| --- | --- | --- |')
|
rep.append('| 本期（权威对照） | %s | %d |' % (os.path.basename(cur_path) if cur_path else '**未找到**', len(cur_sys)))
|
rep.append('| 上月（校验基数） | %s | %d |' % (os.path.basename(pre_path) if pre_path else '**未找到（用户口径 a：需每月放一份）**', len(cur_pre)))
|
rep.append('')
|
|
for cat, spec in TARGETS.items():
|
rep.append('## %s —— 目标明细表' % cat)
|
rep.append('')
|
if not os.path.exists(spec['path']):
|
rep.append('目标表不存在：%s' % spec['path']); rep.append(''); continue
|
tgt = read_target(spec)
|
prow = [r for r in parsed if r['品类'] == cat]
|
tmap = {}
|
for t in tgt:
|
tmap.setdefault(normname(t['项目名称']), []).append(t)
|
pmap = {}
|
for p in prow:
|
pmap.setdefault(normname(p['项目名称']), []).append(p)
|
|
matched = [k for k in tmap if k in pmap]
|
raw_new = [k for k in pmap if k not in tmap] # 源文件有、目标表没有
|
only_tgt = [k for k in tmap if k not in pmap] # 目标表有、源文件没有 → 保持原值
|
# 新增候选：先看是否与目标表某个项目「宽松等价」或高度相似 → 疑似同一项目，不新增，改为弹窗提示
|
loose_tgt = {}
|
for k in tmap:
|
loose_tgt.setdefault(normname(k, loose=True), []).append(k)
|
tkeys = list(tmap.keys())
|
new_in_src, suspicious = [], []
|
for k in raw_new:
|
nl = normname(k, loose=True)
|
if nl in loose_tgt:
|
suspicious.append((k, loose_tgt[nl][0], '宽松等价（去括号/去“项目”二字后相同）'))
|
continue
|
close = difflib.get_close_matches(nl, [normname(x, loose=True) for x in tkeys], n=1, cutoff=0.90)
|
if close:
|
hit = tkeys[[normname(x, loose=True) for x in tkeys].index(close[0])]
|
suspicious.append((k, hit, '高度相似 %.2f' % difflib.SequenceMatcher(None, nl, close[0]).ratio()))
|
continue
|
new_in_src.append(k)
|
rep.append('- 目标表项目行 **%d**；市州月报解析出 **%d** 行' % (len(tgt), len(prow)))
|
rep.append('- 按项目名称**精确匹配**上 **%d**（→ 更新 6 列）' % len(matched))
|
rep.append('- **真正需要新增** **%d**' % len(new_in_src))
|
rep.append('- **疑似同一项目、需人工确认** **%d**（按用户口径：**不自动新增、不动名称，弹窗提示**）' % len(suspicious))
|
rep.append('- 目标表有但本期源文件没有（保持原值）**%d**' % len(only_tgt))
|
rep.append('')
|
if suspicious:
|
rep.append('### ⚠️ 疑似同一项目但名称写法不同 → 需人工确认（程序不自动合并、也不重复新增）')
|
rep.append('')
|
rep.append('| 源文件项目名称 | 目标表已有项目名称 | 判定依据 |')
|
rep.append('| --- | --- | --- |')
|
for a1, b1, why in suspicious:
|
rep.append('| %s | %s | %s |' % (a1, b1, why))
|
rep.append('')
|
if new_in_src:
|
rep.append('### ➕ 将新增的项目（%d 个）' % len(new_in_src))
|
rep.append('')
|
rep.append('| 项目名称 | 来源文件 | 自开始建设累计 | 自年初累计 | 本月完成 | 名称相似度提示 |')
|
rep.append('| --- | --- | --- | --- | --- | --- |')
|
loose_keys = [normname(x, loose=True) for x in tkeys]
|
for k in new_in_src:
|
p0 = pmap[k][0]
|
nl = normname(k, loose=True)
|
close = difflib.get_close_matches(nl, loose_keys, n=1, cutoff=0.80)
|
hint = ''
|
if close:
|
hit = tkeys[loose_keys.index(close[0])]
|
hint = '⚠️ 与「%s」相似 %.2f，请确认是否同一项目' % (
|
hit, difflib.SequenceMatcher(None, nl, close[0]).ratio())
|
rep.append('| %s | %s | %s | %s | %s | %s |' % (
|
p0['项目名称'], os.path.basename(p0['文件'])[:30], p0.get('自开始建设累计', ''),
|
p0.get('自年初累计', ''), p0.get('本月完成', ''), hint))
|
rep.append('')
|
|
# 变更清单
|
for k in matched:
|
for p in pmap[k]:
|
t = tmap[k][0]
|
changes.append({'品类': cat, '动作': '更新', '目标行号': t['目标行号'],
|
'项目名称': t['项目名称'], '写入来源': os.path.basename(p['文件']),
|
**{c: p.get(c, '') for c in WANT}})
|
for k in new_in_src:
|
for p in pmap[k]:
|
changes.append({'品类': cat, '动作': '新增', '目标行号': '',
|
'项目名称': p['项目名称'], '写入来源': os.path.basename(p['文件']),
|
**{c: p.get(c, '') for c in WANT}})
|
|
# 三方数值比对：源文件 vs 投资系统（本期）
|
if cur_sys:
|
diff, missing = [], []
|
for k in matched:
|
p, t = pmap[k][0], tmap[k][0]
|
s = cur_sys.get(k)
|
if not s:
|
missing.append(t['项目名称'])
|
continue
|
for c in ('自开始建设累计', '自年初累计', '本月完成'):
|
pv, sv = to_num(p.get(c)), s.get(c)
|
if pv is None or sv is None:
|
continue
|
if abs(pv - sv) > 0.05:
|
diff.append((t['项目名称'], c, pretty(pv), pretty(sv), pretty(sv - pv)))
|
rep.append('')
|
rep.append('### 源文件 vs 投资系统（本期）：**数值不一致 %d 处**（需弹窗提示）' % len(diff))
|
rep.append('')
|
if diff:
|
rep.append('| 项目名称 | 字段 | 市州月报 | 投资系统 | 差额 |')
|
rep.append('| --- | --- | --- | --- | --- |')
|
for nm1, c1, x1, y1, d1 in diff[:60]:
|
rep.append('| %s | %s | %s | %s | %s |' % (nm1, c1, x1, y1, d1))
|
rep.append('')
|
else:
|
rep.append('（无）'); rep.append('')
|
rep.append('### 投资系统中查不到的项目：**%d** 个（不计为差错；投资系统可能只收录省重点项目）' % len(missing))
|
rep.append('')
|
for nm1 in missing[:60]:
|
rep.append('- %s' % nm1)
|
rep.append('')
|
|
# 月度校验：本月自年初累计 = 上月自年初累计 + 本月完成
|
if cur_pre:
|
bad = []
|
for k in matched:
|
p, t = pmap[k][0], tmap[k][0]
|
nowj = to_num(p.get('自年初累计'))
|
nowk = to_num(p.get('本月完成'))
|
pre = cur_pre.get(k)
|
if nowj is None or nowk is None or not pre:
|
continue
|
prej = pre.get('自年初累计')
|
if prej is None:
|
continue
|
if abs(nowj - (prej + nowk)) > 0.05:
|
bad.append((t['项目名称'], pretty(prej), pretty(nowk), pretty(nowj), pretty(prej + nowk)))
|
rep.append('### 月度校验「本月自年初累计 = 上月自年初累计 + 本月完成」：不符 **%d** 项' % len(bad))
|
rep.append('')
|
if bad:
|
rep.append('| 项目名称 | 上月自年初累计 | 本月完成 | 本月自年初累计 | 等式结果（将纠正为该值） |')
|
rep.append('| --- | --- | --- | --- | --- |')
|
for r1, r2, r3, r4, r5 in bad[:60]:
|
rep.append('| %s | %s | %s | %s | %s |' % (r1, r2, r3, r4, r5))
|
rep.append('')
|
else:
|
rep.append('### 月度校验：**跳过**（未找到上月投资系统数据表，按用户口径 (a) 需每月放一份）')
|
rep.append('')
|
|
cols = ['品类', '动作', '目标行号', '项目名称', '写入来源'] + WANT
|
p_tsv = os.path.join(out, '变更清单_%s.tsv' % a.month)
|
with io.open(p_tsv, 'w', encoding='utf-8-sig') as fh:
|
fh.write('\t'.join(cols) + '\n')
|
for c in changes:
|
fh.write('\t'.join(str(c.get(k, '')) for k in cols) + '\n')
|
p_md = os.path.join(out, '匹配校验报告_%s.md' % a.month)
|
io.open(p_md, 'w', encoding='utf-8').write('\n'.join(rep) + '\n')
|
print('变更条数 = %d' % len(changes))
|
print('wrote', p_md)
|
print('wrote', p_tsv)
|
|
|
if __name__ == '__main__':
|
main()
|
