"""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 个百分点（即 R7 阈值，任总 2026-09-14 18:17 确认 · 正式规则）。
  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")

# R7 阈值（金额 max(1000元, 0.5%×|主值|)、比率 0.5 个百分点）为任总 2026-09-14 18:17 确认的**正式规则**
# （原文为建议值，数值未变）。仅标签/注释类改动不得改 RULE_VERSION——改版本号会误触 R2-⑥
# 「规则变更后首次归集」例外，把本可 T+0 转正的行拦成 pending。
RULE_VERSION = "20260914"
AMOUNT_MIN_YUAN = 1000.0        # R7 金额类下限（元）——任总 2026-09-14 18:17 确认 · 正式规则
AMOUNT_PCT = 0.005              # R7 金额类比例（0.5%×|主值|）——任总 2026-09-14 18:17 确认 · 正式规则
RATIO_PP = 0.005                # R7 比率类（0.5 个百分点；比率值以小数存，13% 存 0.13）——任总 2026-09-14 18:17 确认 · 正式规则
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()