# -*- 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 = ['总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成']
|
CITY17 = ('武汉', '黄石', '十堰', '宜昌', '襄阳', '鄂州', '荆门', '孝感', '荆州', '黄冈',
|
'咸宁', '随州', '恩施', '仙桃', '潜江', '天门', '神农架')
|
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 normname2(s):
|
"""投资系统比对专用：在 normname 基础上再去掉间隔号等纯装饰字符。"""
|
for ch in ('·', '．', '•', '▪', '●', '.', '-'):
|
s = str(s or '').replace(ch, '')
|
return normname(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, notes, infos, merges = [], [], [], []
|
KEYF = ('项目名称', '总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成')
|
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
|
merges = [(r.min_row - 1, r.max_row, r.min_col - 1, r.max_col)
|
for r in ws.merged_cells.ranges]
|
else:
|
try:
|
bk = xlrd.open_workbook(spec['path'], formatting_info=True)
|
except Exception:
|
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
|
merges = list(getattr(sh, 'merged_cells', []) or [])
|
|
# 1) 按表头定位列（两级表头 / 同名列都在 pick_columns 里消歧）
|
cols = {}
|
try:
|
_c, _h = F.pick_columns(lambda r, c: cells(r + 1, c + 1), nrows, ncols, merges)
|
for k in spec['cols']:
|
if k in _c:
|
cols[k] = F.colname(_c[k])
|
except Exception as e:
|
notes.append('表头定位异常(%s: %s)，全部退回母版默认列位' % (type(e).__name__, str(e)[:40]))
|
# 2) 未找到的退回母版默认列位；与默认不一致的记录告警（关键列）/提示（其余列）
|
for k, exp in spec['cols'].items():
|
got = cols.get(k)
|
if got is None:
|
msg = '%s：表头未找到，按母版默认 %s 列' % (k, exp)
|
(notes if k in KEYF else infos).append(msg)
|
cols[k] = exp
|
elif got != exp:
|
msg = '%s：表头定位 %s 列，母版默认 %s 列' % (k, got, exp)
|
(notes if k in KEYF else infos).append(msg)
|
|
namec = F.colindex(cols['项目名称']) + 1
|
anchors = [F.colindex(cols[k]) + 1 for k in ('自开始建设累计', '自年初累计', '总投资') if k in cols]
|
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 in spec['cols']:
|
v = cells(r, F.colindex(cols[k]) + 1)
|
rec[k] = pretty(to_num(v)) if k in NUMF else str(v or '').strip()
|
rows.append(rec)
|
return rows, cols, notes, infos
|
|
|
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 normname3(s):
|
"""再去掉括号及其内容（用于识别「名称多了括注」的同一项目）。"""
|
t = normname(s)
|
t = re.sub(r'\([^)]*\)', '', t)
|
for ch in ('·', '．', '•', '●', '-', '.', '·'):
|
t = t.replace(ch, '')
|
return t
|
|
|
def best_name_hint(key, tkeys, cutoff=0.80):
|
"""在目标表项目名里找与 key 最像的一个（宽松/去括注两种口径取最优）。
|
|
返回 (目标表项目名, 相似度)；找不到返回 (None, 0)。
|
"""
|
best = (None, 0.0)
|
for normf in (lambda s: normname(s, loose=True), normname3):
|
base = normf(key)
|
table = [normf(x) for x in tkeys]
|
hit = difflib.get_close_matches(base, table, n=1, cutoff=cutoff)
|
if hit:
|
ratio = difflib.SequenceMatcher(None, base, hit[0]).ratio()
|
if ratio > best[1]:
|
best = (tkeys[table.index(hit[0])], ratio)
|
return best
|
|
|
def sys_lookup(key, table):
|
"""在投资系统字典里找项目：先精确（含去间隔号），再宽松+相似度（>=0.90）。
|
|
返回 (记录, 说明)；找不到返回 (None, '')。
|
"""
|
if not table:
|
return None, ''
|
s = table.get(key)
|
if s:
|
return s, ''
|
for kk, vv in table.items():
|
if normname2(kk) == normname2(key):
|
return vv, '名称写法不同（投资系统：%s）' % vv['项目名称']
|
idx = {}
|
for kk in table:
|
idx.setdefault(normname2(kk), []).append(kk)
|
cands = difflib.get_close_matches(normname2(key), list(idx.keys()), n=1, cutoff=0.90)
|
if cands:
|
kk = idx[cands[0]][0]
|
return table[kk], '名称写法不同（投资系统：%s）' % table[kk]['项目名称']
|
return None, ''
|
|
|
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 = []
|
alerts = []
|
|
def add_alert(level, kind, cat, city, name, src, detail, action):
|
alerts.append(dict(级别=level, 类型=kind, 品类=cat, 市州=city, 项目名称=name,
|
来源=src, 详情=detail, 建议动作=action))
|
|
# 源文件解析校验（由 市州月报解析.py 产出）
|
checks = []
|
p_ck = os.path.join(out, '解析校验_%s.tsv' % a.month)
|
if os.path.exists(p_ck):
|
_lines = io.open(p_ck, encoding='utf-8-sig').read().splitlines()
|
if _lines:
|
_ckc = _lines[0].split('\t')
|
for _ln in _lines[1:]:
|
_v = _ln.split('\t')
|
_v += [''] * (len(_ckc) - len(_v))
|
checks.append(dict(zip(_ckc, _v)))
|
|
# 投资系统文件（本期 / 上月）
|
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('')
|
|
# —— 弹窗规则：投资系统文件缺失 / 源文件解析异常 / 源文件内部对不上 ——
|
if not cur_path:
|
add_alert('阻断', '缺本期投资系统表', '', '', '', os.path.basename(indir),
|
'未在「输入」目录找到 %s 的投资系统查询结果' % a.month,
|
'把本期投资系统导出的《模板_查询结果》放进 docs\\投资\\输入\\ 后重跑')
|
if not pre_path:
|
add_alert('警告', '缺上月投资系统表', '', '', '', os.path.basename(indir),
|
'未找到 %s 的投资系统查询结果 → 月度校验（本月自年初累计 = 上月自年初累计 + 本月完成）无法执行' % prev_ym(a.month),
|
'每月把上月、本月两份投资系统查询结果都放进 docs\\投资\\输入\\')
|
if not checks:
|
add_alert('阻断', '缺源文件解析校验', '', '', '', os.path.basename(p_ck),
|
'未找到《解析校验》清单，无法判断源文件是否自洽、是否解析失败',
|
'先跑 市州月报解析.py --month %s' % a.month)
|
for ck in checks:
|
st = ck.get('状态', 'OK')
|
if st and st != 'OK':
|
add_alert('阻断' if st.startswith('解析失败') else '警告', '源文件解析异常',
|
ck.get('品类', ''), ck.get('市州', ''), '', ck.get('文件', ''),
|
st, '人工打开源文件确认表头/数据；必要时登记特例后再导入')
|
elif ck.get('列位') and not all(k in ck['列位'] for k in ('自开始建设累计=', '自年初累计=', '本月完成=')):
|
_miss = [k for k in ('自开始建设累计', '自年初累计', '本月完成') if k + '=' not in ck['列位']]
|
add_alert('警告', '源文件缺关键列', ck.get('品类', ''), ck.get('市州', ''), '', ck.get('文件', ''),
|
'按表头没找到列：%s（实际识别到的列位：%s）' % ('、'.join(_miss), ck['列位']),
|
'人工确认源文件表头是否被改过；必要时登记为特例')
|
elif ck.get('一致性', '') == '⚠️ 不一致':
|
add_alert('警告', '源文件内部对不上', ck.get('品类', ''), ck.get('市州', ''), '', ck.get('文件', ''),
|
'项目行求和 %s/%s/%s ≠ %s 行合计 %s/%s/%s' % (
|
ck.get('求和_自开始建设累计', ''), ck.get('求和_自年初累计', ''), ck.get('求和_本月完成', ''),
|
ck.get('合计行', ''), ck.get('合计_自开始建设累计', ''),
|
ck.get('合计_自年初累计', ''), ck.get('合计_本月完成', '')),
|
'让市州核对源文件（合计行或项目行有错），确认后再导入')
|
|
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('')
|
add_alert('阻断', '缺目标明细表', cat, '', '', spec['path'],
|
'目标明细表不存在，本季度该品类无法写入', '把模板表放到 docs\\投资\\输出\\')
|
continue
|
tgt, tcols, tnotes, tinfos = read_target(spec)
|
if tnotes:
|
add_alert('警告', '目标表列位置与预期不符', cat, '', '', os.path.basename(spec['path']),
|
'；'.join(tnotes[:8]),
|
'确认母版表头是否被改动；程序已按表头定位，但请人工核对后再写入')
|
if tinfos:
|
add_alert('提示', '目标表个别列按母版默认', cat, '', '', os.path.basename(spec['path']),
|
'；'.join(tinfos[:8]),
|
'知悉即可（这些列在母版里本来就没有子表头，按默认列位写入）')
|
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)
|
plain_tgt = {}
|
for k in tmap:
|
plain_tgt.setdefault(normname3(k), []).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
|
n3 = normname3(k)
|
if n3 and n3 in plain_tgt:
|
suspicious.append((k, plain_tgt[n3][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('')
|
for k1, hit1, why1 in suspicious:
|
_p1 = pmap[k1][0]
|
add_alert('警告', '疑似同一项目（名称写法不同）', cat, _p1.get('市州', ''), _p1['项目名称'],
|
os.path.basename(_p1['文件']),
|
'与目标表已有项目「%s」%s' % (tmap[hit1][0]['项目名称'], why1),
|
'人工确认是否同一项目：是 → 不改名称、按投资系统名称更新；否 → 整行新增')
|
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('| --- | --- | --- | --- | --- | --- |')
|
for k in new_in_src:
|
p0 = pmap[k][0]
|
_hit, _r = best_name_hint(k, tkeys)
|
hint = ('⚠️ 与「%s」相似 %.2f，请确认是否同一项目' % (_hit, _r)) if _hit else ''
|
_sim = hint
|
add_alert('提示', '将新增项目', cat, p0.get('市州', ''), p0['项目名称'],
|
os.path.basename(p0['文件']),
|
'目标表没有这个项目；自开始建设累计/自年初累计/本月完成 = %s/%s/%s%s' % (
|
p0.get('自开始建设累计', ''), p0.get('自年初累计', ''), p0.get('本月完成', ''),
|
('；' + hint) if hint else ''),
|
'确认是真实新项目后写入（整行插入，复制相邻行格式）')
|
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}})
|
|
# 目标表有、本期源文件没有 → 保持原值（提示，不阻断）
|
if only_tgt:
|
_names = [tmap[k][0]['项目名称'] for k in only_tgt]
|
add_alert('提示', '本期源文件里没有的项目', cat, '', '', '—',
|
'目标表里有 %d 个项目本期市州月报没报：%s%s' % (
|
len(_names), '、'.join(_names[:8]), ' 等' if len(_names) > 8 else ''),
|
'按口径保持上月原值；若市州确实已停建，请人工确认是否保留')
|
|
# 三方数值比对：源文件 vs 投资系统（本期）
|
if cur_sys:
|
diff, missing, shifted = [], [], []
|
for k in matched:
|
p, t = pmap[k][0], tmap[k][0]
|
s, note = sys_lookup(k, cur_sys)
|
if not s:
|
missing.append(t['项目名称'])
|
continue
|
if note:
|
add_alert('提示', '投资系统名称写法不同', cat, p.get('市州', ''), t['项目名称'],
|
os.path.basename(p['文件']), note + '；已按投资系统这条记录做数值对照（项目名称不动）',
|
'知悉即可')
|
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:
|
continue
|
diff.append((t['项目名称'], c, pretty(pv), pretty(sv), pretty(sv - pv)))
|
if abs(pv - sv) > 1.0:
|
add_alert('警告', '源文件与投资系统不一致', cat, p.get('市州', ''), t['项目名称'],
|
os.path.basename(p['文件']),
|
'%s：市州月报 %s ／ 投资系统 %s（差 %s）' % (c, pretty(pv), pretty(sv), pretty(sv - pv)),
|
'人工确认以哪个为准；若是取列错位则修解析规则后重跑')
|
for c2 in ('总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成'):
|
if c2 == c:
|
continue
|
sv2 = s.get(c2)
|
if sv2 is not None and abs(pv - sv2) <= 0.05:
|
shifted.append((t['项目名称'], c, c2))
|
add_alert('警告', '疑似取列错位', cat, p.get('市州', ''), t['项目名称'],
|
os.path.basename(p['文件']),
|
'市州月报「%s」= %s，恰好等于投资系统「%s」' % (c, pretty(pv), c2),
|
'检查源文件表头（组表头合并单元格最容易错位），修取列规则后重跑解析')
|
rep.append('')
|
rep.append('### 源文件 vs 投资系统（本期）：**数值不一致 %d 处**（需弹窗提示），其中疑似取列错位 **%d** 处' % (len(diff), len(shifted)))
|
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('')
|
for nm1 in missing:
|
add_alert('提示', '投资系统查无此项目', cat, '', nm1, '—',
|
'投资系统本期查询结果里没有这个项目名称', '知悉即可；如需以投资系统为准，请人工核对名称')
|
|
# 月度校验：本月自年初累计 = 上月自年初累计 + 本月完成
|
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, _note = sys_lookup(k, cur_pre)
|
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)))
|
add_alert('警告', '月度等式不符', cat, p.get('市州', ''), t['项目名称'],
|
'%s + %s = %s，而本月自年初累计是 %s（差 %s）' % (
|
pretty(prej), pretty(nowk), pretty(prej + nowk), pretty(nowj),
|
pretty(nowj - prej - nowk)),
|
'写入时按等式纠正为 %s，并让市州确认' % 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('')
|
|
# 本期没有任何项目行的市州 → 提示（可能漏报）
|
for cat in TARGETS:
|
_cities = {r['市州'] for r in parsed if r['品类'] == cat and r['市州']}
|
_miss = [c for c in CITY17 if c not in _cities]
|
if _miss:
|
add_alert('提示', '本期无该市州数据', cat, '', '', '—',
|
'共 %d 个市州本期没有任何项目行：%s' % (len(_miss), '、'.join(_miss)),
|
'确认是「该市州确实没有项目」还是「漏报」；漏报则让市州补报')
|
|
# ---- 汇总《弹窗提示》（未来搬进 Java 后端时按此逐条弹窗） ----
|
_lv = {'阻断': 0, '警告': 1, '提示': 2}
|
alerts.sort(key=lambda x: (_lv.get(x['级别'], 9), x['类型'], x['品类'], x['市州'], x['项目名称']))
|
_n = {k: sum(1 for x in alerts if x['级别'] == k) for k in ('阻断', '警告', '提示')}
|
a_cols = ['序号', '级别', '类型', '品类', '市州', '项目名称', '来源', '详情', '建议动作']
|
p_al = os.path.join(out, '弹窗提示_%s.tsv' % a.month)
|
with io.open(p_al, 'w', encoding='utf-8-sig') as fh:
|
fh.write('\t'.join(a_cols) + '\n')
|
for _i, _x in enumerate(alerts, 1):
|
fh.write('\t'.join(str(_i) if _k == '序号' else str(_x.get(_k, '')).replace('\t', ' ')
|
for _k in a_cols) + '\n')
|
al_md = ['## ⚠️ 弹窗提示总览（共 %d 项：阻断 %d ／ 警告 %d ／ 提示 %d）' % (
|
len(alerts), _n['阻断'], _n['警告'], _n['提示']), '']
|
al_md.append('> 级别：**阻断**=不能写表，先处理；**警告**=数据有疑点，人工确认后再写；**提示**=仅告知。')
|
al_md.append('> 机器可读清单：`%s`（未来 Java/网页按此逐条弹窗）。规则见 `docs\\投资\\弹窗提示规则_2026-09-17.md`。'
|
% os.path.basename(p_al))
|
al_md.append('')
|
if alerts:
|
al_md.append('| 序号 | 级别 | 类型 | 品类 | 市州 | 项目名称 | 详情 | 建议动作 |')
|
al_md.append('| --- | --- | --- | --- | --- | --- | --- | --- |')
|
for _i, _x in enumerate(alerts, 1):
|
al_md.append('| %d | %s | %s | %s | %s | %s | %s | %s |' % (
|
_i, _x['级别'], _x['类型'], _x['品类'], _x['市州'], _x['项目名称'], _x['详情'], _x['建议动作']))
|
al_md.append('')
|
else:
|
al_md.append('（无：所有检查项均通过）'); al_md.append('')
|
rep = rep[:4] + al_md + rep[4:]
|
|
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('弹窗提示 = %d（阻断 %d / 警告 %d / 提示 %d）' % (len(alerts), _n['阻断'], _n['警告'], _n['提示']))
|
print('wrote', p_md)
|
print('wrote', p_tsv)
|
print('wrote', p_al)
|
|
|
if __name__ == '__main__':
|
main()
|
