Hermes CI Fix
f1fa781e08
ci/woodpecker/push/main Pipeline was canceled
- db_migrate.py: create_all 后调用 run_structural() → 复用 treasury_migrate（幂等、特权连接） - treasury_migrate.py: 换特权连接（CMA_MIGRATE_DB_USER 或本机 root socket，禁 cma_api）；up() 增 grant_app_account()（cma_api 对 6 张 treasury 表 DML，双 host） - 效果：deploy 链由特权账号一次性完成 建表+补列(cash_plans 4列/1索引)+种子(8规则/6限额)+授权 Ref: 0-P0-project-20260915-cma-treasury-stage1
456 lines
20 KiB
Python
456 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""CMA 数据库结构迁移（特权 DDL 步）—— 与应用启动 create_all 剥离
|
||
|
||
任务：cma-create-all-minpriv-rootfix-20260916（P2）
|
||
|
||
背景（2026-09-16 01:08 发布中断事故）
|
||
--------------------------------------
|
||
应用启动时 `app/database.py::init_db()` 调 `Base.metadata.create_all()`（需要 CREATE 权限），
|
||
而运行库账号 `cma_api` 只有 SELECT + 表级 DML 白名单（**无 DDL**）。新代码引入新模型
|
||
（`tax_filing_log` / `treasury_*`）→ 启动即 `CREATE command denied` → 崩溃重启循环。
|
||
|
||
本脚本 = 部署链里的**独立、幂等、可回滚**的特权迁移步，在应用重启**之前**执行：
|
||
|
||
deploy.sh [4b] → backend/scripts/db_migrate.py --apply → systemctl restart cma-backend
|
||
|
||
于是：DDL 不再随应用 boot 发生；运行库账号可保持最小权限（无需 CREATE）。
|
||
|
||
用法
|
||
----
|
||
python3 scripts/db_migrate.py --check # 只读预检（默认动作，零写入）
|
||
python3 scripts/db_migrate.py --apply # 幂等建表（缺什么建什么）+ 写 journal
|
||
python3 scripts/db_migrate.py --status # 列出历史 journal
|
||
python3 scripts/db_migrate.py --rollback latest # 回滚最近一次 --apply 新建的表
|
||
python3 scripts/db_migrate.py --rollback <journal.json>
|
||
|
||
选项
|
||
----
|
||
--json 机器可读输出（部署链/CI 断言用）
|
||
--force 回滚时允许删除**有数据**的表（默认拒绝，exit 3）
|
||
--no-dump 回滚时不 mysqldump（默认先备份到 journal 目录，dump 失败即拒绝回滚）
|
||
--socket PATH unix socket 路径（默认 /var/run/mysqld/mysqld.sock）
|
||
--db-url URL 显式连接串（覆盖下方全部推导；不含口令时应配合环境变量）
|
||
--db-name NAME 目标库名（默认 env CMA_MIGRATE_DB_NAME → CMA_DB_NAME → cma）
|
||
|
||
连接凭据（按优先级；口令只从环境读，禁止硬编码/回显）
|
||
----------------------------------------------------
|
||
1) CMA_MIGRATE_DB_USER + CMA_MIGRATE_DB_PASS（**专用迁移账号**，TCP）——推荐用于非本机场景
|
||
2) 未设 CMA_MIGRATE_DB_USER → 本机 unix socket + root（socket 认证，无需口令）
|
||
⚠️ 应用运行账号（backend/.env 的 CMA_DB_USER，= cma_api）**不得**被本脚本使用：
|
||
迁移必须走在有 DDL 权限的账号上，本脚本会显式拒绝。
|
||
|
||
幂等契约
|
||
--------
|
||
首跑 --apply：created_new = 缺失表数（本次实测：0 —— 库已由应急 create_all 补齐）
|
||
二跑 --apply：created_new = 0，零副作用
|
||
已存在的表**不** ALTER、**不** DROP、**不**动数据行（列级差异只报告，见下）
|
||
|
||
退出码
|
||
------
|
||
0 成功 ｜ 1 迁移失败 / 迁移后自检不一致 / 连接失败（fail-closed，部署链据此中止）
|
||
2 用法错误 ｜ 3 回滚被拒（目标表有数据且未 --force）
|
||
"""
|
||
import argparse
|
||
import fcntl
|
||
import json
|
||
import os
|
||
import random
|
||
import socket
|
||
import string
|
||
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, inspect, text # noqa: E402
|
||
from sqlalchemy.engine.url import make_url # noqa: E402
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
BACKEND_DIR = os.path.dirname(HERE)
|
||
DEFAULT_JOURNAL_DIR = os.path.join(HERE, "backups", "db-migrate-journal")
|
||
DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock"
|
||
DEFAULT_DB = "cma"
|
||
# 应用运行账号：绝不允许用于本迁移步（无 DDL 权限 = 事故根因）
|
||
FORBIDDEN_RUN_ACCOUNTS = {"cma_api"}
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────
|
||
# 连接（特权通道）
|
||
# ────────────────────────────────────────────────────────────────
|
||
def _resolve_db_name(args) -> str:
|
||
return (
|
||
args.db_name
|
||
or os.getenv("CMA_MIGRATE_DB_NAME")
|
||
or os.getenv("CMA_DB_NAME")
|
||
or DEFAULT_DB
|
||
)
|
||
|
||
|
||
def build_engine(args):
|
||
"""构造特权连接。返回 (engine, 描述串)。描述串不含口令。"""
|
||
db_name = _resolve_db_name(args)
|
||
if args.db_url:
|
||
url = make_url(args.db_url)
|
||
if url.database is None:
|
||
url = url.set(database=db_name)
|
||
return create_engine(url, pool_pre_ping=True), f"db-url({url.drivername}@{url.host or 'socket'}/{url.database})"
|
||
|
||
user = os.getenv("CMA_MIGRATE_DB_USER", "").strip()
|
||
if user:
|
||
if user in FORBIDDEN_RUN_ACCOUNTS:
|
||
raise SystemExit(
|
||
f"[FAIL] CMA_MIGRATE_DB_USER={user} 是**应用运行账号**（无 DDL 权限）——\n"
|
||
f" 迁移步必须使用有 DDL 权限的专用账号（或本机 root socket）。已拒绝执行。"
|
||
)
|
||
password = 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"))
|
||
url = (
|
||
f"mysql+pymysql://{user}:{password}@{host}:{port}/{db_name}"
|
||
f"?charset=utf8mb4"
|
||
)
|
||
return create_engine(url, pool_pre_ping=True), f"tcp({user}@{host}:{port}/{db_name})"
|
||
|
||
# 默认：本机 unix socket + root（socket 认证，无口令落盘）
|
||
sock = args.socket or os.getenv("CMA_MIGRATE_DB_SOCKET") or DEFAULT_SOCKET
|
||
if not os.path.exists(sock):
|
||
raise SystemExit(
|
||
f"[FAIL] 未设 CMA_MIGRATE_DB_USER，且 unix socket 不存在：{sock}\n"
|
||
f" 请提供专用迁移账号（CMA_MIGRATE_DB_USER/PASS）或用 --socket 指定路径。"
|
||
)
|
||
url = f"mysql+pymysql://root@localhost/{db_name}?unix_socket={sock}&charset=utf8mb4"
|
||
return create_engine(url, pool_pre_ping=True), f"socket(root@{sock}/{db_name})"
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────
|
||
# schema 比对（ORM 期望 vs 库内实况）
|
||
# ────────────────────────────────────────────────────────────────
|
||
def _import_models():
|
||
import app.models # noqa: F401 触发全部模型注册
|
||
from app.database import Base
|
||
return Base
|
||
|
||
|
||
def schema_diff(engine):
|
||
"""返回 (missing_tables, missing_columns{表:[列]})。只读。"""
|
||
Base = _import_models()
|
||
insp = inspect(engine)
|
||
existing = set(insp.get_table_names())
|
||
missing_tables, missing_columns = [], {}
|
||
for table in Base.metadata.sorted_tables:
|
||
if table.name not in existing:
|
||
missing_tables.append(table.name)
|
||
continue
|
||
have = {c["name"] for c in insp.get_columns(table.name)}
|
||
miss = [c.name for c in table.columns if c.name not in have]
|
||
if miss:
|
||
missing_columns[table.name] = miss
|
||
return missing_tables, missing_columns
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────
|
||
# journal（迁移留痕：谁/何时/建了什么 → 回滚依据）
|
||
# ────────────────────────────────────────────────────────────────
|
||
def journal_dir() -> str:
|
||
d = os.getenv("CMA_MIGRATE_JOURNAL_DIR") or DEFAULT_JOURNAL_DIR
|
||
os.makedirs(d, exist_ok=True)
|
||
return d
|
||
|
||
|
||
def _git_head() -> str:
|
||
try:
|
||
return subprocess.run(
|
||
["git", "-C", os.path.dirname(BACKEND_DIR), "rev-parse", "--short", "HEAD"],
|
||
capture_output=True, text=True, timeout=10,
|
||
).stdout.strip() or "unknown"
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
|
||
def write_journal(payload: dict) -> str:
|
||
path = os.path.join(
|
||
journal_dir(), f"{datetime.now().strftime('%Y%m%dT%H%M%S')}-{payload.get('status', 'na')}.json"
|
||
)
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
json.dump(payload, fh, ensure_ascii=False, indent=2)
|
||
return path
|
||
|
||
|
||
def list_journals():
|
||
d = journal_dir()
|
||
return sorted(
|
||
(os.path.join(d, f) for f in os.listdir(d) if f.endswith(".json")),
|
||
reverse=True,
|
||
)
|
||
|
||
|
||
def resolve_journal(ref: str) -> str:
|
||
if ref in ("latest", ""):
|
||
items = [p for p in list_journals() if "-applied" in os.path.basename(p)]
|
||
if not items:
|
||
raise SystemExit("[FAIL] 没有可回滚的 journal（无 --apply 记录）")
|
||
return items[0]
|
||
if os.path.exists(ref):
|
||
return ref
|
||
cand = os.path.join(journal_dir(), os.path.basename(ref))
|
||
if os.path.exists(cand):
|
||
return cand
|
||
raise SystemExit(f"[FAIL] journal 不存在：{ref}")
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────
|
||
# 动作
|
||
# ────────────────────────────────────────────────────────────────
|
||
def cmd_check(args, engine, conn_desc) -> int:
|
||
missing_tables, missing_cols = schema_diff(engine)
|
||
out = {
|
||
"action": "check",
|
||
"connection": conn_desc,
|
||
"missing_tables": missing_tables,
|
||
"missing_table_count": len(missing_tables),
|
||
"missing_columns": missing_cols,
|
||
}
|
||
if args.json:
|
||
print(json.dumps(out, ensure_ascii=False))
|
||
else:
|
||
print(f"[check] 连接: {conn_desc}")
|
||
print(f"[check] 缺失表 {len(missing_tables)} 张: {missing_tables if missing_tables else '（无）'}")
|
||
if missing_cols:
|
||
print(f"[check] 列级差异 {len(missing_cols)} 张表: {missing_cols}")
|
||
print(" ⚠️ create_all 不会补列 → 需要独立的 ALTER 迁移（见 journal/回报）")
|
||
else:
|
||
print("[check] 列级差异：无")
|
||
# 列级差异属"存在但结构过期"：部署链上以告警呈现，不阻断（阻断判据 = 缺表未补齐）
|
||
return 0
|
||
|
||
|
||
def run_structural() -> dict:
|
||
"""结构增量（补列/索引/种子/授权）——create_all 覆盖不到。
|
||
复用 `treasury_migrate.py --apply`（幂等；其自身走特权连接）。失败不中断部署，只告警+留痕。"""
|
||
script = os.path.join(HERE, "treasury_migrate.py")
|
||
if not os.path.exists(script):
|
||
return {"status": "skipped", "reason": "treasury_migrate.py 不存在"}
|
||
try:
|
||
proc = subprocess.run([sys.executable, script, "--apply"],
|
||
capture_output=True, text=True, timeout=300)
|
||
except Exception as exc:
|
||
return {"status": "error", "reason": f"{type(exc).__name__}: {exc}"}
|
||
return {
|
||
"status": "ok" if proc.returncode == 0 else "failed",
|
||
"rc": proc.returncode,
|
||
"stdout_tail": (proc.stdout or "").strip()[-600:],
|
||
"stderr_tail": (proc.stderr or "").strip()[-400:],
|
||
}
|
||
|
||
|
||
def cmd_apply(args, engine, conn_desc) -> int:
|
||
Base = _import_models()
|
||
before_missing, before_cols = schema_diff(engine)
|
||
started = datetime.now().isoformat(timespec="seconds")
|
||
payload = {
|
||
"action": "apply",
|
||
"status": "running",
|
||
"started_at": started,
|
||
"finished_at": None,
|
||
"connection": conn_desc,
|
||
"git_head": _git_head(),
|
||
"actor": f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}",
|
||
"db_name": _resolve_db_name(args),
|
||
"missing_tables_before": before_missing,
|
||
"missing_columns_before": before_cols,
|
||
"created_tables": [],
|
||
"created_new": 0,
|
||
"error": None,
|
||
}
|
||
print(f"[apply] 连接: {conn_desc}")
|
||
print(f"[apply] 迁移前缺失表 {len(before_missing)} 张: {before_missing if before_missing else '（无）'}")
|
||
try:
|
||
if before_missing:
|
||
Base.metadata.create_all(bind=engine) # checkfirst=True → 已存在的表零动作
|
||
# 结构增量（补列/索引/种子/授权）：create_all 不补列 → 复用 treasury_migrate（幂等，特权连接）
|
||
structural = run_structural()
|
||
after_missing, after_cols = schema_diff(engine)
|
||
created = [t for t in before_missing if t not in after_missing]
|
||
payload.update(
|
||
status="applied",
|
||
finished_at=datetime.now().isoformat(timespec="seconds"),
|
||
created_tables=created,
|
||
created_new=len(created),
|
||
missing_tables_after=after_missing,
|
||
structural=structural,
|
||
)
|
||
path = write_journal(payload)
|
||
except Exception as exc: # DDL 无事务：失败时记录已建对象，供 --rollback 恢复
|
||
try:
|
||
after_missing, _ = schema_diff(engine)
|
||
except Exception:
|
||
after_missing = before_missing
|
||
created = [t for t in before_missing if t not in after_missing]
|
||
payload.update(
|
||
status="failed",
|
||
finished_at=datetime.now().isoformat(timespec="seconds"),
|
||
created_tables=created,
|
||
created_new=len(created),
|
||
error=f"{type(exc).__name__}: {exc}",
|
||
)
|
||
path = write_journal(payload)
|
||
print(f"[FAIL] 迁移失败：{payload['error']}")
|
||
print(f" 已建对象 {created}（journal: {path}）")
|
||
print(f" 回滚：python3 scripts/db_migrate.py --rollback {os.path.basename(path)}")
|
||
return 1
|
||
|
||
ok = not payload["missing_tables_after"]
|
||
print(f"[apply] 新建 {payload['created_new']} 张表: {created if created else '（无 → 幂等空跑）'}")
|
||
print(f"[apply] 迁移后缺失表 {len(payload['missing_tables_after'])} 张: "
|
||
f"{payload['missing_tables_after'] if payload['missing_tables_after'] else '（无）'}")
|
||
print(f"[apply] 结构增量(treasury_migrate) status={structural.get('status')} rc={structural.get('rc')}")
|
||
if after_cols:
|
||
print(f"[apply] ⚠️ 列级差异仍存（结构步未覆盖）: {after_cols}")
|
||
print(f"[apply] journal: {path}")
|
||
if not ok:
|
||
print("[FAIL] 迁移后自检不一致（仍有缺失表）→ 部署链应中止")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
def _dump_tables(tables, db_name) -> str:
|
||
"""mysqldump 备份（凭据走 600 临时 defaults 文件，不进 argv/日志）。返回文件路径。"""
|
||
fd, cfg = tempfile.mkstemp(prefix="cma-migrate-dump-", suffix=".cnf")
|
||
os.close(fd)
|
||
os.chmod(cfg, 0o600)
|
||
try:
|
||
lines = ["[client]"]
|
||
if os.getenv("CMA_MIGRATE_DB_USER"):
|
||
lines += [
|
||
f"user={os.getenv('CMA_MIGRATE_DB_USER')}",
|
||
f"password={os.getenv('CMA_MIGRATE_DB_PASS', '')}",
|
||
f"host={os.getenv('CMA_MIGRATE_DB_HOST', '127.0.0.1')}",
|
||
f"port={os.getenv('CMA_MIGRATE_DB_PORT', '3306')}",
|
||
]
|
||
else:
|
||
lines += ["user=root", f"socket={os.getenv('CMA_MIGRATE_DB_SOCKET', DEFAULT_SOCKET)}"]
|
||
with open(cfg, "w", encoding="utf-8") as fh:
|
||
fh.write("\n".join(lines) + "\n")
|
||
out = os.path.join(
|
||
journal_dir(), f"dump-{datetime.now().strftime('%Y%m%dT%H%M%S')}.sql"
|
||
)
|
||
cmd = ["mysqldump", f"--defaults-extra-file={cfg}", "--no-tablespaces", "--skip-add-locks", db_name, *tables]
|
||
with open(out, "wb") as fh:
|
||
subprocess.run(cmd, stdout=fh, stderr=subprocess.PIPE, check=True)
|
||
return out
|
||
finally:
|
||
os.unlink(cfg)
|
||
|
||
|
||
def cmd_rollback(args, engine, conn_desc) -> int:
|
||
path = resolve_journal(args.rollback)
|
||
with open(path, encoding="utf-8") as fh:
|
||
payload = json.load(fh)
|
||
tables = payload.get("created_tables") or []
|
||
print(f"[rollback] journal: {path}")
|
||
print(f"[rollback] 该次迁移新建的表: {tables if tables else '（无）'}")
|
||
if not tables:
|
||
print("[rollback] 无新建对象 → 无需回滚（空操作）")
|
||
return 0
|
||
|
||
db_name = payload.get("db_name") or _resolve_db_name(args)
|
||
if not args.no_dump:
|
||
try:
|
||
dump_path = _dump_tables(tables, db_name)
|
||
print(f"[rollback] 已备份: {dump_path}")
|
||
except Exception as exc:
|
||
print(f"[FAIL] 备份失败 → 拒绝回滚（fail-closed）：{type(exc).__name__}: {exc}")
|
||
return 1
|
||
|
||
with engine.connect() as conn:
|
||
rows = {}
|
||
for t in tables:
|
||
try:
|
||
rows[t] = conn.execute(text(f"SELECT COUNT(*) FROM `{t}`")).scalar()
|
||
except Exception:
|
||
rows[t] = 0
|
||
non_empty = {t: n for t, n in rows.items() if n}
|
||
if non_empty and not args.force:
|
||
print(f"[REFUSED] 以下表已有数据，未加 --force 拒绝删除：{non_empty}")
|
||
return 3
|
||
for t in tables:
|
||
conn.execute(text(f"DROP TABLE IF EXISTS `{t}`"))
|
||
conn.commit()
|
||
print(f"[rollback] 已删除 {len(tables)} 张表: {tables}")
|
||
payload["rolled_back_at"] = datetime.now().isoformat(timespec="seconds")
|
||
payload["rolled_back_by"] = f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}"
|
||
payload["rollback_dump"] = None if args.no_dump else "见同目录 dump-*.sql"
|
||
base = os.path.basename(path)
|
||
archived = os.path.join(journal_dir(), base.replace("-applied.json", "-rolled_back.json"))
|
||
with open(archived, "w", encoding="utf-8") as fh:
|
||
json.dump(payload, fh, ensure_ascii=False, indent=2)
|
||
print(f"[rollback] 记录: {archived}")
|
||
return 0
|
||
|
||
|
||
def cmd_status(args) -> int:
|
||
items = list_journals()
|
||
if not items:
|
||
print("[status] 无 journal 记录")
|
||
return 0
|
||
for p in items[:20]:
|
||
try:
|
||
with open(p, encoding="utf-8") as fh:
|
||
j = json.load(fh)
|
||
print(
|
||
f"{os.path.basename(p)} status={j.get('status')} created_new={j.get('created_new')} "
|
||
f"tables={len(j.get('created_tables') or [])} git={j.get('git_head')} by={j.get('actor')}"
|
||
)
|
||
except Exception as exc:
|
||
print(f"{os.path.basename(p)} [unreadable] {exc}")
|
||
return 0
|
||
|
||
|
||
# ────────────────────────────────────────────────────────────────
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(
|
||
description="CMA 特权数据库迁移步（与应用启动 create_all 剥离）",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
g = ap.add_mutually_exclusive_group()
|
||
g.add_argument("--check", action="store_true", help="只读预检（默认）")
|
||
g.add_argument("--apply", action="store_true", help="幂等建表 + journal")
|
||
g.add_argument("--rollback", metavar="JOURNAL", help="回滚某次 --apply 新建的表（latest 或文件名）")
|
||
g.add_argument("--status", action="store_true", help="列出 journal 历史")
|
||
ap.add_argument("--json", action="store_true", help="机器可读输出")
|
||
ap.add_argument("--force", action="store_true", help="回滚允许删有数据的表")
|
||
ap.add_argument("--no-dump", action="store_true", help="回滚前不 dump")
|
||
ap.add_argument("--socket", default=None, help=f"unix socket 路径（默认 {DEFAULT_SOCKET}）")
|
||
ap.add_argument("--db-url", default=None, help="显式连接串（覆盖推导）")
|
||
ap.add_argument("--db-name", default=None, help="目标库名")
|
||
args = ap.parse_args()
|
||
|
||
if not (args.apply or args.rollback or args.status):
|
||
args.check = True # 默认只读预检
|
||
|
||
if args.status:
|
||
return cmd_status(args)
|
||
|
||
engine, conn_desc = build_engine(args)
|
||
|
||
# 迁移互斥：防并发会话/手工操作同时跑 DDL
|
||
lock_path = os.path.join(journal_dir(), ".db-migrate.lock")
|
||
lock_fh = open(lock_path, "a+")
|
||
try:
|
||
fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError:
|
||
print("[FAIL] 另一个迁移正在运行（锁被占用）→ 本次中止")
|
||
return 1
|
||
|
||
if args.rollback:
|
||
return cmd_rollback(args, engine, conn_desc)
|
||
if args.apply:
|
||
return cmd_apply(args, engine, conn_desc)
|
||
return cmd_check(args, engine, conn_desc)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|