labor-arbitration-system/backend/app/migrate.py

from __future__ import annotations

from sqlalchemy import text
from sqlalchemy.engine import Engine


def _table_exists(engine: Engine, table: str) -> bool:
    with engine.connect() as conn:
        r = conn.execute(text("SHOW TABLES LIKE :t"), {"t": table}).fetchone()
        return r is not None


def _column_exists(engine: Engine, table: str, column: str) -> bool:
    with engine.connect() as conn:
        r = conn.execute(text(f"SHOW COLUMNS FROM `{table}` LIKE :c"), {"c": column}).fetchone()
        return r is not None


def _add_column(engine: Engine, table: str, ddl: str) -> None:
    with engine.connect() as conn:
        conn.execute(text(f"ALTER TABLE `{table}` ADD COLUMN {ddl}"))
        conn.commit()


def _column_info(engine: Engine, table: str, column: str) -> dict | None:
    with engine.connect() as conn:
        row = conn.execute(text(f"SHOW COLUMNS FROM `{table}` LIKE :c"), {"c": column}).mappings().fetchone()
        return dict(row) if row else None


def _modify_column(engine: Engine, table: str, ddl: str) -> None:
    with engine.connect() as conn:
        conn.execute(text(f"ALTER TABLE `{table}` MODIFY COLUMN {ddl}"))
        conn.commit()


def ensure_mysql_schema(engine: Engine) -> None:
    """
    轻量迁移（不依赖 Alembic）：
    - 若数据库已有旧版 cases 表，则自动补齐新版所需列
    - 新表（case_files/case_elements_versions/processing_tasks）由 create_all 创建；若表已存在则跳过

    说明：此方法适用于毕业设计原型开发期；正式环境建议使用 Alembic 管理迁移。
    """
    if not _table_exists(engine, "cases"):
        return

    # 旧版 cases 表字段：id, case_name, source_files, raw_text, extracted_elements, case_profile, created_at
    # 新版需要字段：case_no, elements, portrait, applicant_name, respondent_name, case_cause, risk_level
    additions = [
        ("case_no", "`case_no` VARCHAR(64) NULL"),
        ("elements", "`elements` JSON NOT NULL"),
        ("portrait", "`portrait` JSON NOT NULL"),
        ("applicant_name", "`applicant_name` VARCHAR(64) NULL"),
        ("respondent_name", "`respondent_name` VARCHAR(128) NULL"),
        ("case_cause", "`case_cause` VARCHAR(64) NULL"),
        ("risk_level", "`risk_level` VARCHAR(16) NULL"),
        ("application_text", "`application_text` LONGTEXT NULL"),
        ("ruling_result", "`ruling_result` TEXT NULL"),
        ("arbitration_org", "`arbitration_org` VARCHAR(255) NULL"),
        ("created_at", "`created_at` DATETIME NULL"),
        ("material_fingerprint", "`material_fingerprint` VARCHAR(64) NULL"),
    ]

    for col, ddl in additions:
        if not _column_exists(engine, "cases", col):
            _add_column(engine, "cases", ddl)

    if _column_exists(engine, "cases", "material_fingerprint"):
        with engine.connect() as conn:
            idx = conn.execute(
                text(
                    "SELECT 1 FROM information_schema.statistics "
                    "WHERE table_schema = DATABASE() AND table_name = 'cases' AND index_name = 'ix_cases_material_fp'"
                )
            ).fetchone()
            if not idx:
                try:
                    conn.execute(text("CREATE INDEX `ix_cases_material_fp` ON `cases` (`material_fingerprint`)"))
                    conn.commit()
                except Exception:
                    conn.rollback()

    # 旧表约束放宽：避免旧必填字段导致新插入失败
    # 常见旧字段：source_files/raw_text/extracted_elements/case_profile 可能为 NOT NULL 且无默认值
    relax = [
        ("source_files", "`source_files` JSON NULL"),
        ("raw_text", "`raw_text` LONGTEXT NULL"),
        ("extracted_elements", "`extracted_elements` JSON NULL"),
        ("case_profile", "`case_profile` JSON NULL"),
    ]
    for col, ddl in relax:
        info = _column_info(engine, "cases", col)
        if not info:
            continue
        # info: Field, Type, Null, Key, Default, Extra
        if str(info.get("Null", "")).upper() == "NO" and info.get("Default") is None:
            _modify_column(engine, "cases", ddl)

    # 兼容旧字段：如果旧表里没有 elements/portrait，但有 extracted_elements/case_profile，
    # 这里不做数据迁移（避免 JSON 兼容差异）；后续写入会以新字段为主。

    _ensure_case_child_fks_cascade(engine)


def _ensure_case_child_fks_cascade(engine: Engine) -> None:
    """
    将 case_files / case_elements_versions / processing_tasks 指向 cases.id 的外键改为 ON DELETE CASCADE，
    避免手工或 ORM 删除父行时出现 Error 1451。
    """
    child_tables = ("case_files", "case_elements_versions", "processing_tasks")
    with engine.connect() as conn:
        for table in child_tables:
            if not _table_exists(engine, table):
                continue
            rows = conn.execute(
                text(
                    """
                    SELECT rc.CONSTRAINT_NAME, rc.DELETE_RULE
                    FROM information_schema.REFERENTIAL_CONSTRAINTS rc
                    INNER JOIN information_schema.KEY_COLUMN_USAGE kcu
                        ON rc.CONSTRAINT_SCHEMA = kcu.CONSTRAINT_SCHEMA
                        AND rc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
                        AND rc.TABLE_NAME = kcu.TABLE_NAME
                    WHERE rc.CONSTRAINT_SCHEMA = DATABASE()
                      AND rc.TABLE_NAME = :tbl
                      AND kcu.REFERENCED_TABLE_NAME = 'cases'
                      AND kcu.REFERENCED_COLUMN_NAME = 'id'
                    """
                ),
                {"tbl": table},
            ).fetchall()
            for row in rows:
                cname, rule = row[0], str(row[1] or "").upper()
                if rule == "CASCADE":
                    continue
                try:
                    conn.execute(text(f"ALTER TABLE `{table}` DROP FOREIGN KEY `{cname}`"))
                    conn.execute(
                        text(
                            f"ALTER TABLE `{table}` ADD CONSTRAINT `{cname}` "
                            f"FOREIGN KEY (`case_id`) REFERENCES `cases` (`id`) ON DELETE CASCADE"
                        )
                    )
                    conn.commit()
                except Exception:
                    conn.rollback()