feat(cma): treasury stage-1 MVP - migration, models, endpoints, rules, 2 pages

G1 review fixes P0-1/P0-2/P1-3/P1-4/P1-5/P2-6/P2-7/P2-8 applied:
- up() filters comment lines then asserts missing_tables == [] (CREATE hits 6/6)
- created counter = new-object diff via information_schema (idempotent second run = 0)
- down() drops columns per-column, dumps via write-path creds + fail-closed
- seeds expanded to 8 rules + 6 limits; data_status DEFAULT estimated
- no physical FK (app-layer checks), amount FLOAT wan-yuan, round(x,4)

Scope: 6 treasury_* tables + cash_plans 4 added columns; cash.py untouched.
Not pushed: deploy is approval-gated (de851257).
This commit is contained in:
Hermes CI Fix
2026-09-15 20:02:56 +08:00
parent fdf805ab8f
commit e5eb092d20
12 changed files with 3472 additions and 4 deletions
+226
View File
@@ -0,0 +1,226 @@
#!/usr/bin/env python3
"""CMA 司库 · 阶段一 —— 账户池初始化(F1 两份 CSV 导入)
数据源(财务侧前置交付 F1,md5 a49ae4340c05dfa20aba7daad25cc25f 的 F1F2F3 交付件附表):
/root/.hermes/profiles/wecom-finance/reports/CMA_司库P0_F1_账户主数据清单_entity1.csv (12 行,酣客)
/root/.hermes/profiles/wecom-finance/reports/CMA_司库P0_F1_账户主数据清单_entity2.csv (12 行,博海)
表头 14 列与 PRD §3.2 treasury_accounts 逐列一致(tier_basis 不在 CSV → 走 DDL 默认 'nature')。
口径(任务单步骤 6 + G1 §5 + F1 执行建议):
* e1(entity_id=1 酣客)12 个货币资金科目;e2(entity_id=2 博海)12 个账户
* `100107 现金备用卡(唐)` → status=inactive
* `owner_node_id` 首版留空(部门写在 remark;e2 无 org_nodes,禁造虚层级)
* **排除** `1002007 榆林酣客工行存款`(属酣客系,不得入 e2;也不并入 e1——科目码体系不同)
* 期初:e2 用 F1/F2 实测值(treasury_balances,data_status='verified');
e1 分账户 open_balance 留 NULL + 头寸 estimated(F2 分账户未交付,LQ-001 skip_when=estimated)
* 金额纪律:写库前 round(x,4)
用法:
python3 treasury_seed_accounts.py --dry-run # 解析 CSV + 打印将写入的行(零 DB)
python3 treasury_seed_accounts.py --apply # 幂等 upsert(按 (entity_id, account_no) / (account_id, period))
python3 treasury_seed_accounts.py --verify # 只读回查:账户数/排除项/双实体交集
"""
import argparse
import csv
import hashlib
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import text # noqa: E402
from app.database import get_engine # noqa: E402
CSV_DIR = "/root/.hermes/profiles/wecom-finance/reports"
CSV_FILES = {
1: "CMA_司库P0_F1_账户主数据清单_entity1.csv",
2: "CMA_司库P0_F1_账户主数据清单_entity2.csv",
}
EXCLUDED_ACCOUNT_NO = "1002007" # 榆林酣客工行存款(酣客系,不得入 e2)
INACTIVE_ACCOUNT_NOS = {"100107"} # 现金备用卡(唐):首版停用
ACCOUNT_COLS = ["entity_id", "account_no", "account_name", "account_type", "bank_name",
"subject_code", "currency", "tier", "owner_node_id", "open_balance",
"open_period", "is_pool_member", "status", "remark"]
def _load_csv(entity_id: int):
path = os.path.join(CSV_DIR, CSV_FILES[entity_id])
with open(path, encoding="utf-8-sig", newline="") as f:
rows = list(csv.DictReader(f))
return path, rows
def _norm(row: dict, entity_id: int) -> dict:
def s(k):
v = (row.get(k) or "").strip()
return v or None
out = {
"entity_id": entity_id,
"account_no": s("account_no"),
"account_name": s("account_name"),
"account_type": s("account_type") or "other",
"bank_name": s("bank_name"),
"subject_code": s("subject_code"),
"currency": s("currency") or "CNY",
"tier": s("tier") or "grassroots",
"owner_node_id": None, # 首版留空(部门写在 remark)
"open_balance": None,
"open_period": s("open_period"),
"is_pool_member": int(s("is_pool_member") or 0),
"status": s("status") or "active",
"remark": s("remark"),
}
if entity_id == 1:
out["open_balance"] = None # e1 分账户期初留 NULL(F2 分账户未交付)
else:
raw = s("open_balance")
out["open_balance"] = None if raw is None else round(float(raw), 4)
if out["account_no"] in INACTIVE_ACCOUNT_NOS:
out["status"] = "inactive"
return out
def plan():
"""解析两份 CSV → {entity_id: [rows]},并做清单级断言。"""
data, problems = {}, []
for eid in (1, 2):
path, rows = _load_csv(eid)
norms = [_norm(r, eid) for r in rows]
nos = [r["account_no"] for r in norms]
if EXCLUDED_ACCOUNT_NO in nos:
problems.append(f"entity{eid} CSV 含被排除账户 {EXCLUDED_ACCOUNT_NO}")
if len(set(nos)) != len(nos):
problems.append(f"entity{eid} CSV 账号重复")
nos_set = set(nos)
data[eid] = {"csv": path, "rows": norms, "account_nos": sorted(nos_set)}
# 双实体账号集合必须零交集(G1 §5)
inter = set(data[1]["account_nos"]) & set(data[2]["account_nos"])
if inter:
problems.append(f"双实体账号集合交集非空: {sorted(inter)}")
if len(data[1]["rows"]) != 12 or len(data[2]["rows"]) != 12:
problems.append(f"行数异常: e1={len(data[1]['rows'])} e2={len(data[2]['rows'])}(期望 12/12)")
return data, problems
def apply(data):
engine = get_engine()
summary = {"accounts": {}, "balances": {}, "batches": {}}
with engine.begin() as conn:
for eid, payload in data.items():
created = updated = 0
for row in payload["rows"]:
cols = ", ".join(ACCOUNT_COLS)
ph = ", ".join(":" + c for c in ACCOUNT_COLS)
upd = ", ".join(f"{c}=VALUES({c})" for c in ACCOUNT_COLS
if c not in ("entity_id", "account_no"))
found = conn.execute(text(
"SELECT id FROM treasury_accounts WHERE entity_id=:e AND account_no=:a"
), {"e": eid, "a": row["account_no"]}).fetchone()
params = dict(row)
conn.execute(text(
f"INSERT INTO treasury_accounts ({cols}) VALUES ({ph}) "
f"ON DUPLICATE KEY UPDATE {upd}"), params)
created += found is None
updated += found is not None
# e2 期初头寸(F1/F2 实测值 → verified);e1 无分账户期初 → 不写头寸(estimated)
bal_created = 0
if eid == 2:
batch = "f1-f2-e2-" + hashlib.md5(
json.dumps([(r["account_no"], r["open_balance"], r["open_period"])
for r in payload["rows"]], ensure_ascii=False).encode()).hexdigest()[:10]
for row in payload["rows"]:
if row["open_balance"] is None:
continue
acc = conn.execute(text(
"SELECT id FROM treasury_accounts WHERE entity_id=:e AND account_no=:a"
), {"e": eid, "a": row["account_no"]}).fetchone()
if not acc:
continue
exists = conn.execute(text(
"SELECT id FROM treasury_balances WHERE entity_id=:e AND account_id=:i AND period=:p"
), {"e": eid, "i": acc[0], "p": row["open_period"]}).fetchone()
conn.execute(text(
"INSERT INTO treasury_balances (entity_id, account_id, period, balance, "
"available, frozen, source, data_status, batch) VALUES (:e,:i,:p,:b,NULL,0,"
"'import:F1-F2账户主数据清单','verified',:bt) "
"ON DUPLICATE KEY UPDATE balance=VALUES(balance), source=VALUES(source), "
"data_status=VALUES(data_status), batch=VALUES(batch)"),
{"e": eid, "i": acc[0], "p": row["open_period"],
"b": row["open_balance"], "bt": batch})
bal_created += exists is None
summary["batches"][f"entity{eid}"] = batch
summary["accounts"][f"entity{eid}"] = {"rows_in": len(payload["rows"]),
"created": created, "updated": updated}
summary["balances"][f"entity{eid}"] = {"created": bal_created}
return summary
def verify():
"""只读回查:账户数 / 排除项 / 双实体零交集 / 100107 inactive。"""
engine = get_engine()
with engine.connect() as conn:
out = {}
rows = conn.execute(text(
"SELECT entity_id, COUNT(*) FROM treasury_accounts GROUP BY entity_id ORDER BY entity_id"
)).fetchall()
out["accounts_by_entity"] = {str(r[0]): int(r[1]) for r in rows}
e1 = {r[0] for r in conn.execute(text(
"SELECT account_no FROM treasury_accounts WHERE entity_id=1")).fetchall()}
e2 = {r[0] for r in conn.execute(text(
"SELECT account_no FROM treasury_accounts WHERE entity_id=2")).fetchall()}
out["entity1_accounts"] = sorted(e1)
out["entity2_accounts"] = sorted(e2)
out["intersection"] = sorted(e1 & e2)
out["excluded_present"] = sorted({EXCLUDED_ACCOUNT_NO} & (e1 | e2))
out["inactive_100107_status"] = conn.execute(text(
"SELECT status FROM treasury_accounts WHERE entity_id=1 AND account_no='100107'"
)).scalar()
out["open_balance_null_e1"] = conn.execute(text(
"SELECT COUNT(*) FROM treasury_accounts WHERE entity_id=1 AND open_balance IS NULL"
)).scalar()
out["balances_e2"] = conn.execute(text(
"SELECT COUNT(*) FROM treasury_balances WHERE entity_id=2")).scalar()
out["balances_e1"] = conn.execute(text(
"SELECT COUNT(*) FROM treasury_balances WHERE entity_id=1")).scalar()
out["balances_e2_bad_status"] = conn.execute(text(
"SELECT COUNT(*) FROM treasury_balances WHERE entity_id=2 AND data_status<>'verified'"
)).scalar()
out["owner_node_id_nonnull"] = conn.execute(text(
"SELECT COUNT(*) FROM treasury_accounts WHERE owner_node_id IS NOT NULL")).scalar()
print(json.dumps(out, ensure_ascii=False, indent=2))
return out
def main():
ap = argparse.ArgumentParser()
g = ap.add_mutually_exclusive_group()
g.add_argument("--dry-run", action="store_true")
g.add_argument("--apply", action="store_true")
g.add_argument("--verify", action="store_true", help="只读回查(默认)")
args = ap.parse_args()
if args.verify or not (args.dry_run or args.apply):
verify()
return 0
data, problems = plan()
plan_out = {f"entity{e}": {"csv": data[e]["csv"], "rows": len(data[e]["rows"]),
"account_nos": data[e]["account_nos"]} for e in (1, 2)}
if problems:
print(json.dumps({"problems": problems, "plan": plan_out}, ensure_ascii=False, indent=2))
return 3
if args.dry_run:
print(json.dumps({"plan": plan_out,
"sample_entity1": data[1]["rows"][:2],
"sample_entity2": data[2]["rows"][:2]}, ensure_ascii=False, indent=2))
return 0
summary = apply(data)
print(json.dumps({"applied": summary, "plan": plan_out}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())