Hermes CI Fix
fb4609be14
ci/woodpecker/push/main Pipeline failed
- ① kpi_templates instantiate 构造点显式传会话账套（原缺 entity_id ⇒ 静默落 e1），kpi_code 判重按账套收紧
- ② kpi_subject_map 3 个读取点加护栏注释（不加列、不改语义）
- ③ subjects 分层：读口 union（NULL=全局标准准则科目 / N=账套私有），UPSERT 唯一键改 (entity_id, subject_code)
- ④ analysis_results 账套化：读/写/删三端点按会话账套判定归属（列未落地时 kpi_code JOIN 止血），
无归属行进 unattributed 登记、删除一律 404
- 模型层按现网 DDL 补齐：KPIDefinition(template_id/is_system + uk_entity_kpi)、AnalysisResult.entity_id、Subject.entity_id
- 新增 migrate_{subjects,analysis_results}_entity_id_20260923.py（--check/--apply/--down）、
22 条对账用例 test_entity_dimension_convergence_20260923.py
DDL 已由 ops 受控通道落地（subjects 21:31:56 / analysis_results 21:31:59 CST，审批 1e9b4d9a/5ffb1640）
本提交为切片提交，未 push（发布须另取放行）
315 lines
15 KiB
Python
315 lines
15 KiB
Python
#!/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()}
|
||
# ⚠️ kpi_subject_map 护栏（2026-09-23 账套维度收口批 ②，零逻辑改动）：
|
||
# 该表无账套列 —— **账套维度经 `kpi_id` 继承**。此处全表取映射后由 `kpis`（已按
|
||
# `entity_id == ENTITY_ID` 过滤）逐行滤掉非本账套 KPI ⇒ 现状安全。
|
||
# **禁止新增按 `subject_code` 反查的读口，禁止按 `subject_code` 批量写的入口**：
|
||
# 同一科目码可同时属多个账套（标准准则科目全局共享），按科目反查/批量写立刻跨账套。
|
||
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()
|