fix(投资): 全部改为「按表头定位」——目标明细表不再写死列号 + 同名列按数据消歧

zhizhijie

2026-09-17 7b6bc7c126657ae943f3d37e5f19133d8bb187c5

 docs/Ͷ×Ê/¹¤¾ß/Æ¥ÅäÓëУÑé.py

@@ -94,24 +94,55 @@


def read_target(spec):
    """读目标明细表 â†’ (行列表, è¡¨å¤´è¡Œå·, è¯´æ˜Ž)"""
    """读目标明细表 â†’ (行列表, åˆ—位映射, åˆ—位告警)。

    åˆ—位**按表头定位**（母版默认列位只作兜底）；与母版默认不一致时返回告警，由调用方弹窗。
    """
    from openpyxl import load_workbook
    import xlrd
    rows = []
    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:
        bk = xlrd.open_workbook(spec['path'])
        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
    namec = F.colindex(spec['name_col']) + 1
    anchors = [F.colindex(spec['cols'][k]) + 1 for k in ('自开始建设累计', '自年初累计', '总投资')]
        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()
@@ -124,11 +155,11 @@
        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)
        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
    return rows, cols, notes, infos


def read_system(path):
@@ -308,6 +339,11 @@
            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' % (
@@ -324,7 +360,15 @@
            add_alert('阻断', '缺目标明细表', cat, '', '', spec['path'],
                      '目标明细表不存在，本季度该品类无法写入', '把模板表放到 docs\\投资\\输出\\')
            continue
        tgt = read_target(spec)
        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: