Tolan/backend/migrations/migrate_slot_tables_seed.py

"""
迁移脚本：创建槽位化三张表，并初始化 4 个幕布的 6 槽位与商品-槽位规则。

执行：
python migrations/migrate_slot_tables_seed.py
或：
python migrations/migrate_slot_tables_seed.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


BACKDROP_ITEM_CODES = (
    "fresh_green_wallpaper",
    "midnight_blue_wallpaper",
    "mocha_wallpaper",
    "springtime_wallpaper",
)

SLOT_ROWS = [
    ("slot_01_left_col_row2_mid", "左侧中段位", "Left Mid Slot", "floor", "none", 2, 1, 1, 1, "grid", 0, 0, None, None, 5, 10),
    ("slot_02_left_col_row4_5", "左下双格位", "Left Bottom Double Slot", "floor", "none", 4, 1, 2, 1, "grid", 0, 0, None, None, 6, 20),
    ("slot_03_center_2x3", "中部六格位", "Center 2x3 Slot", "wall", "window", 2, 2, 2, 3, "grid", 0, 0, None, None, 2, 30),
    ("slot_04_row4_col2", "第四行第二列位", "Row 4 Column 2 Slot", "floor", "seat", 4, 2, 1, 1, "grid", 0, 0, None, None, 7, 40),
    ("slot_05_row4_col3_4", "第四行右侧双格位", "Row 4 Right Double Slot", "floor", "none", 4, 3, 1, 2, "grid", 0, 0, None, None, 6, 50),
    ("slot_06_row4_col3_4_top_center_48x24", "右侧双格顶部小位", "Top Center Mini Slot", "floor", "none", 4, 3, 1, 2, "pixel", 72, 0, 48, 24, 8, 60),
]

# item_code -> slot_codes
ITEM_SLOT_RULES: dict[str, list[str]] = {
    "stump_stool": ["slot_04_row4_col2"],
    "forest_sofa": ["slot_04_row4_col2"],
    "vintage_table": ["slot_05_row4_col3_4"],
    "stump_table": ["slot_05_row4_col3_4"],
    "mushroom_table": ["slot_05_row4_col3_4"],
    "french_window": ["slot_03_center_2x3"],
    "garden_balcony": ["slot_03_center_2x3"],
    "forest_round_window": ["slot_03_center_2x3"],
    "terra_cotta_window": ["slot_03_center_2x3"],
    "coffee_mug": ["slot_06_row4_col3_4_top_center_48x24"],
    "snacks": ["slot_06_row4_col3_4_top_center_48x24"],
    "books": ["slot_06_row4_col3_4_top_center_48x24"],
    "energy_ball": ["slot_06_row4_col3_4_top_center_48x24"],
    "chinese_new_year_bloom": ["slot_02_left_col_row4_5"],
    "chocolate_tree": ["slot_02_left_col_row4_5"],
    "lilybell_lamp": ["slot_02_left_col_row4_5"],
    "nut_wind_bell": ["slot_01_left_col_row2_mid"],
    "photo_frame": ["slot_01_left_col_row2_mid"],
}


CREATE_SQL = """
CREATE TABLE IF NOT EXISTS `chatbot_backdrop_slots` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `backdrop_item_id` bigint(20) unsigned NOT NULL,
  `slot_code` varchar(64) NOT NULL,
  `slot_name` varchar(100) DEFAULT NULL,
  `slot_name_zh` varchar(100) DEFAULT NULL,
  `slot_name_en` varchar(100) DEFAULT NULL,
  `placement_type` enum('wall','floor') NOT NULL DEFAULT 'floor',
  `anchor_type` varchar(32) NOT NULL DEFAULT 'none',
  `start_row` tinyint(3) unsigned NOT NULL,
  `start_col` tinyint(3) unsigned NOT NULL,
  `size_rows` tinyint(3) unsigned NOT NULL DEFAULT '1',
  `size_cols` tinyint(3) unsigned NOT NULL DEFAULT '1',
  `render_mode` enum('grid','pixel') NOT NULL DEFAULT 'grid',
  `offset_x` smallint(6) NOT NULL DEFAULT '0',
  `offset_y` smallint(6) NOT NULL DEFAULT '0',
  `pixel_width` smallint(5) unsigned DEFAULT NULL,
  `pixel_height` smallint(5) unsigned DEFAULT NULL,
  `z_index` int(11) NOT NULL DEFAULT '0',
  `is_active` tinyint(1) NOT NULL DEFAULT '1',
  `sort_order` int(11) NOT NULL DEFAULT '0',
  `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_backdrop_slot_code` (`backdrop_item_id`,`slot_code`),
  KEY `idx_backdrop_slot_active` (`backdrop_item_id`,`is_active`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE IF NOT EXISTS `shop_item_slot_rules` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `item_id` bigint(20) unsigned NOT NULL,
  `backdrop_item_id` bigint(20) unsigned NOT NULL,
  `slot_code` varchar(64) NOT NULL,
  `is_default` tinyint(1) NOT NULL DEFAULT '0',
  `priority` int(11) NOT NULL DEFAULT '0',
  `is_active` tinyint(1) NOT NULL DEFAULT '1',
  `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_item_backdrop_slot` (`item_id`,`backdrop_item_id`,`slot_code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE IF NOT EXISTS `user_chatbot_slot_placements` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `user_id` bigint(20) unsigned NOT NULL,
  `backdrop_item_id` bigint(20) unsigned NOT NULL,
  `slot_code` varchar(64) NOT NULL,
  `user_asset_id` bigint(20) unsigned NOT NULL,
  `item_id` bigint(20) unsigned NOT NULL,
  `z_index` int(11) NOT NULL DEFAULT '0',
  `image_name_snapshot` varchar(255) NOT NULL,
  `placed_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_user_backdrop_slot` (`user_id`,`backdrop_item_id`,`slot_code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""


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:
            for stmt in [s.strip() for s in CREATE_SQL.split(";") if s.strip()]:
                await cursor.execute(stmt)
            print("✅ 表结构已创建/确认")

            # 兼容历史表结构：补充双语槽位名称字段
            await cursor.execute(
                """
                SELECT COUNT(*)
                FROM information_schema.columns
                WHERE table_schema = %s
                  AND table_name = 'chatbot_backdrop_slots'
                  AND column_name = 'slot_name_zh'
                """,
                (settings.MYSQL_DATABASE,),
            )
            has_slot_name_zh = (await cursor.fetchone() or [0])[0] > 0
            if not has_slot_name_zh:
                await cursor.execute(
                    """
                    ALTER TABLE chatbot_backdrop_slots
                    ADD COLUMN slot_name_zh varchar(100) DEFAULT NULL
                    """
                )

            await cursor.execute(
                """
                SELECT COUNT(*)
                FROM information_schema.columns
                WHERE table_schema = %s
                  AND table_name = 'chatbot_backdrop_slots'
                  AND column_name = 'slot_name_en'
                """,
                (settings.MYSQL_DATABASE,),
            )
            has_slot_name_en = (await cursor.fetchone() or [0])[0] > 0
            if not has_slot_name_en:
                await cursor.execute(
                    """
                    ALTER TABLE chatbot_backdrop_slots
                    ADD COLUMN slot_name_en varchar(100) DEFAULT NULL
                    """
                )

            await cursor.execute(
                f"SELECT id, item_code FROM shop_items WHERE item_code IN ({','.join(['%s']*len(BACKDROP_ITEM_CODES))})",
                BACKDROP_ITEM_CODES,
            )
            backdrop_rows = await cursor.fetchall()
            backdrop_code_to_id = {str(r[1]): int(r[0]) for r in backdrop_rows}
            missing_backdrops = [code for code in BACKDROP_ITEM_CODES if code not in backdrop_code_to_id]
            if missing_backdrops:
                print(f"❌ shop_items 缺少幕布 item_code: {missing_backdrops}")
                await conn.rollback()
                return False
            backdrop_ids = [backdrop_code_to_id[code] for code in BACKDROP_ITEM_CODES]

            for backdrop_id in backdrop_ids:
                for row in SLOT_ROWS:
                    (
                        slot_code,
                        slot_name,
                        slot_name_en,
                        placement_type,
                        anchor_type,
                        start_row,
                        start_col,
                        size_rows,
                        size_cols,
                        render_mode,
                        offset_x,
                        offset_y,
                        pixel_width,
                        pixel_height,
                        z_index,
                        sort_order,
                    ) = row
                    await cursor.execute(
                        """
                        INSERT INTO chatbot_backdrop_slots
                        (backdrop_item_id, slot_code, slot_name, slot_name_zh, slot_name_en, placement_type, anchor_type,
                         start_row, start_col, size_rows, size_cols, render_mode,
                         offset_x, offset_y, pixel_width, pixel_height, z_index, is_active, sort_order)
                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s)
                        ON DUPLICATE KEY UPDATE
                          slot_name=VALUES(slot_name),
                          slot_name_zh=VALUES(slot_name_zh),
                          slot_name_en=VALUES(slot_name_en),
                          placement_type=VALUES(placement_type),
                          anchor_type=VALUES(anchor_type),
                          start_row=VALUES(start_row),
                          start_col=VALUES(start_col),
                          size_rows=VALUES(size_rows),
                          size_cols=VALUES(size_cols),
                          render_mode=VALUES(render_mode),
                          offset_x=VALUES(offset_x),
                          offset_y=VALUES(offset_y),
                          pixel_width=VALUES(pixel_width),
                          pixel_height=VALUES(pixel_height),
                          z_index=VALUES(z_index),
                          is_active=1,
                          sort_order=VALUES(sort_order),
                          updated_at=NOW()
                        """,
                        (
                            backdrop_id,
                            slot_code,
                            slot_name,
                            slot_name,
                            slot_name_en,
                            placement_type,
                            anchor_type,
                            start_row,
                            start_col,
                            size_rows,
                            size_cols,
                            render_mode,
                            offset_x,
                            offset_y,
                            pixel_width,
                            pixel_height,
                            z_index,
                            sort_order,
                        ),
                    )
            print("✅ 槽位初始化完成")

            # 商品映射
            item_codes = list(ITEM_SLOT_RULES.keys())
            await cursor.execute(
                f"SELECT id, item_code FROM shop_items WHERE item_code IN ({','.join(['%s']*len(item_codes))})",
                tuple(item_codes),
            )
            item_rows = await cursor.fetchall()
            code_to_id = {r[1]: int(r[0]) for r in item_rows}
            missing = [c for c in item_codes if c not in code_to_id]
            if missing:
                print(f"❌ shop_items 缺少 item_code: {missing}")
                await conn.rollback()
                return False

            for item_code, slot_codes in ITEM_SLOT_RULES.items():
                item_id = code_to_id[item_code]
                for backdrop_id in backdrop_ids:
                    for idx, slot_code in enumerate(slot_codes):
                        await cursor.execute(
                            """
                            INSERT INTO shop_item_slot_rules
                              (item_id, backdrop_item_id, slot_code, is_default, priority, is_active)
                            VALUES (%s,%s,%s,%s,%s,1)
                            ON DUPLICATE KEY UPDATE
                              is_default=VALUES(is_default),
                              priority=VALUES(priority),
                              is_active=1,
                              updated_at=NOW()
                            """,
                            (item_id, backdrop_id, slot_code, 1 if idx == 0 else 0, 10 + idx * 10),
                        )
            print("✅ 商品-槽位规则初始化完成")

        await conn.commit()
        print("🎉 migrate_slot_tables_seed 完成")
        return True
    except Exception as exc:  # noqa: BLE001
        print(f"❌ 迁移失败: {exc}")
        if conn is not None:
            try:
                await conn.rollback()
            except Exception:
                pass
        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)