# -*- coding: utf-8 -*-
|
"""市州月报解析.py —— 把 17 市州的月报源文件解析成统一的「项目行」记录（只读）
|
|
用法：
|
python docs\投资\工具\市州月报解析.py --month 2026-07
|
python docs\投资\工具\市州月报解析.py --month 2026-07 --body 客运（客运站）
|
输出：
|
docs\投资\输出\解析结果_<YYYY-MM>.tsv 一行一个项目（机器可读，供写库/写表用）
|
docs\投资\输出\解析报告_<YYYY-MM>.md 人读：每个文件的市州/项目数/字段求和/一致性校验
|
|
设计要点（与《市州月报格式差异报告》一致）：
|
1) 列位不写死：按表头关键字定位列（复用 格式探测.py 的 FIELD_PAT 与 scan_fields）；
|
2) 选表：先按“表头区年月标记”命中本期，再按字段命中数打分 —— 不看隐藏与否；
|
3) 选行：先找「市州行」切段，只取本文件对应市州那一段；
|
4) 项目行：名称列非空 + 非小计/合计/分类行 + 有数值锚点；兼容无序号列/缺序号；
|
5) 分类行（一、二、三、四、五、）与「小计/合计」行单独收集：分类行作为项目的归属，
|
小计/合计行只用于**一致性校验**（项目行求和 vs 小计行）。
|
6) 隐藏行/隐藏表一律不作为过滤条件（黄石 R10 是真实项目行）。
|
"""
|
import sys, os, io, re, argparse, json, importlib.util
|
|
_here = os.path.dirname(os.path.abspath(__file__))
|
_root = os.path.abspath(os.path.join(_here, '..', '..', '..'))
|
for _b in (_here, _root, os.getcwd()):
|
for _sub in ('_pylibs',):
|
_lib = os.path.abspath(os.path.join(_b, _sub))
|
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)
|
|
# 需要从源文件取出的字段（与目标明细表列一一对应）
|
FIELDS = ['项目名称', '建设单位', '建设性质', '开工时间', '竣工时间', '总投资',
|
'自开始建设累计', '本年计划投资', '自年初累计', '本月完成', '建设阶段', '形象进度']
|
# 数值字段（用于一致性校验）
|
NUM_FIELDS = ['总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成']
|
|
GROUP_PAT = re.compile(r'^(一|二|三|四|五|六|七|八)、')
|
SKIP_PAT = re.compile(r'^(小\s*计|合\s*计|总\s*计)')
|
TOTAL_PAT = re.compile(r'^(合\s*计|总\s*计)')
|
|
# 少数文件既无文件名城市、也无「市州行」，只能按业务事实人工登记（有据可查：汉口客运中心=武汉）
|
CITY_OVERRIDE = [
|
('站场固投（南山、汉口站）', '武汉'),
|
]
|
|
|
def city_by_override(fname):
|
for key, city in CITY_OVERRIDE:
|
if key in fname:
|
return city
|
return None
|
|
|
def num(v):
|
"""尽量转成数值；转不了返回 None。"""
|
if isinstance(v, (int, float)) and not isinstance(v, bool):
|
return v
|
s = F.norm(v)
|
if not s:
|
return None
|
s = s.replace(',', '').replace('％', '').replace('%', '')
|
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 str(v)
|
|
|
def pick_columns(cell, nrows, ncols, merges=None):
|
"""按表头定位列（含两级表头消歧）——实现见 格式探测.py 的 pick_columns，保证两处口径一致。"""
|
return F.pick_columns(cell, nrows, ncols, merges, fields=FIELDS)
|
|
|
def parse_sheet(cell, nrows, ncols, city=None, merges=None):
|
"""解析一张工作表 → (项目行列表, 段信息, 校验信息)"""
|
cols, hits = pick_columns(cell, nrows, ncols, merges)
|
namec = cols.get('项目名称')
|
if namec is None:
|
return [], {}, {}
|
segs = F.city_segments(cell, nrows, ncols)
|
if city is None and len({s[1] for s in segs}) == 1:
|
city = segs[0][1]
|
lo, hi, city_row = 1, 10 ** 9, None
|
if city:
|
for i, (rr, cc) in enumerate(segs):
|
if cc == city:
|
city_row, lo = rr, rr
|
hi = segs[i + 1][0] if i + 1 < len(segs) else 10 ** 9
|
break
|
project_name_pat = None
|
# 数值锚点列：自开始建设累计 / 自年初累计 / 总投资
|
anchors = [cols[k] for k in ('自开始建设累计', '自年初累计', '总投资') if k in cols]
|
|
rows, group, subtotals, city_vals = [], '', {}, None
|
for r in range(min(nrows, 800)):
|
if not (lo <= r + 1 < hi):
|
continue
|
if city_row and r + 1 == city_row:
|
# 市州行本身通常就带着本市州合计（黄石 R5 / 孝感 R7 / 咸宁 R6 …），留作校验
|
city_vals = {f: num(cell(r, cols[f])) for f in NUM_FIELDS if f in cols}
|
continue
|
a = F.norm(cell(r, 0))
|
name = F.norm(cell(r, namec))
|
joined = a + name
|
if GROUP_PAT.match(joined):
|
group = joined[:40]
|
continue
|
if SKIP_PAT.match(a) or SKIP_PAT.match(name):
|
# 小计/合计行：全部收集，用于一致性校验。
|
# 规则：优先「合计/总计」行；没有合计行时，取「自开始建设累计」最大的那行小计
|
#（源文件常有多个空小计行，如林区客运 R11 空、R14 有值、R17 只有文字）。
|
vals = {}
|
for f in NUM_FIELDS:
|
if f in cols:
|
vals[f] = num(cell(r, cols[f]))
|
vals['_kind'] = 'total' if TOTAL_PAT.match(a) or TOTAL_PAT.match(name) else 'sub'
|
subtotals['R%d' % (r + 1)] = vals
|
continue
|
if not name or len(name) < 4:
|
continue
|
if any(name.startswith(x) for x in F.BAD_NAME):
|
continue
|
if namec == 0:
|
numbered = True # 无序号列（随州）：名称列就是 A 列
|
else:
|
numbered = bool(re.match(r'^\d{1,3}(\.0)?$', a)) or a == ''
|
has_value = any(num(cell(r, c)) is not None for c in anchors) if anchors else True
|
if not (numbered and has_value):
|
continue
|
rec = {'行号': r + 1, '序号': pretty(cell(r, 0)), '分类': group}
|
for f in FIELDS:
|
c = cols.get(f)
|
v = cell(r, c) if c is not None else ''
|
rec[f] = pretty(num(v)) if f in NUM_FIELDS else F.norm(v)
|
rows.append(rec)
|
seg_info = {'市州': city or '', '市州行': city_row or '', '段起': lo,
|
'段止': '' if hi >= 10 ** 9 else hi,
|
'全部市州行': ['R%s:%s' % s for s in segs]}
|
return rows, seg_info, {'列位': {f: F.colname(c) for f, c in cols.items()},
|
'小计合计行': subtotals, '市州行值': city_vals}
|
|
|
def pick_total(subtotals):
|
"""从收集到的小计/合计行里挑出用于校验的那一行：优先合计行，其次自开始建设累计最大的小计行。"""
|
if not subtotals:
|
return None, None
|
totals = [(k, v) for k, v in subtotals.items() if v.get('_kind') == 'total']
|
if totals:
|
return totals[-1]
|
subs = sorted(subtotals.items(), key=lambda kv: (kv[1].get('自开始建设累计') or 0), reverse=True)
|
return subs[0]
|
|
|
def main():
|
ap = argparse.ArgumentParser()
|
ap.add_argument('--month', required=True)
|
ap.add_argument('--root', default=os.path.join('docs', '投资'))
|
ap.add_argument('--body', default=None)
|
ap.add_argument('--out', default=None)
|
a = ap.parse_args()
|
out = a.out or os.path.join(a.root, '输出')
|
os.makedirs(out, exist_ok=True)
|
ym = a.month.replace('-', '')
|
bodies = [a.body] if a.body else ['客运（客运站）', '物流（物流园区）']
|
md = '%d月' % int(a.month.split('-')[1])
|
|
records, report, checks = [], [], []
|
report.append('# 市州月报解析报告（%s）' % a.month)
|
report.append('')
|
report.append('（只读解析；一行一个项目行，明细见同名 `.tsv`）')
|
report.append('')
|
for body in bodies:
|
d = os.path.join(a.root, '输入', body, md)
|
body_key = body.split('（')[0]
|
report.append('## %s' % body)
|
report.append('')
|
report.append('| 文件 | 工作表 | 市州 | 项目行 | 投资系统合计(自开始建设累计/自年初累计/本月完成) | 源文件小计行 | 一致性 |')
|
report.append('| --- | --- | --- | --- | --- | --- | --- |')
|
if not os.path.isdir(d):
|
report.append('| （目录不存在：%s） | | | | | | |' % d)
|
report.append('')
|
continue
|
files = sorted(f for f in os.listdir(d)
|
if os.path.isfile(os.path.join(d, f)) and not f.startswith('~$'))
|
for f in files:
|
path = os.path.join(d, f)
|
city_file = F.city_of_text(f)
|
try:
|
sheets = F.open_sheets(path)
|
sel = F.pick_sheet(sheets, ym=ym)
|
if not sel:
|
report.append('| %s | （未选中工作表） | | | | | |' % f[:44])
|
checks.append({'品类': body_key, '市州': city_file or '', '文件': f, '工作表': '',
|
'状态': '未选中工作表', '项目行数': 0, '一致性': '—'})
|
continue
|
nm, nr, nc, cell, _hits = sel
|
merges = F.sheet_merges(path).get(nm, [])
|
rows, seg, extra = parse_sheet(cell, nr, nc, city=city_file, merges=merges)
|
city = seg.get('市州') or city_file or city_by_override(f) or ''
|
sums = {k: sum(num(r.get(k)) or 0 for r in rows) for k in NUM_FIELDS}
|
sub = extra.get('小计合计行') or {}
|
sub_txt, ok = [], '—'
|
if not sub and extra.get('市州行值'):
|
sub = {'R%s(市州行)' % (seg.get('市州行') or '?'): extra['市州行值']}
|
tok = tval = None
|
if sub:
|
key, val = pick_total(sub)
|
tok, tval = key, val
|
sub_txt.append('%s=%s' % (key, '/'.join(pretty(val.get(k)) for k in ('自开始建设累计', '自年初累计', '本月完成'))))
|
if len(sub) > 1:
|
sub_txt.append('（另有 %d 行小计/合计）' % (len(sub) - 1))
|
same = all(abs((val.get(k) or 0) - sums[k]) < 0.05
|
for k in ('自开始建设累计', '自年初累计', '本月完成'))
|
ok = '✅ 一致' if same else '⚠️ 不一致'
|
ck = {'品类': body_key, '市州': city, '文件': f, '工作表': nm, '状态': 'OK',
|
'项目行数': len(rows), '合计行': '', '一致性': ok,
|
'列位': ','.join('%s=%s' % (k, v) for k, v in (extra.get('列位') or {}).items())}
|
for k in ('自开始建设累计', '自年初累计', '本月完成'):
|
ck['求和_' + k] = sums[k]
|
if sub:
|
ck['合计行'] = tok
|
for k in ('自开始建设累计', '自年初累计', '本月完成'):
|
ck['合计_' + k] = tval.get(k)
|
checks.append(ck)
|
report.append('| %s | %s | %s | %d | %s | %s | %s |' % (
|
f[:44], nm[:14], city, len(rows),
|
'/'.join(pretty(sums[k]) for k in ('自开始建设累计', '自年初累计', '本月完成')),
|
' '.join(sub_txt), ok))
|
for r in rows:
|
rec = {'品类': body_key, '市州': city, '文件': f, '工作表': nm}
|
rec.update(r)
|
records.append(rec)
|
except Exception as e:
|
report.append('| %s | （解析失败：%s: %s） | | | | | |' % (f[:44], type(e).__name__, str(e)[:60]))
|
checks.append({'品类': body_key, '市州': city_file or '', '文件': f, '工作表': '',
|
'状态': '解析失败：%s: %s' % (type(e).__name__, str(e)[:80]),
|
'项目行数': 0, '一致性': '—'})
|
report.append('')
|
|
# 市州汇总（跨文件合并同市州）
|
by_city = {}
|
for r in records:
|
k = (r['品类'], r['市州'])
|
s = by_city.setdefault(k, {f: 0.0 for f in NUM_FIELDS})
|
for f in NUM_FIELDS:
|
s[f] += num(r.get(f)) or 0
|
report.append('## 按市州汇总（项目行求和，跨文件合并）')
|
report.append('')
|
report.append('| 品类 | 市州 | 项目数 | 计划总投资 | 自开始建设累计 | 本年计划 | 自年初累计 | 本月完成 |')
|
report.append('| --- | --- | --- | --- | --- | --- | --- | --- |')
|
for (bk, city), s in sorted(by_city.items()):
|
n = sum(1 for r in records if r['品类'] == bk and r['市州'] == city)
|
report.append('| %s | %s | %d | %s | %s | %s | %s | %s |' % (
|
bk, city, n,
|
*(pretty(s[f]) for f in ('总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成'))))
|
report.append('')
|
|
report.append('## 各源文件识别到的列位（**按表头关键字定位，不是固定列号**）')
|
report.append('')
|
report.append('| 品类 | 文件 | 工作表 | 项目名称 | 总投资 | 自开始建设累计 | 本年计划投资 | 自年初累计 | 本月完成 |')
|
report.append('| --- | --- | --- | --- | --- | --- | --- | --- | --- |')
|
_key5 = ('项目名称', '总投资', '自开始建设累计', '本年计划投资', '自年初累计', '本月完成')
|
for ck in checks:
|
_cm = dict(kv.split('=', 1) for kv in (ck.get('列位') or '').split(',') if '=' in kv)
|
report.append('| %s | %s | %s | %s |' % (ck.get('品类', ''), ck.get('文件', '')[:34], ck.get('工作表', '')[:12],
|
' | '.join(_cm.get(k, '—') for k in _key5)))
|
report.append('')
|
|
ck_cols = ['品类', '市州', '文件', '工作表', '状态', '项目行数', '一致性', '合计行', '列位',
|
'求和_自开始建设累计', '求和_自年初累计', '求和_本月完成',
|
'合计_自开始建设累计', '合计_自年初累计', '合计_本月完成']
|
p_ck = os.path.join(out, '解析校验_%s.tsv' % a.month)
|
with io.open(p_ck, 'w', encoding='utf-8-sig') as fh:
|
fh.write('\t'.join(ck_cols) + '\n')
|
for c in checks:
|
fh.write('\t'.join(pretty(c.get(k, '')) if isinstance(c.get(k), float) else str(c.get(k, ''))
|
for k in ck_cols) + '\n')
|
print('校验记录数 = %d' % len(checks))
|
|
cols = ['品类', '市州', '文件', '工作表', '行号', '序号', '分类'] + FIELDS
|
tsv = os.path.join(out, '解析结果_%s.tsv' % a.month)
|
with io.open(tsv, 'w', encoding='utf-8-sig') as fh:
|
fh.write('\t'.join(cols) + '\n')
|
for r in records:
|
fh.write('\t'.join(str(r.get(c, '')) for c in cols) + '\n')
|
p_md = os.path.join(out, '解析报告_%s.md' % a.month)
|
io.open(p_md, 'w', encoding='utf-8').write('\n'.join(report) + '\n')
|
print('项目行合计 = %d' % len(records))
|
print('wrote', p_md)
|
print('wrote', tsv)
|
print('wrote', p_ck)
|
|
|
if __name__ == '__main__':
|
main()
|
