feat(budget): activate P1 auto-collect + attribution subject mapping (yanxue-20260914-budget-p13p14-activation)

2026-09-14 17:19:49 +08:00
parent b815b7f623
commit a6d5d151c4
4 changed files with 642 additions and 45 deletions
@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""预算偏差告警归因回填 —— yanxue-20260914-budget-p13p14-activation（P1-③ 激活）
+
+两类动作:
+  1) attribution 为空的历史 open 告警 → 走现成接口 GET /budget/deviation-alerts/{id}/attribution
+     （端点在响应前现场组装归因并落库，见 app/api/budget.py:1004-1012）
+  2) 已有旧口径归因的告警 → 用修正后的科目口径重算覆盖（旧值见 /root/WB_WS/_backups/p13p14-20260914/）
+另: scenario_id 为空且 alert_type 命中 scenario_suggestions → 补 scenario_id（让告警"能解释、能关"）
+
+用法:
+  /usr/bin/python3 scripts/kpi_alert_attribution_backfill.py            # dry-run
+  /usr/bin/python3 scripts/kpi_alert_attribution_backfill.py --apply
+"""
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+
+if __name__ == "__main__":
+    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.database import get_session_local
+from app.models import BudgetDeviationAlert, KPIDefinition, ScenarioSuggestion
+from app.utils.alert_attribution import build_attribution, match_scenario
+
+API = "http://127.0.0.1:8010"
+SECRET = "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env"
+OUT = "/tmp/budget-p13p14/attribution-conclusions.json"
+
+
+def load_env(path):
+    vals = {}
+    with open(path) as fh:
+        for line in fh:
+            line = line.strip()
+            if line and not line.startswith("#") and "=" in line:
+                k, v = line.split("=", 1)
+                vals[k.strip()] = v.strip()
+    return vals
+
+
+def api(method, path, body=None, token=None):
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(API + path, data=data, method=method)
+    req.add_header("Content-Type", "application/json")
+    if token:
+        req.add_header("Authorization", f"Bearer {token}")
+    try:
+        with urllib.request.urlopen(req, timeout=90) as r:
+            return r.status, json.loads(r.read().decode())
+    except urllib.error.HTTPError as e:
+        return e.code, {"_error": e.read().decode()[:200]}
+
+
+def conclusion(alert, kpi, attribution, scenario):
+    dims = attribution.get("dimensions") or []
+    subs = attribution.get("subjects") or []
+    trend = attribution.get("trend") or {}
+    vtype = attribution.get("variance_type")
+    hot = [s for s in subs if abs(s.get("amount_diff") or 0) > 0.01][:3]
+    sub_txt = "; ".join(
+        f"{s['subject_code']} {s.get('subject_name','')} {s['amount_diff']:+.2f}元(权重{s.get('weight')})"
+        for s in hot) or "（科目明细为空：该KPI无可用取数映射，见报告§未接通清单）"
+    dim_txt = "; ".join(
+        f"{d.get('kpi_name')} 偏差{d.get('deviation_value')}" for d in dims
+        if d.get("deviation_value") is not None and abs(d["deviation_value"]) > 0.01) or "无下级KPI层级"
+    trend_txt = (f"{trend.get('type')}({trend.get('message')})" if trend.get("anomaly")
+                 else "无连续3期异常")
+    return (
+        f"[{alert.id}] {kpi.kpi_code if kpi else alert.kpi_id} {alert.period}: "
+        f"实际{alert.actual_value} vs 预算{alert.budget_value}（偏差{alert.deviation_rate}%, {alert.alert_level}）| "
+        f"场景={alert.alert_type or '未分类'}{'(建议:' + scenario['title'] + ')' if scenario else ''} | "
+        f"量价={vtype} | 主科目={sub_txt} | 子KPI={dim_txt} | 趋势={trend_txt}"
+    )
+
+
+def main():
+    apply = "--apply" in sys.argv
+    sec = load_env(SECRET)
+    st, login = api("POST", "/api/cma/auth/login",
+                    {"username": sec["CMA_ADMIN_USER"], "password": sec["CMA_ADMIN_PASS"], "entity_id": 1})
+    token = login.get("token")
+    print(f"login: HTTP {st} entity={login.get('entity_short_name')}")
+
+    db = get_session_local()()
+    try:
+        alerts = db.query(BudgetDeviationAlert).filter(
+            BudgetDeviationAlert.entity_id == 1).order_by(BudgetDeviationAlert.id).all()
+        print(f"== 告警: {len(alerts)} 条（open {sum(1 for a in alerts if a.status == 'open')}）")
+
+        api_filled, recomputed, scenario_filled = 0, 0, 0
+        lines = []
+        for a in alerts:
+            kpi = db.query(KPIDefinition).filter(KPIDefinition.id == a.kpi_id).first()
+            had = a.attribution is not None
+
+            if not had and apply:
+                # 1) 现成接口路径：现场组装并落库
+                st_a, resp = api("GET", f"/api/cma/budget/deviation-alerts/{a.id}/attribution", token=token)
+                if st_a == 200:
+                    api_filled += 1
+                db.expire(a)  # 端点已 commit，重读
+                a = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.id == a.id).first()
+            elif had and apply:
+                # 2) 旧口径 → 按修正科目口径重算
+                att, inferred = build_attribution(db, a.kpi_id, a.period, a.alert_type)
+                a.attribution = att
+                if a.alert_type is None:
+                    a.alert_type = inferred
+                recomputed += 1
+
+            scenario = match_scenario(db, a.alert_type)
+            if scenario and a.scenario_id is None and apply:
+                a.scenario_id = scenario["scenario_id"]
+                scenario_filled += 1
+
+            att_now = a.attribution or {}
+            lines.append({
+                "id": a.id, "kpi_id": a.kpi_id, "kpi_code": kpi.kpi_code if kpi else None,
+                "period": a.period, "status": a.status, "alert_level": a.alert_level,
+                "budget_value": a.budget_value, "actual_value": a.actual_value,
+                "deviation_rate": a.deviation_rate, "alert_type": a.alert_type,
+                "scenario_id": a.scenario_id, "has_attribution": bool(a.attribution),
+                "attribution_keys": sorted(att_now.keys()),
+                "view": conclusion(a, kpi, att_now, scenario),
+            })
+
+        if apply:
+            db.commit()
+
+        for row in lines:
+            print(row["view"])
+
+        with open(OUT, "w") as fh:
+            json.dump(lines, fh, ensure_ascii=False, indent=1)
+        print(f"\n== 输出结论: {OUT} ({len(lines)} 条)")
+
+        n_attr = db.query(BudgetDeviationAlert).filter(
+            BudgetDeviationAlert.entity_id == 1,
+            BudgetDeviationAlert.attribution.isnot(None)).count()
+        total = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.entity_id == 1).count()
+        n_open = db.query(BudgetDeviationAlert).filter(
+            BudgetDeviationAlert.entity_id == 1, BudgetDeviationAlert.status == "open").count()
+        n_scen = db.query(BudgetDeviationAlert).filter(
+            BudgetDeviationAlert.entity_id == 1,
+            BudgetDeviationAlert.scenario_id.isnot(None)).count()
+        print(f"== 归因覆盖: {n_attr}/{total}（open {n_open}）| scenario_id 已补: {n_scen} | "
+              f"接口落库 {api_filled} / 重算 {recomputed} / 场景补全 {scenario_filled}"
+              f"{'' if apply else ' [dry-run 未写库]'}")
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()
@@ -71,6 +71,12 @@ def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: boo
    period_field = mapping.period_field or "period"
    unit = mapping.unit_conversion or 1

+    # P1③④ 激活(2026-09-14): 科目组合映射 —— 由 kpi_subject_map 推导的多科目加权归集
+    # filter_rule 形如 {"subject_codes":["5001%"],"measure":"credit","weights":{"5001%":1.0},
+    #                   "calc":"sum|ratio","denominator":"5001%","scale":100}
+    if isinstance(filter_rule, dict) and filter_rule.get("subject_codes"):
+        return collect_subject_group(db, mapping, period, write_kpi=write_kpi)
+
    model = _source_model(table)

    # 构建查询
@@ -138,32 +144,120 @@ def collect_for_mapping(db, mapping: KPIValueSource, period: str, write_kpi: boo
        message += f", 类型{plan_type}"

    if write_kpi:
-        # upsert kpi_values: 同 kpi_id+period 已有 auto_collect 记录则更新
-        existing = db.query(KPIValue).filter(
-            KPIValue.kpi_id == mapping.kpi_id,
-            KPIValue.period == period,
-            KPIValue.source_type == "auto_collect",
-        ).first()
-        if existing:
-            existing.actual_value = value
-            existing.source_batch = _batch_no()
-            existing.remark = f"自动归集: {table}"
-            existing.data_status = "pending"
-        else:
-            db.add(KPIValue(
-                entity_id=mapping.entity_id,
-                kpi_id=mapping.kpi_id,
-                period=period,
-                actual_value=value,
-                source_type="auto_collect",
-                source_batch=_batch_no(),
-                data_status="pending",
-                remark=f"自动归集: {table}",
-            ))
+        _upsert_auto_value(db, mapping, period, value, f"自动归集: {table}")

    return value, message


+def _upsert_auto_value(db, mapping, period: str, value: float, remark: str) -> None:
+    """upsert kpi_values: 同 kpi_id+period 已有 auto_collect 记录则更新。
+
+    人工来源(excel/manual)行不匹配本查询 → 自动归集永不覆盖人工录入值。
+    """
+    existing = db.query(KPIValue).filter(
+        KPIValue.kpi_id == mapping.kpi_id,
+        KPIValue.period == period,
+        KPIValue.source_type == "auto_collect",
+    ).first()
+    if existing:
+        existing.actual_value = value
+        existing.source_batch = _batch_no()
+        existing.remark = remark
+        existing.data_status = "pending"
+    else:
+        db.add(KPIValue(
+            entity_id=mapping.entity_id,
+            kpi_id=mapping.kpi_id,
+            period=period,
+            actual_value=value,
+            source_type="auto_collect",
+            source_batch=_batch_no(),
+            data_status="pending",
+            remark=remark,
+        ))
+
+
+# ── 科目组合归集（P1③④ 激活 · 2026-09-14）─────────────────────────
+# 背景: kpi_subject_map 用新准则科目(6xxx)，voucher_details 是智享通凭证明细(5xxx/1xxx/2xxx)，
+# 编码体系不同 → 由 scripts/kpi_value_source_derive.py 先做「同代码/同名/子目前缀/结转科目」
+# 机械桥接，把**实际存在的科目编码或前缀**写进 filter_rule.subject_codes，再由本函数归集。
+_CREDIT_NORMAL_PREFIXES = ("2", "4", "60", "61", "63")
+
+
+def _is_credit_normal(subject_code: str) -> bool:
+    return str(subject_code).startswith(_CREDIT_NORMAL_PREFIXES)
+
+
+def _subject_amount(db, entity_id: int, pattern: str, period: str, measure: str):
+    """单科目(支持 LIKE 前缀)期间金额 → (金额, 行数)。measure: credit|debit|net|debit_net"""
+    q = db.query(
+        func.coalesce(func.sum(VoucherDetail.credit_amount), 0),
+        func.coalesce(func.sum(VoucherDetail.debit_amount), 0),
+        func.count(VoucherDetail.id),
+    ).filter(VoucherDetail.entity_id == entity_id)
+    if "%" in pattern:
+        q = q.filter(VoucherDetail.subject_code.like(pattern))
+    else:
+        q = q.filter(VoucherDetail.subject_code == pattern)
+    if period:
+        q = q.filter(VoucherDetail.period == period)
+    credit, debit, rows = q.first()
+    credit, debit = float(credit or 0), float(debit or 0)
+    if measure == "debit":
+        amount = debit
+    elif measure == "net":
+        amount = credit - debit
+    elif measure == "debit_net":
+        amount = debit - credit
+    else:  # credit（默认）: 贷方发生额——损益类科目月结"结转"腿与本期发生额同额，取贷方即本期发生额
+        amount = credit
+    return amount, int(rows or 0)
+
+
+def collect_subject_group(db, mapping, period: str, write_kpi: bool = True):
+    """多科目加权归集（filter_rule.subject_codes 路径），返回 (采集值, 说明)。"""
+    rule = mapping.filter_rule or {}
+    codes = list(rule.get("subject_codes") or [])
+    weights = rule.get("weights") or {}
+    measure = rule.get("measure", "credit")
+    calc = rule.get("calc", "sum")
+    scale = float(rule.get("scale", 1) or 1)
+    unit = mapping.unit_conversion or 1
+
+    total, rows, parts, matched = 0.0, 0, [], 0
+    for code in codes:
+        amount, n = _subject_amount(db, mapping.entity_id, code, period, measure)
+        if n:
+            matched += 1
+        rows += n
+        weight = float(weights.get(code, 1) or 0)
+        total += weight * amount
+        parts.append(f"{code}×{weight:g}")
+    if rows == 0:
+        return None, f"源表无记录(period={period}, {len(codes)}个科目均无发生额)"
+
+    if calc == "ratio":
+        denom_code = rule.get("denominator") or (codes[0] if codes else None)
+        if not denom_code:
+            return None, "比率映射缺少 denominator"
+        denom, _ = _subject_amount(db, mapping.entity_id, str(denom_code), period, measure)
+        if not denom:
+            return None, f"比率分母 {denom_code} 期间无发生额"
+        value = round(float(total) / float(denom) * scale * unit, 4)
+    else:
+        value = round(float(total) * unit, 2)
+
+    message = (f"科目组合(period={period}, {measure}, 命中{matched}/{len(codes)}): "
+               + " + ".join(parts))
+    if calc == "ratio":
+        message += f" ÷ {denom_code} × {scale:g}"
+
+    if write_kpi:
+        _upsert_auto_value(db, mapping, period, value,
+                           f"自动归集: {mapping.source_table} 科目组合")
+    return value, message
+
+
 def _batch_no() -> str:
    return f"auto-{datetime.now().strftime('%Y%m%d%H%M%S')}"

@@ -0,0 +1,309 @@
+#!/usr/bin/env python3
+"""kpi_value_sources 机械推导器 —— 预算 P1③④「接通」步（2026-09-14）
+
+任务: yanxue-20260914-budget-p13p14-activation（研学提出，任总批「按推荐三步推进」）
+
+做什么:
+  以现成 kpi_subject_map(89 行) 为据，把 KPI→科目 映射推导成 kpi_value_sources
+  取数映射（source_table/source_field/aggregate/filter_rule/period_field/unit_conversion）。
+
+为什么需要桥接:
+  kpi_subject_map 是"新准则"科目体系(6001/6401/6602...)，
+  voucher_details(智享通凭证明细, 41833 行) 是"智享通"科目体系(5001/5401/5602xx/1001xx...)。
+  两套编码不同 → 机械桥接规则（全部可复算、可审计，无人工猜测）:
+    exact   同科目代码在 voucher_details 存在
+    name    kpi_subject_map.remark(科目名) == voucher_details.subject_name（同名即同科目）
+    prefix  voucher_details 存在以该代码开头的下级子目（如 1001 → 100101/100102）
+    closing 损益类科目组 → 3103 本年利润（月结结转科目，净发生额=本期净利）——仅限"净利润"类KPI
+  桥接不上的科目 → 输出「未匹配清单」，不硬猜（任务书明确要求）。
+
+激活判据（不是"能算就算"）:
+  用生产同一条归集代码路径试算 2026-06/07/08，与现网 kpi_values 的非 auto_collect 权威值
+  对账；≥2/3 期一致（|Δ| ≤ max(0.01, 1%×|ref|)）→ status=active，否则 status=draft（不采集，
+  只登记待业务确认口径）。
+
+用法:
+  /usr/bin/python3 scripts/kpi_value_source_derive.py            # dry-run（只打印）
+  /usr/bin/python3 scripts/kpi_value_source_derive.py --apply    # 落库（幂等，已存在则跳过）
+"""
+import os
+import sys
+import json
+from collections import defaultdict
+
+if __name__ == "__main__":
+    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.database import get_session_local
+from app.models import KPIValueSource, KPISubjectMap, KPIValue, KPIDefinition, VoucherDetail
+from scripts.kpi_value_collector import collect_subject_group
+
+ENTITY_ID = 1
+REF_PERIODS = ["2026-06", "2026-07", "2026-08"]
+TOL_ABS, TOL_REL = 0.01, 0.01
+CLOSING_ACCOUNT = "3103"          # 本年利润（损益类月结结转科目）
+PNL_PREFIXES = ("6",)
+
+
+def load_voucher_index(db):
+    """voucher_details 实际科目索引: code -> name, 以及全部代码集合"""
+    rows = db.query(VoucherDetail.subject_code, VoucherDetail.subject_name).distinct().all()
+    code_name, names, codes = {}, defaultdict(list), set()
+    for code, name in rows:
+        code = (code or "").strip()
+        name = (name or "").strip()
+        codes.add(code)
+        code_name[code] = name
+        names[name].append(code)
+    return code_name, names, codes
+
+
+def load_subject_dict(db):
+    """科目字典(subjects, 207行) → code -> 规范科目名（用于 remark 带前缀修饰时的同名桥接）"""
+    from app.models import Subject
+    return {(s.subject_code or "").strip(): (s.subject_name or "").strip()
+            for s in db.query(Subject).all()}
+
+
+def bridge_subject(code, remark, code_name, names, codes, subject_dict=None):
+    """单科目机械桥接 → (pattern, method) 或 (None, reason 字符串)
+
+    顺序: 同代码 → 本行 remark 同名 → 科目字典规范名同名 → 下级子目前缀。
+    """
+    code = str(code or "").strip()
+    remark = (remark or "").strip()
+    if not code:
+        return None, "科目代码为空"
+    if code in codes:
+        return code, "exact"
+    if remark and remark in names:
+        cand = sorted(names[remark])
+        if len(cand) == 1:
+            return cand[0], "name"
+        return None, f"同名科目多个候选{cand}（歧义，不硬猜）"
+    canonical = (subject_dict or {}).get(code)
+    if canonical and canonical != remark and canonical in names:
+        cand = sorted(names[canonical])
+        if len(cand) == 1:
+            return cand[0], "name(dict)"
+    children = sorted(c for c in codes if c.startswith(code) and len(c) > len(code))
+    if children:
+        return code + "%", "prefix"
+    return None, "voucher_details 无同代码/同名/下级子目（新老科目体系差异）"
+
+
+def build_candidate(db, kpi, rows, code_name, names, codes, subject_dict):
+    """为单个 KPI 构造候选映射 dict（不落库）"""
+    subs, unmatched = [], []
+    for r in rows:
+        pattern, method = bridge_subject(r.subject_code, r.remark, code_name, names, codes, subject_dict)
+        if pattern:
+            subs.append({"new_code": r.subject_code, "pattern": pattern, "method": method,
+                         "calc_type": r.calc_type, "weight": float(r.weight if r.weight is not None else 1.0)})
+        else:
+            unmatched.append({"new_code": r.subject_code, "reason": method})
+
+    all_pnl = all(str(r.subject_code).startswith(PNL_PREFIXES) for r in rows)
+    mixed_measure = len({str(s["new_code"])[0] in ("1", "2", "4") for s in subs}) > 1
+
+    cand = {
+        "kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
+        "unit": kpi.unit, "subjects": subs, "unmatched": unmatched,
+        "total_subjects": len(rows),
+    }
+    if not subs:
+        if all_pnl and "净利润" in (kpi.kpi_name or ""):
+            cand.update(subject_codes=[CLOSING_ACCOUNT + "%"], method="closing",
+                        measure="net", calc="sum", weights={CLOSING_ACCOUNT + "%": 1.0},
+                        unit_conversion=0.0001, reason="损益科目组→3103本年利润结转科目")
+            return cand
+        cand["skip_reason"] = f"0/{len(rows)} 科目可桥接"
+        return cand
+
+    if all_pnl and unmatched and "净利润" in (kpi.kpi_name or ""):
+        cand.update(subject_codes=[CLOSING_ACCOUNT + "%"], method="closing",
+                    measure="net", calc="sum", weights={CLOSING_ACCOUNT + "%": 1.0},
+                    unit_conversion=0.0001,
+                    reason=f"损益科目组 {len(subs)}/{len(rows)} 可桥接，缺费用类科目 → 走结转科目")
+        return cand
+
+    calc_types = {s["calc_type"] for s in subs}
+    calc = "ratio" if calc_types == {"ratio"} else ("sum" if calc_types <= {"sum", None} else None)
+    if calc is None:
+        cand["skip_reason"] = f"calc_type 混合{calc_types}（无法机械推导单一口径）"
+        return cand
+
+    measure = "credit" if all_pnl else ("net" if not mixed_measure else None)
+    if measure is None:
+        cand["skip_reason"] = "同一KPI内混用损益类与余额类科目（无法机械推导）"
+        return cand
+    if not all_pnl and calc == "ratio":
+        cand["skip_reason"] = "余额类科目占比率口径，需业务定义分子分母"
+        return cand
+
+    subject_codes = [s["pattern"] for s in subs]
+    weights = {s["pattern"]: s["weight"] for s in subs}
+    if calc == "ratio":
+        unit_conv = 1.0                      # 比率口径：值本身是 %，不再换算单位
+    else:
+        unit_conv = 1.0 if (kpi.unit or "").strip() == "%" else 0.0001   # 元 → 万元
+    cand.update(subject_codes=subject_codes, method="+".join(sorted({s["method"] for s in subs})),
+                measure=measure, calc=calc, weights=weights, unit_conversion=unit_conv)
+    if calc == "ratio":
+        if len(subject_codes) == 1:
+            cand["skip_reason"] = "单一科目比率：分母与分子同源，无法机械推导（需业务定义）"
+            return cand
+        cand["denominator"] = subject_codes[0]
+        cand["scale"] = 100
+    return cand
+
+
+def reference_value(db, kpi_id, period):
+    row = db.query(KPIValue).filter(
+        KPIValue.kpi_id == kpi_id,
+        KPIValue.period == period,
+        KPIValue.source_type != "auto_collect",
+        KPIValue.actual_value.isnot(None),
+    ).order_by(KPIValue.calculated_at.desc()).first()
+    return float(row.actual_value) if row else None
+
+
+def validate(db, cand):
+    """试算（生产同一代码路径，不写库）+ 与现网权威口径对账"""
+    probe = KPIValueSource(
+        entity_id=ENTITY_ID, kpi_id=cand["kpi_id"], source_table="voucher_details",
+        source_field="credit_amount", aggregate="sum",
+        filter_rule={"subject_codes": cand["subject_codes"], "weights": cand["weights"],
+                     "measure": cand["measure"], "calc": cand["calc"],
+                     "denominator": cand.get("denominator"), "scale": cand.get("scale", 1)},
+        period_field="period", unit_conversion=cand["unit_conversion"], status="active",
+    )
+    checks, ok = [], 0
+    for p in REF_PERIODS:
+        try:
+            val, msg = collect_subject_group(db, probe, p, write_kpi=False)
+        except Exception as e:  # 源表缺列等
+            checks.append({"period": p, "computed": None, "ref": None, "match": False, "error": str(e)})
+            continue
+        ref = reference_value(db, cand["kpi_id"], p)
+        match = (val is not None and ref is not None
+                 and abs(val - ref) <= max(TOL_ABS, TOL_REL * abs(ref)))
+        ok += 1 if match else 0
+        checks.append({"period": p, "computed": val, "ref": ref, "match": bool(match)})
+    return checks, ok
+
+
+def main():
+    apply = "--apply" in sys.argv
+    db = get_session_local()()
+    try:
+        code_name, names, codes = load_voucher_index(db)
+        subject_dict = load_subject_dict(db)
+        kpis = {k.id: k for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == ENTITY_ID).all()}
+        rows = db.query(KPISubjectMap).order_by(KPISubjectMap.kpi_id, KPISubjectMap.id).all()
+        grouped = defaultdict(list)
+        for r in rows:
+            grouped[r.kpi_id].append(r)
+
+        print(f"== kpi_subject_map: {len(rows)} 行 / {len(grouped)} 个 KPI")
+        print(f"== voucher_details 实际科目: {len(codes)} 个")
+        print(f"== mode: {'APPLY(落库)' if apply else 'DRY-RUN(只打印)'}\n")
+
+        existing = {(r.entity_id, r.kpi_id, r.source_table)
+                    for r in db.query(KPIValueSource).all()}
+        created = skipped_kpi = updated = 0
+        active, draft = [], []
+
+        for kpi_id in sorted(grouped):
+            kpi = kpis.get(kpi_id)
+            if not kpi:
+                print(f"-- KPI {kpi_id}: 无定义，跳过")
+                continue
+            cand = build_candidate(db, kpi, grouped[kpi_id], code_name, names, codes, subject_dict)
+            head = f"[KPI {kpi_id} {kpi.kpi_code} {kpi.kpi_name}]"
+            if cand.get("skip_reason"):
+                skipped_kpi += 1
+                print(f"{head} SKIP: {cand['skip_reason']}")
+                for u in cand.get("unmatched", []):
+                    print(f"     未匹配科目 {u['new_code']}: {u['reason']}")
+                continue
+
+            checks, ok = validate(db, cand)
+            complete = (not cand["unmatched"]) or cand["method"] == "closing"
+            if ok >= 2:
+                status = "active"
+            elif complete:
+                status = "draft"
+            else:
+                skipped_kpi += 1
+                print(f"{head} SKIP: 部分覆盖 {len(cand['subjects'])}/{cand['total_subjects']} 科目"
+                      f"且未通过口径对账（{ok}/3）→ 不硬猜、不上线")
+                for c in checks:
+                    print(f"     对账 {c['period']} 算={c['computed']} 现网={c['ref']}")
+                for u in cand["unmatched"]:
+                    print(f"     未匹配科目 {u['new_code']}: {u['reason']}")
+                continue
+            cand["status"] = status
+            cand["validation"] = {"checked_periods": REF_PERIODS, "matched": ok, "checks": checks}
+            (active if status == "active" else draft).append(cand)
+
+            print(f"{head} {status.upper()}  桥接 {len(cand['subjects'])}/{cand['total_subjects']} "
+                  f"method={cand['method']} {cand['calc']}/{cand['measure']} unit×{cand['unit_conversion']}")
+            print(f"     filter_rule={json.dumps({k: v for k, v in cand.items() if k in ('subject_codes','weights','measure','calc','denominator','scale')}, ensure_ascii=False)}")
+            print(f"     对账 {ok}/3: " + " | ".join(
+                f"{c['period']} 算={c['computed']} 现网={c['ref']} {'OK' if c['match'] else 'DIFF'}"
+                for c in checks))
+            for u in cand["unmatched"]:
+                print(f"     未匹配科目 {u['new_code']}: {u['reason']}")
+
+            rule = {"subject_codes": cand["subject_codes"], "weights": cand["weights"],
+                    "measure": cand["measure"], "calc": cand["calc"],
+                    "derivation": {"source": "kpi_subject_map", "method": cand["method"],
+                                   "note": cand.get("reason", "机械桥接(同代码/同名/子目前缀)")},
+                    "validation": {"ref_periods": REF_PERIODS, "matched": ok,
+                                   "checks": [{k: c[k] for k in ("period", "computed", "ref", "match")}
+                                              for c in checks]}}
+            if cand["calc"] == "ratio":
+                rule["denominator"] = cand["denominator"]
+                rule["scale"] = cand["scale"]
+
+            legacy = db.query(KPIValueSource).filter(
+                KPIValueSource.entity_id == ENTITY_ID,
+                KPIValueSource.kpi_id == kpi_id,
+                KPIValueSource.source_table == "voucher_details").first()
+            if legacy and isinstance(legacy.filter_rule, dict) and legacy.filter_rule.get("derivation"):
+                print("     → 已存在派生映射，跳过写入")
+                continue
+            if legacy:
+                print(f"     → 旧映射 id={legacy.id} {legacy.filter_rule} unit×{legacy.unit_conversion}"
+                      f" → 升级为派生映射 unit×{cand['unit_conversion']}")
+                legacy.source_field = "credit_amount"
+                legacy.aggregate = "sum"
+                legacy.period_field = "period"
+                legacy.filter_rule = rule
+                legacy.unit_conversion = cand["unit_conversion"]
+                legacy.status = status
+                updated += 1
+                if apply:
+                    db.commit()
+                continue
+            created += 1
+            if apply:
+                db.add(KPIValueSource(
+                    entity_id=ENTITY_ID, kpi_id=kpi_id, source_table="voucher_details",
+                    source_field="credit_amount", aggregate="sum", filter_rule=rule,
+                    period_field="period", unit_conversion=cand["unit_conversion"], status=status,
+                ))
+
+        if apply:
+            db.commit()
+        total = db.query(KPIValueSource).filter(KPIValueSource.entity_id == ENTITY_ID).count()
+        print(f"\n== 汇总: 候选 {len(active) + len(draft)} 个 KPI（active {len(active)} / draft {len(draft)}）"
+              f"，0 科目可桥接跳过 {skipped_kpi} 个，本次新建 {created} 行 / 升级 {updated} 行")
+        print(f"== kpi_value_sources 现有行数: {total}{' (已落库)' if apply else ' (未落库, dry-run)'}")
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()