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