#!/usr/bin/env python3 # -*- coding: utf-8 -*- """战略章程表迁移（strategy_charter）— 幂等建表 + 回滚，零示例数据 任务：`project-20261021-cma-strategy-charter-p0-fullstack`（P0 战略基座） 用法： python3 migrate_strategy_charter_20261021.py --check # 只读预检（默认）：表/列/唯一键是否齐 python3 migrate_strategy_charter_20261021.py --apply # 建表（CREATE TABLE IF NOT EXISTS）+ 回读断言 python3 migrate_strategy_charter_20261021.py --down # 回滚（先 mysqldump 备份，再 DROP；有数据需 --force） 连接（与 db_migrate.py / treasury_migrate.py 同口径）： env CMA_MIGRATE_DB_USER / CMA_MIGRATE_DB_PASS / CMA_MIGRATE_DB_HOST / CMA_MIGRATE_DB_PORT 优先（TCP）； 否则本机 unix socket + root（/var/run/mysqld/mysqld.sock）。 **显式拒绝**应用运行账号 cma_api（无 DDL/GRANT 权限）。 纪律： · 迁移脚本**零 INSERT** —— 不写任何示例/默认章程文案（任总红线：真实数据非 Demo）； · 幂等：二跑 created=0；列级差异只报告，不改既有列； · 授权面不在本脚本：由部署链 [4b-2] grant_surface_sync 按白名单授予 cma_api INSERT,UPDATE。 """ import argparse import json import os import subprocess import sys from datetime import datetime TABLE = "strategy_charter" NEW_COLUMNS = [ "id", "entity_id", "mission", "vision", "vision_year", "vision_measurable", "values_text", "value_items", "updated_by", "updated_by_name", "created_at", "updated_at", ] UNIQUE_KEY = "uq_strategy_charter_entity" DDL = """ CREATE TABLE IF NOT EXISTS `strategy_charter` ( `id` INT NOT NULL AUTO_INCREMENT, `entity_id` INT NOT NULL COMMENT '企业ID（账套，多租户隔离）', `mission` TEXT NOT NULL COMMENT '使命：我们为何存在（做什么、为谁做）', `vision` TEXT NOT NULL COMMENT '愿景：要成为什么', `vision_year` VARCHAR(20) NOT NULL COMMENT '愿景达成年份 YYYY（不可空）', `vision_measurable` TEXT NOT NULL COMMENT '可验证状态：可核对规模/业务口径描述（不可空）', `values_text` TEXT NOT NULL COMMENT '价值观总述（不可交易的底线）', `value_items` JSON NOT NULL COMMENT '价值观条目数组 [{name, meaning}]', `updated_by` INT NULL COMMENT '最后修改人 user.id', `updated_by_name` VARCHAR(50) NULL COMMENT '最后修改人姓名（快照）', `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP, `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `uq_strategy_charter_entity` (`entity_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='战略章程（使命/愿景/价值观，一账套一行）' """ DB_NAME = os.environ.get("CMA_MIGRATE_DB_NAME", "cma") APP_ACCOUNT = os.environ.get("CMA_APP_ACCOUNT", "cma_api") def connect(args): import pymysql if os.environ.get("CMA_MIGRATE_DB_USER") or args.user: user = args.user or os.environ.get("CMA_MIGRATE_DB_USER") if user == APP_ACCOUNT: raise SystemExit("[FAIL] 迁移身份不得为应用运行账号 cma_api（无 DDL 权限）") return pymysql.connect( host=args.host or os.environ.get("CMA_MIGRATE_DB_HOST", "127.0.0.1"), port=int(args.port or os.environ.get("CMA_MIGRATE_DB_PORT", 3306)), user=user, password=args.password if args.password is not None else os.environ.get("CMA_MIGRATE_DB_PASS", ""), database=DB_NAME, charset="utf8mb4", autocommit=True, ) socket = os.environ.get("CMA_MIGRATE_DB_SOCKET", "/var/run/mysqld/mysqld.sock") if not os.path.exists(socket): raise SystemExit(f"[FAIL] 无迁移账号环境变量且 socket 不存在: {socket}") return pymysql.connect(unix_socket=socket, user="root", database=DB_NAME, charset="utf8mb4", autocommit=True) def table_columns(cur): cur.execute( "SELECT COLUMN_NAME, IS_NULLABLE, DATA_TYPE FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s ORDER BY ORDINAL_POSITION", (DB_NAME, TABLE)) return {r[0]: {"nullable": r[1], "type": r[2]} for r in cur.fetchall()} def unique_keys(cur): cur.execute( "SELECT INDEX_NAME, COLUMN_NAME FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s AND NON_UNIQUE=0", (DB_NAME, TABLE)) keys = {} for idx, col in cur.fetchall(): keys.setdefault(idx, []).append(col) return keys def rel(*parts): return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), *parts) def main(): ap = argparse.ArgumentParser() g = ap.add_mutually_exclusive_group() g.add_argument("--check", action="store_true") g.add_argument("--apply", action="store_true") g.add_argument("--down", action="store_true") ap.add_argument("--force", action="store_true", help="--down 时表内有数据仍删（先 dump）") ap.add_argument("--host") ap.add_argument("--port", type=int) ap.add_argument("--user") ap.add_argument("--password", default=None) args = ap.parse_args() mode = "apply" if args.apply else ("down" if args.down else "check") conn = connect(args) out: dict = {"script": "migrate_strategy_charter_20261021.py", "mode": mode, "db": DB_NAME, "table": TABLE, "started_at": datetime.now().isoformat(timespec="seconds")} try: with conn.cursor() as cur: cur.execute("SHOW TABLES LIKE %s", (TABLE,)) exists = cur.fetchone() is not None out["table_exists_before"] = exists if mode == "check": cols = table_columns(cur) if exists else {} keys = unique_keys(cur) if exists else {} missing = [c for c in NEW_COLUMNS if c not in cols] out.update({ "columns": sorted(cols.keys()), "missing_columns": missing, "unique_keys": keys, "unique_ok": keys.get(UNIQUE_KEY) == ["entity_id"], "ready": exists and not missing and keys.get(UNIQUE_KEY) == ["entity_id"], "note": "只读预检，零写入", }) print(json.dumps(out, ensure_ascii=False, indent=2)) return 0 if mode == "apply": cur.execute(DDL) cols = table_columns(cur) keys = unique_keys(cur) missing = [c for c in NEW_COLUMNS if c not in cols] cur.execute(f"SELECT COUNT(*) FROM `{TABLE}`") rows = cur.fetchone()[0] ok = (not missing) and keys.get(UNIQUE_KEY) == ["entity_id"] out.update({ "created": 0 if exists else 1, "columns": sorted(cols.keys()), "missing_columns": missing, "unique_keys": keys, "rows": rows, "verified": ok, "note": "零示例数据：本脚本不含任何 INSERT", }) print(json.dumps(out, ensure_ascii=False, indent=2)) return 0 if ok else 1 # --down if not exists: out.update({"action": "NOOP", "note": "表不存在，无需回滚"}) print(json.dumps(out, ensure_ascii=False, indent=2)) return 0 cur.execute(f"SELECT COUNT(*) FROM `{TABLE}`") rows = cur.fetchone()[0] out["rows"] = rows if rows > 0 and not args.force: out.update({"action": "REFUSED", "reason": "表内有数据；先确认再 --force"}) print(json.dumps(out, ensure_ascii=False, indent=2)) return 3 backup_dir = rel("backups", "strategy-charter-rollback") os.makedirs(backup_dir, exist_ok=True) dump = os.path.join(backup_dir, f"{TABLE}-{datetime.now():%Y%m%d-%H%M%S}.sql") cmd = ["mysqldump", "-u", "root", "--single-transaction", "--no-tablespaces", DB_NAME, TABLE] with open(dump, "wb") as fh: rc = subprocess.call(cmd, stdout=fh, stderr=subprocess.PIPE) if rc != 0 or os.path.getsize(dump) == 0: out.update({"action": "REFUSED", "reason": f"mysqldump 失败 rc={rc}（fail-closed）"}) print(json.dumps(out, ensure_ascii=False, indent=2)) return 4 out["dump"] = dump cur.execute(f"DROP TABLE `{TABLE}`") cur.execute("SHOW TABLES LIKE %s", (TABLE,)) out.update({"action": "DROPPED", "table_exists_after": cur.fetchone() is not None}) print(json.dumps(out, ensure_ascii=False, indent=2)) return 0 finally: conn.close() if __name__ == "__main__": sys.exit(main())