#!/usr/bin/env python3
"""建表脚本：从各模块 models/*.json 经 json2ddl 生成 DDL，幂等落库。

铁律（2026-09-07 用户定夺）：
  1. DDL 一律由 json2ddl 从 models/*.json 生成——models JSON 是表结构唯一事实源，
     禁止手写 CREATE TABLE（历史手写段已全部移除，含 pipeline-llm mysql.ddl.sql 依赖）。
  2. 全程幂等：CREATE TABLE IF NOT EXISTS，**绝不 DROP**——2026-09-07 14:08 事故：
     原 TABLE_MODULES 走 DROP 重建，一次部署清空 pipeline 库 product/pricing/discount/sms
     全部数据（13 个模型定价丢失、27 个产品映射丢失、全平台记账 failed）。
  3. 存量表 schema 演进靠「通用列同步器」：解析 DDL 的期望列 vs information_schema 实际列，
     缺列 ALTER TABLE ADD COLUMN 补齐。加字段只改 models JSON，部署自动补列。
  4. 索引幂等：按 (表, 列集合, 唯一性) 判重后 CREATE（不按索引名——手写时代的旧索引名
     与 json2ddl 生成名不同，按名判重会重复建同字段索引）。

用法：
  py3/bin/python scripts/create_tables.py
"""
import sys, os, re, json, asyncio, subprocess, glob

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.dirname(SCRIPT_DIR)
sys.path.insert(0, os.path.join(ROOT_DIR, 'py3', 'lib', 'python3.10', 'site-packages'))
sys.path.insert(0, ROOT_DIR)

from sqlor.dbpools import DBPools
from appPublic.jsonConfig import getConfig
from appPublic.folderUtils import ProgramPath
from ahserver.serverenv import ServerEnv
from ahserver.globalEnv import initEnv

# 不建表的模块（无 models/ 或表由宿主/其他机制管理）
SKIP_MODULES = set()

# appbase params 默认参数（数据种子，非 schema；params 表结构走 models JSON）
_PARAMS_INIT = [
    ("workspace_base", "/d/pipeline/workspaces"),
    ("task_max_retry", "3"),
    ("register_open", "1"),
]

# sd_iterations 存量数据修复（数据层，非 schema；seq_no 列由通用列同步器补）
_DATA_FIXUPS = [
    # 「当前迭代」改为显式 status='in_progress'，迭代按 seq_no 顺序编号推进
    ("sd_iterations seq_no 回填",
     "UPDATE sd_iterations t JOIN ("
     "SELECT id, ROW_NUMBER() OVER (PARTITION BY project_id ORDER BY created_at ASC, id ASC) AS rn "
     "FROM sd_iterations WHERE seq_no = 0) x ON x.id = t.id "
     "SET t.seq_no = x.rn"),
    ("sd_iterations 脏状态修复 active→in_progress",
     "UPDATE sd_iterations SET status='in_progress', updated_at=NOW() WHERE status='active'"),
]


def split_ddl(ddl):
    """按分号分割 DDL，跳过注释行和空语句。

    引号感知：单引号字符串内的分号不作为分隔符（实测事故 2026-09-07：
    llm_usage 的 COMMENT '记账状态(三态:...;仅status=SUCCEEDED...)' 注释含分号，
    裸 split(';') 把 CREATE TABLE 拦腰截断 → 语法错误 → 该模块建表整段中止）。
    """
    stmts = []
    cur = []
    in_str = False
    i = 0
    n = len(ddl)
    while i < n:
        ch = ddl[i]
        if in_str:
            cur.append(ch)
            if ch == "'":
                if i + 1 < n and ddl[i + 1] == "'":   # '' 转义，仍在串内
                    cur.append("'")
                    i += 1
                else:
                    in_str = False
            elif ch == '\\' and i + 1 < n:
                cur.append(ddl[i + 1])                # 反斜杠转义（MySQL 风格）
                i += 1
        elif ch == "'":
            in_str = True
            cur.append(ch)
        elif ch == ';':
            raw = ''.join(cur)
            lines = [l for l in raw.split('\n') if not l.strip().startswith('--')]
            s = '\n'.join(lines).strip()
            if s:
                stmts.append(s)
            cur = []
        else:
            cur.append(ch)
        i += 1
    raw = ''.join(cur)
    lines = [l for l in raw.split('\n') if not l.strip().startswith('--')]
    s = '\n'.join(lines).strip()
    if s:
        stmts.append(s)
    return stmts


def parse_create_table(stmt):
    """从 CREATE TABLE 语句解析 (表名, {列名: 列定义DDL})。

    列定义直接取自 json2ddl 输出——类型映射逻辑单一来源（sqlor 模板），不在此重复实现。
    """
    m = re.search(r'(?is)CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?\s*\(', stmt)
    if not m:
        return None, {}
    table = m.group(1)
    body = stmt[m.end():]
    # 截到最后一个 ) 之前（表选项在括号外）
    depth = 1
    end = -1
    in_str = False
    for i, ch in enumerate(body):
        if ch == "'" and (i == 0 or body[i - 1] != '\\'):
            in_str = not in_str
        elif not in_str:
            if ch == '(':
                depth += 1
            elif ch == ')':
                depth -= 1
                if depth == 0:
                    end = i
                    break
    if end < 0:
        return table, {}
    cols = {}
    for line in body[:end].split('\n'):
        line = line.strip().rstrip(',')
        cm = re.match(r'`(\w+)`\s+(.+)$', line)
        if cm:
            cols[cm.group(1)] = cm.group(2).strip()
    return table, cols


def parse_create_index(stmt):
    """解析 CREATE [UNIQUE] INDEX 语句 → (索引名, 表名, 唯一?, 列列表)。"""
    m = re.match(r'(?is)CREATE\s+(UNIQUE\s+)?INDEX\s+`?(\w+)`?\s+ON\s+`?(\w+)`?\s*\(([^)]+)\)', stmt)
    if not m:
        return None
    unique = bool(m.group(1))
    name, table = m.group(2), m.group(3)
    cols = [c.strip().strip('`') for c in m.group(4).split(',')]
    return name, table, unique, cols


async def get_existing_indexes(sor, table):
    """现网索引集合：{(唯一?, (列...)): 索引名}——按列集合判重，免疫索引改名。"""
    recs = await sor.sqlExe(
        "SELECT index_name, non_unique, column_name, seq_in_index "
        "FROM information_schema.statistics "
        "WHERE table_schema=DATABASE() AND table_name=${t}$ "
        "ORDER BY index_name, seq_in_index", {"t": table})
    await sor.sqlExe("COMMIT", {})
    acc = {}
    for r in (recs or []):
        name = getattr(r, 'index_name', '')
        acc.setdefault((name, str(getattr(r, 'non_unique', ''))), []).append(
            getattr(r, 'column_name', ''))
    return {(nu == '0', tuple(cols)): name for (name, nu), cols in acc.items()}


async def get_existing_columns(sor, table):
    recs = await sor.sqlExe(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": table})
    await sor.sqlExe("COMMIT", {})
    return {getattr(r, 'column_name', '') for r in (recs or [])}


async def table_exists(sor, table):
    recs = await sor.sqlExe(
        "SELECT COUNT(*) AS c FROM information_schema.tables "
        "WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": table})
    await sor.sqlExe("COMMIT", {})
    return bool(recs) and int(getattr(recs[0], 'c', 0) or 0) > 0


async def apply_module_ddl(sor, mod, models_dir, stats):
    """单模块：json2ddl → 幂等建表 → 列同步 → 索引幂等。"""
    json2ddl = os.path.join(ROOT_DIR, 'py3', 'bin', 'json2ddl')
    r = subprocess.run([json2ddl, 'mysql', models_dir], capture_output=True, text=True)
    if r.returncode != 0 or not r.stdout.strip():
        print(f'  skip {mod}: json2ddl failed ({(r.stderr or "").strip()[:120]})')
        return

    creates, indexes = [], []
    for stmt in split_ddl(r.stdout):
        head = stmt.lstrip('(').lstrip().split(None, 2)
        kw = (head[0] + ' ' + (head[1] if len(head) > 1 else '')).upper()
        if re.match(r'(?i)^\s*DROP\s+TABLE', stmt):
            continue                      # 铁律：绝不 DROP（2026-09-07 清库事故根治）
        if re.match(r'(?i)^\s*CREATE\s+TABLE', stmt):
            creates.append(stmt)
        elif re.match(r'(?i)^\s*CREATE\s+(UNIQUE\s+)?INDEX', stmt):
            indexes.append(stmt)
        elif re.match(r'(?i)^\s*ALTER\s+TABLE', stmt):
            pass                          # json2ddl 不产 ALTER；存量演进走列同步器
        else:
            stats['other'] += 1

    # 1) 建表（IF NOT EXISTS）+ 收集期望列
    expected_cols = {}
    for stmt in creates:
        table, cols = parse_create_table(stmt)
        if not table:
            continue
        expected_cols[table] = cols
        s = re.sub(r'(?i)CREATE\s+TABLE\s+', 'CREATE TABLE IF NOT EXISTS ', stmt, count=1)
        existed = await table_exists(sor, table)
        try:
            await sor.execute(s, {})
            if not existed:
                stats['created'] += 1
                print(f'    + table {table} (新建)')
        except Exception as e:
            print(f'    ! table {table} 建表失败: {str(e)[:160]}')
            stats['errors'] += 1

    # 2) 列同步（存量表补新列——models JSON 是唯一事实源）
    for table, cols in expected_cols.items():
        if not await table_exists(sor, table):
            continue
        have = await get_existing_columns(sor, table)
        for col, colddl in cols.items():
            if col in have:
                continue
            try:
                await sor.execute(
                    f'ALTER TABLE `{table}` ADD COLUMN `{col}` {colddl}', {})
                stats['cols_added'] += 1
                print(f'    + column {table}.{col} (存量表补列)')
            except Exception as e:
                print(f'    ! column {table}.{col} 补列失败: {str(e)[:160]}')
                stats['errors'] += 1

    # 3) 索引幂等（按 列集合+唯一性 判重，免疫手写时代的旧索引名）
    idx_cache = {}
    for stmt in indexes:
        parsed = parse_create_index(stmt)
        if not parsed:
            continue
        name, table, unique, cols = parsed
        if not await table_exists(sor, table):
            continue
        if table not in idx_cache:
            idx_cache[table] = await get_existing_indexes(sor, table)
        if (unique, tuple(cols)) in idx_cache[table]:
            stats['idx_skipped'] += 1
            continue
        try:
            await sor.execute(stmt, {})
            idx_cache[table][(unique, tuple(cols))] = name
            stats['idx_created'] += 1
            print(f'    + index {table}({",".join(cols)}){" UNIQUE" if unique else ""}')
        except Exception as e:
            msg = str(e)[:160]
            if 'Duplicate key name' in msg:
                stats['idx_skipped'] += 1
            else:
                print(f'    ! index {name} 失败: {msg}')
                stats['errors'] += 1


async def main():
    config = getConfig(ROOT_DIR, NS={'workdir': ROOT_DIR, 'ProgramPath': ProgramPath()})
    DBPools(config.databases)
    initEnv()
    env = ServerEnv()
    env.get_module_dbname = lambda m: 'pipeline' if 'pipeline' in m else 'sage'

    # 动态扫描 pkgs/*/models——不再维护静态清单
    #（历史教训：清单遗漏 → pricing 定价管理 500、product/pricing 走 DROP 清库）
    mods = []
    for d in sorted(glob.glob(os.path.join(ROOT_DIR, 'pkgs', '*', 'models'))):
        mod = os.path.basename(os.path.dirname(d))
        if mod in SKIP_MODULES:
            continue
        if not glob.glob(os.path.join(d, '*.json')):
            continue
        mods.append((mod, d))
    print(f'  modules with models/: {len(mods)}')

    stats = {'created': 0, 'cols_added': 0, 'idx_created': 0, 'idx_skipped': 0,
             'errors': 0, 'other': 0}
    async with DBPools().sqlorContext('pipeline') as sor:
        for mod, models_dir in mods:
            print(f'  === {mod} ===')
            await apply_module_ddl(sor, mod, models_dir, stats)

        # appbase params 默认参数（数据种子）
        try:
            for pname, pval in _PARAMS_INIT:
                await sor.execute(
                    "INSERT INTO params (id, params_name, params_value) "
                    "VALUES (${id}$, ${n}$, ${v}$) "
                    "ON DUPLICATE KEY UPDATE params_name=params_name",
                    {"id": pname, "n": pname, "v": pval})
            print('  seed: appbase params ensured (workspace_base/task_max_retry/register_open)')
        except Exception as e:
            print(f'  WARN: appbase params seed failed: {e}')

        # 存量数据修复（非 schema）
        for label, sql in _DATA_FIXUPS:
            try:
                await sor.execute(sql, {})
                print(f'  fixup: {label}')
            except Exception as e:
                print(f'  WARN: {label} failed: {str(e)[:120]}')

    print(f'  DONE: 新建表={stats["created"]} 补列={stats["cols_added"]} '
          f'新建索引={stats["idx_created"]} 索引已存在跳过={stats["idx_skipped"]} '
          f'错误={stats["errors"]}')
    if stats['errors']:
        sys.exit(1)


if __name__ == '__main__':
    asyncio.run(main())