Hermes CI Fix
df2be301b3
- spec read from _cma_write_table_whitelist.json table_ops (single source of truth) - revoke non-whitelist table privs before grant (idempotent; 1141 handled as no-op) - add zero-DB --grant-plan mode; post-apply convergence assertion via information_schema - task: yanxue-20260918-treasury-grant-idempotent-fix
647 lines
29 KiB
Python
647 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
"""CMA 司库（资金）阶段一 —— 迁移脚本（阶段 1 落地版）
|
||
|
||
落地路径：/root/cma-management/backend/scripts/treasury_migrate.py
|
||
评审稿 ：wecom-fullstack/pending-tasks/_reports/treasury-stage1-migration-draft-20260915.py（md5 ceaa3b3bf563b8d664ddd7224f01a5bf，归档不回改）
|
||
G1 修正项 ：P0-1 up() 语句过滤器 / P0-2 计数口径 / P1-3 down() 部分态 / P1-4 down() dump 凭据 /
|
||
P1-5 种子展开 / P2-6 data_status fail-safe（DDL 侧）/ P2-7 import_logs 指纹行 / P2-8 金额纪律
|
||
|
||
用法:
|
||
python3 treasury_migrate.py --parse-only # 零 DB：只解析 DDL，打印命中语句与期望计数（P0-1 复算证据）
|
||
python3 treasury_migrate.py --grant-plan # 零 DB：打印授权收敛语句计划（授权面幂等自证证据）
|
||
python3 treasury_migrate.py --check # 只读预检：缺表/缺列/缺索引/种子行数（零写入，默认动作）
|
||
python3 treasury_migrate.py --apply # 建表 + 加列 + 索引 + 种子 + 授权收敛（幂等）
|
||
python3 treasury_migrate.py --down # 回滚（先 dump 后删；有数据时要求 --yes）
|
||
|
||
幂等契约（P0-2 计数口径 = 新增对象数，不是执行语句数）:
|
||
首跑 --apply 期望：created_new=6 / added_cols=4 / added_indexes=1 / rules=8 行 / limits=6 行
|
||
二跑 --apply 期望：created_new=0 / added_cols=0 / added_indexes=0 / 种子行数不变
|
||
指纹行：import_logs(batch=treasury-ddl-<sha1[:10]>)，按 (import_type,batch) 先查后写，
|
||
不依赖表上是否有唯一键（P2-7：entity_id=1 为约定——迁移指纹属实体无关元数据）
|
||
|
||
授权面契约（2026-09-18 F-1 根因侧：部署管道不再扩面，而是「按白名单最小面 + 收敛」）:
|
||
旧形态（本文件 2026-09-18 前）= 每次 --apply 无条件 `GRANT SELECT,INSERT,UPDATE,DELETE`
|
||
于 6 表 × 2 host = 12 条 → 白名单外的权限（表级 SELECT / balances·payments 的 DELETE /
|
||
hits 的 UPDATE,DELETE）每次部署回涨，基线巡检恒 BASELINE_MISMATCH。
|
||
新形态 = 授权集合取自白名单文件 table_ops（授权面唯一事实源），并对白名单外权限发 REVOKE：
|
||
· 语句计划 grant_plan() 是**纯函数**（只依赖 spec，不读现网）⇒ 连跑两次逐字相同；
|
||
· 重复执行结果一致（幂等）：白名单内 GRANT 幂等，白名单外 REVOKE 在收敛态报 1141
|
||
（no such grant）→ 视为幂等空操作记账，不报错；
|
||
· 执行后回读 information_schema.TABLE_PRIVILEGES 断言逐表 == 白名单，不符则非零退出。
|
||
白名单文件路径：env `CMA_WRITE_WHITELIST` 优先，默认 /root/WB_WS/_reports/_cma_write_table_whitelist.json；
|
||
文件不可用时退回 TREASURY_GRANT_SPEC_FALLBACK（须与白名单一致，否则 stderr 告警）；文件在位但
|
||
不合规（缺表裁定 / 不可解析 / 空集 / 含表级 SELECT）→ fail-closed，不猜不降级。
|
||
|
||
边界：只动 6 张 treasury_* 新表 + cash_plans 加列/加索引 + cma_api 授权收敛；
|
||
不碰 cash.py / 既有端点契约 / 其他模块表结构 / 司库业务读写路径。
|
||
"""
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from datetime import datetime
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from sqlalchemy import create_engine, text # noqa: E402
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
BACKUP_DIR = os.path.join(HERE, "backups")
|
||
DDL_FILE = os.path.join(HERE, "treasury_ddl.sql")
|
||
|
||
NEW_TABLES = [
|
||
"treasury_accounts",
|
||
"treasury_balances",
|
||
"treasury_control_rules",
|
||
"treasury_payments",
|
||
"treasury_control_hits",
|
||
"treasury_limits",
|
||
]
|
||
CASH_PLAN_NEW_COLS = ["account_id", "tier", "pool_id", "is_locked"]
|
||
CASH_PLAN_NEW_INDEX = "ix_cashplans_entity_acct"
|
||
EXPECTED_RULE_ROWS = 8 # e1 4 + e2 4
|
||
EXPECTED_LIMIT_ROWS = 6 # e1 3 + e2 3
|
||
CASH_PLAN_ALTER = """ALTER TABLE cash_plans
|
||
ADD COLUMN account_id INT NULL COMMENT '资金来源/去向账户 treasury_accounts.id（软链，无物理外键）',
|
||
ADD COLUMN tier VARCHAR(20) NULL COMMENT 'hq/platform/grassroots（三位一体视角；可由 org_nodes.level 派生）',
|
||
ADD COLUMN pool_id INT NULL COMMENT '归属资金池（阶段三启用，本阶段恒 NULL）',
|
||
ADD COLUMN is_locked TINYINT DEFAULT 0 COMMENT '是否已被预算/内控锁定'"""
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 特权迁移连接（2026-09-16 并入）：与应用运行账号 cma_api 剥离，走有 DDL 权限的通道
|
||
# ─────────────────────────────────────────────────────────────
|
||
DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock"
|
||
FORBIDDEN_RUN_ACCOUNTS = {"cma_api"}
|
||
GRANT_APP_ACCOUNT = "cma_api"
|
||
GRANT_HOSTS = ("localhost", "127.0.0.1")
|
||
# 授权面域：本迁移步只可能授出这 4 种（旧形态的全集）；收敛 = 该域内白名单外的项一律 REVOKE
|
||
GRANT_PRIV_DOMAIN = ("SELECT", "INSERT", "UPDATE", "DELETE")
|
||
# 白名单文件 = 授权面唯一事实源（后续白名单变更只改该文件一处）
|
||
WHITELIST_PATH = os.getenv(
|
||
"CMA_WRITE_WHITELIST", "/root/WB_WS/_reports/_cma_write_table_whitelist.json")
|
||
# 兜底常量：与白名单 table_ops 的 6 表裁定逐表一致（文件不可用时的退路；不一致时 stderr 告警）
|
||
TREASURY_GRANT_SPEC_FALLBACK = {
|
||
"treasury_accounts": ("INSERT", "UPDATE", "DELETE"),
|
||
"treasury_control_rules": ("INSERT", "UPDATE", "DELETE"),
|
||
"treasury_limits": ("INSERT", "UPDATE", "DELETE"),
|
||
"treasury_balances": ("INSERT", "UPDATE"),
|
||
"treasury_payments": ("INSERT", "UPDATE"),
|
||
"treasury_control_hits": ("INSERT",),
|
||
}
|
||
|
||
|
||
def get_engine():
|
||
"""特权引擎：CMA_MIGRATE_DB_USER/PASS（TCP）优先，否则本机 root unix socket（无口令）。
|
||
绝不使用应用运行账号（无 DDL）——与 db_migrate.py 同口径。"""
|
||
from app import database as db
|
||
db_name = (os.getenv("CMA_MIGRATE_DB_NAME") or getattr(db, "DB_NAME", None) or "cma")
|
||
user = os.getenv("CMA_MIGRATE_DB_USER", "").strip()
|
||
if user:
|
||
if user in FORBIDDEN_RUN_ACCOUNTS:
|
||
raise SystemExit(
|
||
"[FAIL] CMA_MIGRATE_DB_USER=cma_api 是应用运行账号（无 DDL 权限），已拒绝执行")
|
||
pw = os.getenv("CMA_MIGRATE_DB_PASS", "")
|
||
host = os.getenv("CMA_MIGRATE_DB_HOST", "127.0.0.1")
|
||
port = int(os.getenv("CMA_MIGRATE_DB_PORT", "3306"))
|
||
return create_engine(
|
||
f"mysql+pymysql://{user}:{pw}@{host}:{port}/{db_name}?charset=utf8mb4",
|
||
pool_pre_ping=True)
|
||
sock = os.getenv("CMA_MIGRATE_DB_SOCKET", DEFAULT_SOCKET)
|
||
if not os.path.exists(sock):
|
||
raise SystemExit(
|
||
f"[FAIL] 未设 CMA_MIGRATE_DB_USER 且 unix socket 不存在：{sock}")
|
||
return create_engine(
|
||
f"mysql+pymysql://root@localhost/{db_name}?unix_socket={sock}&charset=utf8mb4",
|
||
pool_pre_ping=True)
|
||
|
||
|
||
def _resolve_dbn():
|
||
"""库名：env CMA_MIGRATE_DB_NAME 优先，否则应用配置（app.database.DB_NAME），最后 'cma'。"""
|
||
name = (os.getenv("CMA_MIGRATE_DB_NAME") or "").strip()
|
||
if name:
|
||
return name
|
||
try:
|
||
from app import database as db
|
||
return getattr(db, "DB_NAME", None) or "cma"
|
||
except Exception:
|
||
return "cma"
|
||
|
||
|
||
def _parse_priv_text(raw):
|
||
"""白名单 table_ops 值 → 权限集合。值形如 'INSERT,UPDATE,DELETE (中文说明（含嵌套括号）)'。"""
|
||
head = str(raw)
|
||
for sep in ("(", "（"):
|
||
i = head.find(sep)
|
||
if i >= 0:
|
||
head = head[:i]
|
||
return {p.strip().upper() for p in head.split(",") if p.strip()}
|
||
|
||
|
||
def load_grant_spec():
|
||
"""读白名单 table_ops 的 6 张 treasury 表裁定 → (spec, source, note)。
|
||
|
||
spec : {table: (priv, ...)}，权限按 GRANT_PRIV_DOMAIN 顺序归一（确定性）
|
||
source : 'whitelist' | 'fallback-constant'
|
||
note : 常量兜底与白名单不一致时的告警文本（否则 None）
|
||
fail-closed：文件在位但不可解析 / 缺表裁定 / 空集 / 含表级 SELECT → SystemExit，不猜不降级。
|
||
"""
|
||
fallback = {t: tuple(TREASURY_GRANT_SPEC_FALLBACK[t]) for t in NEW_TABLES}
|
||
if not os.path.exists(WHITELIST_PATH):
|
||
return fallback, "fallback-constant", None
|
||
try:
|
||
with open(WHITELIST_PATH, encoding="utf-8") as fh:
|
||
ops = (json.load(fh) or {}).get("table_ops") or {}
|
||
except Exception as exc:
|
||
raise SystemExit(f"[FAIL] 白名单不可解析 {WHITELIST_PATH}: {exc}")
|
||
spec = {}
|
||
for t in NEW_TABLES:
|
||
raw = ops.get(t)
|
||
if raw is None:
|
||
raise SystemExit(f"[FAIL] 白名单 table_ops 缺 {t} 裁定（{WHITELIST_PATH}）")
|
||
privs = tuple(p for p in GRANT_PRIV_DOMAIN if p in _parse_priv_text(raw))
|
||
if not privs:
|
||
raise SystemExit(f"[FAIL] 白名单 {t} 授权集为空（{raw!r}）")
|
||
if "SELECT" in privs:
|
||
raise SystemExit(
|
||
f"[FAIL] 白名单 {t} 含表级 SELECT —— 读面由库级 SELECT ON cma.* 覆盖，"
|
||
"表级 SELECT 正是本单要收敛的冗余；如需改基线口径请回报调度方另单")
|
||
spec[t] = privs
|
||
note = None
|
||
if spec != fallback:
|
||
note = (f"兜底常量与白名单不一致（以白名单为准）: fallback={fallback} whitelist={spec}")
|
||
return spec, "whitelist", note
|
||
|
||
|
||
def grant_plan(spec, dbn):
|
||
"""授权收敛语句计划（纯函数：只依赖 spec/dbn，不读现网）⇒ 连跑两次逐字相同。
|
||
|
||
每 (host, 表)：先 REVOKE 白名单外权限（部署成为「收敛」通道），再 GRANT 白名单集合。
|
||
收敛态下 REVOKE 会报 1141（no such grant）→ 执行侧记为幂等空操作，非错误。
|
||
"""
|
||
stmts = []
|
||
for host in GRANT_HOSTS:
|
||
for t in sorted(NEW_TABLES):
|
||
wanted = tuple(p for p in GRANT_PRIV_DOMAIN if p in spec[t])
|
||
extras = tuple(p for p in GRANT_PRIV_DOMAIN if p not in wanted)
|
||
if extras:
|
||
stmts.append(
|
||
f"REVOKE {', '.join(extras)} ON `{dbn}`.`{t}` FROM '{GRANT_APP_ACCOUNT}'@{host}")
|
||
stmts.append(
|
||
f"GRANT {', '.join(wanted)} ON `{dbn}`.`{t}` TO '{GRANT_APP_ACCOUNT}'@{host}")
|
||
return stmts
|
||
|
||
|
||
def _is_nonexistent_grant(exc) -> bool:
|
||
"""MySQL 1141/1147（no such grant）：REVOKE 未持有权限 —— 收敛态的正常返回。"""
|
||
orig = getattr(exc, "orig", exc)
|
||
args = getattr(orig, "args", ()) or ()
|
||
if args and args[0] in (1141, 1147):
|
||
return True
|
||
return "no such grant" in str(exc).lower()
|
||
|
||
|
||
_GRANT_STMT_RE = re.compile(
|
||
r"^(?P<verb>REVOKE|GRANT)\s+(?P<privs>.+?)\s+ON\s+`(?P<db>[^`]+)`\.`(?P<table>[^`]+)`"
|
||
r"\s+(?:FROM|TO)\s+'(?P<user>[^']+)'@(?P<host>\S+)$")
|
||
|
||
|
||
def _stmt_entry(stmt: str) -> dict:
|
||
"""授权语句 → {verb, privs, db, table, user, host}（解析失败即抛，不静默）。"""
|
||
m = _GRANT_STMT_RE.match(stmt.strip())
|
||
if not m:
|
||
raise RuntimeError(f"授权语句形态非预期，无法解析: {stmt!r}")
|
||
d = m.groupdict()
|
||
d["privs"] = ", ".join(p.strip() for p in d["privs"].split(","))
|
||
return d
|
||
|
||
|
||
def _current_table_privs(conn, dbn):
|
||
"""只读回查 information_schema.TABLE_PRIVILEGES：{(host, table): {priv}}（仅 cma_api）。"""
|
||
rows = conn.execute(text(
|
||
"SELECT GRANTEE, TABLE_NAME, PRIVILEGE_TYPE FROM information_schema.TABLE_PRIVILEGES "
|
||
"WHERE TABLE_SCHEMA = :d"), {"d": dbn}).fetchall()
|
||
out = {}
|
||
for grantee, table, priv in rows:
|
||
if table not in NEW_TABLES:
|
||
continue
|
||
user, _sep, host = str(grantee).rpartition("@")
|
||
if user.strip().strip("'") != GRANT_APP_ACCOUNT:
|
||
continue
|
||
out.setdefault((host.strip().strip("'"), table), set()).add(str(priv).upper())
|
||
return out
|
||
|
||
|
||
def verify_grant_sync(conn, spec, dbn):
|
||
"""收敛断言：逐 (host, 表) 现网表级权限集合 == 白名单集合；返回差异清单（空 = 已收敛）。"""
|
||
current = _current_table_privs(conn, dbn)
|
||
bad = []
|
||
for host in GRANT_HOSTS:
|
||
for t in sorted(NEW_TABLES):
|
||
have = current.get((host, t), set())
|
||
want = set(spec[t])
|
||
if have != want:
|
||
bad.append(f"{t}@{host}: 现持={sorted(have)} 期望={sorted(want)}")
|
||
return bad
|
||
|
||
|
||
def apply_grant_sync(conn, spec, source):
|
||
"""执行授权收敛（按白名单最小面 + 幂等）：REVOKE 白名单外 + GRANT 白名单，回读断言。"""
|
||
dbn = _resolve_dbn()
|
||
granted, revoked, absent = [], [], []
|
||
for stmt in grant_plan(spec, dbn):
|
||
try:
|
||
conn.execute(text(stmt))
|
||
except Exception as exc:
|
||
if stmt.startswith("REVOKE") and _is_nonexistent_grant(exc):
|
||
absent.append(_stmt_entry(stmt))
|
||
continue
|
||
raise
|
||
(granted if stmt.startswith("GRANT") else revoked).append(_stmt_entry(stmt))
|
||
conn.commit()
|
||
mismatch = verify_grant_sync(conn, spec, dbn)
|
||
return {
|
||
"spec_source": source,
|
||
"spec": {t: ",".join(spec[t]) for t in sorted(spec)},
|
||
"granted": [f"{e['table']}@{e['host']}" for e in granted],
|
||
"revoked": [f"{e['table']}@{e['host']} ({e['privs']}) [REVOKE]" for e in revoked],
|
||
"revoke_absent": [f"{e['table']}@{e['host']} ({e['privs']}) [REVOKE-ABSENT]"
|
||
for e in absent],
|
||
"statements": len(granted) + len(revoked) + len(absent),
|
||
"converged": not mismatch,
|
||
"converged_mismatch": mismatch,
|
||
}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# DDL 解析（P0-1：先剔除注释行再判定语句类型，不再用 startswith 裸判）
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def _strip_comment_lines(sql_text: str) -> str:
|
||
"""剔除整行注释（-- 开头）+ 行尾注释，避免注释行带动导致语句被跳过/误判。"""
|
||
out = []
|
||
for raw in sql_text.splitlines():
|
||
line = raw
|
||
idx = line.find("--")
|
||
if idx >= 0:
|
||
line = line[:idx]
|
||
if not line.strip():
|
||
continue
|
||
out.append(line)
|
||
return "\n".join(out)
|
||
|
||
|
||
def _split_statements(sql_text: str) -> list:
|
||
return [s.strip() for s in _strip_comment_lines(sql_text).split(";") if s.strip()]
|
||
|
||
|
||
def _stmt_table_name(stmt: str, keyword: str):
|
||
"""取 `KEYWORD TABLE [IF NOT EXISTS] <name>` 中的表名。"""
|
||
toks = stmt.replace("(", " ").replace("\n", " ").split()
|
||
up = [t.upper() for t in toks]
|
||
try:
|
||
i = up.index(keyword)
|
||
except ValueError:
|
||
return None
|
||
j = i + 1
|
||
while j < len(up) and up[j] in ("IF", "NOT", "EXISTS"):
|
||
j += 1
|
||
return toks[j].strip("`") if j < len(toks) else None
|
||
|
||
|
||
def parse_ddl(path=DDL_FILE):
|
||
"""解析 DDL → {creates: [(table, stmt)], seeds: [stmt], alter: str|None, index: str|None}"""
|
||
sql = open(path, encoding="utf-8").read()
|
||
creates, seeds, alter, index_stmt = [], [], None, None
|
||
for stmt in _split_statements(sql):
|
||
head = stmt.upper()
|
||
if head.startswith("CREATE TABLE"):
|
||
creates.append((_stmt_table_name(stmt, "TABLE"), stmt))
|
||
elif head.startswith("INSERT INTO"):
|
||
seeds.append(stmt)
|
||
elif head.startswith("ALTER TABLE CASH_PLANS ADD KEY"):
|
||
index_stmt = stmt
|
||
elif head.startswith("ALTER TABLE CASH_PLANS"):
|
||
alter = stmt
|
||
return {"creates": creates, "seeds": seeds, "alter": alter, "index": index_stmt,
|
||
"ddl_text": sql}
|
||
|
||
|
||
def assert_ddl_parseable(p):
|
||
"""P0-1 断言：6 条 CREATE 全命中且表名与 NEW_TABLES 一致；种子 2 条；ALTER/B1 各 1 条。"""
|
||
names = [t for t, _ in p["creates"]]
|
||
errs = []
|
||
if len(p["creates"]) != len(NEW_TABLES):
|
||
errs.append(f"CREATE TABLE 命中数 {len(p['creates'])} != {len(NEW_TABLES)}（{names}）")
|
||
if set(names) != set(NEW_TABLES):
|
||
errs.append(f"CREATE TABLE 表名集合不一致: {sorted(names)} vs {sorted(NEW_TABLES)}")
|
||
if len(p["seeds"]) != 2:
|
||
errs.append(f"种子 INSERT 语句数 {len(p['seeds'])} != 2")
|
||
if not p["alter"]:
|
||
errs.append("DDL §B 未解析到 ALTER TABLE cash_plans 加列语句")
|
||
if not p["index"]:
|
||
errs.append("DDL §B1 未解析到 ALTER TABLE cash_plans ADD KEY 语句")
|
||
if p["alter"] and p["alter"].strip() not in CASH_PLAN_ALTER.strip():
|
||
errs.append("DDL §B ALTER 文本与脚本 CASH_PLAN_ALTER 常量漂移（指纹口径会不一致）")
|
||
return errs
|
||
|
||
|
||
def _fingerprint(texts):
|
||
canon = json.dumps(sorted(texts), ensure_ascii=False, separators=(",", ":"))
|
||
return hashlib.sha1(canon.encode("utf-8")).hexdigest()[:10]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# information_schema 内省（只读）
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def _introspect(conn):
|
||
rows = conn.execute(text(
|
||
"SELECT TABLE_NAME FROM information_schema.TABLES "
|
||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'treasury%'"
|
||
)).fetchall()
|
||
tables = {r[0] for r in rows}
|
||
cols = conn.execute(text(
|
||
"SELECT COLUMN_NAME FROM information_schema.COLUMNS "
|
||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='cash_plans'"
|
||
)).fetchall()
|
||
colset = {r[0] for r in cols}
|
||
idx = conn.execute(text(
|
||
"SELECT DISTINCT INDEX_NAME FROM information_schema.STATISTICS "
|
||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='cash_plans'"
|
||
)).fetchall()
|
||
return tables, colset, {r[0] for r in idx}
|
||
|
||
|
||
def _seed_counts(conn):
|
||
counts = {}
|
||
for tbl, key in (("treasury_control_rules", "rules_rows"), ("treasury_limits", "limits_rows")):
|
||
try:
|
||
counts[key] = conn.execute(text(f"SELECT COUNT(*) FROM {tbl}")).scalar()
|
||
except Exception:
|
||
counts[key] = None
|
||
return counts
|
||
|
||
|
||
def check(conn):
|
||
"""只读预检 —— 缺表 / 缺列 / 缺索引 / 存量行数 / 种子行数；零写入。"""
|
||
tables, cols, idx = _introspect(conn)
|
||
cash_rows = conn.execute(text("SELECT COUNT(*) FROM cash_plans")).scalar()
|
||
report = {
|
||
"missing_tables": [t for t in NEW_TABLES if t not in tables],
|
||
"existing_tables": sorted(tables),
|
||
"missing_cash_plan_cols": [c for c in CASH_PLAN_NEW_COLS if c not in cols],
|
||
"missing_cash_plan_index": [] if CASH_PLAN_NEW_INDEX in idx else [CASH_PLAN_NEW_INDEX],
|
||
"cash_plans_rows": cash_rows,
|
||
}
|
||
report.update(_seed_counts(conn))
|
||
print(json.dumps(report, ensure_ascii=False, indent=2))
|
||
return report
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# UP / DOWN
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def _write_fingerprint(conn, batch, tables_created):
|
||
"""P2-7：指纹行按 (import_type,batch) 先查后写（不依赖唯一键；entity_id=1 为约定）。"""
|
||
cols = {r[0] for r in conn.execute(text(
|
||
"SELECT COLUMN_NAME FROM information_schema.COLUMNS "
|
||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='import_logs'"
|
||
)).fetchall()}
|
||
need = {"entity_id", "filename", "batch", "total_rows", "success_rows", "import_type"}
|
||
if not need <= cols:
|
||
raise RuntimeError(f"import_logs 结构不符，缺列 {sorted(need - cols)}（fail-closed，不静默跳过）")
|
||
row = conn.execute(text(
|
||
"SELECT id FROM import_logs WHERE import_type='treasury-ddl' AND batch=:b ORDER BY id LIMIT 1"
|
||
), {"b": batch}).fetchone()
|
||
if row:
|
||
conn.execute(text(
|
||
"UPDATE import_logs SET total_rows=:t, success_rows=:s WHERE id=:i"
|
||
), {"t": tables_created, "s": tables_created, "i": row[0]})
|
||
return "updated"
|
||
conn.execute(text(
|
||
"INSERT INTO import_logs (entity_id, filename, batch, total_rows, success_rows, import_type, created_by) "
|
||
"VALUES (1, 'treasury_ddl.sql', :b, :t, :s, 'treasury-ddl', 'treasury_migrate')"
|
||
), {"b": batch, "t": tables_created, "s": tables_created})
|
||
return "inserted"
|
||
|
||
|
||
def up(conn):
|
||
p = parse_ddl()
|
||
errs = assert_ddl_parseable(p)
|
||
if errs:
|
||
for e in errs:
|
||
print("❌ DDL 解析失败:", e, file=sys.stderr)
|
||
return 3
|
||
|
||
before_tables, before_cols, before_idx = _introspect(conn)
|
||
|
||
# ① 建表（6 条 CREATE，IF NOT EXISTS 幂等）
|
||
for _table, stmt in p["creates"]:
|
||
conn.execute(text(stmt))
|
||
|
||
# ② 条件加列（列级幂等）
|
||
missing_cols = [c for c in CASH_PLAN_NEW_COLS if c not in before_cols]
|
||
if missing_cols:
|
||
conn.execute(text(p["alter"]))
|
||
|
||
# ③ 条件加索引（D-7）
|
||
if CASH_PLAN_NEW_INDEX not in before_idx:
|
||
conn.execute(text(p["index"]))
|
||
|
||
# ③b 授权收敛（按白名单最小面 + 幂等，须特权连接）：REVOKE 白名单外 + GRANT 白名单
|
||
# 旧形态为无条件 12 条全量 GRANT（含表级 SELECT 等白名单外权限）→ 每次部署扩面，
|
||
# 基线巡检恒 BASELINE_MISMATCH（2026-09-18 F-1）。现改为收敛：重复执行结果不变。
|
||
spec, spec_source, spec_note = load_grant_spec()
|
||
if spec_note:
|
||
print("[warn] " + spec_note, file=sys.stderr)
|
||
grant_sync = apply_grant_sync(conn, spec, spec_source)
|
||
|
||
# ④ 种子（ON DUPLICATE KEY UPDATE，数据级幂等）
|
||
for stmt in p["seeds"]:
|
||
conn.execute(text(stmt))
|
||
|
||
conn.commit()
|
||
|
||
after_tables, after_cols, after_idx = _introspect(conn)
|
||
created_new = len(after_tables - before_tables)
|
||
added_cols = len(after_cols - before_cols)
|
||
added_indexes = len(after_idx - before_idx)
|
||
counts = _seed_counts(conn)
|
||
|
||
# ⑤ 自检断言：缺表即非零退出（P0-1，禁静默成功）
|
||
post = check(conn)
|
||
if post["missing_tables"]:
|
||
print(f"❌ up() 后仍有缺表 {post['missing_tables']}（非零退出）", file=sys.stderr)
|
||
return 4
|
||
|
||
# ⑥ 指纹行
|
||
batch = "treasury-ddl-" + _fingerprint([p["ddl_text"], CASH_PLAN_ALTER])
|
||
action = _write_fingerprint(conn, batch, created_new or len(NEW_TABLES))
|
||
conn.commit()
|
||
|
||
print(json.dumps({
|
||
"phase": "up",
|
||
"created_new": created_new,
|
||
"added_cols": added_cols,
|
||
"added_indexes": added_indexes,
|
||
"seeded_statements": len(p["seeds"]),
|
||
"rules_rows": counts["rules_rows"],
|
||
"limits_rows": counts["limits_rows"],
|
||
"import_logs_row": action,
|
||
"batch": batch,
|
||
"granted": grant_sync["granted"],
|
||
"grant_sync": grant_sync,
|
||
"missing_tables": post["missing_tables"],
|
||
"cash_plans_rows": post["cash_plans_rows"],
|
||
}, ensure_ascii=False))
|
||
if not grant_sync["converged"]:
|
||
print(f"❌ 授权收敛未达成 {grant_sync['converged_mismatch']}（非零退出）", file=sys.stderr)
|
||
return 6
|
||
return 0
|
||
|
||
|
||
def _dump_cash_plans(out_path):
|
||
"""P1-4：写路径账号 + 临时 defaults-file 注入口令（不硬编码、不回显）；失败即抛异常（fail-closed）。"""
|
||
from app import database as db
|
||
|
||
fd, cnf = tempfile.mkstemp(prefix=".cma-dump-", suffix=".cnf")
|
||
try:
|
||
with os.fdopen(fd, "w") as f:
|
||
f.write("[client]\nuser=%s\npassword=%s\nhost=%s\nport=%s\n"
|
||
% (db.DB_USER, db.DB_PASS, db.DB_HOST, db.DB_PORT))
|
||
os.chmod(cnf, 0o600)
|
||
with open(out_path, "w") as fh:
|
||
proc = subprocess.run(
|
||
["mysqldump", f"--defaults-extra-file={cnf}", "--single-transaction",
|
||
"--skip-comments", db.DB_NAME, "cash_plans"],
|
||
stdout=fh, stderr=subprocess.PIPE, text=True)
|
||
if proc.returncode != 0:
|
||
raise RuntimeError("mysqldump 失败(rc=%s): %s" % (proc.returncode, (proc.stderr or "").strip()[:300]))
|
||
finally:
|
||
try:
|
||
os.remove(cnf)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def down(conn):
|
||
"""回滚：先 dump（fail-closed）→ 逐列删扩展列（P1-3）→ 删索引 → 删表（IF EXISTS）。"""
|
||
os.makedirs(BACKUP_DIR, exist_ok=True)
|
||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||
dump = os.path.join(BACKUP_DIR, f"cash_plans-before-rollback-{stamp}.sql")
|
||
try:
|
||
_dump_cash_plans(dump)
|
||
except Exception as exc: # fail-closed：dump 不成功就不允许删列
|
||
print(f"❌ 备份失败，回滚中止（未做任何结构变更）: {exc}", file=sys.stderr)
|
||
return 5
|
||
print("backup:", dump, os.path.getsize(dump), "bytes")
|
||
|
||
tables, cols, idx = _introspect(conn)
|
||
dropped = []
|
||
for col in CASH_PLAN_NEW_COLS: # P1-3：按列存在性逐列 DROP（部分态不再 1091）
|
||
if col in cols:
|
||
conn.execute(text(f"ALTER TABLE cash_plans DROP COLUMN {col}"))
|
||
dropped.append(col)
|
||
if CASH_PLAN_NEW_INDEX in idx:
|
||
conn.execute(text(f"ALTER TABLE cash_plans DROP INDEX {CASH_PLAN_NEW_INDEX}"))
|
||
dropped.append(CASH_PLAN_NEW_INDEX)
|
||
for t in reversed(NEW_TABLES):
|
||
conn.execute(text(f"DROP TABLE IF EXISTS {t}"))
|
||
conn.commit()
|
||
print(json.dumps({"phase": "down", "dropped_cash_plan_objects": dropped,
|
||
"dropped_tables": list(reversed(NEW_TABLES)), "backup": dump},
|
||
ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 入口
|
||
# ─────────────────────────────────────────────────────────────
|
||
|
||
def grant_plan_only():
|
||
"""零 DB：打印授权收敛语句计划（幂等自证证据：连跑两次逐字相同）。"""
|
||
spec, source, note = load_grant_spec()
|
||
dbn = _resolve_dbn()
|
||
stmts = grant_plan(spec, dbn)
|
||
out = {
|
||
"mode": "grant-plan",
|
||
"whitelist_path": WHITELIST_PATH,
|
||
"whitelist_present": os.path.exists(WHITELIST_PATH),
|
||
"spec_source": source,
|
||
"spec_note": note,
|
||
"spec": {t: ",".join(spec[t]) for t in sorted(spec)},
|
||
"hosts": list(GRANT_HOSTS),
|
||
"db_name": dbn,
|
||
"statement_count": len(stmts),
|
||
"statements": stmts,
|
||
"statements_md5": hashlib.md5("\n".join(stmts).encode("utf-8")).hexdigest(),
|
||
}
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
|
||
def parse_only():
|
||
"""零 DB 的 DDL 解析复算（P0-1 证据）：命中 6 条 CREATE + 表名清单 + 种子语句。"""
|
||
p = parse_ddl()
|
||
errs = assert_ddl_parseable(p)
|
||
out = {
|
||
"ddl_file": DDL_FILE,
|
||
"ddl_md5": hashlib.md5(open(DDL_FILE, "rb").read()).hexdigest(),
|
||
"statements_total": len(_split_statements(p["ddl_text"])),
|
||
"create_table_hits": len(p["creates"]),
|
||
"create_table_names": [t for t, _ in p["creates"]],
|
||
"seed_statements": len(p["seeds"]),
|
||
"has_cash_plan_alter": bool(p["alter"]),
|
||
"has_cash_plan_index": bool(p["index"]),
|
||
"expected_first_run": {"created_new": 6, "added_cols": 4, "added_indexes": 1,
|
||
"rules_rows": EXPECTED_RULE_ROWS, "limits_rows": EXPECTED_LIMIT_ROWS},
|
||
"expected_second_run": {"created_new": 0, "added_cols": 0, "added_indexes": 0,
|
||
"rules_rows": EXPECTED_RULE_ROWS, "limits_rows": EXPECTED_LIMIT_ROWS},
|
||
"errors": errs,
|
||
}
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||
return 1 if errs else 0
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
g = ap.add_mutually_exclusive_group()
|
||
g.add_argument("--parse-only", action="store_true", help="零 DB：只解析 DDL（P0-1 复算证据）")
|
||
g.add_argument("--grant-plan", action="store_true", help="零 DB：打印授权收敛语句计划")
|
||
g.add_argument("--check", action="store_true", help="只读预检（默认）")
|
||
g.add_argument("--apply", action="store_true")
|
||
g.add_argument("--down", action="store_true")
|
||
ap.add_argument("--yes", action="store_true", help="回滚确认（表内有数据时必需）")
|
||
args = ap.parse_args()
|
||
|
||
if args.parse_only:
|
||
return parse_only()
|
||
if args.grant_plan:
|
||
return grant_plan_only()
|
||
|
||
engine = get_engine()
|
||
with engine.connect() as conn:
|
||
if args.apply:
|
||
return up(conn)
|
||
if args.down:
|
||
tables, _c, _i = _introspect(conn)
|
||
payload = 0
|
||
if {"treasury_payments", "treasury_balances"} <= tables:
|
||
payload = conn.execute(text(
|
||
"SELECT (SELECT COUNT(*) FROM treasury_payments) + "
|
||
"(SELECT COUNT(*) FROM treasury_balances)")).scalar()
|
||
if payload and not args.yes:
|
||
print(f"REFUSE: 已有 {payload} 行业务数据，回滚需显式 --yes", file=sys.stderr)
|
||
return 2
|
||
return down(conn)
|
||
return 0 if check(conn) is not None else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|