baodan/api/insurance/db/migrate_023.py

"""迁移 023: 产品小册子异步解析 + 保司多Logo。

- 扩展 insurance_ppt_products：新增小册子解析状态字段
- 新建 insurance_ppt_company_logos：保司多Logo表
- 清理卡住的 parsing 数据

兼容 PostgreSQL 和 MySQL。
"""
import logging
from sqlalchemy import text

logger = logging.getLogger(__name__)


def _column_exists(db, table: str, column: str) -> bool:
    """检查列是否存在（兼容 PostgreSQL 和 MySQL）。"""
    try:
        result = db.session.execute(text(
            "SELECT COUNT(*) FROM information_schema.columns "
            "WHERE table_name = :table AND column_name = :column"
        ), {"table": table, "column": column})
        return result.scalar() > 0
    except Exception:
        return False


def _table_exists(db, table_name: str) -> bool:
    """检查表是否存在（兼容 PostgreSQL 和 MySQL）。"""
    try:
        result = db.session.execute(text(
            "SELECT COUNT(*) FROM information_schema.tables "
            "WHERE table_name = :table"
        ), {"table": table_name})
        return result.scalar() > 0
    except Exception:
        return False


def _index_exists(db, index_name: str) -> bool:
    """检查索引是否存在（兼容 PostgreSQL 和 MySQL）。"""
    try:
        result = db.session.execute(text(
            "SELECT COUNT(*) FROM pg_indexes WHERE indexname = :name"
        ), {"name": index_name})
        return result.scalar() > 0
    except Exception:
        return False


def migrate():
    """执行迁移。"""
    from insurance.db.compat import db

    # ─── 扩展 PptProduct 小册子字段 ───────────────────────────
    manual_columns = [
        ("manual_parse_message", "VARCHAR(500) DEFAULT ''"),
        ("manual_parse_error", "TEXT"),
        ("manual_parse_task_id", "VARCHAR(200)"),
        ("manual_parse_started_at", "TIMESTAMP"),
        ("manual_parse_finished_at", "TIMESTAMP"),
    ]

    for col_name, col_def in manual_columns:
        if not _column_exists(db, "insurance_ppt_products", col_name):
            db.session.execute(text(
                f"ALTER TABLE insurance_ppt_products ADD COLUMN {col_name} {col_def}"
            ))
            logger.info(f"[migrate_023] 已为 insurance_ppt_products 添加 {col_name} 列")

    # ─── 清理卡住的 parsing 数据 ─────────────────────────────
    # 将 manual_parse_status='parsing' 且无有效 manual_parsed_rules 的记录改为 failed
    result = db.session.execute(text(
        "UPDATE insurance_ppt_products "
        "SET manual_parse_status = 'failed', "
        "    manual_parse_error = '任务因服务重启而中断，请重新解析', "
        "    manual_parse_finished_at = CURRENT_TIMESTAMP "
        "WHERE manual_parse_status = 'parsing' "
        "AND (manual_parsed_rules IS NULL OR manual_parsed_rules = '')"
    ))
    if result.rowcount > 0:
        logger.info(f"[migrate_023] 已恢复 {result.rowcount} 个卡住的 parsing 记录为 failed")

    # ─── 新建 insurance_ppt_company_logos ────────────────────
    if not _table_exists(db, "insurance_ppt_company_logos"):
        db.session.execute(text("""
            CREATE TABLE insurance_ppt_company_logos (
                id VARCHAR(36) PRIMARY KEY,
                company_id VARCHAR(50) NOT NULL,
                file_path VARCHAR(500) NOT NULL,
                file_url VARCHAR(500),
                original_name VARCHAR(200),
                mime_type VARCHAR(50),
                file_size INT,
                is_primary BOOLEAN DEFAULT false,
                sort_order INT DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """))
        logger.info("[migrate_023] 已创建 insurance_ppt_company_logos 表")

        # 创建索引
        indexes = [
            ("idx_company_logos_company", "company_id"),
            ("idx_company_logos_primary", "company_id, is_primary"),
        ]
        for idx_name, idx_cols in indexes:
            if not _index_exists(db, idx_name):
                db.session.execute(text(
                    f"CREATE INDEX {idx_name} ON insurance_ppt_company_logos ({idx_cols})"
                ))
                logger.info(f"[migrate_023] 已创建索引 {idx_name}")

    db.session.commit()
    logger.info("[migrate_023] 产品小册子异步解析 + 保司多Logo 迁移完成")