diff --git a/backend/scripts/kpi_review_gate.py b/backend/scripts/kpi_review_gate.py new file mode 100644 index 00000000..a196af8b --- /dev/null +++ b/backend/scripts/kpi_review_gate.py @@ -0,0 +1,430 @@ +"""auto_collect 行复核闸门 — 管理会计OS(财务裁定 2026-09-14) + +依据:`fullstack-20260914-autocollect-reviewflow-decision`(财务Bot 裁定) + 「规则闸门自动转正 + 例外人工复核」,规则 R1-R7;落地方 `finance-20260914-autocollect-reviewflow-impl.md`。 + +职责(单一实现,采集器与回填 CLI 共用): + - 采集落库时同事务判定 data_status(不再写死 pending); + - 对存量 pending 的 auto_collect 行做回填判定(例外清单 + 转正/归并); + - 例外清单输出(不新增表;落 JSON/文本,可由采集链路或人工触发)。 + +判定口径(R2 六条 + R3 七类例外): + R2-① `kpi_value_sources.status='active'` 且该期间 validation 锚点 match=true(锚点缺失/为假 → 例外); + 规则级 matched 数一并记录(kpi3 规则 matched=2/3:2026-07 match=false、2026-08 match=true)。 + R2-② 与同 (kpi_id, period) 非 auto 主值对账:金额类 |Δ| ≤ max(1000元, 0.5%×|主值|);比率类 |Δ| ≤ 0.5 个百分点。 + R2-③ 主值 calculated_at ≥ 该期取数科目凭证明细末次同步时间(无批次时点差)。 + 实现口径:**按重大性适用**——Δ 在容差内时主值陈旧不改变对外数字,仍按「同一事实重复行」归并; + Δ 超容差且主值陈旧 → 例外子类 batch_lag(批次时点差,非差错);Δ 超容差且主值不陈旧 → caliber_diff(口径差异)。 + 依据:裁定 §二 依据3 把「批次时点差」作为超差行的**性质判定**(本例净利 2026-07)。 + R2-④ 同 (kpi_id, period) 无 manual/excel/excel-import 行(人工值优先)。 + R2-⑤ 该期已结账(复用 `kpi_bep_deriver._is_period_closed`,不造第二套判定)。 + R2-⑥ 规则版本自上次转正后未变更(版本号 = filter_rule.version,缺省取规则体短哈希;从既往转正 remark 回读)。 + R3-⑥ 派生指标(filter_rule.calc='cashflow_direct' 或映射 status='draft')根本不自动转正。 + R3-⑦ 同 (kpi_id, period) 出现双主值 → 例外。 + +转正动作(同一事务): + - 值一致/容差内 → **保留原主值**,自动行置 `error`(同一事实重复行,留痕不删行),主值 remark 回写复算一致; + - 无主值 → 自动行置 `verified` 成为首个主值,remark 按 R5 格式; + - 例外 → 不动(保持 pending),进例外清单。 + +用法: + venv/bin/python3 scripts/kpi_review_gate.py --dry-run --period 2026-07 2026-08 + venv/bin/python3 scripts/kpi_review_gate.py --apply --period 2026-07 2026-08 + venv/bin/python3 scripts/kpi_review_gate.py --dry-run --value-id 2130 + venv/bin/python3 scripts/kpi_review_gate.py --apply --exception-out /path/exceptions.json +""" +import argparse +import hashlib +import json +import logging +import os +import sys +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import func # noqa: E402 + +from app.database import get_session_local # noqa: E402 +from app.models import KPIValue, KPIValueSource, VoucherDetail # noqa: E402 + +try: # 采集器/脚本同目录导入 + from scripts.kpi_bep_deriver import _is_period_closed +except ImportError: # pragma: no cover - 直接以脚本目录运行时 + from kpi_bep_deriver import _is_period_closed + +logging.basicConfig(level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") +logger = logging.getLogger("cma.kpi_review_gate") + +RULE_VERSION = "20260914" +AMOUNT_MIN_YUAN = 1000.0 # R7 金额类下限(元) +AMOUNT_PCT = 0.005 # R7 金额类比例(0.5%×|主值|) +RATIO_PP = 0.005 # R7 比率类(0.5 个百分点;比率值以小数存,13% 存 0.13) +AUTO_SOURCE = "auto_collect" +MANUAL_SOURCES = ("manual", "excel", "excel-import") +DERIVED_CALCS = ("cashflow_direct",) + +VERDICT_MERGE = "merge_duplicate" # 与主值一致 → 归并(自动行不作主值) +VERDICT_PROMOTE = "promote_first" # 无主值 → 自动行转正成为主值 +VERDICT_EXCEPTION = "exception" # 例外人工 + + +# ── 规则版本 / 工具 ──────────────────────────────────────────────── +def rule_version(mapping: KPIValueSource) -> str: + """规则版本号:filter_rule.version 优先,缺省取规则体短哈希(不含 validation 结果)。""" + rule = mapping.filter_rule or {} + if isinstance(rule, dict) and rule.get("version"): + return str(rule["version"]) + core = {k: v for k, v in (rule or {}).items() if k != "validation"} + body = json.dumps(core, sort_keys=True, ensure_ascii=False, default=str) + return hashlib.sha256(body.encode("utf-8")).hexdigest()[:8] + + +def _rule_calc(mapping: KPIValueSource) -> str: + rule = mapping.filter_rule or {} + return str(rule.get("calc") or "sum") if isinstance(rule, dict) else "sum" + + +def _subject_patterns(mapping: KPIValueSource): + rule = mapping.filter_rule or {} + codes = list(rule.get("subject_codes") or []) if isinstance(rule, dict) else [] + return [str(c) for c in codes] + + +def tolerance(mapping: KPIValueSource, main_value: float): + """返回 (容差, 类别)。比率类按百分点;金额类按 max(1000元, 0.5%×|主值|) 换算到存放单位。""" + if _rule_calc(mapping) == "ratio": + return RATIO_PP, "ratio" + unit = float(mapping.unit_conversion or 1) or 1.0 + tol_yuan = max(AMOUNT_MIN_YUAN, AMOUNT_PCT * abs(float(main_value or 0)) / unit) + return round(tol_yuan * unit, 6), "amount" + + +# ── 查询 ────────────────────────────────────────────────────────── +def get_mapping(db, entity_id: int, kpi_id: int): + q = db.query(KPIValueSource).filter( + KPIValueSource.entity_id == entity_id, KPIValueSource.kpi_id == kpi_id) + return (q.filter(KPIValueSource.status == "active").first()) or q.first() + + +def anchor_check(mapping: KPIValueSource, period: str): + """R2-①:该期锚点 match=true?同时回报规则级 matched/锚点数。""" + rule = mapping.filter_rule or {} + validation = rule.get("validation") if isinstance(rule, dict) else None + if not validation: + return False, "规则无 validation 锚点", None + checks = validation.get("checks") or [] + anchors = len(validation.get("ref_periods") or checks) + matched = validation.get("matched") + hit = [c for c in checks if str(c.get("period")) == str(period)] + if not hit: + return False, f"该期({period})无锚点记录", {"matched": matched, "anchors": anchors} + ok = all(bool(c.get("match")) for c in hit) + return ok, f"锚点 match={ok} (computed={hit[0].get('computed')} ref={hit[0].get('ref')})", \ + {"matched": matched, "anchors": anchors} + + +def last_source_sync(db, entity_id: int, period: str, patterns=None): + """该期取数科目凭证明细末次入库时间(无科目模式时退化为全期)。""" + q = db.query(func.max(VoucherDetail.created_at)).filter( + VoucherDetail.entity_id == entity_id, VoucherDetail.period == period) + if patterns: + from sqlalchemy import or_ + conds = [] + for p in patterns: + conds.append(VoucherDetail.subject_code.like(p if "%" in p else p)) + q = q.filter(or_(*conds)) + val = q.scalar() + return val + + +def main_rows(db, entity_id: int, kpi_id: int, period: str): + """非 auto 的 verified 行(按 id desc,主值取最新)。""" + return db.query(KPIValue).filter( + KPIValue.entity_id == entity_id, KPIValue.kpi_id == kpi_id, + KPIValue.period == period, KPIValue.data_status == "verified", + KPIValue.source_type != AUTO_SOURCE, + ).order_by(KPIValue.id.desc()).all() + + +def manual_rows(db, entity_id: int, kpi_id: int, period: str): + return db.query(KPIValue).filter( + KPIValue.entity_id == entity_id, KPIValue.kpi_id == kpi_id, + KPIValue.period == period, KPIValue.source_type.in_(MANUAL_SOURCES)).all() + + +def last_promoted_version(db, entity_id: int, kpi_id: int, exclude_id=None): + """R2-⑥:从既往转正行 remark 回读上次转正的规则版本(`rule= v`)。""" + q = db.query(KPIValue).filter( + KPIValue.entity_id == entity_id, KPIValue.kpi_id == kpi_id, + KPIValue.remark.like("%自动归集复算一致%")) + if exclude_id: + q = q.filter(KPIValue.id != exclude_id) + row = q.order_by(KPIValue.id.desc()).first() + if not row or not row.remark: + return None, None + import re + m = re.search(r"rule=(\S+)\s+v(\S+?)[,)]", row.remark) + return (m.group(2) if m else None), row.id + + +# ── 判定 ────────────────────────────────────────────────────────── +def evaluate(db, kv: KPIValue, mapping=None, write_check=True) -> dict: + """对单条 auto_collect 行判定。返回 decision dict(不写库)。""" + entity_id, kpi_id, period = kv.entity_id, kv.kpi_id, kv.period + mapping = mapping or get_mapping(db, entity_id, kpi_id) + value = float(kv.actual_value if kv.actual_value is not None else 0.0) + + d = { + "value_id": kv.id, "entity_id": entity_id, "kpi_id": kpi_id, "period": period, + "auto_value": value, "source_batch": kv.source_batch, + "verdict": VERDICT_EXCEPTION, "subtype": None, "reasons": [], + "main_id": None, "main_value": None, "main_batch": None, "delta": None, + "tolerance": None, "unit_class": None, "main_stale": False, + "source_sync_at": None, "main_calculated_at": None, + "rule_id": getattr(mapping, "id", None), "rule_version": rule_version(mapping) if mapping else None, + "rule_calc": _rule_calc(mapping) if mapping else None, + "period_closed": None, "period_closed_by": None, + "anchor": None, "rule_matched": None, "rule_anchors": None, + "manual_rows": 0, "prev_rule_version": None, "prev_rule_value_id": None, + } + if mapping is None: + d["reasons"].append("无取数映射(kpi_value_sources 缺失)") + d["subtype"] = "no_mapping" + return d + + # R3-⑥ 派生指标:不自动转正 + if _rule_calc(mapping) in DERIVED_CALCS or (mapping.status or "") != "active": + d["subtype"] = "derived_kpi" if _rule_calc(mapping) in DERIVED_CALCS else "source_not_active" + d["reasons"].append(f"R3-⑥ 派生指标/映射非 active(calc={_rule_calc(mapping)}, status={mapping.status})不自动转正") + return d + + # R2-① 锚点 + ok, detail, agg = anchor_check(mapping, period) + d["anchor"] = detail + if agg: + d["rule_matched"], d["rule_anchors"] = agg.get("matched"), agg.get("anchors") + if not ok: + d["reasons"].append(f"R2-①/R3-② 锚点未全通过:{detail}(规则级 matched={d['rule_matched']}/{d['rule_anchors']})") + d["subtype"] = "anchor_mismatch" + + # R2-⑤ 结账 + try: + closed, by = _is_period_closed(db, entity_id, period) + except Exception as e: # 结账判定异常 → 不转正(fail-safe) + closed, by = False, f"判定异常: {e}" + d["period_closed"], d["period_closed_by"] = bool(closed), by + if not closed: + d["reasons"].append(f"R2-⑤/R3-④ 未结账({by or '无结账信号'})") + + # R2-④ 人工/导入行优先 + man = manual_rows(db, entity_id, kpi_id, period) + d["manual_rows"] = len(man) + if man: + d["reasons"].append(f"R2-④/R3 存在人工/导入行 {len(man)} 条(source_type={man[0].source_type})") + + # R3-⑦ 双主值 + mains = main_rows(db, entity_id, kpi_id, period) + if len(mains) > 1: + d["reasons"].append(f"R3-⑦ 同 (kpi_id, period) 双主值 {len(mains)} 条(id={[m.id for m in mains]})") + d["subtype"] = d["subtype"] or "double_main" + + # R2-③ 批次时点 + sync_at = last_source_sync(db, entity_id, period, _subject_patterns(mapping)) + d["source_sync_at"] = str(sync_at) if sync_at else None + if mains: + main = mains[0] + d["main_id"], d["main_value"], d["main_batch"] = main.id, float(main.actual_value or 0), main.source_batch + d["main_calculated_at"] = str(main.calculated_at) if main.calculated_at else None + if sync_at and main.calculated_at and main.calculated_at < sync_at: + d["main_stale"] = True + + # R2-② 对账 + if mains: + tol, cls = tolerance(mapping, d["main_value"]) + d["tolerance"], d["unit_class"] = tol, cls + d["delta"] = round(value - d["main_value"], 6) + within = abs(d["delta"]) <= tol + if within: + d["verdict"] = VERDICT_MERGE + if d["main_stale"]: + d["reasons"].append( + f"R2-③ 主值批次({d['main_calculated_at']})早于该期源末次同步({d['source_sync_at']})," + f"但 Δ={d['delta']} 在容差 {tol} 内 → 主值陈旧不改数,按同一事实重复行归并") + else: + d["subtype"] = d["subtype"] or ("batch_lag" if d["main_stale"] else "caliber_diff") + d["reasons"].append( + f"R2-②/R3-① Δ={d['delta']} 超容差 {tol}({cls})→ 例外" + + (";R3-③ 主值批次早于源末次同步 = 批次时点差(非差错)" if d["main_stale"] else "")) + else: + # 无主值:无对账基准,自动行转正成为首个主值(否则该链路产出恒不可见) + d["verdict"] = VERDICT_PROMOTE + d["reasons"].append("无同 (kpi_id, period) 非 auto 主值 → 转正为首个主值") + + # R2-⑥ 规则版本 + prev_ver, prev_id = last_promoted_version(db, entity_id, kpi_id, exclude_id=kv.id) + d["prev_rule_version"], d["prev_rule_value_id"] = prev_ver, prev_id + if prev_ver and d["rule_version"] and str(prev_ver) != str(d["rule_version"]): + d["verdict"] = VERDICT_EXCEPTION + d["subtype"] = d["subtype"] or "rule_changed" + d["reasons"].append(f"R2-⑥/R3-⑤ 规则版本变更(上次 v{prev_ver} @id{prev_id} → 本次 v{d['rule_version']})首次归集不自动转正") + + # 锚点/结账/人工/双主值 → 强制例外(R3) + if d["subtype"] in ("anchor_mismatch", "double_main") or d["manual_rows"] or not d["period_closed"]: + if d["verdict"] == VERDICT_MERGE: + d["verdict"] = VERDICT_EXCEPTION + return d + + +# ── 落库动作 ────────────────────────────────────────────────────── +def _with_recheck_remark(existing: str, seg: str, limit: int = 500, keep_recheck: int = 3) -> str: + """主值 remark 回写(幂等 + 长度安全)。 + + - 同一自动行的既往「复算一致」段按 id 替换(重跑不重复追加 → 不撑爆 varchar(500)); + - 最多保留最近 keep_recheck 条复算留痕(更早的留痕在各自自动行 remark 与审计文件中); + - 超出 limit 时截断旧正文(旧正文另存改前备份/报告)。 + """ + import re + seg = seg[:limit] + m = re.search(r"自动归集id=(\d+)", seg) + sid = m.group(1) if m else None + parts = [p.strip() for p in (existing or "").split(" || ") if p.strip()] + rechecks, others = [], [] + for p in parts: + r = re.match(r"^\d{4}-\d{2}-\d{2} voucher_details 复算一致\(自动归集id=(\d+)", p) + if r: + if sid is None or r.group(1) != sid: + rechecks.append(p) + else: + others.append(p) + rechecks = rechecks[-(keep_recheck - 1):] if keep_recheck > 1 else [] + tail = " || ".join(rechecks + [seg]) + out = tail + if others: + room = limit - len(tail) - 4 + if room >= 60: + body = " || ".join(others) + if len(body) > room: + body = body[:room - 3] + "..." + out = body + " || " + tail + if len(out) > limit: + out = tail[:limit] + return out + + +def apply_decision(db, kv: KPIValue, d: dict, mapping: KPIValueSource, today=None): + """按判定落库(不 commit,由调用方同事务提交)。返回动作说明。""" + today = today or datetime.now().strftime("%Y-%m-%d") + if d["verdict"] == VERDICT_MERGE: + delta_txt = f"{d['delta']:.4f}" if d["delta"] is not None else "-" + kv.data_status = "error" + kv.remark = (f"自动归集复算一致,与主值(id={d['main_id']})为同一事实重复行" + f"(rule={d['rule_id']} v{d['rule_version']}, Δ={delta_txt}, batch={d['source_batch']})" + f" 复核:财务Bot {today}")[:500] + main = db.query(KPIValue).filter(KPIValue.id == d["main_id"]).first() + if main is not None: + if d["main_stale"]: + stale = f";主值批次{d['main_calculated_at']}早于源同步{d['source_sync_at']}(时点差,不改数)" + else: + stale = "" + seg = (f"{today} voucher_details 复算一致(自动归集id={kv.id}, Δ={delta_txt}, " + f"rule={d['rule_id']} v{d['rule_version']}){stale}") + main.remark = _with_recheck_remark(main.remark, seg) + return f"merge: 自动行 id={kv.id} → error;主值 id={d['main_id']} 保留原值" + if d["verdict"] == VERDICT_PROMOTE: + delta_txt = "0.0000" + kv.data_status = "verified" + kv.remark = (f"自动归集复算一致(rule={d['rule_id']} v{d['rule_version']}, Δ={delta_txt}, " + f"batch={d['source_batch']}) 复核:财务Bot {today}")[:500] + return f"promote: 自动行 id={kv.id} → verified(首个主值)" + return f"exception({d['subtype']}): 自动行 id={kv.id} 保持 pending,不动" + + +def review_auto_value(db, mapping, kv, period=None): + """采集链路入口(同事务,由采集器在落 pending 后调用)。返回 decision。""" + d = evaluate(db, kv, mapping=mapping, write_check=False) + if d["verdict"] == VERDICT_EXCEPTION: + logger.info("复核门控: 例外 kpi=%s period=%s 自动=%.4f 主值=%s Δ=%s 原因=%s", + d["kpi_id"], d["period"], d["auto_value"], d["main_value"], + d["delta"], "; ".join(d["reasons"])[:200]) + return d + logger.info("%s kpi=%s period=%s 自动=%.4f 主值=%s Δ=%s", + d["verdict"], d["kpi_id"], d["period"], d["auto_value"], d["main_value"], d["delta"]) + apply_decision(db, kv, d, mapping) + return d + + +# ── 回填 / CLI ──────────────────────────────────────────────────── +def scan(db, entity_id=None, periods=None, kpi_id=None, value_id=None): + q = db.query(KPIValue).filter(KPIValue.source_type == AUTO_SOURCE) + if value_id: + q = q.filter(KPIValue.id == value_id) + else: + q = q.filter(KPIValue.data_status == "pending") + if entity_id is not None: + q = q.filter(KPIValue.entity_id == entity_id) + if periods: + q = q.filter(KPIValue.period.in_(periods)) + if kpi_id is not None: + q = q.filter(KPIValue.kpi_id == kpi_id) + rows = q.order_by(KPIValue.kpi_id, KPIValue.period, KPIValue.id).all() + out = [] + for kv in rows: + mapping = get_mapping(db, kv.entity_id, kv.kpi_id) + out.append((kv, mapping, evaluate(db, kv, mapping=mapping))) + return out + + +def run_cli(args): + db = get_session_local()() + try: + results = scan(db, entity_id=args.entity, periods=args.period, kpi_id=args.kpi, + value_id=args.value_id) + actions, exceptions = [], [] + for kv, mapping, d in results: + tag = "EXCEPTION" if d["verdict"] == VERDICT_EXCEPTION else d["verdict"].upper() + print(f"[{tag}/{d['subtype'] or '-'}] value_id={kv.id} kpi={d['kpi_id']} {d['period']} " + f"自动={d['auto_value']} 主值={d['main_value']} Δ={d['delta']} 容差={d['tolerance']} " + f"({d['unit_class']}) 主值id={d['main_id']} 规则v={d['rule_version']}") + for r in d["reasons"]: + print(f" · {r}") + if d["verdict"] == VERDICT_EXCEPTION: + exceptions.append(d) + elif mapping is not None: + actions.append(apply_decision(db, kv, d, mapping)) + if args.apply: + db.commit() + print(f"\n已提交: {len(actions)} 条动作, {len(exceptions)} 条例外") + else: + db.rollback() + print(f"\n[dry-run] 未提交: 待执行 {len(actions)} 条, 例外 {len(exceptions)} 条") + for a in actions: + print(f" - {a}") + if args.exception_out: + payload = {"generated_at": datetime.now().isoformat(timespec="seconds"), + "rule_version": RULE_VERSION, "mode": "apply" if args.apply else "dry-run", + "periods": args.period or [], "exceptions": exceptions} + with open(args.exception_out, "w", encoding="utf-8") as fh: + json.dump(payload, fh, ensure_ascii=False, indent=2, default=str) + print(f"例外清单 → {args.exception_out}") + return 0 + finally: + db.close() + + +def main(): + ap = argparse.ArgumentParser(description="auto_collect 行复核闸门(财务裁定 2026-09-14)") + ap.add_argument("--period", action="append", help="期间(可多次),缺省扫全部 pending") + ap.add_argument("--kpi", type=int, help="限定 KPI id") + ap.add_argument("--entity", type=int, default=1, help="实体 id(缺省 1=酣客)") + ap.add_argument("--value-id", type=int, help="限定单条 kpi_values.id") + ap.add_argument("--apply", action="store_true", help="落库(缺省 dry-run)") + ap.add_argument("--dry-run", action="store_true", help="显式 dry-run(与缺省同义)") + ap.add_argument("--exception-out", help="例外清单 JSON 输出路径") + args = ap.parse_args() + sys.exit(run_cli(args)) + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/kpi_value_collector.py b/backend/scripts/kpi_value_collector.py index 0b31871f..954c4759 100644 --- a/backend/scripts/kpi_value_collector.py +++ b/backend/scripts/kpi_value_collector.py @@ -11,11 +11,15 @@ /usr/bin/python3 scripts/kpi_value_collector.py 2026-08 # 指定期间 /usr/bin/python3 scripts/kpi_value_collector.py 2026-08 5 # 指定期间+KPI """ +import os import sys import logging from datetime import datetime from typing import Optional, Tuple +# 复核闸门开关(R2/R3,财务裁定 2026-09-14;KPI_REVIEW_GATE=off 可关闭,仅供测试) +_REVIEW_GATE_ENABLED = os.environ.get("KPI_REVIEW_GATE", "on").strip().lower() not in ("0", "off", "false", "no") + logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", @@ -161,6 +165,9 @@ def _upsert_auto_value(db, mapping, period: str, value: float, remark: str) -> N """upsert kpi_values: 同 kpi_id+period 已有 auto_collect 记录则更新。 人工来源(excel/manual)行不匹配本查询 → 自动归集永不覆盖人工录入值。 + + 复核闸门(财务裁定 2026-09-14,规则 R2/R3):落 pending 后**同事务**判定是否转正 / + 归并 / 进例外;判定异常一律 fail-safe 保持 pending(N 条转正条件未证成就不转正)。 """ existing = db.query(KPIValue).filter( KPIValue.kpi_id == mapping.kpi_id, @@ -172,8 +179,9 @@ def _upsert_auto_value(db, mapping, period: str, value: float, remark: str) -> N existing.source_batch = _batch_no() existing.remark = remark existing.data_status = "pending" + kv = existing else: - db.add(KPIValue( + kv = KPIValue( entity_id=mapping.entity_id, kpi_id=mapping.kpi_id, period=period, @@ -182,7 +190,29 @@ def _upsert_auto_value(db, mapping, period: str, value: float, remark: str) -> N source_batch=_batch_no(), data_status="pending", remark=remark, - )) + ) + db.add(kv) + db.flush() # 取 id(幂等去重与 remark 留痕需要) + _review_gate(db, mapping, period, kv) + + +def _review_gate(db, mapping, period: str, kv) -> None: + """R2/R3 复核闸门:同事务判定并落状态。任何异常都不阻断采集(保持 pending)。""" + if not _REVIEW_GATE_ENABLED: + return + try: + try: + from scripts.kpi_review_gate import review_auto_value + except ImportError: + from kpi_review_gate import review_auto_value + except ImportError as e: + logger.warning("复核闸门模块不可用(保持 pending): %s", e) + return + try: + review_auto_value(db, mapping, kv, period) + except Exception as e: + logger.warning("复核闸门判定异常 → 保持 pending (kpi=%s period=%s): %s", + mapping.kpi_id, period, e) # ── 科目组合归集(P1③④ 激活 · 2026-09-14)───────────────────────── diff --git a/backend/tests/test_kpi_review_gate_20260914.py b/backend/tests/test_kpi_review_gate_20260914.py new file mode 100644 index 00000000..d18e9f7b --- /dev/null +++ b/backend/tests/test_kpi_review_gate_20260914.py @@ -0,0 +1,89 @@ +"""auto_collect 复核闸门(kpi_review_gate.py)纯函数单测 — 2026-09-14 + +覆盖:阈值换算(金额/比率)、规则版本指纹、锚点判定、主值 remark 回写幂等与长度安全。 +不连库(只构造 KPIValueSource 内存对象)。 +""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.models import KPIValueSource # noqa: E402 +from scripts.kpi_review_gate import ( # noqa: E402 + RATIO_PP, _with_recheck_remark, anchor_check, rule_version, tolerance, +) + + +def _mapping(rule, unit=1.0, status="active", kpi_id=1): + m = KPIValueSource() + m.entity_id = 1 + m.kpi_id = kpi_id + m.source_table = "voucher_details" + m.source_field = "credit_amount" + m.aggregate = "sum" + m.period_field = "period" + m.unit_conversion = unit + m.status = status + m.filter_rule = rule + return m + + +def test_tolerance_amount_class(): + # 万元口径(unit=0.0001):max(1000元, 0.5%×主值) → 换算回万元 + m = _mapping({"calc": "sum", "subject_codes": ["5001"]}, unit=0.0001) + tol, cls = tolerance(m, 81.08) + assert cls == "amount" + assert abs(tol - 0.4054) < 1e-9 # 0.5%×810800元=4054元=0.4054万 + tol_small, _ = tolerance(m, 1.0) # 主值 1 万元 → 0.5%=50元 < 下限 1000元 + assert abs(tol_small - 0.1) < 1e-9 + + +def test_tolerance_ratio_class(): + m = _mapping({"calc": "ratio", "subject_codes": ["5001", "5401"]}, unit=1.0, kpi_id=2) + tol, cls = tolerance(m, 0.05) + assert cls == "ratio" and tol == RATIO_PP == 0.005 + # 0.34 个百分点在容差内、0.6 个百分点超容差 + assert abs(0.3766 - 0.38) <= tol + assert abs(0.3860 - 0.38) > tol + + +def test_rule_version_stable_and_changes_on_rule_edit(): + r1 = {"calc": "sum", "subject_codes": ["5001"], "validation": {"matched": 3}} + r2 = {"calc": "sum", "subject_codes": ["5001"], "validation": {"matched": 2}} + v1a = rule_version(_mapping(r1)) + v1b = rule_version(_mapping(r1)) + assert v1a == v1b # 确定性 + validation 不参与指纹 + assert rule_version(_mapping(r2)) == v1a + r3 = {"calc": "sum", "subject_codes": ["5001", "6001"], "validation": {"matched": 3}} + assert rule_version(_mapping(r3)) != v1a # 科目组合变更 → 版本变 + assert rule_version(_mapping({"version": 7, "calc": "sum"})) == "7" + + +def test_anchor_check_period_level(): + rule = {"calc": "sum", "validation": {"matched": 2, "ref_periods": ["2026-06", "2026-07", "2026-08"], + "checks": [{"period": "2026-06", "match": True}, + {"period": "2026-07", "match": False}, + {"period": "2026-08", "match": True}]}} + m = _mapping(rule, unit=0.0001, kpi_id=3) + ok_07, detail_07, agg = anchor_check(m, "2026-07") + ok_08, _, _ = anchor_check(m, "2026-08") + ok_09, detail_09, _ = anchor_check(m, "2026-09") + assert ok_07 is False and "match=False" in detail_07 + assert ok_08 is True # 该期锚点通过(规则级 matched=2/3 一并回报) + assert agg == {"matched": 2, "anchors": 3} + assert ok_09 is False and "无锚点记录" in detail_09 + + +def test_main_remark_idempotent_and_bounded(): + seg = ("2026-09-14 voucher_details 复算一致(自动归集id=2128, Δ=0.0000, rule=1 v6eac76a3)" + ";主值批次2026-08-24 15:50:01早于源同步2026-09-11 16:35:14(时点差,不改数)") + once = _with_recheck_remark(None, seg) + twice = _with_recheck_remark(once, seg) + assert once == twice, "同一自动行重跑不得重复追加(否则撑爆 varchar(500))" + long_old = "旧正文" * 300 # 远超 500 + got = _with_recheck_remark(long_old, seg) + assert len(got) <= 500 and got.endswith(seg) + # 两条不同自动行的留痕都要在(不同 id → 不互相覆盖) + seg2 = seg.replace("id=2128", "id=2129") + both = _with_recheck_remark(once, seg2) + assert "id=2128" in both and "id=2129" in both and len(both) <= 500