Hermes CI Fix
ae6f71dbc0
- M1 时点报表口径：余额 = 期初(report_opening_balances) + 累计发生额；期初只作用于取数层
（不生成凭证/不写 voucher_details/不改 kpi_values）；basis=period 不套期初；受 effective_date 约束
- M2 期初未补 → direction_check_unreliable=true：报表/底稿带"期初未补，方向判定不可信"水印，
重分类触发只进底稿诊断（diagnostic_only），不落地对外行项目
- M7 权益行科目集 = {3001,3002,3101,3103,3104,3105}（含 4001/4002/4101/4103/4104 别名）：补实收资本 3001
- 1901 待处理财产损溢过渡列报「其他流动资产」以保恒等式（M6 归因文案）
- 只读验收脚本 + 幂等导入器（dry-run 默认，--commit 门控）
263 lines
13 KiB
Python
263 lines
13 KiB
Python
#!/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())
|