cma-management/backend/scripts/verify_opening_balance_live_20260916.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""建账期初层（M1/M2）现网只读验收 —— 2026-09-16

判据（任务书 `finance-20260916-opening-balance-layer.md` 验收标准）：
  ① 13 科目期末余额 = 期初 + 累计，与源系统《科目余额表》2026.08 / 2026.07 **逐科目 0 差异**（容差 ≤0.01）
  ② 恒等式：资产 = 负债 + 权益，勾稽差额 0.00（含 1901 按「其他流动资产」列报）
  ③ M7：权益合计 10,932,589.50（含实收资本 3001 = 2,000,000.00）
  ④ 期初补齐后 R1~R12 触发数 = 0（R2/R4 伪触发消失）
  ⑤ M2 两态：期初未补 → direction_check_unreliable=True；期初注入 → 自动 False

**零写入**：全程只读 SELECT；期初行不落库（用 monkeypatch 注入内存期初，不建表、不 INSERT、不 commit）。
用法：cd /root/cma-management/backend && venv/bin/python3 scripts/verify_opening_balance_live_20260916.py [period]
"""
from __future__ import annotations

import csv
import io
import json
import os
import sys
from decimal import Decimal

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 VoucherDetail  # noqa: E402
from app.services import report_reclass as rr  # noqa: E402
from app.utils.voucher_guard import apply_voucher_guard  # noqa: E402

PERIOD = sys.argv[1] if len(sys.argv) > 1 else "2026-08"
PREV = "2026-07"
ENTITY = 1
SRC = "/root/WB_WS/_data/zxt-hanke/%s/kmyeb.csv"
OPENING_CSV = "/root/.hermes/profiles/wecom-finance/reports/CMA_酣客建账期初余额_反解清单_v1.csv"
OUT_JSON = "/root/.hermes/profiles/wecom-fullstack/pending-tasks/_reports/finance-20260916-opening-balance-verify.json"
TOL = Decimal("0.01")

# 任务书给出的 13 科目期末余额（借−贷，负=贷方）
EXPECT = {
    "1001": Decimal("6331.30"), "1002": Decimal("46729.79"), "1122": Decimal("1449489.54"),
    "1123": Decimal("2066724.00"), "1405": Decimal("5462698.72"), "1601": Decimal("9333477.39"),
    "1602": Decimal("-516957.04"), "1801": Decimal("0.00"), "2203": Decimal("-836505.86"),
    "2221": Decimal("-4295885.13"), "3001": Decimal("-2000000.00"), "3103": Decimal("-14918101.66"),
    "3104": Decimal("5985512.16"),
}
SRC_TOTALS = {"assets": Decimal("18203980.49"), "liabilities": Decimal("7271390.99"),
              "equity": Decimal("10932589.50")}

FINDINGS = []


def chk(status: str, name: str, detail: str):
    FINDINGS.append({"status": status, "judge": name, "detail": detail})
    print(f"[{status}] {name} — {detail}")


def load_opening():
    rows = []
    with io.open(OPENING_CSV, encoding="utf-8-sig") as f:
        for r in csv.DictReader(f):
            rows.append({
                "subject_code": (r["subject_code"] or "").strip(),
                "subject_name": (r["subject_name"] or "").strip(),
                "balance_side": (r["balance_side"] or "").strip().lower(),
                "amount": Decimal(str(r["opening_amount"]).strip()),
                "effective_date": "2021-03",
                "status": "derived", "source": "reverse-calc-20260916",
            })
    return rows


OPENING = load_opening()


def src_rows(period: str):
    """源系统《科目余额表》4 位科目行 → {code: (name, 期末借−期末贷)}"""
    path = SRC % period.replace("-", ".")
    out = {}
    if not os.path.exists(path):
        return out
    with io.open(path, encoding="utf-8-sig") as f:
        for r in csv.DictReader(f):
            code = (r["科目编码"] or "").strip()
            if len(code) != 4:
                continue
            d = Decimal(str(r["期末借"] or "0"))
            c = Decimal(str(r["期末贷"] or "0"))
            out[code] = ((r["科目名称"] or "").strip(), d - c)
    return out


def src_totals(rows):
    t = {"assets": Decimal(0), "liabilities": Decimal(0), "equity": Decimal(0)}
    for code, (_n, net) in rows.items():
        if code.startswith("1"):
            t["assets"] += net
        elif code.startswith("2"):
            t["liabilities"] += -net
        elif code.startswith("3"):
            t["equity"] += -net
    return t


def cma_cum(db, code: str, period: str) -> Decimal:
    """CMA voucher_details 累计（借−贷，作废守卫生效）"""
    q = db.query(func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount)).filter(
        VoucherDetail.entity_id == ENTITY,
        VoucherDetail.subject_code.like(f"{code}%"),
        VoucherDetail.period <= period,
    )
    v = apply_voucher_guard(db, q, VoucherDetail).scalar()
    return Decimal(str(v or 0))


def opening_signed(code: str) -> Decimal:
    tot = Decimal(0)
    for r in OPENING:
        if r["subject_code"].startswith(code):
            tot += r["amount"] if r["balance_side"] == "debit" else -r["amount"]
    return tot


def main():
    db = get_session_local()()
    evidence = {"period": PERIOD, "prev_period": PREV, "entity_id": ENTITY, "mode": "read-only",
                "opening_rows": len(OPENING), "subjects": [], "checks": []}
    try:
        # ── ① 13 科目：期初 + 累计 vs 源系统期末（2026.08 / 2026.07 双期） ──
        for period in (PERIOD, PREV):
            sr = src_rows(period)
            bad = 0
            print(f"\n=== [①] {period} 13 科目 期初 + 累计  vs  源系统《科目余额表》期末（借−贷）===")
            print(f"{'科目':<6}{'名称':<12}{'期初':>16}{'CMA累计':>16}{'期初+累计':>16}"
                  f"{'源系统期末':>16}{'差异':>12}  判定")
            for code in list(EXPECT) if period == PERIOD else list(EXPECT):
                name, snet = sr.get(code, ("", None))
                cum = cma_cum(db, code, period)
                op = opening_signed(code)
                total = cum + op
                diff = (total - snet) if snet is not None else None
                ok = diff is not None and abs(diff) <= TOL
                if period == PERIOD:
                    evidence["subjects"].append({
                        "subject_code": code, "subject_name": name, "period": period,
                        "opening": float(op), "cma_cumulative": float(cum),
                        "opening_plus_cumulative": float(total),
                        "source_end_balance": float(snet) if snet is not None else None,
                        "diff": float(diff) if diff is not None else None, "pass": bool(ok),
                    })
                if not ok:
                    bad += 1
                print(f"{code:<6}{name:<12}{float(op):>16,.2f}{float(cum):>16,.2f}{float(total):>16,.2f}"
                      f"{(float(snet) if snet is not None else float('nan')):>16,.2f}"
                      f"{(float(diff) if diff is not None else float('nan')):>12,.2f}  {'OK' if ok else 'FAIL'}")
            chk("PASS" if bad == 0 else "FAIL",
                f"{period} 13 科目逐科目 0 差异",
                f"不一致 {bad} 项（容差 {TOL}）")

        # ── 任务书硬判据：13 科目期末余额逐科目对表 ──
        bad_exp = 0
        for code, want in EXPECT.items():
            cum = cma_cum(db, code, PERIOD) + opening_signed(code)
            if abs(cum - want) > TOL:
                bad_exp += 1
                print(f"  [差异] {code}: {cum} != {want}")
        chk("PASS" if bad_exp == 0 else "FAIL", "任务书 13 科目期末余额硬判据",
            f"不一致 {bad_exp} 项")

        # ── 源系统合计（4 位科目自算，独立于财务给出值） ──
        st = src_totals(src_rows(PERIOD))
        ok_src = all(abs(st[k] - SRC_TOTALS[k]) <= TOL for k in SRC_TOTALS)
        chk("PASS" if ok_src else "FAIL", "源系统合计复算可比",
            f"资产 {st['assets']:,.2f} / 负债 {st['liabilities']:,.2f} / 权益 {st['equity']:,.2f}")
        evidence["source_totals_selfcalc"] = {k: float(v) for k, v in st.items()}

        # ── ②③④ 报表层：期初注入（内存，不落库）后跑引擎 ──
        from app.api.reports import BALANCE_SHEET_SECTIONS

        def analyze_with(opening_rows):
            orig = rr.load_opening_rows
            rr.load_opening_rows = lambda db_, entity_id: list(opening_rows)
            try:
                return rr.analyze(db, ENTITY, PERIOD, basis="end",
                                  template_sections=BALANCE_SHEET_SECTIONS)
            finally:
                rr.load_opening_rows = orig

        p_no = analyze_with([])                 # 态①：期初未补（现网实况）
        p_yes = analyze_with(OPENING)           # 态②：期初补齐（F1 反解值）
        evidence["state_no_opening"] = {
            "direction_check_unreliable": p_no["direction_check_unreliable"],
            "entries_count": p_no["invariants"]["entries_count"],
            "entries": [(e["rule"], e["reclass_amount"], e["source_code"]) for e in p_no["entries"]],
            "totals_after": p_no["totals_after"],
        }
        evidence["state_with_opening"] = {
            "direction_check_unreliable": p_yes["direction_check_unreliable"],
            "opening": p_yes["opening"],
            "entries_count": p_yes["invariants"]["entries_count"],
            "entries": [(e["rule"], e["reclass_amount"], e["source_code"]) for e in p_yes["entries"]],
            "totals_after": p_yes["totals_after"],
            "invariants": {k: p_yes["invariants"][k] for k in (
                "assets_after", "liabilities_after", "equity_after", "residual_after",
                "identity_holds_after", "opening_imported", "opening_balanced")},
        }

        print(f"\n=== [②③④] 报表层（basis=end, {PERIOD}）===")
        print(f"  期初未补：可信度不可信={p_no['direction_check_unreliable']} 触发={p_no['invariants']['entries_count']} "
              f"资产={p_no['totals_after']['assets']:,.2f} 负债={p_no['totals_after']['liabilities']:,.2f} "
              f"权益={p_no['totals_after']['equity']:,.2f} 差额={p_no['totals_after']['residual']:,.2f}")
        ta = p_yes["totals_after"]
        print(f"  期初补齐：可信度不可信={p_yes['direction_check_unreliable']} 触发={p_yes['invariants']['entries_count']} "
              f"资产={ta['assets']:,.2f} 负债={ta['liabilities']:,.2f} 权益={ta['equity']:,.2f} "
              f"差额={ta['residual']:,.2f}")
        print(f"  期初层：{p_yes['opening']}")

        ok_tot = (abs(Decimal(str(ta["assets"])) - SRC_TOTALS["assets"]) <= TOL
                  and abs(Decimal(str(ta["liabilities"])) - SRC_TOTALS["liabilities"]) <= TOL
                  and abs(Decimal(str(ta["equity"])) - SRC_TOTALS["equity"]) <= TOL)
        chk("PASS" if ok_tot else "FAIL", "② 三合计与源系统一致",
            f"资产 {ta['assets']:,.2f}(源 18,203,980.49) / 负债 {ta['liabilities']:,.2f}(源 7,271,390.99) "
            f"/ 权益 {ta['equity']:,.2f}(源 10,932,589.50)")
        chk("PASS" if abs(Decimal(str(ta["residual"]))) <= TOL else "FAIL", "② 勾稽差额 0.00",
            f"residual_after={ta['residual']}")
        chk("PASS" if abs(Decimal(str(ta["equity"])) - SRC_TOTALS["equity"]) <= TOL else "FAIL",
            "③ M7 权益合计（含 3001 实收资本 2,000,000）",
            f"权益合计 {ta['equity']:,.2f}（实收资本行 raw={p_yes['raw'].get('4001')}）")
        chk("PASS" if p_yes["invariants"]["entries_count"] == 0 else "FAIL", "④ R1~R12 触发数 = 0（R2/R4 伪触发消失）",
            f"期初补齐后触发 {p_yes['invariants']['entries_count']} 条 / 期初未补时 {p_no['invariants']['entries_count']} 条（诊断）")
        chk("PASS" if (p_no["direction_check_unreliable"] and not p_yes["direction_check_unreliable"]) else "FAIL",
            "⑤ M2 两态（期初未补=不可信 → 补齐=可信）",
            f"态①{ p_no['direction_check_unreliable'] } → 态②{ p_yes['direction_check_unreliable'] }")

        # ── 零持久化自证 ──
        try:
            from app.models import ReportOpeningBalance
            n_rows = db.query(ReportOpeningBalance).count()
            has_table = True
        except Exception as exc:  # noqa: BLE001
            n_rows, has_table = None, False
            print(f"  [INFO] report_opening_balances 未建（{type(exc).__name__}）")
        chk("PASS", "零持久化（期初未写入现网库）",
            f"本次仅内存注入；表存在={has_table} 行数={n_rows}（未落库）")

        evidence["checks"] = FINDINGS
        evidence["pass"] = all(f["status"] == "PASS" for f in FINDINGS)
        os.makedirs(os.path.dirname(OUT_JSON), exist_ok=True)
        with io.open(OUT_JSON, "w", encoding="utf-8") as f:
            json.dump(evidence, f, ensure_ascii=False, indent=2)
        print(f"\n证据 JSON: {OUT_JSON} ({os.path.getsize(OUT_JSON)} B)")
        n_fail = sum(1 for f in FINDINGS if f["status"] == "FAIL")
        print(f"总判定：PASS {len(FINDINGS) - n_fail} / FAIL {n_fail}")
        return 1 if n_fail else 0
    finally:
        db.close()


if __name__ == "__main__":
    sys.exit(main())