diff --git a/backend/app/utils/alert_attribution.py b/backend/app/utils/alert_attribution.py index 810017bd..fa7e0681 100644 --- a/backend/app/utils/alert_attribution.py +++ b/backend/app/utils/alert_attribution.py @@ -78,39 +78,77 @@ def build_dimension_attribution(db: Session, kpi_id: int, period: str) -> list: return result +def _subject_patterns(db: Session, kpi_id: int) -> list: + """科目取数口径解析（2026-09-14 P1③④ 激活）: + + 优先用已推导的取数映射 kpi_value_sources.filter_rule.subject_codes —— 那是把 + kpi_subject_map(新准则 6xxx) 机械桥接到 voucher_details 实际科目(5xxx/1xxx/2xxx) 的结果； + 无映射时退回 kpi_subject_map 原始科目（新老科目体系不一致时明细多为 0）。 + """ + from app.models import KPIValueSource, KPISubjectMap + + src = db.query(KPIValueSource).filter( + KPIValueSource.kpi_id == kpi_id, + KPIValueSource.source_table == "voucher_details", + KPIValueSource.status.in_(("active", "draft")), + ).order_by(KPIValueSource.status.asc()).first() + rule = (src.filter_rule or {}) if src else {} + labels = {m.subject_code: (m.remark or m.subject_code) + for m in db.query(KPISubjectMap).filter(KPISubjectMap.kpi_id == kpi_id).all()} + if rule.get("subject_codes"): + return [{ + "code": code, + "label": labels.get(str(code).rstrip("%"), str(code)), + "weight": float((rule.get("weights") or {}).get(code, 1.0)), + "measure": rule.get("measure", "credit"), + "source": "kpi_value_sources", + } for code in rule["subject_codes"]] + return [{ + "code": m.subject_code, + "label": m.remark or m.subject_code, + "weight": float(m.weight if m.weight is not None else 1.0), + "measure": "credit", + "source": "kpi_subject_map", + } for m in db.query(KPISubjectMap).filter(KPISubjectMap.kpi_id == kpi_id).all()] + + def build_subject_attribution(db: Session, kpi_id: int, period: str) -> list: - """科目明细拆解 — 查 kpi_subject_map → voucher_details 汇总科目发生额""" - mappings = db.query(KPISubjectMap).filter(KPISubjectMap.kpi_id == kpi_id).all() - if not mappings: + """科目明细拆解 — 科目口径取 _subject_patterns（优先修正后的取数映射），按 entity 隔离""" + patterns = _subject_patterns(db, kpi_id) + if not patterns: return [] + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() + entity_id = kpi.entity_id if kpi else None + result = [] - for m in mappings: + for p in patterns: q = db.query( - func.coalesce(func.sum(VoucherDetail.debit_amount), 0), func.coalesce(func.sum(VoucherDetail.credit_amount), 0), - ).filter( - VoucherDetail.subject_code == m.subject_code, - VoucherDetail.period == period, - ) + func.coalesce(func.sum(VoucherDetail.debit_amount), 0), + func.count(VoucherDetail.id), + ).filter(VoucherDetail.period == period) + if "%" in p["code"]: + q = q.filter(VoucherDetail.subject_code.like(p["code"])) + else: + q = q.filter(VoucherDetail.subject_code == p["code"]) + if entity_id is not None: + q = q.filter(VoucherDetail.entity_id == entity_id) row = q.first() - debit_sum = float(row[0] or 0) - credit_sum = float(row[1] or 0) - # 方向: credit贷方(收入/流入) / debit借方(费用/流出) - if m.calc_type == "ratio": - amount = credit_sum - debit_sum - elif m.calc_type in ("avg", "other"): - amount = (credit_sum - debit_sum) / 2 - else: # sum - amount = credit_sum - debit_sum - amount = round(amount * float(m.weight or 1.0), 2) + credit_sum = float(row[0] or 0) + debit_sum = float(row[1] or 0) + rows = int(row[2] or 0) + # 口径: 损益类取贷方发生额(月结结转腿=本期发生额) / 余额类取净额(借-贷) + amount = (debit_sum - credit_sum) if p["measure"] == "net" else credit_sum + amount = round(amount * p["weight"], 2) result.append({ - "subject_code": m.subject_code, - "subject_name": m.remark or m.subject_code, + "subject_code": p["code"], + "subject_name": p["label"], "amount_diff": amount, - "calc_type": m.calc_type, - "weight": float(m.weight or 1.0), + "calc_type": p["measure"], + "weight": p["weight"], + "rows": rows, }) total = sum(abs(r["amount_diff"]) for r in result) or 0 diff --git a/backend/scripts/kpi_alert_attribution_backfill.py b/backend/scripts/kpi_alert_attribution_backfill.py new file mode 100644 index 00000000..f81e42db --- /dev/null +++ b/backend/scripts/kpi_alert_attribution_backfill.py @@ -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() diff --git a/backend/scripts/kpi_value_collector.py b/backend/scripts/kpi_value_collector.py index 769c4055..c3772242 100644 --- a/backend/scripts/kpi_value_collector.py +++ b/backend/scripts/kpi_value_collector.py @@ -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')}" diff --git a/backend/scripts/kpi_value_source_derive.py b/backend/scripts/kpi_value_source_derive.py new file mode 100644 index 00000000..5851518d --- /dev/null +++ b/backend/scripts/kpi_value_source_derive.py @@ -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()