import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.sql.*;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * MySQL -> 达梦(DM8) 通用结构+数据迁移工具（JDBC 驱动层，零 SQL 文本转义）。
 *
 * 设计要点（兼容规则来自 DM8 8.1.2.128 实测）：
 *  - 结构：解析源库 SHOW CREATE TABLE（结构化元数据，非 dump 文本），生成达梦 DDL；
 *  - 数据：ResultSet -> PreparedStatement 批量直搬，类型转换全部由 JDBC 驱动完成；
 *  - 次序：建表(PK/CHECK/普通UK) -> 索引 -> 数据 -> 外键 -> 注释，外键后置消除表间依赖；
 *  - 幂等：--drop-existing 先删目标模式全部外键与表，可反复重跑；
 *  - 韧性：批次失败自动降级为逐行重试，坏行记入错误日志不中断整体；
 *  - 通用：不针对特定表名/列名；类型映射覆盖 MySQL 常用类型族；分页策略适配任意主键形态。
 *
 * 达梦兼容实测规则：
 *  1) AUTO_INCREMENT -> IDENTITY(1,1)；自增列插显式值需 SET IDENTITY_INSERT ON + 显式列清单；
 *  2) 生成列 GENERATED ALWAYS AS (IF(cond,1,NULL)) VIRTUAL -> TINYINT AS (CASE WHEN cond THEN 1 ELSE NULL END)；
 *  3) 含生成列的内联 UNIQUE 不被达梦接受(-2034)，改后置 CREATE UNIQUE INDEX；
 *     NULL 判重语义差异：MySQL 任一键列 NULL 即不判重，达梦(Oracle 语义)仅全 NULL 键不判重——
 *     对齐方式为全部键成分套同一守卫 CASE，非判重行整体置全 NULL 键；单标记生成列唯一键用单成分 CASE；
 *  4) ON DELETE RESTRICT 不被支持，剥离（默认 NO ACTION 语义等价）；
 *  5) 达梦无 FULLTEXT/SPATIAL 索引与前缀索引，跳过/降级并告警。
 *
 * 用法（JDK17 单文件运行，cp 需 MySQL 与达梦 JDBC 驱动 jar）：
 *   java -Dfile.encoding=UTF-8 -cp "mysql.jar;dm.jar" DbMigrator.java migrate \
 *     --source-url "jdbc:mysql://host:3306/db?useSSL=false&characterEncoding=utf8" \
 *     --source-user u --source-pwd p \
 *     --target-url jdbc:dm://host:5236 --target-user U --target-pwd P --schema SCHEMA \
 *     [--create-schema] [--drop-existing] [--tables t1,t2] [--exclude t3] \
 *     [--batch 1000] [--no-verify] [--err-log migrate_errors.log]
 *   ... verify  <同 migrate 参数>     仅校验行数与主键水位
 *   ... drop-schema --target-url U --target-user u --target-pwd p --schema S
 */
public class DbMigrator {

    // ==================== 数据模型 ====================

    record Col(String name, String rawType, boolean notNull, String def,
               boolean autoInc, String genExpr, String comment) {}

    static final class Idx {
        final String name; final List<String> cols = new ArrayList<>();
        final List<String> prefixes = new ArrayList<>();   // 前缀长度，null=整列
        boolean unique; boolean fulltext; boolean spatial;
        Idx(String name) { this.name = name; }
    }

    record Fk(String name, List<String> cols, String refTable, List<String> refCols, String onDelete) {}

    record Chk(String name, String expr) {}

    static final class TableDef {
        String name; final List<Col> cols = new ArrayList<>();
        final List<String> pk = new ArrayList<>();
        final List<Idx> indexes = new ArrayList<>();
        final List<Fk> fks = new ArrayList<>();
        final List<Chk> checks = new ArrayList<>();
        String comment; boolean hasIdentity;
        List<String> dataCols() {                            // 可搬运列 = 非生成列
            return cols.stream().filter(c -> c.genExpr() == null).map(Col::name).toList();
        }
        List<Col> generatedCols() {
            return cols.stream().filter(c -> c.genExpr() != null).toList();
        }
        Col findCol(String name) {
            return cols.stream().filter(c -> c.name().equalsIgnoreCase(name)).findFirst().orElse(null);
        }
        boolean singleNumericPk() {
            if (pk.size() != 1) return false;
            return cols.stream().filter(c -> c.name().equals(pk.get(0))).findFirst()
                    .map(c -> c.rawType().matches("(big|int|medium)int.*")).orElse(false);
        }
    }

    // ==================== 入口 ====================

    public static void main(String[] args) throws Exception {
        if (args.length == 0) { usage(); System.exit(2); }
        String mode = args[0];
        Map<String, String> opt = parseOpts(args);
        switch (mode) {
            case "migrate", "verify" -> run(mode, opt);
            case "drop-schema" -> {
                String tUrl = require(opt, "target-url"), tUser = require(opt, "target-user"), tPwd = require(opt, "target-pwd");
                String schema = require(opt, "schema").toUpperCase();
                try (Connection dst = DriverManager.getConnection(tUrl, tUser, tPwd)) {
                    dropSchemaObjects(dst, schema);
                }
            }
            default -> { usage(); System.exit(2); }
        }
    }

    static void usage() {
        System.err.println("""
                用法:
                  DbMigrator migrate  --source-url U --source-user u --source-pwd p \\
                                      --target-url U --target-user u --target-pwd p --schema S \\
                                      [--create-schema] [--drop-existing] [--tables a,b] [--exclude c] \\
                                      [--batch 1000] [--no-verify] [--err-log FILE]
                  DbMigrator verify   <同 migrate 参数，仅校验>
                  DbMigrator drop-schema --target-url U --target-user u --target-pwd p --schema S""");
    }

    static Map<String, String> parseOpts(String[] args) {
        Map<String, String> m = new HashMap<>();
        for (int i = 1; i < args.length; i++) {
            String k = args[i];
            if (!k.startsWith("--")) continue;
            String v = i + 1 < args.length && !args[i + 1].startsWith("--") ? args[++i] : "true";
            m.put(k.substring(2), v);
        }
        return m;
    }

    static void run(String mode, Map<String, String> opt) throws Exception {
        String sUrl = require(opt, "source-url"), sUser = require(opt, "source-user"), sPwd = require(opt, "source-pwd");
        String tUrl = require(opt, "target-url"), tUser = require(opt, "target-user"), tPwd = require(opt, "target-pwd");
        String schema = require(opt, "schema").toUpperCase();
        int batch = Integer.parseInt(opt.getOrDefault("batch", "1000"));
        Path errLog = Path.of(opt.getOrDefault("err-log", "migrate_errors.log"));
        if (sUrl.startsWith("jdbc:mysql") && !sUrl.contains("characterEncoding")) {
            sUrl += (sUrl.contains("?") ? "&" : "?") + "characterEncoding=utf8";
        }
        long t0 = System.currentTimeMillis();
        try (Connection src = DriverManager.getConnection(sUrl, sUser, sPwd);
             Connection dst = DriverManager.getConnection(tUrl, tUser, tPwd)) {
            System.out.println("[源] " + src.getMetaData().getDatabaseProductVersion());
            System.out.println("[目标] " + dst.getMetaData().getDatabaseProductVersion() + "  模式: " + schema);

            if (opt.containsKey("create-schema") && mode.equals("migrate")) createSchema(dst, schema);
            if (opt.containsKey("drop-existing") && mode.equals("migrate")) dropSchemaObjects(dst, schema);

            String catalog = src.getCatalog();
            Set<String> only = opt.containsKey("tables")
                    ? Arrays.stream(opt.get("tables").split(",")).map(String::trim).collect(Collectors.toSet()) : null;
            Set<String> excl = opt.containsKey("exclude")
                    ? Arrays.stream(opt.get("exclude").split(",")).map(String::trim).collect(Collectors.toSet()) : null;
            List<String> tables = listTables(src, catalog).stream()
                    .filter(t -> (only == null || only.contains(t)) && (excl == null || !excl.contains(t)))
                    .sorted().toList();
            if (tables.isEmpty()) { System.out.println("无待迁移表"); return; }

            Map<String, TableDef> defs = new LinkedHashMap<>();
            for (String t : tables) defs.put(t, parseShowCreate(src, t));
            System.out.println("待迁移表 " + defs.size() + " 张: " + tables);

            List<String> warns = new ArrayList<>();
            if (mode.equals("migrate")) {
                dst.setAutoCommit(true);
                for (TableDef d : defs.values()) {
                    for (String ddl : buildCreateTable(d, schema, warns))
                        exec(dst, ddl, "结构 " + d.name);
                }
                dst.setAutoCommit(false);
                for (TableDef d : defs.values()) copyData(src, dst, d, schema, batch, errLog, warns);
                dst.setAutoCommit(true);
                for (TableDef d : defs.values())
                    for (Fk fk : d.fks)
                        exec(dst, buildFk(d, fk, schema), "外键 " + d.name + "." + fk.name());
                for (TableDef d : defs.values())
                    for (String c : buildComments(d, schema)) execQuiet(dst, c);
            }

            if (opt.containsKey("no-verify")) {
                System.out.println("(跳过校验)");
            } else {
                System.out.println("==================== 校验 ====================");
                int bad = 0;
                for (TableDef d : defs.values()) if (!verifyTable(src, dst, d, schema)) bad++;
                if (!warns.isEmpty()) {
                    System.out.println("==================== 告警 ====================");
                    warns.forEach(w -> System.out.println("  [WARN] " + w));
                }
                System.out.println("=============================================");
                System.out.println("表: " + defs.size() + " | 校验不一致: " + bad + " | 告警: " + warns.size()
                        + " | 耗时 " + (System.currentTimeMillis() - t0) / 1000 + "s"
                        + (Files.exists(errLog) ? " | 错误日志: " + errLog.toAbsolutePath() : ""));
                if (bad > 0) System.exit(1);
            }
        }
    }

    static String require(Map<String, String> opt, String key) {
        String v = opt.get(key);
        if (v == null || v.isBlank()) { System.err.println("缺少参数 --" + key); usage(); System.exit(2); }
        return v;
    }

    // ==================== 源库元数据 ====================

    static List<String> listTables(Connection c, String catalog) throws SQLException {
        List<String> out = new ArrayList<>();
        try (ResultSet rs = c.getMetaData().getTables(catalog, null, "%", new String[]{"TABLE"})) {
            while (rs.next()) out.add(rs.getString("TABLE_NAME"));
        }
        return out;
    }

    /** 解析 SHOW CREATE TABLE 输出为结构化定义。 */
    static TableDef parseShowCreate(Connection src, String table) throws SQLException {
        String ddl;
        try (Statement st = src.createStatement(); ResultSet rs = st.executeQuery("SHOW CREATE TABLE `" + table + "`")) {
            rs.next();
            ddl = rs.getString(2);
        }
        TableDef d = new TableDef();
        d.name = table;
        String body = ddl.substring(ddl.indexOf('(') + 1, ddl.lastIndexOf(')'));
        for (String seg : splitTop(body)) {
            String s = seg.trim();
            if (s.isEmpty()) continue;
            if (s.startsWith("`")) parseColumn(d, s);
            else if (s.regionMatches(true, 0, "PRIMARY KEY", 0, 11)) d.pk.addAll(identList(cutParen(s)));
            else if (s.regionMatches(true, 0, "UNIQUE KEY", 0, 10)) parseIndex(d, s, true);
            else if (s.regionMatches(true, 0, "FULLTEXT", 0, 8) || s.regionMatches(true, 0, "SPATIAL", 0, 7)) parseIndex(d, s, false);
            else if (s.regionMatches(true, 0, "KEY", 0, 3)) parseIndex(d, s, false);
            else if (s.regionMatches(true, 0, "CONSTRAINT", 0, 10)) parseConstraint(d, s);
            else warns0.add(table + ": 未识别片段(忽略) " + abbrev(s));
        }
        Matcher m = Pattern.compile("\\)\\s*(?:ENGINE|AUTO_INCREMENT).*?(?:COMMENT='([^']*)')?", Pattern.DOTALL).matcher(ddl);
        if (m.find() && m.group(1) != null) d.comment = m.group(1);
        return d;
    }

    // parseShowCreate 早期告警通道（结构解析阶段无 warns 列表）
    static final List<String> warns0 = new ArrayList<>();

    static void parseColumn(TableDef d, String s) {
        Matcher head = Pattern.compile("^`([^`]+)`\\s+(.*)$", Pattern.DOTALL).matcher(s);
        if (!head.matches()) return;
        String name = head.group(1), rest = head.group(2);
        String type = firstTypeToken(rest);
        String tail = rest.substring(type.length());
        // 生成列: GENERATED ALWAYS AS (...) VIRTUAL|STORED —— 括号配对提取
        String gen = null;
        Matcher g = Pattern.compile("\\bGENERATED\\s+ALWAYS\\s+AS\\s*\\(", Pattern.CASE_INSENSITIVE).matcher(tail);
        if (g.find()) {
            int open = g.end() - 1, depth = 0, end = -1;
            for (int i = open; i < tail.length(); i++) {
                char ch = tail.charAt(i);
                if (ch == '(') depth++;
                else if (ch == ')') { depth--; if (depth == 0) { end = i; break; } }
            }
            if (end > 0) gen = tail.substring(open + 1, end);
        }
        boolean notNull = gen == null && Pattern.compile("\\bNOT\\s+NULL\\b").matcher(tail).find();
        String def = null;
        Matcher dm = Pattern.compile("\\bDEFAULT\\s+(NULL|'(?:[^']|'')*'|[\\w.()+-]+)", Pattern.CASE_INSENSITIVE).matcher(tail);
        if (gen == null && dm.find()) {
            String v = dm.group(1);
            def = v.equalsIgnoreCase("NULL") ? null : v;
        }
        boolean auto = gen == null
                && Pattern.compile("\\bAUTO_INCREMENT\\b", Pattern.CASE_INSENSITIVE).matcher(tail).find();
        String comment = null;
        Matcher cm = Pattern.compile("\\bCOMMENT\\s+'((?:[^']|'')*)'").matcher(tail);
        if (cm.find()) comment = cm.group(1).replace("''", "'");
        if (auto) d.hasIdentity = true;
        d.cols.add(new Col(name, type.toLowerCase(), notNull, def, auto, gen, comment));
    }

    /** 提取类型词：双词类型走白名单（避免吞掉 NOT/DEFAULT/GENERATED 等后续关键字），并归并 unsigned/zerofill。 */
    static String firstTypeToken(String rest) {
        Matcher m = Pattern.compile(
                "^(?:double\\s+precision|character\\s+varying|[a-zA-Z]+)(?:\\s*\\([^)]*\\))?(?:\\s+(?:unsigned|zerofill))*",
                Pattern.CASE_INSENSITIVE).matcher(rest);
        return m.find() ? m.group() : rest;
    }

    static void parseIndex(TableDef d, String s, boolean unique) {
        Matcher m = Pattern.compile("^(?:UNIQUE\\s+|FULLTEXT\\s+|SPATIAL\\s+)?KEY\\s+`([^`]+)`\\s*\\((.*)\\)$",
                Pattern.DOTALL).matcher(s.trim());
        if (!m.matches()) { warns0.add(d.name + ": 索引未解析(忽略) " + abbrev(s)); return; }
        Idx idx = new Idx(m.group(1));
        idx.unique = unique;
        idx.fulltext = s.trim().regionMatches(true, 0, "FULLTEXT", 0, 8);
        idx.spatial = s.trim().regionMatches(true, 0, "SPATIAL", 0, 7);
        for (String part : splitTop(m.group(2))) {
            Matcher p = Pattern.compile("^`([^`]+)`(?:\\((\\d+)\\))?$").matcher(part.trim());
            if (p.matches()) { idx.cols.add(p.group(1)); idx.prefixes.add(p.group(2)); }
        }
        d.indexes.add(idx);
    }

    static void parseConstraint(TableDef d, String s) {
        Matcher m = Pattern.compile("^CONSTRAINT\\s+`([^`]+)`\\s+(.*)$", Pattern.DOTALL).matcher(s.trim());
        if (!m.matches()) return;
        String name = m.group(1), body = m.group(2);
        if (body.regionMatches(true, 0, "FOREIGN KEY", 0, 11)) {
            // 列清单取第一个括号对（cutParen 的 first/last 会跨越 REFERENCES 段）
            List<String> cols = identList(cutOuterParen(body.substring(body.indexOf('('))));
            Matcher ref = Pattern.compile("REFERENCES\\s+`([^`]+)`\\s*\\(([^)]*)\\)(?:\\s+ON\\s+DELETE\\s+(\\w+(?:\\s+\\w+)?))?",
                    Pattern.CASE_INSENSITIVE).matcher(body);
            if (!ref.find()) { warns0.add(d.name + ": 外键未解析 " + name); return; }
            d.fks.add(new Fk(name, cols, ref.group(1), identList(ref.group(2)), ref.group(3)));
        } else if (body.regionMatches(true, 0, "CHECK", 0, 5)) {
            d.checks.add(new Chk(name, cutOuterParen(body.substring(5).trim())));
        }
    }

    static String cutParen(String s) { return s.substring(s.indexOf('(') + 1, s.lastIndexOf(')')); }

    static String cutOuterParen(String s) {
        s = s.trim();
        int depth = 0;
        for (int i = 0; i < s.length(); i++) {
            if (s.charAt(i) == '(') depth++;
            else if (s.charAt(i) == ')') { depth--; if (depth == 0) return s.substring(1, i); }
        }
        return s;
    }

    static List<String> identList(String s) {
        return splitTop(s).stream().map(x -> {
            Matcher m = Pattern.compile("^`([^`]+)`").matcher(x.trim());
            return m.matches() ? m.group(1) : x.trim();
        }).toList();
    }

    /** 顶层逗号切分（跳过括号嵌套与单/双引号、反引号字符串，含成对转义）。 */
    static List<String> splitTop(String s) {
        List<String> out = new ArrayList<>();
        StringBuilder cur = new StringBuilder();
        int depth = 0;
        for (int i = 0; i < s.length(); i++) {
            char c = s.charAt(i);
            if (c == '\'' || c == '"' || c == '`') {
                char quote = c;
                cur.append(c);
                for (i++; i < s.length(); i++) {
                    cur.append(s.charAt(i));
                    if (s.charAt(i) == quote) {
                        if (i + 1 < s.length() && s.charAt(i + 1) == quote) { cur.append(quote); i++; }
                        else break;
                    }
                }
            } else if (c == '(') { depth++; cur.append(c); }
            else if (c == ')') { depth--; cur.append(c); }
            else if (c == ',' && depth == 0) { out.add(cur.toString()); cur.setLength(0); }
            else cur.append(c);
        }
        out.add(cur.toString());
        return out;
    }

    // ==================== 达梦 DDL 生成 ====================

    static final Set<String> DM_RESERVED = Set.of(
            "USER", "VERSION", "TYPE", "STATUS", "LEVEL", "COMMENT", "VALUES", "OPTION",
            "PASSWORD", "ROLE", "TABLE", "COLUMN", "INDEX", "DEFAULT", "CHECK", "CONSTRAINT",
            "REFERENCES", "PRIMARY", "FOREIGN", "UNIQUE", "VIEW", "SEQUENCE", "TRIGGER",
            "FUNCTION", "PROCEDURE", "DECLARE", "BEGIN", "END", "IF", "ELSE", "WHILE", "LOOP",
            "FOR", "SELECT", "INSERT", "UPDATE", "DELETE", "FROM", "WHERE", "GROUP", "ORDER",
            "BY", "HAVING", "DISTINCT", "ALL", "ANY", "EXISTS", "BETWEEN", "LIKE", "IS",
            "NULL", "NOT", "AND", "OR", "AS", "ON", "JOIN", "LEFT", "RIGHT", "INNER", "OUTER",
            "FULL", "CROSS", "UNION", "INTERSECT", "MINUS", "CASE", "WHEN", "THEN", "GRANT",
            "REVOKE", "COMMIT", "ROLLBACK", "SAVEPOINT", "SET", "INTO", "ROW", "ROWS",
            "DATE", "TIME", "TIMESTAMP", "INTERVAL", "PUBLIC", "SESSION", "SYSTEM", "AUDIT");

    static String ident(String name) {
        String u = name.toUpperCase();
        return (DM_RESERVED.contains(u) || !u.matches("[A-Z_][A-Z0-9_$#]*")) ? '"' + u + '"' : u;
    }

    /** MySQL 类型 -> 达梦类型；返回 null 表示不支持（调用方告警跳过列）。 */
    static String mapType(String mysqlType, List<String> warns, String ctx) {
        String t = mysqlType.toLowerCase().replaceAll("\\s+", " ").trim();
        Matcher m = Pattern.compile("^([a-z ]+?)(?:\\((\\d+)(?:,(\\d+))?\\))?( unsigned| zerofill)?$").matcher(t);
        if (!m.matches()) { warns.add(ctx + ": 无法识别类型 " + mysqlType); return null; }
        String base = m.group(1).trim();
        String len = m.group(2), scale = m.group(3);
        boolean uns = m.group(4) != null && m.group(4).contains("unsigned");
        if (uns && !base.equals("smallint") && !base.equals("tinyint"))
            warns.add(ctx + ": UNSIGNED 不被达梦支持，已忽略（" + mysqlType + "）");
        return switch (base) {
            case "bigint" -> "BIGINT";
            case "int", "integer", "mediumint" -> "INT";
            case "smallint" -> uns ? "INT" : "SMALLINT";        // unsigned 超界升 INT
            case "tinyint" -> uns ? "SMALLINT" : "TINYINT";     // (n) 为显示宽度，不代表值域
            case "decimal", "numeric" -> scale != null ? "DECIMAL(" + len + "," + scale + ")"
                    : len != null ? "DECIMAL(" + len + ",0)" : "DECIMAL(65,30)";
            case "float" -> "FLOAT";
            case "double", "double precision", "real" -> "DOUBLE";
            case "bit" -> len != null && Integer.parseInt(len) > 1
                    ? unsupport(warns, ctx, mysqlType) : "BIT";
            case "date" -> "DATE";
            case "time" -> "TIME";
            case "year" -> { warns.add(ctx + ": YEAR 映射为 SMALLINT"); yield "SMALLINT"; }
            case "datetime", "timestamp" -> len != null ? "TIMESTAMP(" + len + ")" : "TIMESTAMP";
            case "char", "character" -> "CHAR(" + (len == null ? 1 : len) + ")";
            case "varchar", "character varying" -> {
                if (len == null) yield "VARCHAR(255)";
                int n = Integer.parseInt(len);
                if (n > 8188) { warns.add(ctx + ": VARCHAR(" + n + ") 超达梦页上限，降级 TEXT"); yield "TEXT"; }
                yield "VARCHAR(" + n + ")";
            }
            case "tinytext", "text" -> "TEXT";
            case "mediumtext", "longtext" -> "CLOB";
            case "json" -> "TEXT";
            case "enum", "set" -> { warns.add(ctx + ": " + base.toUpperCase()
                    + " 映射为 VARCHAR(255)，枚举元素约束不迁移"); yield "VARCHAR(255)"; }
            case "binary" -> "BINARY(" + (len == null ? 1 : len) + ")";
            case "varbinary" -> "VARBINARY(" + (len == null ? 255 : len) + ")";
            case "tinyblob", "blob", "mediumblob", "longblob" -> "BLOB";
            default -> unsupport(warns, ctx, mysqlType);
        };
    }

    static String unsupport(List<String> warns, String ctx, String t) {
        warns.add(ctx + ": 不支持的类型 " + t + "，列已跳过，请手工处理");
        return null;
    }

    /** 生成建表语句 + 后置索引语句（首元素 CREATE TABLE，其余 CREATE [UNIQUE] INDEX）。 */
    static List<String> buildCreateTable(TableDef d, String schema, List<String> warns) {
        Set<String> genNames = d.generatedCols().stream().map(c -> c.name().toUpperCase()).collect(Collectors.toSet());
        List<String> out = new ArrayList<>();
        StringBuilder sb = new StringBuilder("CREATE TABLE ").append(schema).append('.').append(ident(d.name)).append(" (\n");
        List<String> parts = new ArrayList<>();
        for (Col c : d.cols) {
            String ctx = d.name + "." + c.name();
            String colDdl;
            if (c.genExpr() != null) {
                colDdl = buildGeneratedColumn(c, ctx, warns);
            } else {
                String type = mapType(c.rawType(), warns, ctx);
                if (type == null) continue;
                StringBuilder cb = new StringBuilder("  ").append(ident(c.name())).append(' ').append(type);
                if (c.notNull()) cb.append(" NOT NULL");
                if (c.autoInc()) cb.append(" IDENTITY(1,1)");
                if (c.def() != null) {
                    String dv = mapDefault(c.def(), warns, ctx);
                    if (dv != null) cb.append(" DEFAULT ").append(dv);
                }
                colDdl = cb.toString();
            }
            if (colDdl != null) parts.add(colDdl);
        }
        if (!d.pk.isEmpty())
            parts.add("  PRIMARY KEY (" + d.pk.stream().map(DbMigrator::ident).collect(Collectors.joining(",")) + ")");
        // 不含生成列的唯一约束 inline；含生成列的转后置（达梦 -2034）
        Set<String> deferredUk = new LinkedHashSet<>();
        for (Idx idx : d.indexes) {
            if (!idx.unique) continue;
            boolean hasGen = idx.cols.stream().anyMatch(cn -> genNames.contains(cn.toUpperCase()));
            if (!hasGen) {
                if (idx.cols.size() > 1 && idx.cols.stream().map(d::findCol).filter(Objects::nonNull)
                        .anyMatch(bc -> bc.genExpr() == null && !bc.notNull()))
                    warns.add(d.name + "." + idx.name + ": 多列唯一键含可空列，达梦对部分NULL键判重（MySQL任一NULL不判重），NULL组合重复的数据将冲突");
                parts.add("  CONSTRAINT " + ident(idx.name) + " UNIQUE ("
                        + idx.cols.stream().map(DbMigrator::ident).collect(Collectors.joining(",")) + ")");
            } else deferredUk.add(idx.name);
        }
        for (Chk chk : d.checks)
            parts.add("  CONSTRAINT " + ident(chk.name()) + " CHECK (" + exprToDm(chk.expr()) + ")");
        sb.append(String.join(",\n", parts)).append("\n)");
        out.add(sb.toString());

        for (Idx idx : d.indexes) {
            if (idx.unique) {
                if (!deferredUk.contains(idx.name)) continue;
                Col marker = findMarkerCol(d, idx, genNames);
                String bizCol = idx.cols.stream()
                        .filter(cn -> !genNames.contains(cn.toUpperCase())).findFirst().orElse(null);
                if (bizCol == null) {
                    // 单标记生成列唯一键（"全局唯一默认项"形态）：单成分 CASE 守卫即可对齐 MySQL 部分唯一语义
                    if (idx.cols.size() == 1 && marker != null) {
                        out.add("CREATE UNIQUE INDEX " + ident(idx.name) + " ON " + schema + '.' + ident(d.name)
                                + " ((CASE WHEN " + markerCond(marker) + " THEN 1 END))");
                    } else {
                        warns.add(d.name + "." + idx.name + ": 唯一键全部由生成列构成，已跳过");
                    }
                    continue;
                }
                String key;
                if (marker != null) {
                    // NULL 判重语义对齐：MySQL 任一键列 NULL 即不判重；达梦(Oracle 语义)仅全 NULL 键不判重、
                    // 部分 NULL 键的 NULL 视为相等参与判重。对齐方式：全部键成分套同一守卫 CASE——
                    // 判重条件不满足时整行键置全 NULL（达梦对全 NULL 键不判重），满足时展开原值。
                    String guard = idx.cols.stream()
                            .filter(cn -> !genNames.contains(cn.toUpperCase()))
                            .map(d::findCol).filter(Objects::nonNull)
                            .filter(bc -> !bc.notNull())
                            .map(bc -> ident(bc.name()) + " IS NOT NULL")
                            .collect(Collectors.joining(" AND "));
                    String cond = guard.isEmpty() ? markerCond(marker)
                            : guard + " AND " + markerCond(marker);
                    key = idx.cols.stream()
                            .map(cn -> genNames.contains(cn.toUpperCase())
                                    ? "(CASE WHEN " + cond + " THEN 1 END)"
                                    : "(CASE WHEN " + cond + " THEN " + ident(cn) + " END)")
                            .collect(Collectors.joining(","));
                } else {
                    key = idx.cols.stream().filter(cn -> !genNames.contains(cn.toUpperCase()))
                            .map(DbMigrator::ident).collect(Collectors.joining(","));
                    warns.add(d.name + "." + idx.name + ": 唯一键含生成列，已剔除生成列降级为普通唯一索引");
                }
                out.add("CREATE UNIQUE INDEX " + ident(idx.name) + " ON " + schema + '.' + ident(d.name)
                        + " (" + key + ")");
            } else if (idx.fulltext || idx.spatial) {
                warns.add(d.name + "." + idx.name + ": 达梦无 FULLTEXT/SPATIAL 索引，已跳过");
            } else {
                if (idx.prefixes.stream().anyMatch(Objects::nonNull))
                    warns.add(d.name + "." + idx.name + ": 达梦不支持前缀索引，已按整列建索引");
                out.add("CREATE INDEX " + ident(idx.name) + " ON " + schema + '.' + ident(d.name) + " ("
                        + idx.cols.stream().map(DbMigrator::ident).collect(Collectors.joining(",")) + ")");
            }
        }
        return out;
    }

    /** 生成列 -> 达梦 AS(...) 列：IF(a,b,c) -> CASE WHEN a THEN b ELSE c END。 */
    static String buildGeneratedColumn(Col c, String ctx, List<String> warns) {
        String type = mapType(c.rawType(), warns, ctx);
        if (type == null) type = "INT";
        if (type.equals("INT") || type.equals("BIGINT")) type = "TINYINT";   // 1/NULL 标记形态实测 TINYINT
        String expr = exprToDm(c.genExpr());
        if (expr == null || expr.isEmpty()) {
            warns.add(ctx + ": 生成列表达式无法转译(" + c.genExpr() + ")，列已跳过");
            return null;
        }
        if (Pattern.compile("\\bIF\\s*\\(").matcher(expr).find())
            warns.add(ctx + ": 生成列表达式含未完全转译的 IF，请人工核验");
        return "  " + ident(c.name()) + " " + type + " AS (" + expr + ")";
    }

    /** 唯一键中的 1/NULL 活跃标记生成列（IF((cond),1,NULL) 形态）。 */
    static Col findMarkerCol(TableDef d, Idx idx, Set<String> genNames) {
        for (String cn : idx.cols) {
            if (!genNames.contains(cn.toUpperCase())) continue;
            Col gc = d.cols.stream().filter(x -> x.name().equalsIgnoreCase(cn)).findFirst().orElse(null);
            if (gc != null && markerCond(gc) != null) return gc;
        }
        return null;
    }

    /** 提取生成列 IF((cond),1,NULL) 的 cond（达梦表达式）；非该形态返回 null。 */
    static String markerCond(Col gc) {
        Matcher m = Pattern.compile("^if\\s*\\(\\((.*)\\)\\s*,\\s*1\\s*,\\s*null\\s*\\)$",
                Pattern.CASE_INSENSITIVE | Pattern.DOTALL).matcher(gc.genExpr().trim());
        return m.matches() ? exprToDm(m.group(1)) : null;
    }

    static String mapDefault(String def, List<String> warns, String ctx) {
        String d = def.trim();
        if (d.regionMatches(true, 0, "CURRENT_TIMESTAMP", 0, 17)) return "CURRENT_TIMESTAMP";
        if (d.startsWith("_utf8mb4")) d = d.substring(8);
        if (d.startsWith("'")) return d;
        if (d.matches("-?\\d+(\\.\\d+)?")) return d;
        warns.add(ctx + ": 默认值 " + def + " 无法映射，已丢弃");
        return null;
    }

    /** 表达式转译：_utf8mb4 剥离、`x` -> X、IF(a,b,c) -> CASE WHEN、CURRENT_TIMESTAMP(n) 归一。 */
    static String exprToDm(String e) {
        String out = e.replace("_utf8mb4", "");
        out = Pattern.compile("`([^`]+)`").matcher(out).replaceAll(r -> Matcher.quoteReplacement(ident(r.group(1))));
        out = out.replaceAll("CURRENT_TIMESTAMP\\(\\d+\\)", "CURRENT_TIMESTAMP");
        Matcher ifm = Pattern.compile("^if\\s*\\((.*)\\)$", Pattern.CASE_INSENSITIVE | Pattern.DOTALL).matcher(out.trim());
        if (ifm.matches()) {
            List<String> args = splitTop(ifm.group(1));
            if (args.size() == 3)
                out = "CASE WHEN " + args.get(0).trim() + " THEN " + args.get(1).trim()
                        + " ELSE " + args.get(2).trim() + " END";
        }
        return out.trim();
    }

    static String buildFk(TableDef d, Fk fk, String schema) {
        String onDel = fk.onDelete() == null ? "" : fk.onDelete().trim().toUpperCase();
        if (onDel.equals("RESTRICT")) onDel = "";                // 达梦不支持，NO ACTION 等价
        return "ALTER TABLE " + schema + '.' + ident(d.name) + " ADD CONSTRAINT " + ident(fk.name())
                + " FOREIGN KEY (" + fk.cols().stream().map(DbMigrator::ident).collect(Collectors.joining(",")) + ")"
                + " REFERENCES " + schema + '.' + ident(fk.refTable())
                + " (" + fk.refCols().stream().map(DbMigrator::ident).collect(Collectors.joining(",")) + ")"
                + (onDel.isEmpty() ? "" : " ON DELETE " + onDel);
    }

    static List<String> buildComments(TableDef d, String schema) {
        List<String> out = new ArrayList<>();
        if (d.comment != null)
            out.add("COMMENT ON TABLE " + schema + '.' + ident(d.name) + " IS '" + d.comment.replace("'", "''") + "'");
        for (Col c : d.cols)
            if (c.comment() != null)
                out.add("COMMENT ON COLUMN " + schema + '.' + ident(d.name) + '.' + ident(c.name())
                        + " IS '" + c.comment().replace("'", "''") + "'");
        return out;
    }

    // ==================== 模式管理 ====================

    static void createSchema(Connection dst, String schema) throws SQLException {
        try (Statement st = dst.createStatement()) {
            st.execute("CREATE SCHEMA " + schema + " AUTHORIZATION \"" + dst.getMetaData().getUserName() + "\"");
            System.out.println("[建模式] " + schema + " OK");
        } catch (SQLException e) {
            try (ResultSet rs = dst.getMetaData().getSchemas()) {
                while (rs.next())
                    if (rs.getString(1).equalsIgnoreCase(schema)) {
                        System.out.println("[建模式] " + schema + " 已存在，跳过");
                        return;
                    }
            }
            throw e;
        }
    }

    /** 幂等清理：先删目标模式全部外键（解除依赖），再删全部表。 */
    static void dropSchemaObjects(Connection dst, String schema) throws SQLException {
        List<String[]> fks = new ArrayList<>();
        try (ResultSet rs = dst.createStatement().executeQuery(
                "SELECT CONSTRAINT_NAME, TABLE_NAME FROM ALL_CONSTRAINTS WHERE OWNER='" + schema
                        + "' AND CONSTRAINT_TYPE='R'")) {
            while (rs.next()) fks.add(new String[]{rs.getString(1), rs.getString(2)});
        }
        int fkOk = 0;
        for (String[] fk : fks)
            try (Statement st = dst.createStatement()) {
                st.execute("ALTER TABLE " + schema + ".\"" + fk[1] + "\" DROP CONSTRAINT \"" + fk[0] + "\"");
                fkOk++;
            } catch (SQLException e) {
                System.out.println("  [外键删除跳过] " + fk[1] + "." + fk[0] + ": " + e.getMessage().trim());
            }
        List<String> tables = new ArrayList<>();
        try (ResultSet rs = dst.getMetaData().getTables(null, schema, "%", new String[]{"TABLE"})) {
            while (rs.next()) tables.add(rs.getString("TABLE_NAME"));
        }
        int tOk = 0;
        for (String t : tables)
            try (Statement st = dst.createStatement()) {
                st.execute("DROP TABLE " + schema + ".\"" + t + "\"");
                tOk++;
            } catch (SQLException e) {
                System.out.println("  [表删除失败] " + t + ": " + e.getMessage().trim());
            }
        System.out.println("[清理] 外键删除 " + fkOk + "/" + fks.size() + "，表删除 " + tOk + "/" + tables.size());
    }

    static void exec(Connection c, String sql, String label) {
        try (Statement st = c.createStatement()) {
            st.execute(sql);
            System.out.println("[" + label + "] OK");
        } catch (SQLException e) {
            System.out.println("[" + label + "] 失败: " + e.getErrorCode() + " " + abbrev(e.getMessage()));
            System.out.println("  SQL: " + abbrev(sql));
            throw new RuntimeException(label + " 失败", e);
        }
    }

    static void execQuiet(Connection c, String sql) {
        try (Statement st = c.createStatement()) { st.execute(sql); } catch (SQLException ignored) { }
    }

    // ==================== 数据搬运 ====================

    /**
     * 单表搬运：单列数值主键走窗口分页（WHERE pk > ? ORDER BY pk LIMIT n，内存友好），
     * 其余（复合/非数值/无主键）一次性流式读。失败自动降级逐行抢救。
     */
    static void copyData(Connection src, Connection dst, TableDef d, String schema,
                         int batch, Path errLog, List<String> warns) {
        List<String> cols = d.dataCols();
        if (cols.isEmpty()) { System.out.println("[数据] " + d.name + ": 无可搬运列，跳过"); return; }
        boolean windowed = d.singleNumericPk();
        if (!windowed) warns.add(d.name + ": 非单列数值主键，整表加载（大表请评估内存）");
        String srcCols = cols.stream().map(cn -> '`' + cn + '`').collect(Collectors.joining(","));
        String target = schema + '.' + ident(d.name);
        String ins = "INSERT INTO " + target + " (" + cols.stream().map(DbMigrator::ident)
                .collect(Collectors.joining(",")) + ") VALUES (" + String.join(",", Collections.nCopies(cols.size(), "?")) + ")";
        long t0 = System.currentTimeMillis();
        try {
            dst.setAutoCommit(false);
            if (d.hasIdentity) execDdl(dst, "SET IDENTITY_INSERT " + target + " ON");
            long copied = 0;
            if (windowed) {
                String pk = d.pk.get(0);
                long last = Long.MIN_VALUE;
                while (true) {
                    int n;
                    try (PreparedStatement q = src.prepareStatement(
                            "SELECT " + srcCols + " FROM `" + d.name + "` WHERE `" + pk + "` > ? ORDER BY `"
                                    + pk + "` LIMIT " + batch)) {
                        q.setLong(1, last);
                        try (ResultSet rs = q.executeQuery();
                             PreparedStatement ps = dst.prepareStatement(ins)) {
                            n = 0;
                            while (rs.next()) {
                                for (int i = 1; i <= cols.size(); i++) ps.setObject(i, rs.getObject(i));
                                last = Math.max(last, rs.getLong(pk));
                                ps.addBatch();
                                if (++n >= batch) break;
                            }
                            ps.executeBatch();
                        }
                    }
                    if (n == 0) break;
                    copied += n;
                    dst.commit();
                }
            } else {
                try (PreparedStatement q = src.prepareStatement(
                        "SELECT " + srcCols + " FROM `" + d.name + "`"
                                + (d.pk.isEmpty() ? "" : " ORDER BY " + d.pk.stream()
                                .map(p -> '`' + p + '`').collect(Collectors.joining(","))));
                     ResultSet rs = q.executeQuery();
                     PreparedStatement ps = dst.prepareStatement(ins)) {
                    int n = 0;
                    while (rs.next()) {
                        for (int i = 1; i <= cols.size(); i++) ps.setObject(i, rs.getObject(i));
                        ps.addBatch();
                        copied++;
                        if (++n >= batch) { ps.executeBatch(); dst.commit(); n = 0; }
                    }
                    if (n > 0) { ps.executeBatch(); dst.commit(); }
                }
            }
            if (d.hasIdentity) execDdl(dst, "SET IDENTITY_INSERT " + target + " OFF");
            System.out.printf("[数据] %-30s %8d 行  %.1fs%n", d.name, copied,
                    (System.currentTimeMillis() - t0) / 1000.0);
        } catch (Exception e) {
            try { dst.rollback(); } catch (SQLException ignored) { }
            try { if (d.hasIdentity) execDdl(dst, "SET IDENTITY_INSERT " + target + " OFF"); } catch (SQLException ignored) { }
            long salvaged = copyRowByRow(src, dst, d, schema, cols, errLog);
            System.out.printf("[数据] %s 批量失败(%s)，逐行抢救完成 %d 行%n", d.name, abbrev(e.getMessage()), salvaged);
        } finally {
            try { dst.setAutoCommit(true); } catch (SQLException ignored) { }
        }
    }

    static long totalRows(Connection src, TableDef d) throws SQLException {
        try (Statement st = src.createStatement(); ResultSet rs = st.executeQuery(
                "SELECT COUNT(*) FROM `" + d.name + "`")) { rs.next(); return rs.getLong(1); }
    }

    /** 批次失败后的逐行抢救：成功行逐行提交，坏行写错误日志并跳过。 */
    static long copyRowByRow(Connection src, Connection dst, TableDef d, String schema,
                             List<String> cols, Path errLog) {
        String srcCols = cols.stream().map(cn -> '`' + cn + '`').collect(Collectors.joining(","));
        String target = schema + '.' + ident(d.name);
        String ins = "INSERT INTO " + target + " (" + cols.stream().map(DbMigrator::ident)
                .collect(Collectors.joining(",")) + ") VALUES (" + String.join(",", Collections.nCopies(cols.size(), "?")) + ")";
        long ok = 0;
        List<String> bad = new ArrayList<>();
        try {
            dst.setAutoCommit(true);
            if (d.hasIdentity) execDdl(dst, "SET IDENTITY_INSERT " + target + " ON");
            try (PreparedStatement q = src.prepareStatement("SELECT " + srcCols + " FROM `" + d.name + "`");
                 ResultSet rs = q.executeQuery();
                 PreparedStatement ps = dst.prepareStatement(ins)) {
                while (rs.next()) {
                    try {
                        for (int i = 1; i <= cols.size(); i++) ps.setObject(i, rs.getObject(i));
                        ps.executeUpdate();
                        ok++;
                    } catch (SQLException e) {
                        bad.add("[行丢弃] " + d.name + " pk≈" + safePkValue(rs, d) + " : "
                                + e.getErrorCode() + " " + e.getMessage().trim());
                    }
                }
            }
            if (d.hasIdentity) execDdl(dst, "SET IDENTITY_INSERT " + target + " OFF");
        } catch (Exception fatal) {
            throw new RuntimeException("逐行抢救失败: " + fatal.getMessage(), fatal);
        }
        if (!bad.isEmpty()) {
            try {
                Files.write(errLog, bad, StandardCharsets.UTF_8,
                        StandardOpenOption.CREATE, StandardOpenOption.APPEND);
            } catch (IOException io) { bad.forEach(System.out::println); }
            System.out.println("  坏行 " + bad.size() + " 行已记录到 " + errLog.toAbsolutePath());
        }
        return ok;
    }

    static String safePkValue(ResultSet rs, TableDef d) {
        if (d.pk.size() != 1) return "?";
        try { return String.valueOf(rs.getObject(d.pk.get(0))); } catch (SQLException e) { return "?"; }
    }

    static void execDdl(Connection c, String sql) throws SQLException {
        try (Statement st = c.createStatement()) { st.execute(sql); }
    }

    // ==================== 校验 ====================

    static boolean verifyTable(Connection src, Connection dst, TableDef d, String schema) {
        try {
            long a, b;
            try (Statement st = src.createStatement(); ResultSet rs = st.executeQuery(
                    "SELECT COUNT(*) FROM `" + d.name + "`")) { rs.next(); a = rs.getLong(1); }
            try (Statement st = dst.createStatement(); ResultSet rs = st.executeQuery(
                    "SELECT COUNT(*) FROM " + schema + '.' + ident(d.name))) { rs.next(); b = rs.getLong(1); }
            String extra = "";
            if (d.singleNumericPk()) {
                String pk = d.pk.get(0);
                try (Statement st = src.createStatement(); ResultSet rs = st.executeQuery(
                        "SELECT COALESCE(MAX(`" + pk + "`),0) FROM `" + d.name + "`")) {
                    rs.next(); extra = " maxPk 源=" + rs.getLong(1);
                }
                try (Statement st = dst.createStatement(); ResultSet rs = st.executeQuery(
                        "SELECT COALESCE(MAX(" + ident(pk) + "),0) FROM " + schema + '.' + ident(d.name))) {
                    rs.next(); extra += " 目标=" + rs.getLong(1);
                }
            }
            boolean same = a == b;
            System.out.printf("  %-30s 源=%-8d 目标=%-8d %s%s%n", d.name, a, b, same ? "OK" : "!!不一致", extra);
            return same;
        } catch (SQLException e) {
            System.out.println("  " + d.name + " 校验失败: " + e.getMessage().trim());
            return false;
        }
    }

    static String abbrev(String s) {
        String one = s.replaceAll("\\s+", " ");
        return one.length() <= 200 ? one : one.substring(0, 200) + " ...(" + one.length() + " chars)";
    }
}