Tolan/backend/migrations/migrate_shop_items_bilingual_name.py

"""
迁移脚本：为 shop_items 增加 name_zh / name_en 并回填

执行：
python migrations/migrate_shop_items_bilingual_name.py
或：
python migrations/migrate_shop_items_bilingual_name.py --force
"""

import asyncio
import sys
from pathlib import Path
from typing import Any

import aiomysql

backend_dir = Path(__file__).resolve().parents[1]
if str(backend_dir) not in sys.path:
    sys.path.insert(0, str(backend_dir))

from config import settings


async def _column_exists(conn: aiomysql.Connection, table_name: str, column_name: str) -> bool:
    async with conn.cursor() as cursor:
        await cursor.execute(
            """
            SELECT COUNT(*)
            FROM information_schema.columns
            WHERE table_schema = %s
              AND table_name = %s
              AND column_name = %s
            """,
            (settings.MYSQL_DATABASE, table_name, column_name),
        )
        row: Any = await cursor.fetchone()
        return bool(row and row[0] > 0)


async def migrate() -> bool:
    conn: aiomysql.Connection | None = None
    try:
        conn = await aiomysql.connect(
            host=settings.MYSQL_HOST,
            port=settings.MYSQL_PORT,
            user=settings.MYSQL_USER,
            password=settings.MYSQL_PASSWORD,
            db=settings.MYSQL_DATABASE,
            charset="utf8mb4",
            autocommit=False,
        )
        async with conn.cursor() as cursor:
            has_name_zh = await _column_exists(conn, "shop_items", "name_zh")
            if not has_name_zh:
                await cursor.execute(
                    """
                    ALTER TABLE `shop_items`
                      ADD COLUMN `name_zh` VARCHAR(100) NOT NULL DEFAULT ''
                      COMMENT '商品中文名称'
                      AFTER `name`
                    """
                )
                print("✅ 已新增列 shop_items.name_zh")
            else:
                print("ℹ️ 列 shop_items.name_zh 已存在，跳过新增")

            has_name_en = await _column_exists(conn, "shop_items", "name_en")
            if not has_name_en:
                await cursor.execute(
                    """
                    ALTER TABLE `shop_items`
                      ADD COLUMN `name_en` VARCHAR(100) NOT NULL DEFAULT ''
                      COMMENT '商品英文名称'
                      AFTER `name_zh`
                    """
                )
                print("✅ 已新增列 shop_items.name_en")
            else:
                print("ℹ️ 列 shop_items.name_en 已存在，跳过新增")

            await cursor.execute(
                """
                UPDATE `shop_items`
                SET `name_zh` = `name`
                WHERE (`name_zh` IS NULL OR `name_zh` = '')
                """
            )
            print(f"✅ 已回填 name_zh：{cursor.rowcount or 0} 行")

            await cursor.execute(
                """
                UPDATE `shop_items`
                SET `name_en` = CONCAT(
                  UPPER(LEFT(REPLACE(`item_code`, '_', ' '), 1)),
                  LOWER(SUBSTRING(REPLACE(`item_code`, '_', ' '), 2))
                )
                WHERE (`name_en` IS NULL OR `name_en` = '')
                """
            )
            print(f"✅ 已回填 name_en：{cursor.rowcount or 0} 行")

        await conn.commit()
        print("🎉 双语名称迁移完成")
        return True
    except Exception as exc:  # noqa: BLE001
        if conn is not None:
            try:
                await conn.rollback()
            except Exception:
                pass
        print(f"❌ 迁移失败: {exc}")
        return False
    finally:
        if conn is not None:
            conn.close()


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--force":
        confirm = "y"
    else:
        confirm = input("\n是否继续执行双语名称迁移? (y/n): ").strip().lower()

    if confirm != "y":
        sys.exit(0)

    ok = asyncio.run(migrate())
    sys.exit(0 if ok else 1)