# -*- coding: utf-8 -*-
|
"""格式探测.py —— 投资月报源文件格式探测（只读，不修改任何源文件）
|
|
用法：
|
python docs\投资\工具\格式探测.py --month 2026-07
|
python docs\投资\工具\格式探测.py --month 2026-07 --body 物流（物流园区）
|
输出：
|
<out>\格式探测_<YYYY-MM>.md 人读报告
|
<out>\格式探测_<YYYY-MM>.tsv 机器可读明细（供程序回归比对）
|
|
原则：
|
1) 列位不写死：按表头关键字在“前 14 行 × 前 34 列”里匹配定位；
|
2) 工作表先按“表头区是否带本期年月标记”优先，再按命中字段数打分，
|
避免取到空模板表 / 历史月表（如潜江、林区、襄阳的多月工作簿）；
|
3) 项目行判定：名称列非空且不是小计/合计/分类行，且有数值锚点（自开始建设累计等），
|
兼容“无序号列”（随州）与“缺序号”（林区 7 月）两种情况；
|
4) 市州分段：先找出「市州行」（首 5 列里恰好只有 1 个单元格是市州名，同行其余列全空），
|
再只取目标市州段内的项目行。湖北省十堰交通物流…表的 88 行里，R11-R63 是隐藏的
|
武汉历史区块，若不按市州分段就会把武汉的历史项目当成十堰的取进来；
|
5) 隐藏行 / 隐藏工作表不作为判定依据：黄石物流 R10 是隐藏行，但它是一条真实项目行
|
（阳新县焦山物流仓储中心），且已计入黄石市州汇总行，必须取。
|
"""
|
import sys, os, io, re, argparse, datetime as _dt
|
|
for _base in (os.path.dirname(os.path.abspath(__file__)), os.getcwd()):
|
for _sub in ('_pylibs', os.path.join('..', '_pylibs')):
|
_lib = os.path.abspath(os.path.join(_base, _sub))
|
if os.path.isdir(_lib) and _lib not in sys.path:
|
sys.path.insert(0, _lib)
|
|
try:
|
import xlrd
|
except ImportError:
|
xlrd = None
|
try:
|
import openpyxl
|
except ImportError:
|
openpyxl = None
|
|
FIELD_PAT = [
|
('项目名称', r'^(项目名称|项目名称及站级|项目及站级|项目名称及战级)$'),
|
('建设单位', r'(投资建设\s*单位名称|投资建设单位名称|项目业主|建设单位)'),
|
('建设性质', r'^(建设性质|建设 性质)$'),
|
('开工时间', r'^(开工时间|开工年)$'),
|
('竣工时间', r'^(竣工时间|建成时间|完工年)$'),
|
('总投资', r'(计划总投资|总投资)'),
|
('自开始建设累计', r'(自开始建设|自开始建设至本年)'),
|
('本年计划投资', r'(本年计划投资|本年计划)'),
|
('自年初累计', r'((自)?年初累计完成投资|自年初累计|自年初-累计完成投资)'),
|
('本月完成', r'(本月完成投资|本月完成|当月完成投资|^本月$)'),
|
('建设阶段', r'^(项目建设阶段|建设阶段)$'),
|
('形象进度', r'^(形象进度|进度情况)$'),
|
('工可', r'^工可'),
|
('初设', r'^初设'),
|
('市州', r'^(市州|市州名称)$'),
|
('日期', r'^(20\d\d[-./]\d{1,2})'),
|
]
|
KEY = ['项目名称', '建设单位', '建设性质', '开工时间', '竣工时间', '总投资',
|
'自开始建设累计', '本年计划投资', '自年初累计', '本月完成', '建设阶段', '形象进度']
|
BAD_NAME = ('填报单位', '签章', '单位负责人', '统计负责人', '审核', '说明', '注：', '月报', '报表',
|
'序 号', '序号', '项目名称', '项目业主')
|
|
CITIES = ('武汉', '黄石', '十堰', '宜昌', '襄阳', '鄂州', '荆门', '孝感', '荆州', '黄冈',
|
'咸宁', '随州', '恩施', '仙桃', '潜江', '天门', '神农架', '林区')
|
|
|
def city_segments(cell, nrows, ncols, maxc=5):
|
# 市州行 = 首 maxc 列里恰好只有 1 个单元格是市州名（同行其余列为空）。
|
# 返回 [(行号1基, 市州名), ...]，已按行号排序。
|
segs = []
|
for r in range(min(nrows, 800)):
|
vals = [norm(cell(r, c)) for c in range(min(ncols, maxc))]
|
hit = [i for i, v in enumerate(vals) if v in CITIES]
|
if len(hit) == 1 and sum(1 for v in vals if v) == 1:
|
segs.append((r + 1, vals[hit[0]]))
|
return segs
|
|
|
def city_of_text(text):
|
for c in CITIES:
|
if c in str(text or ''):
|
return c
|
return None
|
|
|
def norm(s):
|
s = str(s or '')
|
for ch in (' ', '\u3000', '\n', '\r', '\t'):
|
s = s.replace(ch, '')
|
return s
|
|
|
def colname(i):
|
s = ''
|
i += 1
|
while i:
|
i, r = divmod(i - 1, 26)
|
s = chr(65 + r) + s
|
return s
|
|
|
def colindex(letters):
|
n = 0
|
for ch in letters:
|
n = n * 26 + (ord(ch) - 64)
|
return n - 1
|
|
|
def open_sheets(path):
|
"""返回 [(sheet_name, nrows, ncols, cell_fn), ...]。
|
|
注意：`.et`（WPS）实为 OLE2/BIFF8 复合文档，xlrd 可直接读，不必走 Excel COM。
|
"""
|
with open(path, 'rb') as fh:
|
magic = fh.read(4)
|
if magic[:4] == b'PK\x03\x04':
|
if openpyxl is None:
|
raise RuntimeError('需要 openpyxl 读取 .xlsx')
|
wb = openpyxl.load_workbook(path, data_only=True)
|
out = []
|
for nm in wb.sheetnames:
|
ws = wb[nm]
|
out.append((nm, ws.max_row, ws.max_column,
|
(lambda r, c, _w=ws: _w.cell(row=r + 1, column=c + 1).value)))
|
wb.close()
|
return out
|
if xlrd is None:
|
raise RuntimeError('需要 xlrd 读取 .xls')
|
bk = xlrd.open_workbook(path)
|
out = []
|
for si in range(bk.nsheets):
|
sh = bk.sheet_by_index(si)
|
out.append((sh.name, sh.nrows, sh.ncols,
|
(lambda r, c, _s=sh: _s.cell_value(r, c) if (r < _s.nrows and c < _s.ncols) else '')))
|
return out
|
|
|
def scan_fields(cell, nrows, ncols, maxrow=14, maxcol=34):
|
hits = {}
|
for r in range(min(nrows, maxrow)):
|
for c in range(min(ncols, maxcol)):
|
t = norm(cell(r, c))
|
if not t or len(t) > 30:
|
continue
|
for f, p in FIELD_PAT:
|
if re.search(p, t):
|
hits.setdefault(f, []).append((r + 1, colname(c), t[:24]))
|
return hits
|
|
|
def is_num(v):
|
if isinstance(v, (int, float)) and not isinstance(v, bool):
|
return True
|
return bool(re.match(r'^-?\d+(\.\d+)?$', norm(v)))
|
|
|
def profile(cell, nrows, ncols, hits, city=None):
|
namecol = None
|
if '项目名称' in hits:
|
namecol = hits['项目名称'][0][1]
|
data_start = data_end = None
|
project_rows = 0
|
groups = []
|
segs = 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
|
if namecol is not None:
|
nc = colindex(namecol)
|
anchors = [colindex(hits[k][0][1]) for k in ('自开始建设累计', '自年初累计', '总投资') if k in hits]
|
for r in range(min(nrows, 800)):
|
if not (lo <= r + 1 < hi):
|
continue
|
a = norm(cell(r, 0))
|
name = norm(cell(r, nc))
|
is_group = bool(re.match(r'^(一|二|三|四|五|六)、', a + name)) \
|
or a in ('小计', '合计') or name in ('小计', '合计')
|
if is_group:
|
if name:
|
groups.append((r + 1, name[:20]))
|
continue
|
if not name or len(name) < 4 or any(name.startswith(x) for x in BAD_NAME):
|
continue
|
if nc == 0:
|
numbered = True # 无序号列（随州）：名称列就是 A 列
|
else:
|
numbered = bool(re.match(r'^\d{1,3}(\.0)?$', a)) or a == ''
|
has_value = any(is_num(cell(r, c)) for c in anchors) if anchors else True
|
if numbered and has_value:
|
data_start = data_start or r + 1
|
data_end = r + 1
|
project_rows += 1
|
return dict(namecol=namecol, data_start=data_start, data_end=data_end,
|
project_rows=project_rows, groups=groups[:6],
|
city=city or '', city_row=city_row or '', segments=segs[:8],
|
header_rows=sorted({h[0] for v in hits.values() for h in v}))
|
|
|
def hdr_dates(cell, nrows, ncols):
|
"""表头区（前 14 行 × 前 32 列）出现的年月标记，如 '202607'（兼容文本日期与 Excel 日期序列号）。"""
|
out = set()
|
for r in range(min(nrows, 14)):
|
for c in range(min(ncols, 32)):
|
v = cell(r, c)
|
t = str(v or '')
|
for m in re.finditer(r'(20\d{2})\s*[-./年]\s*(\d{1,2})', t):
|
out.add('%s%02d' % (m.group(1), int(m.group(2))))
|
if isinstance(v, (int, float)) and not isinstance(v, bool) and 40000 <= v <= 60000:
|
d = _dt.date(1899, 12, 30) + _dt.timedelta(days=int(v))
|
out.add('%04d%02d' % (d.year, d.month))
|
return out
|
|
|
def pick_sheet(sheets, ym=None):
|
best = None
|
for (nm, nr, nc, cell) in sheets:
|
hits = scan_fields(cell, nr, nc)
|
score = sum(3 if k == '项目名称' else 2 for k in hits if k in KEY)
|
if ym and ym in hdr_dates(cell, nr, nc):
|
score += 20
|
if best is None or score > best[1]:
|
best = ((nm, nr, nc, cell, hits), score)
|
return best[0] if best else None
|
|
|
def month_dir(month):
|
return '%d月' % int(month.split('-')[1])
|
|
|
def main():
|
ap = argparse.ArgumentParser()
|
ap.add_argument('--month', required=True, help='如 2026-07')
|
ap.add_argument('--root', default=os.path.join('docs', '投资'))
|
ap.add_argument('--out', default=None)
|
ap.add_argument('--body', default=None, help='只扫一个品类目录')
|
args = ap.parse_args()
|
out = args.out or os.path.join(args.root, '输出')
|
os.makedirs(out, exist_ok=True)
|
md = month_dir(args.month)
|
ym = args.month.replace('-', '')
|
bodies = [args.body] if args.body else ['客运（客运站）', '物流（物流园区）']
|
rows, report = [], []
|
for body in bodies:
|
d = os.path.join(args.root, '输入', body, md)
|
if not os.path.isdir(d):
|
report.append('!! 目录不存在：%s' % d)
|
continue
|
files = sorted(f for f in os.listdir(d)
|
if os.path.isfile(os.path.join(d, f)) and not f.startswith('~$'))
|
report.append('## %s（%d 个文件）' % (body, len(files)))
|
for f in files:
|
p = os.path.join(d, f)
|
rec = dict(品类=body.split('（')[0], 文件=f, 工作表='', 表头行='', 数据起='', 数据止='',
|
项目行数=0, 名称列='', 分组行='', 市州='', 市州行='', 市州段='')
|
file_city = city_of_text(f)
|
try:
|
sel = pick_sheet(open_sheets(p), ym=ym)
|
if sel:
|
nm, nr, nc, cell, hits = sel
|
pr = profile(cell, nr, nc, hits, city=file_city)
|
rec.update(工作表=nm, 表头行=','.join(map(str, pr['header_rows'])),
|
数据起=pr['data_start'], 数据止=pr['data_end'],
|
项目行数=pr['project_rows'], 名称列=pr['namecol'] or '',
|
分组行=';'.join('R%s:%s' % g for g in pr['groups']),
|
市州=pr['city'] or (file_city or ''), 市州行=pr['city_row'],
|
市州段=';'.join('R%s:%s' % s for s in pr['segments']))
|
for k in KEY:
|
if k in hits:
|
rec[k + '列'] = hits[k][0][1]
|
except Exception as e:
|
rec['错误'] = '%s: %s' % (type(e).__name__, e)
|
rows.append(rec)
|
report.append(' %-46s sheet=%-12s 市州=%s@R%s 表头=%-12s 数据=%s~%s 项目行=%-3s 名称列=%s %s'
|
% (rec['文件'][:46], str(rec['工作表'])[:12], rec['市州'], rec['市州行'],
|
rec['表头行'], rec['数据起'],
|
rec['数据止'], rec['项目行数'], rec['名称列'],
|
''.join('%s=%s ' % (k, rec.get(k + '列', '-')) for k in KEY)
|
+ (' [%s]' % rec['错误'] if rec.get('错误') else '')))
|
cols = ['品类', '文件', '工作表', '市州', '市州行', '市州段', '表头行', '数据起', '数据止', '项目行数',
|
'名称列'] + [k + '列' for k in KEY] + ['分组行', '错误']
|
tsv = os.path.join(out, '格式探测_%s.tsv' % args.month)
|
with io.open(tsv, 'w', encoding='utf-8-sig') as fh:
|
fh.write('\t'.join(cols) + '\n')
|
for r in rows:
|
fh.write('\t'.join(str(r.get(c, '')) for c in cols) + '\n')
|
mdpath = os.path.join(out, '格式探测_%s.md' % args.month)
|
io.open(mdpath, 'w', encoding='utf-8').write(
|
'# 投资月报源文件格式探测（%s）\n\n' % args.month + '\n'.join(report) +
|
'\n\n明细表：`%s`\n' % tsv)
|
print('wrote', mdpath)
|
print('wrote', tsv)
|
|
|
if __name__ == '__main__':
|
main()
|
