#!/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] 并中止；
    人工评审后把迁移文件标记 "destructive": true，用 `apply-approved <batch>` 显式放行
    （备份与台账照常，缺声明仍拒绝——审批是显式动作，不是绕过开关）。
"""

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):
    """已处理批次：{batch: (max_step, has_failed)}。applied+skipped 都算已处理，
    含 failed 记录的批次不算完成。"""
    out, err = mysql(conf, "SELECT batch, MAX(step), SUM(status='failed') FROM %s "
                           "WHERE status IN ('applied','skipped','failed') "
                           "GROUP BY batch" % LEDGER_TABLE)
    applied = {}
    for ln in (out or "").splitlines():
        p = ln.split("\t")
        if len(p) == 3:
            applied[p[0]] = (int(p[1]), int(p[2]) > 0)
    return applied


def _esc(s):
    """转义 SQL 单引号（detail/backup_file 入台账前必用）。"""
    return str(s).replace("\\", "\\\\").replace("'", "''")


def detect_env():
    """当前环境：DEPLOY_ENV 环境变量优先；否则按应用根路径推断（/doit/ → prod）。"""
    env = os.environ.get("DEPLOY_ENV", "").strip().lower()
    if env in ("test", "prod", "dev"):
        return env
    if "/doit/" in APP_ROOT:
        return "prod"
    return "test"


def env_match(mig, env):
    """迁移文件可选 "env": "prod"/"test"/"all"（缺省=all），环境不匹配的批次跳过。"""
    want = (mig.get("env") or "all").lower()
    return want in ("all", env)


def load_migrations():
    migs = {}
    if not os.path.isdir(MIG_DIR):
        return migs
    env = detect_env()
    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
            if not env_match(d, env):
                continue  # 环境不匹配：本环境不可见
            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))
        # DROP INDEX 幂等守卫（2026-09-17 m0038 部署实抓的解析缺陷）：
        # 必须放在 DROP COLUMN 守卫**之前**——后者的可选 `(COLUMN\s+)?` 分支对
        # `ALTER TABLE t DROP INDEX idx` 同样匹配，且 group(3) 捕获到的是字面词
        # 'INDEX' 而非索引名 → column_exists(t,'INDEX') 恒 False → 整步被静默跳过
        # （索引永远删不掉、也不报错）。删索引的正确幂等判据是 index_exists。
        m = re.match(r"ALTER\s+TABLE\s+`?(\w+)`?\s+DROP\s+INDEX\s+`?(\w+)`?", sql, re.I)
        if m:
            if not table_exists(conf, m.group(1)):
                return [], "表不存在: %s" % m.group(1)
            if not index_exists(conf, m.group(1), m.group(2)):
                return [], "索引已不存在: %s.%s" % (m.group(1), m.group(2))
        # 独立形态 DROP INDEX idx ON table（MySQL/MariaDB 同样合法，历史迁移
        # m0025/m0031 等 13 条 down 段用的就是它）——同款幂等守卫，否则重复
        # rollback 会 1091 中止。
        m = re.match(r"DROP\s+INDEX\s+`?(\w+)`?\s+ON\s+`?(\w+)`?", sql, re.I)
        if m:
            if not table_exists(conf, m.group(2)):
                return [], "表不存在: %s" % m.group(2)
            if not index_exists(conf, m.group(2), m.group(1)):
                return [], "索引已不存在: %s.%s" % (m.group(2), m.group(1))
        # DROP COLUMN 幂等守卫：列已不存在则跳过（生产表由 create_tables.py 按最新
        # models 建出、天然没有旧列，m0009/m0010 类删列迁移直跑会 1091 中止）。
        # ⚠️ 正则组数是 2（table, col）——原写法带可选 `(COLUMN\s+)?` 组时列名是
        # group(3)，收紧为强制 COLUMN 关键字后列名变成 group(2)，索引必须同步改，
        # 否则 IndexError（2026-09-17 本地测试抓到）。历史迁移 22 条 DROP COLUMN
        # 全部显式带 COLUMN 关键字、bare 形态 0 条，收紧无回归。
        m = re.match(r"ALTER\s+TABLE\s+`?(\w+)`?\s+DROP\s+COLUMN\s+`?(\w+)`?", sql, re.I)
        if m:
            if not table_exists(conf, m.group(1)):
                return [], "表不存在: %s" % m.group(1)
            if not column_exists(conf, m.group(1), m.group(2)):
                return [], "列已不存在: %s.%s" % (m.group(1), m.group(2))
        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("=== 迁移台账（当前环境: %s）===" % detect_env())
    for mid in sorted(migs):
        m = migs[mid]
        if mid in applied:
            mx, failed = applied[mid]
            st = "失败(步骤≤%d)" % mx if failed else "已处理(步骤≤%d)" % mx
        else:
            st = "待执行"
        print("  %s  %-16s  %s" % (mid, st, m.get("title", "")[:50]))
    pending = [mid for mid in migs
               if mid not in applied or applied[mid][1]]
    print("\n待执行批次数:", len(pending), ("(%s)" % ",".join(pending)) if pending else "")


def _pending(migs, applied):
    """未完成的批次：未执行过，或含 failed 记录（修复后可重试）。"""
    return [b for b in sorted(migs) if b not in applied or applied[b][1]]


def _ledger_write(conf, mid, step, status, detail, backup_file=""):
    _, err = mysql(conf,
                   "INSERT INTO %s (id, batch, step, status, backup_file, detail) "
                   "VALUES ('%s', '%s', %d, '%s', '%s', '%s') "
                   "ON DUPLICATE KEY UPDATE status='%s', detail='%s'"
                   % (LEDGER_TABLE, _new_id(), mid, step, status,
                      _esc(backup_file[:200]), _esc(detail[:200]), status, _esc(detail[:200])))
    if err:
        print("   [WARN] 台账写入失败(不影响库变更): %s" % err[:150])


def cmd_plan(conf, batches):
    ensure_ledger(conf)
    applied = ledger_state(conf)
    migs = load_migrations()
    targets = [b for b in (batches or _pending(migs, applied)) if b not in applied or applied[b][1]]
    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, allow_destructive=False):
    ensure_ledger(conf)
    applied = ledger_state(conf)
    migs = load_migrations()
    targets = [b for b in (batches or _pending(migs, applied)) if b not in applied or applied[b][1]]
    if not targets:
        print("无待执行批次")
        return
    _, _, _, _, db = conf
    for mid in targets:
        m = migs[mid]
        print("\n══ 应用 %s: %s" % (mid, m.get("title", "")))
        # 破坏性守卫（先扫全部语句再动手）。
        # apply-approved 通道（2026-09-10）：迁移文件显式声明 "destructive": true
        # 且人工执行 apply-approved 才放行；缺任一条件照旧拒绝。
        # 放行的批次同样先备份 backup_tables、逐步写台账——审批≠裸奔。
        has_destructive = False
        for step in m.get("up", []):
            sqls, skip = render_step(conf, step, "up")
            if skip:
                continue
            for s in sqls:
                if DESTRUCTIVE.search(s):
                    has_destructive = True
                    if not allow_destructive:
                        die("up 含破坏性语句，拒绝自动执行（人工复核后用 apply-approved）:\n    %s" % s[:200])
        if has_destructive and allow_destructive and not m.get("destructive"):
            die("%s 含破坏性语句但迁移文件未声明 \"destructive\": true —— 拒绝执行"
                "（声明即承诺已人工评审）" % mid)
        if has_destructive and allow_destructive:
            print("   [APPROVED] 破坏性语句经人工审批通道放行（destructive: true）")
        # 备份
        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:
                _ledger_write(conf, mid, i, "skipped", skip)
                print("   步骤%d 跳过(%s)" % (i, skip))
                continue
            for s in sqls:
                out, err = mysql(conf, s)
                if err:
                    _ledger_write(conf, mid, i, "failed", err)
                    die("步骤%d 失败: %s\n（已完成步骤可用 `dmig.py rollback %s` 回退）"
                        % (i, err, mid))
                print("   步骤%d OK: %s" % (i, s[:90]))
            _ledger_write(conf, mid, i, "applied", str(step), bk)
        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", "apply-approved", "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 == "apply-approved":
        # 破坏性批次人工审批通道：必须显式指定批次号，禁默认全量（防止把
        # 未来新增的 destructive 迁移顺手带跑）。
        if not args:
            die("apply-approved 需要显式批次号（禁默认全量）: dmig.py apply-approved mNNNN")
        cmd_apply(conf, args, allow_destructive=True)
    elif cmd == "rollback":
        if not args:
            die("rollback 需要批次号: dmig.py rollback m0001")
        cmd_rollback(conf, args[0])


if __name__ == "__main__":
    main()