feat(deploy): 测试→生产增量部署机制——dmig迁移引擎+ddiff差异对账+dbackup增量备份+四步规范

- METHODOLOGY.md: 备份/操作/验证/回退四大步操作规范
- dmig.py: 迁移引擎(台账表pipeline_deploy_ledger+幂等+破坏性守卫+down回退)
- ddiff.py: 环境schema/参数/模块版本差异对账
- dbackup.py: mysqldump增量表备份+恢复命令生成
- migrations/m0001-m0004: 首批迁移(今日发现的生产库缺口)

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/<db>_<table>_<日期时间>.sql，同目录生成 RESTORE 命令文件
+```
+
+检查点：
+- [ ] 备份文件存在且非空（`ls -lh backups/`）
+- [ ] 备份文件头部有 `CREATE TABLE` 且行数与线上一致（`dbackup.py verify <file>`）
+
+### 第 2 步：操作（按序执行增量）
+
+```bash
+# 2.1 差异核对（必做）：打印生产与测试的差异清单，与本次批次对照
+./py3/bin/python deploy/ddiff.py --checklist
+
+# 2.2 代码增量（先代码后库，或按迁移说明顺序）
+cd /d/doit/pipeline-app/pkgs/<module> && git pull --ff-only origin main
+cd /d/doit/pipeline-app && ./py3/bin/pip install pkgs/<module>
+find py3/lib/python3.10/site-packages/<module_pkg> -name __pycache__ -exec rm -rf {} +
+nohup bash restart-pipeline.sh </dev/null > 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/<module> && git log --oneline -5   # 找到上线前的提交
+git checkout <上线前提交> && cd ../.. && ./py3/bin/pip install pkgs/<module> && 重启
+```
+
+回退后重新走第 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]`，人工执行。

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   # 生成并执行恢复（危险）
+
+备份位置：<APP_ROOT>/backups/<库名>_<表名>_<YYYYmmdd_HHMMSS>.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()

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()

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 <batch>    # 回退：执行该批次 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()

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"}
+  ]
+}

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"}
+  ]
+}

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"}
+  ]
+}

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 '创建人'"}
+  ]
+}