#!/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())