From dd40df6337a7ba031eeca5951bfa55f0ac985958 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 1 Sep 2026 16:00:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(deploy):=20=E6=B5=8B=E8=AF=95=E2=86=92?= =?UTF-8?q?=E7=94=9F=E4=BA=A7=E5=A2=9E=E9=87=8F=E9=83=A8=E7=BD=B2=E6=9C=BA?= =?UTF-8?q?=E5=88=B6=E2=80=94=E2=80=94dmig=E8=BF=81=E7=A7=BB=E5=BC=95?= =?UTF-8?q?=E6=93=8E+ddiff=E5=B7=AE=E5=BC=82=E5=AF=B9=E8=B4=A6+dbackup?= =?UTF-8?q?=E5=A2=9E=E9=87=8F=E5=A4=87=E4=BB=BD+=E5=9B=9B=E6=AD=A5?= =?UTF-8?q?=E8=A7=84=E8=8C=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - METHODOLOGY.md: 备份/操作/验证/回退四大步操作规范 - dmig.py: 迁移引擎(台账表pipeline_deploy_ledger+幂等+破坏性守卫+down回退) - ddiff.py: 环境schema/参数/模块版本差异对账 - dbackup.py: mysqldump增量表备份+恢复命令生成 - migrations/m0001-m0004: 首批迁移(今日发现的生产库缺口) --- deploy/METHODOLOGY.md | 162 +++++++++ deploy/dbackup.py | 172 ++++++++++ deploy/ddiff.py | 180 ++++++++++ deploy/dmig.py | 316 ++++++++++++++++++ .../m0001_sd_projects_dir_model.json | 13 + .../m0002_pipeline_agent_instances.json | 12 + .../m0003_conversations_session_id.json | 11 + .../m0004_deliverables_created_by_widen.json | 11 + 8 files changed, 877 insertions(+) create mode 100644 deploy/METHODOLOGY.md create mode 100755 deploy/dbackup.py create mode 100755 deploy/ddiff.py create mode 100755 deploy/dmig.py create mode 100644 deploy/migrations/m0001_sd_projects_dir_model.json create mode 100644 deploy/migrations/m0002_pipeline_agent_instances.json create mode 100644 deploy/migrations/m0003_conversations_session_id.json create mode 100644 deploy/migrations/m0004_deliverables_created_by_widen.json diff --git a/deploy/METHODOLOGY.md b/deploy/METHODOLOGY.md new file mode 100644 index 0000000..0a3c794 --- /dev/null +++ b/deploy/METHODOLOGY.md @@ -0,0 +1,162 @@ +# 测试 → 生产 增量部署规范 + +> 原则：生产上线永远是增量，没有全量。每次上线 = 一批「增量单元」（代码 + 数据库变更 + 配置）， +> 每个单元必须可幂等重放、可验证、可回退。本规范固化四步：备份 → 操作 → 验证 → 故障回退。 + +## 0. 目录约定（在 pipeline-app 仓库） + +``` +deploy/ +├── METHODOLOGY.md # 本文档 +├── dmig.py # 迁移引擎：台账表 + up/down + 幂等 +├── ddiff.py # 环境差异比对：代码模块版本 / schema / 参数 +├── dbackup.py # 增量表备份（mysqldump）+ 恢复命令生成 +└── migrations/ + ├── m0001_xxx.json # 迁移单元（编号递增，字典序 = 执行序） + └── ... +``` + +## 1. 增量单元（一次上线携带什么） + +每次从测试推向生产，变更分成三类，各走各的通道，**不允许混**： + +| 类型 | 载体 | 通道 | +|---|---|---| +| 代码 | 各模块仓库提交 | 生产机 `git pull --ff-only` + `pip install`（非 editable 模块）+ 清 `__pycache__` + 重启 | +| 数据库变更 | `deploy/migrations/mNNNN_*.json` | `dmig.py`（测试先跑 → 生产复核后执行） | +| 系统配置 | `params` 表 / 环境变量 | 走 `dmig.py` 的 `params_set`/`env` 类型，或记录在迁移说明里人工核对 | + +**铁律**： +1. 测试环境执行过的库变更，必须在提交测试代码的同时登记成迁移文件——**没有登记就没有测试通过**。 +2. 生产执行迁移前必须先跑 `ddiff.py`，拿差异清单与迁移批次对照，**清单里没有对应迁移的差异 = 遗漏**。 +3. 破坏性操作（DROP/DELETE/TRUNCATE）永远生成命令由人工在生产执行，自动化只做到「打印 [NEEDS_APPROVAL]」。 + +## 2. 迁移单元格式（migrations/mNNNN_名称.json） + +```json +{ + "id": "m0001", + "title": "说明（一句话：这次改了什么、为什么）", + "backup_tables": ["sd_projects"], + "up": [ + {"op": "sql", "sql": "ALTER TABLE ..."}, + {"op": "create_table", "name": "xxx", "columns": "...完整 CREATE TABLE 语句..."}, + {"op": "params_set", "key": "tender_api_base", "value": "http://..."} + ], + "down": [ + {"op": "sql", "sql": "DROP TABLE xxx"}, + {"op": "params_del", "key": "tender_api_base"} + ] +} +``` + +规则： +- `id` 唯一且递增；执行顺序 = 文件名字典序。 +- `up` 的每一步都必须**幂等**：`CREATE TABLE IF NOT EXISTS`、`ALTER TABLE ... ADD COLUMN IF NOT EXISTS`（MySQL 8 用先查 information_schema 的方式，dmig 会自动做存在性检查）、`INSERT ... ON DUPLICATE KEY UPDATE`。 +- `down` 是 `up` 的精确逆操作；无逆操作的步骤（如纯数据修复）`down` 写空数组并在 title 注明「不可逆」。 +- `backup_tables` 列出本次会改动的表——`dmig.py apply` 前自动备份这些表（生产备份由人工确认后执行）。 + +## 3. 四步流程 + +### 第 1 步：数据备份（上线前必做，不可跳） + +```bash +# 生产机执行。dmig 会自动列出本次批次涉及的表；也可手工指定。 +cd /d/doit/pipeline-app +./py3/bin/python deploy/dbackup.py backup --batch # 备份本批次所有 backup_tables +./py3/bin/python deploy/dbackup.py backup --tables sd_projects,pipeline_tasks # 手工补充 +# 备份文件：/d/doit/pipeline-app/backups/__<日期时间>.sql，同目录生成 RESTORE 命令文件 +``` + +检查点： +- [ ] 备份文件存在且非空（`ls -lh backups/`） +- [ ] 备份文件头部有 `CREATE TABLE` 且行数与线上一致（`dbackup.py verify `） + +### 第 2 步：操作（按序执行增量） + +```bash +# 2.1 差异核对（必做）：打印生产与测试的差异清单，与本次批次对照 +./py3/bin/python deploy/ddiff.py --checklist + +# 2.2 代码增量（先代码后库，或按迁移说明顺序） +cd /d/doit/pipeline-app/pkgs/ && git pull --ff-only origin main +cd /d/doit/pipeline-app && ./py3/bin/pip install pkgs/ +find py3/lib/python3.10/site-packages/ -name __pycache__ -exec rm -rf {} + +nohup bash restart-pipeline.sh logs/restart.log 2>&1 & +sleep 15; curl -s -o /dev/null -w "health:%{http_code}\n" http://127.0.0.1:9090/ + +# 2.3 数据库增量（迁移引擎） +./py3/bin/python deploy/dmig.py status # 看台账：哪些已执行、哪些待执行 +./py3/bin/python deploy/dmig.py plan # 预演：打印本批次将执行的 SQL（不落库） +./py3/bin/python deploy/dmig.py apply # 执行（自动先备份本批次表，逐条落库，逐条记账） +``` + +`dmig.py apply` 的行为： +- 每条语句执行前查存在性（表/列/索引已有 → 跳过，保证幂等）； +- 执行成功才写台账 `pipeline_deploy_ledger`（批次、步骤、状态、时间）； +- 任一步失败 → 停止后续步骤，打印已完成步骤清单和对应 `down` 操作（供回退），**不自动回滚**。 + +### 第 3 步：验证（不验证 = 没上线） + +```bash +./py3/bin/python deploy/ddiff.py --checklist # 差异清零（或只剩本批次明确豁免项） +./py3/bin/python deploy/dmig.py status # 台账：本批次全部 applied +curl -s -o /dev/null -w "health:%{http_code}\n" http://127.0.0.1:9090/ # 服务 200 +grep -a "poller started" logs/pipeline.log | tail -5 # 各 poller 正常 +# + 功能验收：真实浏览器走一遍本次变更影响的功能（铁律：功能验收必须真浏览器执行） +``` + +检查点： +- [ ] ddiff 差异清零 +- [ ] dmig status 全 applied、无 failed +- [ ] 服务健康 + poller 正常 +- [ ] 功能浏览器实测通过（记录证据：页面/请求/数据三选一） + +### 第 4 步：故障回退（任一步失败时） + +**判断**：回退的是「这次增量」，不是整个系统。定位到失败的批次号。 + +```bash +# 4.1 库回退：用迁移自带的 down +./py3/bin/python deploy/dmig.py rollback m0003 # 执行 m0003 的 down + 台账置 rolled_back + +# 4.2 数据回退：down 不足以恢复时，用第 1 步的备份 +./py3/bin/python deploy/dbackup.py restore --file backups/pipeline_sd_projects_20260901_120000.sql --confirm +# 注意：恢复会先 DROP 再导入该表——执行前人工确认没有比备份点更新的写入（或先导出增量） + +# 4.3 代码回退 +cd /d/doit/pipeline-app/pkgs/ && git log --oneline -5 # 找到上线前的提交 +git checkout <上线前提交> && cd ../.. && ./py3/bin/pip install pkgs/ && 重启 +``` + +回退后重新走第 3 步验证，确认系统回到上线前状态。 + +## 4. 台账表（自动创建） + +```sql +CREATE TABLE IF NOT EXISTS pipeline_deploy_ledger ( + id VARCHAR(32) NOT NULL, + batch VARCHAR(16) NOT NULL, -- 迁移 id：m0001 + step INT NOT NULL, -- up 内步骤序号 + status VARCHAR(16) NOT NULL, -- applied / rolled_back / skipped + backup_file VARCHAR(255), + detail VARCHAR(4000), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_ledger_batch_step (batch, step) +); +``` + +## 5. 测试环境的角色 + +测试环境是**迁移的演练场**：每个迁移文件必须在测试环境 `dmig.py apply` 跑通（含幂等重跑验证） +才允许进入生产批次。测试环境的台账独立（同库不同环境实例），生产台账只反映生产执行记录。 + +## 6. 常见坑（历史教训） + +1. **改库没登记** → 生产是老库。8/31-9/1 连续发现：`pipeline_agent_instances` 缺表、 + `sd_projects.directory_name` 缺列、`pipeline_deliverables.created_by` 还是 32 宽。 + 根治就是本规范：改库必须落成迁移文件。 +2. **只在测试跑 DDL 口头说"生产一样"** → 必然漂移。用 `ddiff.py` 强制对账。 +3. **回退没有 down** → 不可逆变更（DROP 列、删数据）必须有备份兜底，备份在操作前。 +4. **破坏性语句进自动化** → 永远 `[NEEDS_APPROVAL]`，人工执行。 diff --git a/deploy/dbackup.py b/deploy/dbackup.py new file mode 100755 index 0000000..33f5711 --- /dev/null +++ b/deploy/dbackup.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""dbackup.py — 增量表备份与恢复命令生成（四步流程第 1 步/第 4 步配套）。 + +用法（在应用根目录执行）： + ./py3/bin/python deploy/dbackup.py backup --tables sd_projects,pipeline_tasks + ./py3/bin/python deploy/dbackup.py backup --batch # 备份待执行迁移批次声明的所有表 + ./py3/bin/python deploy/dbackup.py verify <备份文件> # 校验备份文件完整性 + ./py3/bin/python deploy/dbackup.py restore --file <备份文件> --confirm # 生成并执行恢复（危险） + +备份位置：/backups/<库名>_<表名>_.sql +恢复 = DROP TABLE + 导入，执行前打印 [NEEDS_APPROVAL]，必须 --confirm 才真执行。 +""" + +import json +import os +import subprocess +import sys +from datetime import datetime + +APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +BACKUP_DIR = os.path.join(APP_ROOT, "backups") +MIG_DIR = os.path.join(APP_ROOT, "deploy", "migrations") + + +def get_db_conf(): + sys.path.insert(0, APP_ROOT) + try: + from appPublic.jsonConfig import getConfig + from appPublic.aes import aes_decode_b64 + cfg = getConfig(APP_ROOT, {"workdir": APP_ROOT}) + kw = cfg.databases["pipeline"].kwargs + pwd = aes_decode_b64(cfg.password_key, kw.password) + return str(kw.host), str(kw.port), str(kw.user), pwd, str(kw.db) + except Exception as e: + print("[FATAL] 读配置失败:", e) + sys.exit(1) + + +def table_exists(conf, name): + host, port, user, pwd, db = conf + r = subprocess.run(["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd, + db, "-N", "-e", "SHOW TABLES LIKE '%s'" % name], + capture_output=True, text=True) + return bool((r.stdout or "").strip()) + + +def batch_tables(): + """收集所有待执行迁移批次声明的 backup_tables（去重）。""" + # 读台账判断哪些批次已执行（简化：直接扫所有迁移文件的 backup_tables 并集） + tables = [] + if os.path.isdir(MIG_DIR): + for f in sorted(os.listdir(MIG_DIR)): + if not f.endswith(".json"): + continue + try: + d = json.load(open(os.path.join(MIG_DIR, f))) + for t in d.get("backup_tables") or []: + if t not in tables: + tables.append(t) + except Exception: + pass + return tables + + +def backup(conf, tables): + host, port, user, pwd, db = conf + os.makedirs(BACKUP_DIR, exist_ok=True) + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + done = [] + for t in tables: + if not table_exists(conf, t): + print(" 跳过(表不存在): %s" % t) + continue + out_file = os.path.join(BACKUP_DIR, "%s_%s_%s.sql" % (db, t, ts)) + r = subprocess.run( + ["mysqldump", "-h", host, "-P", port, "-u", user, "-p" + pwd, + db, t, "--single-transaction", "--skip-lock-tables", + "--result-file", out_file], + capture_output=True, text=True) + err = (r.stderr or "").replace( + "mysqldump: [Warning] Using a password on the command line interface can be insecure.\n", "") + if r.returncode != 0 or not os.path.isfile(out_file) or os.path.getsize(out_file) == 0: + print("[FAIL] %s: %s" % (t, err[:200])) + sys.exit(1) + size = os.path.getsize(out_file) + print(" OK %s (%d bytes)" % (out_file, size)) + done.append(out_file) + # 生成对应的恢复命令文件 + if done: + restore_file = os.path.join(BACKUP_DIR, "RESTORE_%s.sh" % ts) + with open(restore_file, "w") as f: + f.write("#!/bin/bash\n# 恢复命令（危险！先人工核对）\n") + for bf in done: + f.write('mysql -h%s -P%s -u%s -p"$PASS" %s < %s\n' % (host, port, user, db, bf)) + os.chmod(restore_file, 0o700) + print(" 恢复命令: %s" % restore_file) + return done + + +def verify(conf, path): + if not os.path.isfile(path): + print("[FAIL] 文件不存在: %s" % path) + sys.exit(1) + size = os.path.getsize(path) + head = open(path, encoding="utf-8", errors="replace").read(4000) + has_create = "CREATE TABLE" in head + # 尾部完整性标记 + tail = subprocess.run(["tail", "-c", "200", path], capture_output=True, text=True).stdout + has_dump_info = "Dump completed" in tail + print("文件: %s (%d bytes)" % (path, size)) + print(" 含 CREATE TABLE: %s" % ("是" if has_create else "否[异常]")) + print(" 含 Dump completed: %s" % ("是" if has_dump_info else "否[可能被截断]")) + if not (has_create and has_dump_info and size > 0): + sys.exit(1) + print(" ✅ 备份文件完整") + + +def restore(conf, path): + """恢复 = 从备份文件重建表。打印命令，--confirm 才执行。""" + host, port, user, pwd, db = conf + if not os.path.isfile(path): + print("[FAIL] 文件不存在: %s" % path) + sys.exit(1) + print("[NEEDS_APPROVAL] 恢复操作将 DROP 并重建备份中的表！") + print(" 备份文件: %s" % path) + print(" 目标库: %s@%s" % (db, host)) + if "--confirm" not in sys.argv: + print(" （预演模式，未执行。确认后追加 --confirm）") + print(" 手工命令: mysql -h%s -P%s -u%s -p<密码> %s < %s" % (host, port, user, db, path)) + return + r = subprocess.run(["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd, db], + stdin=open(path), capture_output=True, text=True) + err = (r.stderr or "").replace( + "mysql: [Warning] Using a password on the command line interface can be insecure.\n", "") + if r.returncode != 0: + print("[FAIL] 恢复失败: %s" % err[:300]) + sys.exit(1) + print(" ✅ 恢复完成") + + +def main(): + if len(sys.argv) < 2 or sys.argv[1] not in ("backup", "verify", "restore"): + print(__doc__) + sys.exit(1) + conf = get_db_conf() + cmd = sys.argv[1] + if cmd == "backup": + tables = [] + if "--tables" in sys.argv: + tables = [t.strip() for t in sys.argv[sys.argv.index("--tables") + 1].split(",") if t.strip()] + elif "--batch" in sys.argv: + tables = batch_tables() + if not tables: + print("无表可备份（--tables a,b,c 或 --batch）") + sys.exit(1) + print("备份 %d 张表: %s" % (len(tables), ",".join(tables))) + backup(conf, tables) + elif cmd == "verify": + if len(sys.argv) < 3: + print("verify 需要备份文件路径") + sys.exit(1) + verify(conf, sys.argv[2]) + elif cmd == "restore": + if "--file" not in sys.argv: + print("restore 需要 --file <备份文件>") + sys.exit(1) + restore(conf, sys.argv[sys.argv.index("--file") + 1]) + + +if __name__ == "__main__": + main() diff --git a/deploy/ddiff.py b/deploy/ddiff.py new file mode 100755 index 0000000..838b224 --- /dev/null +++ b/deploy/ddiff.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""ddiff.py — 测试/生产环境差异比对（增量部署的"对账单"）。 + +用法： + # 在生产机（或测试机）执行，拉取本环境快照后与对照快照比较： + ./py3/bin/python deploy/ddiff.py snapshot > /tmp/schema_here.json # 导出本环境快照 + ./py3/bin/python deploy/ddiff.py compare --against /tmp/schema_test.json + # 打印差异清单：缺表/缺列/缺索引/参数差异/模块git版本差异 + + # 一步到位：先在对照机跑 snapshot 存文件，scp 过来，再 compare。 + # 每次生产上线前：先 snapshot 测试环境 → compare 生产，差异必须全部有对应迁移或豁免。 + +设计原则： + - 只读：本工具只查 information_schema / params / git log，绝不写库。 + - 输出面向人工核对：按 类型 分组，标注 [需迁移] / [可忽略]。 +""" + +import json +import os +import subprocess +import sys + +APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +PKGS_DIR = os.path.join(APP_ROOT, "pkgs") + + +def get_db_conf(): + sys.path.insert(0, APP_ROOT) + try: + from appPublic.jsonConfig import getConfig + from appPublic.aes import aes_decode_b64 + cfg = getConfig(APP_ROOT, {"workdir": APP_ROOT}) + kw = cfg.databases["pipeline"].kwargs + pwd = aes_decode_b64(cfg.password_key, kw.password) + return str(kw.host), str(kw.port), str(kw.user), pwd, str(kw.db) + except Exception as e: + print("[FATAL] 读配置失败:", e) + sys.exit(1) + + +def mysql(conf, sql): + host, port, user, pwd, db = conf + r = subprocess.run(["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd, + db, "-N", "-e", sql], capture_output=True, text=True) + return r.stdout if r.returncode == 0 else "" + + +def snapshot(conf): + _, _, _, _, db = conf + s = {} + s["tables"] = sorted(mysql(conf, "SHOW TABLES").split()) + s["cols"] = sorted(x for x in mysql(conf, + "SELECT CONCAT(table_name,'|',column_name,'|',column_type) FROM information_schema.columns " + "WHERE table_schema='%s'" % db).split("\n") if x) + s["idx"] = sorted(x for x in mysql(conf, + "SELECT CONCAT(table_name,'|',index_name,'|',GROUP_CONCAT(column_name ORDER BY seq_in_index)) " + "FROM information_schema.statistics WHERE table_schema='%s' " + "GROUP BY table_name,index_name" % db).split("\n") if x) + s["params"] = sorted(x for x in mysql(conf, + "SELECT CONCAT(params_name,'=',LEFT(params_value,60)) FROM params").split("\n") if x) + # 模块版本（pkgs 下各仓库的 HEAD） + mods = {} + if os.path.isdir(PKGS_DIR): + for d in sorted(os.listdir(PKGS_DIR)): + gitdir = os.path.join(PKGS_DIR, d, ".git") + if os.path.isdir(gitdir): + r = subprocess.run(["git", "-C", os.path.join(PKGS_DIR, d), + "log", "--oneline", "-1"], capture_output=True, text=True) + mods[d] = r.stdout.strip() + s["modules"] = mods + return s + + +def compare(base, target): + """base = 对照环境(测试), target = 本环境(生产)。打印 base 有 / target 无 的缺口。""" + print("═══ 差异清单（对照环境有 / 本环境缺）═══\n") + + bt, tt = set(base["tables"]), set(target["tables"]) + missing_tables = sorted(bt - tt) + extra_tables = sorted(tt - bt) + print("── 缺表 [%d] %s" % (len(missing_tables), "【需迁移】" if missing_tables else "")) + for t in missing_tables: + print(" -", t) + if extra_tables: + print(" （本环境多出的表，通常可忽略）:", ", ".join(extra_tables[:10])) + + bc, tc = set(base["cols"]), set(target["cols"]) + from collections import defaultdict + miss_cols = defaultdict(list) + for c in bc - tc: + p = c.split("|") + if len(p) == 3: + miss_cols[p[0]].append(p[1] + " " + p[2]) + print("\n── 缺列 [%d 表] %s" % (len(miss_cols), "【需迁移】" if miss_cols else "")) + for t in sorted(miss_cols): + print(" %s: %s" % (t, ", ".join(sorted(miss_cols[t])))) + # 列宽/类型不同（同名不同型） + base_colmap = {} + for c in base["cols"]: + p = c.split("|") + if len(p) == 3: + base_colmap[(p[0], p[1])] = p[2] + type_diff = [] + for c in target["cols"]: + p = c.split("|") + if len(p) == 3 and (p[0], p[1]) in base_colmap: + bt2 = base_colmap[(p[0], p[1])] + if bt2 != p[2]: + type_diff.append("%s.%s: 对照=%s 本环境=%s" % (p[0], p[1], bt2, p[2])) + print("\n── 列类型不同 [%d] %s" % (len(type_diff), "【需迁移】" if type_diff else "")) + for d in type_diff: + print(" ", d) + + bi, ti = set(base["idx"]), set(target["idx"]) + miss_idx = defaultdict(list) + for i in bi - ti: + p = i.split("|") + if len(p) == 3: + miss_idx[p[0]].append(p[1] + "(" + p[2] + ")") + # 名字不同但列相同的索引视为等价（如 uk_project_role vs uk_sd_project_role_models_pr） + target_idxcols = {} + for i in target["idx"]: + p = i.split("|") + if len(p) == 3: + target_idxcols.setdefault(p[0], set()).add(p[2]) + real_miss = {} + for t, lst in miss_idx.items(): + real = [x for x in lst if x.split("(")[1].rstrip(")") not in target_idxcols.get(t, set())] + if real: + real_miss[t] = real + print("\n── 缺索引 [%d 表]（已排除列等价的改名索引）%s" + % (len(real_miss), "【建议迁移】" if real_miss else "")) + for t in sorted(real_miss): + print(" %s: %s" % (t, ", ".join(sorted(real_miss[t])[:8]))) + + bp, tp = set(base["params"]), set(target["params"]) + miss_params = sorted(bp - tp) + print("\n── 缺参数/参数值不同 [%d] %s" % (len(miss_params), "【需迁移】" if miss_params else "")) + for p in miss_params: + print(" -", p[:80]) + + bm, tm = base.get("modules", {}), target.get("modules", {}) + mod_diff = [] + for k in sorted(set(bm) | set(tm)): + if bm.get(k) != tm.get(k): + mod_diff.append(" %s:\n 对照: %s\n 本环境: %s" % (k, bm.get(k, "(无)"), tm.get(k, "(无)"))) + print("\n── 模块版本不同 [%d] %s" % (len(mod_diff), "【需 pull+install】" if mod_diff else "")) + for d in mod_diff: + print(d) + + total = len(missing_tables) + len(miss_cols) + len(type_diff) + len(miss_params) + len(mod_diff) + print("\n═══ 汇总：需处理 %d 项（缺表 %d / 缺列 %d / 类型不同 %d / 参数 %d / 模块 %d），" + "缺索引 %d 建议项 ═══" + % (total, len(missing_tables), len(miss_cols), len(type_diff), + len(miss_params), len(mod_diff), len(real_miss))) + return total + + +def main(): + if len(sys.argv) < 2 or sys.argv[1] not in ("snapshot", "compare"): + print(__doc__) + sys.exit(1) + conf = get_db_conf() + if sys.argv[1] == "snapshot": + print(json.dumps(snapshot(conf), ensure_ascii=False, indent=1)) + return + # compare + if "--against" not in sys.argv: + print("compare 需要 --against <对照快照.json>") + sys.exit(1) + path = sys.argv[sys.argv.index("--against") + 1] + base = json.load(open(path)) + target = snapshot(conf) + total = compare(base, target) + sys.exit(0 if total == 0 else 2) + + +if __name__ == "__main__": + main() diff --git a/deploy/dmig.py b/deploy/dmig.py new file mode 100755 index 0000000..01fe693 --- /dev/null +++ b/deploy/dmig.py @@ -0,0 +1,316 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""dmig.py — 数据库增量迁移引擎（测试→生产增量部署规范配套工具）。 + +用法（在应用根目录执行，用应用 venv）： + ./py3/bin/python deploy/dmig.py status # 台账 + 待执行批次 + ./py3/bin/python deploy/dmig.py plan [batch ...] # 预演：打印将执行的语句（不落库） + ./py3/bin/python deploy/dmig.py apply [batch ...] # 执行（默认全部待执行；先自动备份涉及表） + ./py3/bin/python deploy/dmig.py rollback # 回退：执行该批次 down + 台账置 rolled_back + +迁移文件：deploy/migrations/mNNNN_名称.json（字典序 = 执行序），格式见 METHODOLOGY.md。 +设计原则： + - 幂等：每步执行前查存在性（表/列/索引/参数），已有则跳过；重跑安全。 + - 台账：pipeline_deploy_ledger 记录每步 applied/skipped/rolled_back。 + - 破坏性守卫：up 里的 DROP TABLE / TRUNCATE / DELETE 打印 [NEEDS_APPROVAL] 并中止， + 需人工执行（--force 也无效，这是有意设计）。 +""" + +import json +import os +import re +import subprocess +import sys +from datetime import datetime + +APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +MIG_DIR = os.path.join(APP_ROOT, "deploy", "migrations") +LEDGER_TABLE = "pipeline_deploy_ledger" +DESTRUCTIVE = re.compile(r"\b(DROP\s+TABLE|TRUNCATE|DELETE\s+FROM)\b", re.I) + +LEDGER_DDL = """CREATE TABLE IF NOT EXISTS %s ( + id VARCHAR(32) NOT NULL, + batch VARCHAR(16) NOT NULL, + step INT NOT NULL, + status VARCHAR(16) NOT NULL, + backup_file VARCHAR(255), + detail VARCHAR(4000), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uk_ledger_batch_step (batch, step) +) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci""" % LEDGER_TABLE + + +def die(msg): + print("[FATAL]", msg) + sys.exit(1) + + +def get_db_conf(): + """从应用配置读 pipeline 库连接（复用应用 venv 的 appPublic 解密）。""" + sys.path.insert(0, APP_ROOT) + conf_path = os.path.join(APP_ROOT, "conf", "config.json") + if not os.path.isfile(conf_path): + die("找不到 %s（必须在应用根目录的 venv 里跑）" % conf_path) + try: + from appPublic.jsonConfig import getConfig + from appPublic.aes import aes_decode_b64 + cfg = getConfig(APP_ROOT, {"workdir": APP_ROOT}) + kw = cfg.databases["pipeline"].kwargs + pwd = aes_decode_b64(cfg.password_key, kw.password) + return str(kw.host), str(kw.port), str(kw.user), pwd, str(kw.db) + except Exception as e: + die("读配置失败: %s" % e) + + +def mysql(conf, sql, database=None): + host, port, user, pwd, db = conf + cmd = ["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd, + database or db, "-N", "-e", sql] + r = subprocess.run(cmd, capture_output=True, text=True) + err = (r.stderr or "").replace( + "mysql: [Warning] Using a password on the command line interface can be insecure.\n", "") + if r.returncode != 0: + return None, err.strip()[:300] + return r.stdout, None + + +def table_exists(conf, name): + out, _ = mysql(conf, "SHOW TABLES LIKE '%s'" % name) + return bool((out or "").strip()) + + +def column_exists(conf, table, col): + _, _, _, _, db = conf + out, _ = mysql(conf, "SELECT 1 FROM information_schema.columns WHERE table_schema='%s' " + "AND table_name='%s' AND column_name='%s'" % (db, table, col)) + return bool((out or "").strip()) + + +def index_exists(conf, table, idx): + _, _, _, _, db = conf + out, _ = mysql(conf, "SELECT 1 FROM information_schema.statistics WHERE table_schema='%s' " + "AND table_name='%s' AND index_name='%s'" % (db, table, idx)) + return bool((out or "").strip()) + + +def ensure_ledger(conf): + out, err = mysql(conf, LEDGER_DDL) + if err: + die("建台账表失败: %s" % err) + + +def ledger_state(conf): + out, err = mysql(conf, "SELECT batch, MAX(step) FROM %s WHERE status='applied' " + "GROUP BY batch" % LEDGER_TABLE) + applied = {} + for ln in (out or "").splitlines(): + p = ln.split("\t") + if len(p) == 2: + applied[p[0]] = int(p[1]) + return applied + + +def load_migrations(): + migs = {} + if not os.path.isdir(MIG_DIR): + return migs + for f in sorted(os.listdir(MIG_DIR)): + if not f.endswith(".json"): + continue + try: + d = json.load(open(os.path.join(MIG_DIR, f))) + d["_file"] = f + migs[d["id"]] = d + except Exception as e: + die("迁移文件解析失败 %s: %s" % (f, e)) + return migs + + +def render_step(conf, step, direction): + """把一个 op 渲染成 (sql 列表, skip_reason|None)。幂等判断在这里。""" + op = step.get("op", "") + if op == "sql": + sql = step["sql"].strip().rstrip(";") + m = re.match(r"CREATE\s+TABLE\s+(IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?", sql, re.I) + if m: + if table_exists(conf, m.group(2)): + return [], "表已存在: %s" % m.group(2) + m = re.match(r"ALTER\s+TABLE\s+`?(\w+)`?\s+ADD\s+(COLUMN\s+)?`?(\w+)`?", sql, re.I) + if m: + if column_exists(conf, m.group(1), m.group(3)): + return [], "列已存在: %s.%s" % (m.group(1), m.group(3)) + m = re.match(r"CREATE\s+(UNIQUE\s+)?INDEX\s+`?(\w+)`?\s+ON\s+`?(\w+)`?", sql, re.I) + if m: + if index_exists(conf, m.group(3), m.group(2)): + return [], "索引已存在: %s.%s" % (m.group(3), m.group(2)) + return [sql], None + if op == "create_table": + name = step["name"] + if table_exists(conf, name): + return [], "表已存在: %s" % name + return [step["sql"].strip().rstrip(";")], None + if op == "params_set": + out, _ = mysql(conf, "SELECT id FROM params WHERE params_name='%s'" % step["key"]) + val = str(step["value"]).replace("'", "''") + if (out or "").strip(): + return ["UPDATE params SET params_value='%s' WHERE params_name='%s'" + % (val, step["key"])], None + import uuid + return ["INSERT INTO params (id, params_name, params_value) " + "VALUES ('%s', '%s', '%s')" % (uuid.uuid4().hex, step["key"], val)], None + if op == "params_del": + return ["DELETE FROM params WHERE params_name='%s'" % step["key"]], None + die("未知 op: %s" % op) + + +def backup_tables(conf, tables): + if not tables: + return "" + r = subprocess.run( + [sys.executable, os.path.join(APP_ROOT, "deploy", "dbackup.py"), + "backup", "--tables", ",".join(tables)], + capture_output=True, text=True, cwd=APP_ROOT) + if r.returncode != 0: + die("备份失败（中止迁移）: %s" % (r.stderr or r.stdout)[-300:]) + return (r.stdout or "").strip().splitlines()[-1] if r.stdout.strip() else "" + + +def cmd_status(conf): + ensure_ledger(conf) + applied = ledger_state(conf) + migs = load_migrations() + print("=== 迁移台账 ===") + for mid in sorted(migs): + m = migs[mid] + st = "applied(步骤%d)" % applied[mid] if mid in applied else "待执行" + print(" %s %-10s %s" % (mid, st, m.get("title", "")[:50])) + print("\n待执行批次数:", len([m for m in migs if m not in applied])) + + +def cmd_plan(conf, batches): + ensure_ledger(conf) + applied = ledger_state(conf) + migs = load_migrations() + targets = [b for b in (batches or sorted(migs)) if b not in applied] + if not targets: + print("无待执行批次") + return + for mid in targets: + m = migs[mid] + print("\n── %s %s" % (mid, m.get("title", ""))) + if m.get("backup_tables"): + print(" [自动备份表]: %s" % ",".join(m["backup_tables"])) + for i, step in enumerate(m.get("up", []), 1): + sqls, skip = render_step(conf, step, "up") + if skip: + print(" 步骤%d 跳过(%s)" % (i, skip)) + else: + for s in sqls: + mark = "[NEEDS_APPROVAL] " if DESTRUCTIVE.search(s) else "" + print(" 步骤%d %s%s" % (i, mark, s[:150])) + + +def cmd_apply(conf, batches): + ensure_ledger(conf) + applied = ledger_state(conf) + migs = load_migrations() + targets = [b for b in (batches or sorted(migs)) if b not in applied] + if not targets: + print("无待执行批次") + return + _, _, _, _, db = conf + for mid in targets: + m = migs[mid] + print("\n══ 应用 %s: %s" % (mid, m.get("title", ""))) + # 破坏性守卫（先扫全部语句再动手） + for step in m.get("up", []): + sqls, skip = render_step(conf, step, "up") + if skip: + continue + for s in sqls: + if DESTRUCTIVE.search(s): + die("up 含破坏性语句，拒绝自动执行（人工复核后手工跑）:\n %s" % s[:200]) + # 备份 + bk = backup_tables(conf, m.get("backup_tables") or []) + # 逐步执行 + 记账 + for i, step in enumerate(m.get("up", []), 1): + sqls, skip = render_step(conf, step, "up") + if skip: + mysql(conf, "INSERT INTO %s (id, batch, step, status, detail) " + "VALUES ('%s', '%s', %d, 'skipped', '%s') " + "ON DUPLICATE KEY UPDATE status='skipped'" + % (LEDGER_TABLE, _new_id(), mid, i, skip[:200])) + print(" 步骤%d 跳过(%s)" % (i, skip)) + continue + for s in sqls: + out, err = mysql(conf, s) + if err: + mysql(conf, "INSERT INTO %s (id, batch, step, status, detail) " + "VALUES ('%s', '%s', %d, 'failed', '%s')" + % (LEDGER_TABLE, _new_id(), mid, i, err[:200])) + die("步骤%d 失败: %s\n（已完成步骤可用 `dmig.py rollback %s` 回退）" + % (i, err, mid)) + print(" 步骤%d OK: %s" % (i, s[:90])) + mysql(conf, "INSERT INTO %s (id, batch, step, status, backup_file, detail) " + "VALUES ('%s', '%s', %d, 'applied', '%s', '%s')" + % (LEDGER_TABLE, _new_id(), mid, i, bk[:200], str(step)[:200])) + print(" ✅ %s 完成" % mid) + print("\n全部完成。验证: ./py3/bin/python deploy/dmig.py status && ddiff.py --checklist") + + +def cmd_rollback(conf, batch): + ensure_ledger(conf) + applied = ledger_state(conf) + migs = load_migrations() + if batch not in migs: + die("迁移 %s 不存在" % batch) + if batch not in applied: + print("台账中 %s 未执行过，无需回退" % batch) + return + m = migs[batch] + downs = m.get("down", []) + if not downs: + die("%s 无 down 定义（不可自动回退，用 dbackup.py restore 恢复备份）" % batch) + print("══ 回退 %s（down %d 步，倒序）" % (batch, len(downs))) + for i, step in enumerate(reversed(downs), 1): + sqls, skip = render_step(conf, step, "down") + if skip: + print(" 回退%d 跳过(%s)" % (i, skip)) + continue + for s in sqls: + out, err = mysql(conf, s) + if err: + die("回退%d 失败: %s（剩余步骤请人工处理）" % (i, err)) + print(" 回退%d OK: %s" % (i, s[:90])) + mysql(conf, "UPDATE %s SET status='rolled_back' WHERE batch='%s' AND status='applied'" + % (LEDGER_TABLE, batch)) + print("✅ %s 已回退（台账已标记）" % batch) + + +def _new_id(): + import uuid + return uuid.uuid4().hex + + +def main(): + if len(sys.argv) < 2 or sys.argv[1] not in ("status", "plan", "apply", "rollback"): + print(__doc__) + sys.exit(1) + conf = get_db_conf() + cmd = sys.argv[1] + args = sys.argv[2:] + if cmd == "status": + cmd_status(conf) + elif cmd == "plan": + cmd_plan(conf, args) + elif cmd == "apply": + cmd_apply(conf, args) + elif cmd == "rollback": + if not args: + die("rollback 需要批次号: dmig.py rollback m0001") + cmd_rollback(conf, args[0]) + + +if __name__ == "__main__": + main() diff --git a/deploy/migrations/m0001_sd_projects_dir_model.json b/deploy/migrations/m0001_sd_projects_dir_model.json new file mode 100644 index 0000000..a1bc5cb --- /dev/null +++ b/deploy/migrations/m0001_sd_projects_dir_model.json @@ -0,0 +1,13 @@ +{ + "id": "m0001", + "title": "sd_projects 补 directory_name + default_model 列（项目目录统一机制依赖，9/1 测试已上线）", + "backup_tables": ["sd_projects"], + "up": [ + {"op": "sql", "sql": "ALTER TABLE sd_projects ADD COLUMN directory_name VARCHAR(200) NULL COMMENT '项目目录名(创建时定死,解析只读此字段)'"}, + {"op": "sql", "sql": "ALTER TABLE sd_projects ADD COLUMN default_model VARCHAR(128) NULL COMMENT '项目默认模型'"} + ], + "down": [ + {"op": "sql", "sql": "ALTER TABLE sd_projects DROP COLUMN default_model"}, + {"op": "sql", "sql": "ALTER TABLE sd_projects DROP COLUMN directory_name"} + ] +} diff --git a/deploy/migrations/m0002_pipeline_agent_instances.json b/deploy/migrations/m0002_pipeline_agent_instances.json new file mode 100644 index 0000000..8dc6d41 --- /dev/null +++ b/deploy/migrations/m0002_pipeline_agent_instances.json @@ -0,0 +1,12 @@ +{ + "id": "m0002", + "title": "建 pipeline_agent_instances 表（poller 实例 id 机制，9/1 修复 1406/审计根因）", + "backup_tables": [], + "up": [ + {"op": "sql", "sql": "CREATE TABLE IF NOT EXISTS pipeline_agent_instances (id VARCHAR(32) NOT NULL COMMENT '主键ID', project_id VARCHAR(32) NOT NULL DEFAULT '' COMMENT '项目ID', role VARCHAR(64) NOT NULL COMMENT '角色(agent.xxx)', agent_id VARCHAR(32) NOT NULL COMMENT '实例ID(ag.xxxx)', status VARCHAR(16) NOT NULL DEFAULT 'active' COMMENT '状态', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL COMMENT '更新时间', PRIMARY KEY(id)) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci engine=innodb COMMENT '角色agent实例注册(项目×角色→稳定实例id)'"}, + {"op": "sql", "sql": "CREATE UNIQUE INDEX pipeline_agent_instances_idx_pai_pr ON pipeline_agent_instances(project_id, role)"} + ], + "down": [ + {"op": "sql", "sql": "DROP TABLE pipeline_agent_instances"} + ] +} diff --git a/deploy/migrations/m0003_conversations_session_id.json b/deploy/migrations/m0003_conversations_session_id.json new file mode 100644 index 0000000..81ed4e4 --- /dev/null +++ b/deploy/migrations/m0003_conversations_session_id.json @@ -0,0 +1,11 @@ +{ + "id": "m0003", + "title": "pipeline_conversations 补 session_id 列（会话隔离机制依赖）", + "backup_tables": ["pipeline_conversations"], + "up": [ + {"op": "sql", "sql": "ALTER TABLE pipeline_conversations ADD COLUMN session_id VARCHAR(32) NULL COMMENT '会话ID(多会话隔离)'"} + ], + "down": [ + {"op": "sql", "sql": "ALTER TABLE pipeline_conversations DROP COLUMN session_id"} + ] +} diff --git a/deploy/migrations/m0004_deliverables_created_by_widen.json b/deploy/migrations/m0004_deliverables_created_by_widen.json new file mode 100644 index 0000000..7ebd6ec --- /dev/null +++ b/deploy/migrations/m0004_deliverables_created_by_widen.json @@ -0,0 +1,11 @@ +{ + "id": "m0004", + "title": "pipeline_deliverables.created_by 加宽 32→64（根治 1406 溢出：技术写者交付件落不了库）", + "backup_tables": ["pipeline_deliverables"], + "up": [ + {"op": "sql", "sql": "ALTER TABLE pipeline_deliverables MODIFY created_by VARCHAR(64) COMMENT '创建人(agent实例id)'"} + ], + "down": [ + {"op": "sql", "sql": "ALTER TABLE pipeline_deliverables MODIFY created_by VARCHAR(32) COMMENT '创建人'"} + ] +}