feat(kpi): auto_collect review gate (finance ruling 2026-09-14)

R2 rule-gate auto-promotion + R3 exception interception in the same collect transaction; merge duplicate rows, keep single verified main per (kpi_id, period); reuse kpi_bep_deriver._is_period_closed; no schema change (data_status/remark only).

Task: finance-20260914-autocollect-reviewflow-impl

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=<id> v<ver>`）。"""
+    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()

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）─────────────────────────

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