#!/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()