wecom-fullstack
0a6525edab
ci/woodpecker/push/main Pipeline was successful
- 表 strategy_charter（一账套一行，entity_id 唯一）：mission/vision/vision_year/vision_measurable/values_text/value_items(JSON) 全 NOT NULL
- 迁移脚本 backend/scripts/migrate_strategy_charter_20261021.py：--check/--apply/--down（回滚先 mysqldump，有数据需 --force），零 INSERT
- API：GET 无行 → filled=false 空态（禁示例数据）；PUT 字段白名单 + 账套隔离三道 + vision_year 4位年份 + value_items [{name,meaning}] 结构校验
- 前端：P 组「战略章程」入口 + MyDialog 编辑（ceo/finance 可写，其余只读）；战略地图页顶部只读章程条
- 用例 43 条 + auth-surface 58 条 = 101 passed；vite build rc=0
197 lines
8.8 KiB
Python
197 lines
8.8 KiB
Python
#!/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())
|