diff --git a/backend/scripts/db_migrate.py b/backend/scripts/db_migrate.py index de7c9369..5cd1c890 100644 --- a/backend/scripts/db_migrate.py +++ b/backend/scripts/db_migrate.py @@ -226,6 +226,25 @@ def cmd_check(args, engine, conn_desc) -> int: 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) @@ -250,6 +269,8 @@ def cmd_apply(args, engine, conn_desc) -> int: 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( @@ -258,6 +279,7 @@ def cmd_apply(args, engine, conn_desc) -> int: created_tables=created, created_new=len(created), missing_tables_after=after_missing, + structural=structural, ) path = write_journal(payload) except Exception as exc: # DDL 无事务：失败时记录已建对象，供 --rollback 恢复 @@ -283,8 +305,9 @@ def cmd_apply(args, engine, conn_desc) -> int: 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] ⚠️ 列级差异（create_all 不补列，需独立 ALTER）: {after_cols}") + print(f"[apply] ⚠️ 列级差异仍存（结构步未覆盖）: {after_cols}") print(f"[apply] journal: {path}") if not ok: print("[FAIL] 迁移后自检不一致（仍有缺失表）→ 部署链应中止") diff --git a/backend/scripts/treasury_migrate.py b/backend/scripts/treasury_migrate.py index b9562046..06c1837f 100644 --- a/backend/scripts/treasury_migrate.py +++ b/backend/scripts/treasury_migrate.py @@ -30,8 +30,7 @@ from datetime import datetime 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 +from sqlalchemy import create_engine, text # noqa: E402 HERE = os.path.dirname(os.path.abspath(__file__)) BACKUP_DIR = os.path.join(HERE, "backups") @@ -55,6 +54,53 @@ CASH_PLAN_ALTER = """ALTER TABLE cash_plans 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" + + +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 grant_app_account(conn): + """授权应用运行账号 cma_api 对 6 张 treasury 新表的 DML（幂等 GRANT，双 host）。 + 必须由特权连接执行；否则应用端点读不到/写不进新表（1142/1044）。""" + from app import database as db + dbn = (os.getenv("CMA_MIGRATE_DB_NAME") or getattr(db, "DB_NAME", None) or "cma") + granted = [] + for host in ("localhost", "127.0.0.1"): + for t in NEW_TABLES: + conn.execute(text( + f"GRANT SELECT, INSERT, UPDATE, DELETE ON `{dbn}`.`{t}` " + f"TO '{GRANT_APP_ACCOUNT}'@{host}")) + granted.append(f"{t}@{host}") + return granted + # ───────────────────────────────────────────────────────────── # DDL 解析（P0-1：先剔除注释行再判定语句类型，不再用 startswith 裸判） @@ -233,6 +279,9 @@ def up(conn): if CASH_PLAN_NEW_INDEX not in before_idx: conn.execute(text(p["index"])) + # ③b 授权应用账号 cma_api 对本 6 表 DML（幂等 GRANT，须特权连接） + granted = grant_app_account(conn) + # ④ 种子（ON DUPLICATE KEY UPDATE，数据级幂等） for stmt in p["seeds"]: conn.execute(text(stmt)) @@ -266,6 +315,7 @@ def up(conn): "limits_rows": counts["limits_rows"], "import_logs_row": action, "batch": batch, + "granted": granted, "missing_tables": post["missing_tables"], "cash_plans_rows": post["cash_plans_rows"], }, ensure_ascii=False))