Hermes CI Fix
e5eb092d20
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).
227 lines
11 KiB
Python
227 lines
11 KiB
Python
#!/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())
|