Tolan/backend/services/user_service.py

"""
用户服务模块（SQLAlchemy ORM 版本）
"""
import uuid
import random
import asyncio
import logging
from typing import Dict, Optional, Any, Tuple
from datetime import datetime, date
from sqlalchemy import select, update, func, insert, desc, text
from sqlalchemy.ext.asyncio import AsyncSession
from database import db_manager
from config import settings
from models_db import (
    User,
    UserUIState,
    UserBackgroundImage,
    BackgroundImage,
    UserAvatar,
    VerificationCode,
    UserCommunityChoice,
    UserEarlyCommunity,
    PageVisitLog,
    UserIntimacy,
    UserEvent,
)
from error_codes import (
    PHONE_ALREADY_REGISTERED,
    USER_ID_NOT_FOUND,
    ALREADY_REGISTERED,
    EMAIL_ALREADY_REGISTERED,
    QQ_ALREADY_BOUND,
    EMAIL_REQUIRED,
    PHONE_REQUIRED,
    CODE_REQUIRED,
    CODE_INVALID,
)
from shop_item_image_utils import scene_image_name_from_shop_row

logger = logging.getLogger(__name__)


class AuthError(Exception):
    """业务校验错误，带错误码供 API 层返回 detail={"code": "XXX"}"""

    def __init__(self, code: str):
        self.code = code
        super().__init__(code)

# 默认头像 ID
DEFAULT_AVATAR_ID = "avatar_default"
REGISTERED_AVATAR_ID = "avatar_registered_cap"

LEVEL_THRESHOLDS_MS: Tuple[Tuple[int, int], ...] = (
    (5, 240_000),
    (4, 120_000),
    (3, 60_000),
    (2, 30_000),
    (1, 10_000),
)


class UserService:
    """用户管理服务"""

    @staticmethod
    def _calculate_user_level(total_ms: int) -> int:
        """
        根据累计说话时长计算用户等级
        """
        for level, threshold in LEVEL_THRESHOLDS_MS:
            if total_ms >= threshold:
                return level
        return 0
    
    async def _init_user_background_image(self, session: AsyncSession, user_id: str) -> None:
        """
        初始化用户背景图片（添加到已获得列表）
        
        Args:
            session: 数据库会话
            user_id: 用户ID
        """
        try:
            default_image_id = "default"
            
            # 检查是否已存在，避免重复添加
            result = await session.execute(
                select(UserBackgroundImage).where(
                    UserBackgroundImage.user_id == user_id,
                    UserBackgroundImage.image_id == default_image_id
                )
            )
            existing = result.scalar_one_or_none()
            
            if not existing:
                user_bg_image = UserBackgroundImage(
                    user_id=user_id,
                    image_id=default_image_id,
                    obtained_type="default",
                    obtained_source="system_init"
                )
                session.add(user_bg_image)
                logger.debug(f"✅ 为用户初始化背景图片: user_id={user_id}, image_id={default_image_id}")
            else:
                logger.debug(f"⏭️ 用户已拥有默认背景图片: user_id={user_id}")
                
        except Exception as e:
            # 背景图片初始化失败不应该影响用户创建，只记录日志
            logger.warning(f"⚠️ 初始化用户背景图片失败: user_id={user_id}, error={e}")
    
    async def _init_user_avatar(self, session: AsyncSession, user_id: str, user_type: str = "guest") -> None:
        """
        初始化用户头像（添加到已获得列表）
        
        Args:
            session: 数据库会话
            user_id: 用户ID
            user_type: 用户类型（guest 或 registered）
        """
        try:
            # 游客用户只分配默认头像
            # 注册用户同时分配默认头像和注册赠送头像
            avatars_to_add = []
            
            if user_type == "registered":
                # 注册用户：默认头像 + 注册赠送头像
                avatars_to_add = [
                    (DEFAULT_AVATAR_ID, "default", "system_init"),
                    (REGISTERED_AVATAR_ID, "register", "register_bonus"),
                ]
            else:
                # 游客用户：只分配默认头像
                avatars_to_add = [
                    (DEFAULT_AVATAR_ID, "default", "system_init"),
                ]
            
            for avatar_id, obtained_type, obtained_source in avatars_to_add:
                # 检查是否已存在，避免重复添加
                result = await session.execute(
                    select(UserAvatar).where(
                        UserAvatar.user_id == user_id,
                        UserAvatar.avatar_id == avatar_id
                    )
                )
                existing = result.scalar_one_or_none()
                
                if not existing:
                    user_avatar = UserAvatar(
                        user_id=user_id,
                        avatar_id=avatar_id,
                        obtained_type=obtained_type,
                        obtained_source=obtained_source
                    )
                    session.add(user_avatar)
                    logger.debug(f"✅ 为用户初始化头像: user_id={user_id}, avatar_id={avatar_id}, user_type={user_type}")
                else:
                    logger.debug(f"⏭️ 用户已拥有头像: user_id={user_id}, avatar_id={avatar_id}")
                
        except Exception as e:
            # 头像初始化失败不应该影响用户创建，只记录日志
            logger.warning(f"⚠️ 初始化用户头像失败: user_id={user_id}, error={e}")

    async def _mysql_table_exists(self, session: AsyncSession, table_name: str) -> bool:
        try:
            r = await session.execute(
                text(
                    """
                    SELECT COUNT(*) AS c
                    FROM information_schema.tables
                    WHERE table_schema = :db AND table_name = :table
                    """
                ),
                {"db": settings.MYSQL_DATABASE, "table": table_name},
            )
            row = r.mappings().first()
            return bool(row and int(row.get("c") or 0) > 0)
        except Exception:
            return False

    async def _mysql_column_data_type(
        self, session: AsyncSession, table_name: str, column_name: str
    ) -> str:
        try:
            r = await session.execute(
                text(
                    """
                    SELECT DATA_TYPE
                    FROM information_schema.columns
                    WHERE table_schema = :db
                      AND table_name = :table
                      AND column_name = :column
                    """
                ),
                {"db": settings.MYSQL_DATABASE, "table": table_name, "column": column_name},
            )
            row = r.mappings().first()
            return str((row or {}).get("DATA_TYPE") or "")
        except Exception:
            return ""

    async def _chatbot_mysql_user_id_value(
        self, session: AsyncSession, table_name: str, user_id_uuid: str
    ) -> str | int | None:
        dt = await self._mysql_column_data_type(session, table_name, "user_id")
        if not dt:
            return None
        if "bigint" in dt.lower():
            r = await session.execute(select(User.id).where(User.user_id == user_id_uuid))
            rid = r.scalar_one_or_none()
            return int(rid) if rid is not None else None
        return user_id_uuid

    async def _init_registered_user_chatbot_shop(self, session: AsyncSession, user_id_uuid: str) -> None:
        """
        新注册用户：赠送小木桩/落地窗/清新绿幕布商品，并写入默认幕布与 5x4 场景布局。
        表或商品不存在时跳过，不阻断注册。
        """
        required = (
            "shop_items",
            "user_assets",
            "chatbot_layouts",
            "user_chatbot_backdrops",
        )
        # 显式逐个 await，避免把 await 放进生成器表达式导致的潜在迭代异常
        for t in required:
            exists = await self._mysql_table_exists(session, t)
            if not exists:
                logger.warning(
                    "⚠️ Chatbot 注册礼包：缺少必需表，跳过 user_id=%s missing_table=%s",
                    user_id_uuid,
                    t,
                )
                return

        gift_codes = ("stump_stool", "french_window", "fresh_green_wallpaper")
        used_qty_by_code: Dict[str, int] = {
            "stump_stool": 1,
            "french_window": 1,
            "fresh_green_wallpaper": 0,
        }

        try:
            ua_uid = await self._chatbot_mysql_user_id_value(session, "user_assets", user_id_uuid)
            layout_uid = await self._chatbot_mysql_user_id_value(session, "chatbot_layouts", user_id_uuid)
            backdrop_uid = await self._chatbot_mysql_user_id_value(
                session, "user_chatbot_backdrops", user_id_uuid
            )
            has_slot_placement_table = await self._mysql_table_exists(
                session, "user_chatbot_slot_placements"
            )
            slot_uid = (
                await self._chatbot_mysql_user_id_value(
                    session, "user_chatbot_slot_placements", user_id_uuid
                )
                if has_slot_placement_table
                else None
            )
            if ua_uid is None or layout_uid is None or backdrop_uid is None:
                return

            codes_in = ", ".join([f"'{c}'" for c in gift_codes])
            shop_res = await session.execute(
                text(
                    f"""
                    SELECT id, item_code, image_name, render_image_name, default_z_index,
                           size_rows, size_cols, placement_type
                    FROM shop_items
                    WHERE item_code IN ({codes_in}) AND is_active = 1
                    """
                )
            )
            shop_rows = list(shop_res.mappings().all())
            if len(shop_rows) < len(gift_codes):
                logger.warning(
                    "⚠️ Chatbot 注册礼包：shop_items 缺少商品，跳过。已有=%s",
                    [r["item_code"] for r in shop_rows],
                )
                return

            by_code: Dict[str, Any] = {str(r["item_code"]): r for r in shop_rows}

            for code in gift_codes:
                row = by_code.get(code)
                if not row:
                    continue
                item_id = int(row["id"])
                owned = 1
                used = used_qty_by_code.get(code, 0)
                chk = await session.execute(
                    text(
                        """
                        SELECT id FROM user_assets
                        WHERE user_id = :uid AND item_id = :iid
                        LIMIT 1
                        """
                    ),
                    {"uid": ua_uid, "iid": item_id},
                )
                if chk.scalar_one_or_none() is None:
                    await session.execute(
                        text(
                            """
                            INSERT INTO user_assets
                              (user_id, item_id, owned_qty, used_qty, last_acquired_at, created_at, updated_at)
                            VALUES
                              (:uid, :iid, :owned, :used, NOW(), NOW(), NOW())
                            """
                        ),
                        {"uid": ua_uid, "iid": item_id, "owned": owned, "used": used},
                    )

            backdrop_row = by_code.get("fresh_green_wallpaper")
            stump_row = by_code.get("stump_stool")
            window_row = by_code.get("french_window")
            if not backdrop_row or not stump_row or not window_row:
                return

            backdrop_chk = await session.execute(
                text(
                    """
                    SELECT id FROM user_chatbot_backdrops
                    WHERE user_id = :uid
                    LIMIT 1
                    """
                ),
                {"uid": backdrop_uid},
            )
            if backdrop_chk.scalar_one_or_none() is None:
                await session.execute(
                    text(
                        """
                        INSERT INTO user_chatbot_backdrops
                          (user_id, backdrop_item_id, z_index, created_at, updated_at)
                        VALUES
                          (:uid, :bid, 0, NOW(), NOW())
                        """
                    ),
                    {"uid": backdrop_uid, "bid": int(backdrop_row["id"])},
                )

            layout_chk = await session.execute(
                text(
                    """
                    SELECT id FROM chatbot_layouts
                    WHERE user_id = :uid
                    LIMIT 1
                    """
                ),
                {"uid": layout_uid},
            )
            layout_existing_id = layout_chk.scalar_one_or_none()
            layout_id = int(layout_existing_id) if layout_existing_id is not None else 0

            await session.execute(
                text(
                    """
                    UPDATE user_assets
                    SET used_qty = 1, updated_at = NOW()
                    WHERE user_id = :uid AND item_id IN (:i1, :i2)
                    """
                ),
                {
                    "uid": ua_uid,
                    "i1": int(stump_row["id"]),
                    "i2": int(window_row["id"]),
                },
            )

            if layout_id <= 0:
                await session.execute(
                    text(
                        """
                        INSERT INTO chatbot_layouts
                          (user_id, grid_rows, grid_cols, cell_width, cell_height,
                           design_width, design_height, decor_area_width, decor_area_height, version,
                           created_at, updated_at)
                        VALUES
                          (:uid, 5, 4, 96, 96, 393, 852, 393, 482, 1, NOW(), NOW())
                        """
                    ),
                    {"uid": layout_uid},
                )
                lid_res = await session.execute(text("SELECT LAST_INSERT_ID() AS lid"))
                layout_id_row = lid_res.mappings().first()
                layout_id = int((layout_id_row or {}).get("lid") or 0)
                if layout_id <= 0:
                    logger.warning("⚠️ Chatbot 注册礼包：无法取得 layout_id，跳过后续明细")
                    return

            stump_item_id = int(stump_row["id"])
            window_item_id = int(window_row["id"])
            ua_stump = await session.execute(
                text(
                    "SELECT id FROM user_assets WHERE user_id = :uid AND item_id = :iid LIMIT 1"
                ),
                {"uid": ua_uid, "iid": stump_item_id},
            )
            ua_window = await session.execute(
                text(
                    "SELECT id FROM user_assets WHERE user_id = :uid AND item_id = :iid LIMIT 1"
                ),
                {"uid": ua_uid, "iid": window_item_id},
            )
            stump_ua_id = ua_stump.scalar_one_or_none()
            window_ua_id = ua_window.scalar_one_or_none()
            if stump_ua_id is None or window_ua_id is None:
                logger.warning("⚠️ Chatbot 注册礼包：缺少 user_assets 行，跳过布局明细")
                return

            if has_slot_placement_table and slot_uid is not None:
                backdrop_item_id = int(backdrop_row["id"])
                await session.execute(
                    text(
                        """
                        INSERT INTO user_chatbot_slot_placements
                          (user_id, backdrop_item_id, slot_code, user_asset_id, item_id, z_index, image_name_snapshot, placed_at, updated_at)
                        VALUES
                          (:uid, :bid, 'slot_04_row4_col2', :uaid, :iid, :z, :img, NOW(), NOW())
                        ON DUPLICATE KEY UPDATE
                          user_asset_id = VALUES(user_asset_id),
                          item_id = VALUES(item_id),
                          z_index = VALUES(z_index),
                          image_name_snapshot = VALUES(image_name_snapshot),
                          updated_at = NOW()
                        """
                    ),
                    {
                        "uid": slot_uid,
                        "bid": backdrop_item_id,
                        "uaid": int(stump_ua_id),
                        "iid": stump_item_id,
                        "z": int(stump_row["default_z_index"] or 0),
                        "img": scene_image_name_from_shop_row(stump_row),
                    },
                )
                await session.execute(
                    text(
                        """
                        INSERT INTO user_chatbot_slot_placements
                          (user_id, backdrop_item_id, slot_code, user_asset_id, item_id, z_index, image_name_snapshot, placed_at, updated_at)
                        VALUES
                          (:uid, :bid, 'slot_03_center_2x3', :uaid, :iid, :z, :img, NOW(), NOW())
                        ON DUPLICATE KEY UPDATE
                          user_asset_id = VALUES(user_asset_id),
                          item_id = VALUES(item_id),
                          z_index = VALUES(z_index),
                          image_name_snapshot = VALUES(image_name_snapshot),
                          updated_at = NOW()
                        """
                    ),
                    {
                        "uid": slot_uid,
                        "bid": backdrop_item_id,
                        "uaid": int(window_ua_id),
                        "iid": window_item_id,
                        "z": int(window_row["default_z_index"] or 0),
                        "img": scene_image_name_from_shop_row(window_row),
                    },
                )
            elif await self._mysql_table_exists(session, "chatbot_layout_items"):
                # 旧模型兼容：仅在新槽位表不存在时写入旧布局明细
                legacy_stump_exists = await session.execute(
                    text(
                        """
                        SELECT id FROM chatbot_layout_items
                        WHERE layout_id = :lid AND item_id = :iid
                        LIMIT 1
                        """
                    ),
                    {"lid": layout_id, "iid": stump_item_id},
                )
                if legacy_stump_exists.scalar_one_or_none() is None:
                    await session.execute(
                        text(
                            """
                            INSERT INTO chatbot_layout_items
                              (layout_id, user_asset_id, item_id, placement_type,
                               start_row, start_col, size_rows, size_cols, z_index, image_name_snapshot, created_at)
                            VALUES
                              (:lid, :uaid, :iid, 'floor', 4, 2, 1, 1, :z, :img, NOW())
                            """
                        ),
                        {
                            "lid": layout_id,
                            "uaid": int(stump_ua_id),
                            "iid": stump_item_id,
                            "z": int(stump_row["default_z_index"] or 0),
                            "img": scene_image_name_from_shop_row(stump_row),
                        },
                    )

                legacy_window_exists = await session.execute(
                    text(
                        """
                        SELECT id FROM chatbot_layout_items
                        WHERE layout_id = :lid AND item_id = :iid
                        LIMIT 1
                        """
                    ),
                    {"lid": layout_id, "iid": window_item_id},
                )
                if legacy_window_exists.scalar_one_or_none() is None:
                    await session.execute(
                        text(
                            """
                            INSERT INTO chatbot_layout_items
                              (layout_id, user_asset_id, item_id, placement_type,
                               start_row, start_col, size_rows, size_cols, z_index, image_name_snapshot, created_at)
                            VALUES
                              (:lid, :uaid, :iid, 'wall', 2, 2, 2, 3, :z, :img, NOW())
                            """
                        ),
                        {
                            "lid": layout_id,
                            "uaid": int(window_ua_id),
                            "iid": window_item_id,
                            "z": int(window_row["default_z_index"] or 0),
                            "img": scene_image_name_from_shop_row(window_row),
                        },
                    )

            logger.debug(
                "✅ Chatbot 注册礼包：已写入幕布与默认布局 user_id=%s layout_id=%s",
                user_id_uuid,
                layout_id,
            )
        except Exception as e:
            logger.warning(
                "⚠️ 初始化 Chatbot 商店/布局失败: user_id=%s error=%s",
                user_id_uuid,
                e,
            )
    
    async def _generate_temp_username(self, session: AsyncSession) -> str:
        """
        生成临时用户名（格式：User + 7位随机数字）
        确保用户名不重复
        
        Args:
            session: 数据库会话
        
        Returns:
            临时用户名
        """
        max_attempts = 100  # 最大重试次数，避免无限循环
        
        for attempt in range(max_attempts):
            # 生成7位随机数字（1000000-9999999）
            random_number = random.randint(1000000, 9999999)
            temp_username = f"User{random_number}"
            
            # 检查用户名是否已存在
            result = await session.execute(
                select(User).where(User.username == temp_username)
            )
            existing_user = result.scalar_one_or_none()
            
            if not existing_user:
                logger.debug(f"生成临时用户名: {temp_username} (尝试次数: {attempt + 1})")
                return temp_username
        
        # 如果100次都重复，使用时间戳的后7位作为备选方案
        # 使用当前时间戳的毫秒数取模，确保是7位数字
        for fallback_attempt in range(10):  # 备选方案也最多尝试10次
            timestamp_suffix = int(datetime.now().timestamp() * 1000) % 10000000
            # 确保至少是7位数字（如果不足7位，前面补0）
            fallback_number = str(timestamp_suffix).zfill(7)
            temp_username = f"User{fallback_number}"
            
            # 再次检查是否重复
            result = await session.execute(
                select(User).where(User.username == temp_username)
            )
            existing_user = result.scalar_one_or_none()
            
            if not existing_user:
                logger.warning(f"⚠️ 临时用户名生成达到最大重试次数，使用备选方案: {temp_username}")
                return temp_username
            
            # 如果还是重复，等待1毫秒后重试（确保时间戳变化）
            await asyncio.sleep(0.001)
        
        # 最后的备选方案：使用UUID的十六进制转数字（取后7位）
        uuid_hex = uuid.uuid4().hex
        final_number = str(int(uuid_hex[:8], 16) % 10000000).zfill(7)
        temp_username = f"User{final_number}"
        logger.error(f"❌ 临时用户名生成异常，使用最终备选方案: {temp_username}")
        return temp_username
    
    async def create_guest_user(
        self,
        username: str = "",
        bot_name: Optional[str] = None,
        *,
        user_segment: str = "individual",
        org_id: Optional[str] = None,
        is_test_user: bool = False,
        region: Optional[str] = None,
    ) -> Dict:
        """
        创建游客用户
        
        Args:
            username: 用户名（默认为空字符串，为空时自动生成临时用户名）
            bot_name: 用户给机器人取的名字（可选）
        
        Returns:
            用户信息字典（包含user_id）
        """
        try:
            # 生成唯一用户ID
            user_id = str(uuid.uuid4())
            
            # 使用 ORM 创建用户
            async with db_manager.get_session() as session:
                # 如果用户名为空，生成临时用户名
                if not username or username.strip() == "":
                    username = await self._generate_temp_username(session)
                    logger.info(f"🔧 自动生成临时用户名: {username}")
                
                user = User(
                    user_id=user_id,
                    username=username,
                    user_type="guest",
                    user_segment=user_segment or "individual",
                    org_id=org_id,
                    is_test_user=is_test_user,
                    region=region,
                    bot_name=bot_name,
                    current_background_image_id="default",  # 设置默认背景图片
                )
                session.add(user)
                await session.flush()  # 先 flush 确保 user 有 ID，但不提交事务
                
                # 初始化背景图片
                await self._init_user_background_image(session, user_id)
                
                # 初始化默认头像
                await self._init_user_avatar(session, user_id, user_type="guest")
                
                await session.commit()
                await session.refresh(user)
            
            # 初始化 user_ui_states 的 current_avatar_id（游客用户使用默认头像）
            await self.upsert_user_ui_state(
                user_id=user_id,
                current_avatar_id=DEFAULT_AVATAR_ID
            )
            
            logger.info(f"✅ 创建游客用户成功: user_id={user_id}, username={username}, bot_name={bot_name}")
            
            return {
                "user_id": user_id,
                "username": username,
                "user_type": "guest",
                "bot_name": bot_name,
                "user_level": 0,
                "current_background_image_id": "default",
                "created_at": datetime.now().isoformat()
            }
            
        except Exception as e:
            logger.error(f"❌ 创建用户失败: {e}")
            raise
    
    async def get_user(self, user_id: str) -> Optional[Dict]:
        """
        获取用户信息
        
        Args:
            user_id: 用户ID
        
        Returns:
            用户信息字典，如果不存在返回None
            包含 current_background_image_metadata 字段（背景图片的metadata配置）
        """
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(User).where(User.user_id == user_id)
                )
                user = result.scalar_one_or_none()
            
            if user:
                user_dict = user.to_dict()
                
                # 如果用户有背景图片ID，查询背景图片的metadata
                if user.current_background_image_id:
                    try:
                        bg_result = await session.execute(
                            select(BackgroundImage).where(
                                BackgroundImage.image_id == user.current_background_image_id
                            )
                        )
                        bg_image = bg_result.scalar_one_or_none()
                        if bg_image and bg_image.metadata_json:
                            user_dict["current_background_image_metadata"] = bg_image.metadata_json
                    except Exception as e:
                        logger.warning(f"⚠️ 获取背景图片metadata失败: image_id={user.current_background_image_id}, error={e}")
                
                return user_dict
            return None
            
        except Exception as e:
            logger.error(f"❌ 获取用户信息失败: user_id={user_id}, error={e}")
            return None
    
    async def update_user_last_active(self, user_id: str):
        """
        更新用户最后活跃时间
        
        Args:
            user_id: 用户ID
        """
        try:
            async with db_manager.get_session() as session:
                await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(last_active_at=func.now())
                )
                await session.commit()
        except Exception as e:
            logger.warning(f"⚠️ 更新用户活跃时间失败: user_id={user_id}, error={e}")
    
    async def update_username(self, user_id: str, username: str) -> bool:
        """
        更新用户名
        
        Args:
            user_id: 用户ID
            username: 用户名
        
        Returns:
            是否更新成功
        """
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(username=username)
                )
                await session.commit()
                
                if result.rowcount > 0:
                    logger.info(f"✅ 更新用户名成功: user_id={user_id}, username={username}")
                    return True
                return False
        except Exception as e:
            logger.error(f"❌ 更新用户名失败: user_id={user_id}, error={e}")
            return False
    
    async def update_current_background_image_id(self, user_id: str, background_image_id: str) -> bool:
        """
        更新用户当前使用的背景图片ID
        
        Args:
            user_id: 用户ID
            background_image_id: 背景图片ID
        
        Returns:
            是否更新成功
        """
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(current_background_image_id=background_image_id)
                )
                await session.commit()
                
                if result.rowcount > 0:
                    logger.info(f"✅ 更新用户背景图片ID成功: user_id={user_id}, background_image_id={background_image_id}")
                    return True
                return False
        except Exception as e:
            logger.error(f"❌ 更新用户背景图片ID失败: user_id={user_id}, error={e}")
            return False
    
    async def update_bot_name(self, user_id: str, bot_name: str) -> bool:
        """
        更新用户给机器人取的名字
        
        Args:
            user_id: 用户ID
            bot_name: 机器人名字
        
        Returns:
            是否更新成功
        """
        import time
        start_time = time.time()
        
        try:
            async with db_manager.get_session() as session:
                # 🔥 先查询当前值，避免不必要的更新
                query_start = time.time()
                result = await session.execute(
                    select(User.bot_name).where(User.user_id == user_id)
                )
                current_bot_name = result.scalar_one_or_none()
                query_duration = (time.time() - query_start) * 1000
                
                # 如果值相同，直接返回（避免不必要的 UPDATE）
                if current_bot_name == bot_name:
                    total_duration = (time.time() - start_time) * 1000
                    logger.info(
                        f"⏭️ 机器人名字未变化，跳过更新: user_id={user_id}, "
                        f"bot_name={bot_name}, 耗时={total_duration:.1f}ms"
                    )
                    return True
                
                # 执行更新
                update_start = time.time()
                update_result = await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(bot_name=bot_name)
                )
                await session.commit()
                update_duration = (time.time() - update_start) * 1000
                total_duration = (time.time() - start_time) * 1000
                
                if update_result.rowcount > 0:
                    logger.info(
                        f"✅ 更新机器人名字成功: user_id={user_id}, "
                        f"bot_name={bot_name}, "
                        f"查询耗时={query_duration:.1f}ms, "
                        f"更新耗时={update_duration:.1f}ms, "
                        f"总耗时={total_duration:.1f}ms"
                    )
                    return True
                else:
                    logger.warning(
                        f"⚠️ 更新机器人名字失败: 未找到用户 user_id={user_id}, "
                        f"总耗时={total_duration:.1f}ms"
                    )
                    return False
        except asyncio.TimeoutError:
            total_duration = (time.time() - start_time) * 1000
            logger.error(
                f"❌ 更新机器人名字超时: user_id={user_id}, "
                f"耗时={total_duration:.1f}ms"
            )
            return False
        except Exception as e:
            total_duration = (time.time() - start_time) * 1000
            logger.error(
                f"❌ 更新机器人名字失败: user_id={user_id}, "
                f"error={e}, 耗时={total_duration:.1f}ms"
            )
            return False

    async def update_preferred_language(self, user_id: str, preferred_language: str) -> bool:
        """更新用户首选界面语言（zh/en）"""
        if preferred_language not in ("zh", "en"):
            return False
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(preferred_language=preferred_language)
                )
                await session.commit()
                return result.rowcount > 0
        except Exception as e:
            logger.error(f"❌ 更新首选语言失败: user_id={user_id}, error={e}")
            return False
    
    async def update_speech_duration(self, user_id: str, duration_ms: int) -> Optional[Dict[str, int]]:
        """
        更新用户累积说话时长
        
        Args:
            user_id: 用户ID
            duration_ms: 本次说话时长（毫秒）
        
        Returns:
            包含 total_duration_ms 与 user_level 的字典，失败时返回 None
        """
        try:
            if duration_ms <= 0:
                return None  # 时长无效，跳过更新

            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(User.total_speech_duration_ms, User.user_level)
                    .where(User.user_id == user_id)
                    .with_for_update()
                )
                row = result.one_or_none()
                if row is None:
                    logger.warning(f"⚠️ 更新说话时长失败，未找到用户: user_id={user_id}")
                    return None

                current_total = (row[0] or 0)
                current_level = (row[1] or 0)
                new_total = current_total + duration_ms
                new_level = self._calculate_user_level(new_total)

                update_payload: Dict[str, Any] = {
                    "total_speech_duration_ms": new_total,
                }
                if new_level != current_level:
                    update_payload["user_level"] = new_level

                await session.execute(
                    update(User)
                    .where(User.user_id == user_id)
                    .values(**update_payload)
                )
                await session.commit()

            logger.info(
                f"✅ 更新用户说话时长成功: user_id={user_id}, "
                f"duration_ms={duration_ms}, total_ms={new_total}, level={new_level}"
            )
            return {
                "total_duration_ms": new_total,
                "user_level": new_level,
            }
        except Exception as e:
            logger.error(f"❌ 更新用户说话时长失败: user_id={user_id}, error={e}")
            return None

    async def get_user_ui_state(self, user_id: str) -> Optional[Dict[str, Any]]:
        """
        获取用户前端可视化状态
        """
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(UserUIState).where(UserUIState.user_id == user_id)
                )
                state = result.scalar_one_or_none()

            if state is None:
                return None
            return state.to_dict()
        except Exception as e:  # noqa: BLE001
            logger.error(f"❌ 获取用户UI状态失败: user_id={user_id}, error={e}")
            return None

    async def upsert_user_ui_state(
        self,
        user_id: str,
        *,
        current_page: Optional[str] = None,
        visited_routes: Optional[list[str]] = None,
        has_shown_tap_to_chat: Optional[bool] = None,
        has_shown_cancel_tooltip: Optional[bool] = None,
        has_shown_send_tooltip: Optional[bool] = None,
        has_shown_progress_bar_tooltip: Optional[bool] = None,
        has_shown_mission_stone_modal: Optional[bool] = None,
        has_shown_conversation_stone_modal: Optional[bool] = None,
        has_clicked_mission_stone: Optional[bool] = None,
        has_clicked_conversation_stone: Optional[bool] = None,
        has_shown_first_shop_tooltip: Optional[bool] = None,
        has_clicked_shop_in_settings: Optional[bool] = None,
        avatar_c_unlocked: Optional[bool] = None,
        current_avatar_id: Optional[str] = None,
        has_shared_avatar: Optional[bool] = None,
        subtitle_on: Optional[bool] = None,
        sfx_enabled: Optional[bool] = None,
        bgm_enabled: Optional[bool] = None,
    ) -> Optional[Dict[str, Any]]:
        """
        创建或更新用户前端可视化状态（部分字段更新）
        """
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(UserUIState).where(UserUIState.user_id == user_id).with_for_update()
                )
                state: Optional[UserUIState] = result.scalar_one_or_none()

                if state is None:
                    # 新建记录，仅写入非空字段
                    data: Dict[str, Any] = {"user_id": user_id}
                    if current_page is not None:
                        data["current_page"] = current_page
                    if visited_routes is not None:
                        data["visited_routes"] = visited_routes
                    if has_shown_tap_to_chat is not None:
                        data["has_shown_tap_to_chat"] = has_shown_tap_to_chat
                    if has_shown_cancel_tooltip is not None:
                        data["has_shown_cancel_tooltip"] = has_shown_cancel_tooltip
                    if has_shown_send_tooltip is not None:
                        data["has_shown_send_tooltip"] = has_shown_send_tooltip
                    if has_shown_progress_bar_tooltip is not None:
                        data["has_shown_progress_bar_tooltip"] = has_shown_progress_bar_tooltip
                    if has_shown_mission_stone_modal is not None:
                        data["has_shown_mission_stone_modal"] = has_shown_mission_stone_modal
                    if has_shown_conversation_stone_modal is not None:
                        data["has_shown_conversation_stone_modal"] = has_shown_conversation_stone_modal
                    if has_clicked_mission_stone is not None:
                        data["has_clicked_mission_stone"] = has_clicked_mission_stone
                    if has_clicked_conversation_stone is not None:
                        data["has_clicked_conversation_stone"] = has_clicked_conversation_stone
                    if has_shown_first_shop_tooltip is not None:
                        data["has_shown_first_shop_tooltip"] = has_shown_first_shop_tooltip
                    if has_clicked_shop_in_settings is not None:
                        data["has_clicked_shop_in_settings"] = has_clicked_shop_in_settings
                    if avatar_c_unlocked is not None:
                        data["avatar_c_unlocked"] = avatar_c_unlocked
                    if current_avatar_id is not None:
                        data["current_avatar_id"] = current_avatar_id
                    if has_shared_avatar is not None:
                        data["has_shared_avatar"] = has_shared_avatar
                    if subtitle_on is not None:
                        data["subtitle_on"] = subtitle_on
                    if sfx_enabled is not None:
                        data["sfx_enabled"] = sfx_enabled
                    if bgm_enabled is not None:
                        data["bgm_enabled"] = bgm_enabled

                    stmt = insert(UserUIState).values(**data)
                    await session.execute(stmt)
                else:
                    # 更新现有记录（仅覆盖非空字段）
                    if current_page is not None:
                        state.current_page = current_page
                    if visited_routes is not None:
                        state.visited_routes = visited_routes
                    if has_shown_tap_to_chat is not None:
                        state.has_shown_tap_to_chat = has_shown_tap_to_chat
                    if has_shown_cancel_tooltip is not None:
                        state.has_shown_cancel_tooltip = has_shown_cancel_tooltip
                    if has_shown_send_tooltip is not None:
                        state.has_shown_send_tooltip = has_shown_send_tooltip
                    if has_shown_progress_bar_tooltip is not None:
                        state.has_shown_progress_bar_tooltip = has_shown_progress_bar_tooltip
                    if has_shown_mission_stone_modal is not None:
                        state.has_shown_mission_stone_modal = has_shown_mission_stone_modal
                    if has_shown_conversation_stone_modal is not None:
                        state.has_shown_conversation_stone_modal = has_shown_conversation_stone_modal
                    if has_clicked_mission_stone is not None:
                        state.has_clicked_mission_stone = has_clicked_mission_stone
                    if has_clicked_conversation_stone is not None:
                        state.has_clicked_conversation_stone = has_clicked_conversation_stone
                    if has_shown_first_shop_tooltip is not None:
                        state.has_shown_first_shop_tooltip = has_shown_first_shop_tooltip
                    if has_clicked_shop_in_settings is not None:
                        state.has_clicked_shop_in_settings = has_clicked_shop_in_settings
                    if avatar_c_unlocked is not None:
                        state.avatar_c_unlocked = avatar_c_unlocked
                    if current_avatar_id is not None:
                        state.current_avatar_id = current_avatar_id
                    if has_shared_avatar is not None:
                        state.has_shared_avatar = has_shared_avatar
                    if subtitle_on is not None:
                        state.subtitle_on = subtitle_on
                    if sfx_enabled is not None:
                        state.sfx_enabled = sfx_enabled
                    if bgm_enabled is not None:
                        state.bgm_enabled = bgm_enabled

                    session.add(state)

                await session.commit()
                # 重新读取最新状态
                result = await session.execute(
                    select(UserUIState).where(UserUIState.user_id == user_id)
                )
                latest_state = result.scalar_one_or_none()

            if latest_state is None:
                return None
            return latest_state.to_dict()
        except Exception as e:  # noqa: BLE001
            logger.error(f"❌ 更新用户UI状态失败: user_id={user_id}, error={e}")
            return None

    async def log_page_visit(self, user_id: Optional[str], path: str) -> bool:
        """记录一次页面访问（未登录时 user_id 为 None）"""
        try:
            async with db_manager.get_session() as session:
                stmt = insert(PageVisitLog).values(user_id=user_id, path=path[:512])
                await session.execute(stmt)
                await session.commit()
            return True
        except Exception as e:  # noqa: BLE001
            logger.warning(f"记录页面访问失败: path={path}, error={e}")
            return False

    async def _get_last_visit_date_for_user(
        self,
        session: AsyncSession,
        user_id: str,
        on_or_after_date: Optional[date] = None,
        before_date: Optional[date] = None,
    ) -> Optional[date]:
        """
        查询用户最后一次页面访问的日期。
        
        Args:
            session: 数据库会话
            user_id: 用户 ID
            on_or_after_date: 仅统计该日期及之后的访问
            before_date: 仅统计早于该日期（严格小于）的访问，用于排除当日访问
        """
        stmt = (
            select(func.date(PageVisitLog.visited_at))
            .where(PageVisitLog.user_id == user_id)
            .order_by(desc(PageVisitLog.visited_at))
            .limit(1)
        )
        if on_or_after_date is not None:
            stmt = stmt.where(func.date(PageVisitLog.visited_at) >= on_or_after_date)
        if before_date is not None:
            stmt = stmt.where(func.date(PageVisitLog.visited_at) < before_date)
        result = await session.execute(stmt)
        val = result.scalar_one_or_none()
        if val is None:
            return None
        if isinstance(val, date):
            return val
        if isinstance(val, str):
            return datetime.strptime(val[:10], "%Y-%m-%d").date()
        return val

    async def save_user_event(self, payload: dict) -> bool:
        """
        保存单条用户事件（user_events）

        payload 字段应与 UserEventPayload 对齐：
        - event_id, user_id, event_name, event_label, event_time, session_id,
          platform, app_version, timezone_offset_minutes, properties
        """
        try:
            async with db_manager.get_session() as session:
                event_time = payload.get("event_time")
                if isinstance(event_time, str):
                    event_time = datetime.fromisoformat(event_time.replace("Z", "+00:00"))

                user_event = UserEvent(
                    event_id=payload["event_id"],
                    user_id=payload["user_id"],
                    event_name=payload["event_name"],
                    event_label=payload.get("event_label"),
                    event_time=event_time or datetime.utcnow(),
                    session_id=payload.get("session_id"),
                    platform=payload["platform"],
                    app_version=payload.get("app_version"),
                    timezone_offset_minutes=payload.get("timezone_offset_minutes"),
                    properties=payload.get("properties"),
                )
                session.add(user_event)
                await session.commit()
            return True
        except Exception as e:  # noqa: BLE001
            logger.warning(f"记录用户事件失败: event_id={payload.get('event_id')}, error={e}")
            return False

    def _parse_effective_date(self) -> date:
        """解析配置中的亲密度规则生效日"""
        s = getattr(settings, "INTIMACY_RULE_EFFECTIVE_DATE", "2026-02-13") or "2026-02-13"
        return datetime.strptime(s.strip()[:10], "%Y-%m-%d").date()

    async def get_intimacy(self, user_id: str) -> Dict[str, Any]:
        """
        获取用户亲密度：0-3 颗心。扣心由每日定时脚本完成，此处仅读库并计算展示用 days_since_last_visit。
        只统计 INTIMACY_RULE_EFFECTIVE_DATE 及之后的页面访问。
        返回: {"heart_count": int, "days_since_last_visit": int}
        """
        today = date.today()
        effective = self._parse_effective_date()
        try:
            async with db_manager.get_session() as session:
                # 所有访问记录中最晚的一次（含今天），用于判断是否为新用户
                last_any_visit = await self._get_last_visit_date_for_user(
                    session,
                    user_id,
                    on_or_after_date=effective,
                )
                # 排除今天之后，获取“上一次见面”的日期
                last_visit_before_today = await self._get_last_visit_date_for_user(
                    session,
                    user_id,
                    on_or_after_date=effective,
                    before_date=today,
                )

                if last_any_visit is None:
                    # 从未有过访问记录：视为首次见面
                    days_since = 0
                elif last_visit_before_today is None:
                    # 只有今天的访问记录（例如久未登录后今天第一次回来）：
                    # 文案希望展示“已经过去 X 天”，这里按 0 天处理
                    days_since = 0
                else:
                    days_since = (today - last_visit_before_today).days

                result = await session.execute(select(UserIntimacy).where(UserIntimacy.user_id == user_id))
                intimacy = result.scalar_one_or_none()
                if intimacy is None:
                    intimacy = UserIntimacy(user_id=user_id, heart_count=3)
                    session.add(intimacy)
                    await session.flush()
                await session.commit()
                await session.refresh(intimacy)
                return {
                    "heart_count": intimacy.heart_count,
                    "days_since_last_visit": max(0, days_since),
                }
        except Exception as e:  # noqa: BLE001
            logger.error(f"get_intimacy 失败: user_id={user_id}, error={e}")
            return {"heart_count": 3, "days_since_last_visit": 0}

    async def ensure_user_intimacy(self, session: AsyncSession, user_id: str, heart_count: int = 3) -> None:
        """注册时为新用户写入亲密度记录（若不存在则插入）。"""
        result = await session.execute(select(UserIntimacy).where(UserIntimacy.user_id == user_id))
        if result.scalar_one_or_none() is not None:
            return
        session.add(UserIntimacy(user_id=user_id, heart_count=heart_count))
        await session.flush()

    async def light_up_intimacy(self, user_id: str) -> Dict[str, Any]:
        """
        点亮一颗心：heart_count = min(3, heart_count + 1)。
        规则：同一用户每天最多点亮一次（按服务端当天日期判断）。
        返回最新 heart_count。
        """
        try:
            today_str = date.today().isoformat()[:10]
            async with db_manager.get_session() as session:
                result = await session.execute(select(UserIntimacy).where(UserIntimacy.user_id == user_id))
                intimacy = result.scalar_one_or_none()
                if intimacy is None:
                    # 首次点亮：给 1 颗心，并记录今天为 last_light_up_date
                    intimacy = UserIntimacy(user_id=user_id, heart_count=1, last_light_up_date=today_str)
                    session.add(intimacy)
                else:
                    # 如果今天已经点亮过，则直接返回当前亲密度，不再增加
                    if intimacy.last_light_up_date == today_str:
                        await session.commit()
                        await session.refresh(intimacy)
                        return {"heart_count": intimacy.heart_count}

                    # 今日尚未点亮：不满 3 颗心时才增加一颗，并更新 last_light_up_date
                    if intimacy.heart_count < 3:
                        intimacy.heart_count = min(3, intimacy.heart_count + 1)
                        intimacy.last_light_up_date = today_str
                await session.commit()
                await session.refresh(intimacy)
                return {"heart_count": intimacy.heart_count}
        except Exception as e:  # noqa: BLE001
            logger.error(f"light_up_intimacy 失败: user_id={user_id}, error={e}")
            return {"heart_count": 0}

    async def get_intimacy_status(self, user_id: str) -> Dict[str, Any]:
        """
        获取用户当天是否可以点亮亲密度。
        返回:
        {
            "heart_count": int,           # 当前心数 0-3
            "can_light_up_today": bool,  # 今天是否还能点亮一次
        }
        """
        today_str = date.today().isoformat()[:10]
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(select(UserIntimacy).where(UserIntimacy.user_id == user_id))
                intimacy = result.scalar_one_or_none()
                if intimacy is None:
                    # 没有记录时视为 3 颗心、今日尚未点亮（但前端通常会基于 heart_count<3 决定是否展示弹窗）
                    intimacy = UserIntimacy(user_id=user_id, heart_count=3)
                    session.add(intimacy)
                    await session.flush()
                    await session.commit()
                    await session.refresh(intimacy)

                can_light_up_today = intimacy.heart_count < 3 and intimacy.last_light_up_date != today_str
                return {
                    "heart_count": intimacy.heart_count,
                    "can_light_up_today": can_light_up_today,
                }
        except Exception as e:  # noqa: BLE001
            logger.error(f"get_intimacy_status 失败: user_id={user_id}, error={e}")
            # 兜底：认为不能点亮，避免无限点亮
            return {
                "heart_count": 3,
                "can_light_up_today": False,
            }

    async def register_with_phone(
        self,
        phone: str,
        username: str = "",
        user_id: Optional[str] = None,
        area_code: Optional[str] = None,
        preferred_language: Optional[str] = None,
    ) -> Dict:
        """
        使用手机号注册
        
        Args:
            phone: 手机号
            username: 用户名（可选）
            user_id: 可选的现有用户ID，如果提供则更新该用户而不是创建新用户
        
        Returns:
            用户信息字典
        """
        try:
            lang = preferred_language if preferred_language in ("zh", "en") else None
            async with db_manager.get_session() as session:
                # 检查手机号是否已被其他用户注册
                result = await session.execute(
                    select(User).where(User.phone == phone)
                )
                existing_phone_user = result.scalar_one_or_none()
                
                if existing_phone_user:
                    # 如果提供了 user_id，检查是否是同一个用户
                    if user_id and existing_phone_user.user_id == user_id:
                        # 同一个用户，可能是重新注册，允许继续
                        pass
                    else:
                        raise AuthError(PHONE_ALREADY_REGISTERED)
                
                # 如果提供了 user_id，使用现有用户
                final_user_id = None
                if user_id:
                    result = await session.execute(
                        select(User).where(User.user_id == user_id)
                    )
                    existing_user = result.scalar_one_or_none()
                    
                    if not existing_user:
                        raise AuthError(USER_ID_NOT_FOUND)
                    
                    if existing_user.user_type == "registered":
                        raise AuthError(ALREADY_REGISTERED)
                    
                    # 更新现有用户
                    if not username or username.strip() == "":
                        # 如果用户已有用户名，保留；否则生成新用户名
                        if not existing_user.username or existing_user.username.strip() == "":
                            username = await self._generate_temp_username(session)
                        else:
                            username = existing_user.username
                    
                    existing_user.phone = phone
                    if area_code is not None:
                        existing_user.area_code = area_code
                    existing_user.username = username
                    existing_user.user_type = "registered"
                    existing_user.is_verified = True
                    existing_user.registered_at = datetime.now()
                    if lang is not None:
                        existing_user.preferred_language = lang
                    
                    user = existing_user
                    final_user_id = user_id
                    await session.flush()
                    
                    # 如果用户还没有背景图片，初始化背景图片
                    if not existing_user.current_background_image_id or existing_user.current_background_image_id == "default":
                        await self._init_user_background_image(session, user_id)
                    
                    # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                    await self._init_user_avatar(session, user_id, user_type="registered")
                    await self.ensure_user_intimacy(session, user_id, heart_count=3)
                    await self._init_registered_user_chatbot_shop(session, user_id)
                    await session.commit()
                    await session.refresh(user)
                else:
                    # 创建新用户（原有逻辑）
                    new_user_id = str(uuid.uuid4())
                    if not username or username.strip() == "":
                        username = await self._generate_temp_username(session)
                    
                    user = User(
                        user_id=new_user_id,
                        username=username,
                        phone=phone,
                        area_code=area_code,
                        user_type="registered",
                        is_verified=True,
                        registered_at=datetime.now(),
                        current_background_image_id="default",  # 设置默认背景图片
                        preferred_language=lang,
                    )
                    session.add(user)
                    await session.flush()  # 先 flush 确保 user 有 ID，但不提交事务
                    
                    final_user_id = new_user_id
                    
                    # 初始化背景图片
                    await self._init_user_background_image(session, new_user_id)
                    
                    # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                    await self._init_user_avatar(session, new_user_id, user_type="registered")
                    await self.ensure_user_intimacy(session, new_user_id, heart_count=3)
                    await self._init_registered_user_chatbot_shop(session, new_user_id)
                    await session.commit()
                    await session.refresh(user)
            
            # 初始化 user_ui_states 的 current_avatar_id（注册用户优先使用注册赠送头像）
            final_user_id = final_user_id if final_user_id else user.user_id
            await self.upsert_user_ui_state(
                user_id=final_user_id,
                current_avatar_id=REGISTERED_AVATAR_ID
            )
            
            logger.info(f"✅ 手机号注册成功: user_id={final_user_id}, area_code={area_code}, phone={phone}")
            return user.to_dict()
        except Exception as e:
            logger.error(f"❌ 手机号注册失败: {e}")
            raise
    
    async def register_with_email(
        self,
        email: str,
        username: str = "",
        user_id: Optional[str] = None,
        preferred_language: Optional[str] = None,
    ) -> Dict:
        """
        使用邮箱注册
        
        Args:
            email: 邮箱
            username: 用户名（可选）
            user_id: 可选的现有用户ID，如果提供则更新该用户而不是创建新用户
        
        Returns:
            用户信息字典
        """
        try:
            lang = preferred_language if preferred_language in ("zh", "en") else None
            async with db_manager.get_session() as session:
                # 检查邮箱是否已被其他用户注册
                result = await session.execute(
                    select(User).where(User.email == email)
                )
                existing_email_user = result.scalar_one_or_none()
                
                if existing_email_user:
                    # 如果提供了 user_id，检查是否是同一个用户
                    if user_id and existing_email_user.user_id == user_id:
                        # 同一个用户，可能是重新注册，允许继续
                        pass
                    else:
                        raise AuthError(EMAIL_ALREADY_REGISTERED)
                
                # 如果提供了 user_id，使用现有用户
                final_user_id = None
                if user_id:
                    result = await session.execute(
                        select(User).where(User.user_id == user_id)
                    )
                    existing_user = result.scalar_one_or_none()
                    
                    if not existing_user:
                        raise AuthError(USER_ID_NOT_FOUND)
                    
                    if existing_user.user_type == "registered":
                        raise AuthError(ALREADY_REGISTERED)
                    
                    # 更新现有用户
                    if not username or username.strip() == "":
                        # 如果用户已有用户名，保留；否则生成新用户名
                        if not existing_user.username or existing_user.username.strip() == "":
                            username = await self._generate_temp_username(session)
                        else:
                            username = existing_user.username
                    
                    existing_user.email = email
                    existing_user.username = username
                    existing_user.user_type = "registered"
                    existing_user.is_verified = True
                    existing_user.registered_at = datetime.now()
                    if lang is not None:
                        existing_user.preferred_language = lang
                    
                    user = existing_user
                    final_user_id = user_id
                    await session.flush()
                    
                    # 如果用户还没有背景图片，初始化背景图片
                    if not existing_user.current_background_image_id or existing_user.current_background_image_id == "default":
                        await self._init_user_background_image(session, user_id)
                    
                    # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                    await self._init_user_avatar(session, user_id, user_type="registered")
                    await self.ensure_user_intimacy(session, user_id, heart_count=3)
                    await self._init_registered_user_chatbot_shop(session, user_id)
                    await session.commit()
                    await session.refresh(user)
                else:
                    # 创建新用户（原有逻辑）
                    new_user_id = str(uuid.uuid4())
                    if not username or username.strip() == "":
                        username = await self._generate_temp_username(session)
                    
                    user = User(
                        user_id=new_user_id,
                        username=username,
                        email=email,
                        user_type="registered",
                        is_verified=True,
                        registered_at=datetime.now(),
                        current_background_image_id="default",  # 设置默认背景图片
                        preferred_language=lang,
                    )
                    session.add(user)
                    await session.flush()  # 先 flush 确保 user 有 ID，但不提交事务
                    
                    final_user_id = new_user_id
                    
                    # 初始化背景图片
                    await self._init_user_background_image(session, new_user_id)
                    
                    # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                    await self._init_user_avatar(session, new_user_id, user_type="registered")
                    await self.ensure_user_intimacy(session, new_user_id, heart_count=3)
                    await self._init_registered_user_chatbot_shop(session, new_user_id)
                    await session.commit()
                    await session.refresh(user)
            
            # 初始化 user_ui_states 的 current_avatar_id（注册用户优先使用注册赠送头像）
            final_user_id = final_user_id if final_user_id else user.user_id
            await self.upsert_user_ui_state(
                user_id=final_user_id,
                current_avatar_id=REGISTERED_AVATAR_ID
            )
            
            logger.info(f"✅ 邮箱注册成功: user_id={final_user_id}, email={email}")
            return user.to_dict()
        except Exception as e:
            logger.error(f"❌ 邮箱注册失败: {e}")
            raise
    
    async def login_with_wechat(self, openid: str, username: str = "") -> Dict:
        """
        使用微信OpenID登录/注册
        
        Args:
            openid: 微信OpenID
            username: 用户名（可选，仅新用户时使用）
        
        Returns:
            用户信息字典
        """
        try:
            async with db_manager.get_session() as session:
                # 检查是否已存在
                result = await session.execute(
                    select(User).where(User.wechat_openid == openid)
                )
                existing_user = result.scalar_one_or_none()
                
                if existing_user:
                    # 已存在，直接返回
                    logger.info(f"✅ 微信登录成功: user_id={existing_user.user_id}, openid={openid}")
                    return existing_user.to_dict()
                
                # 新用户，创建账号
                user_id = str(uuid.uuid4())
                if not username or username.strip() == "":
                    username = await self._generate_temp_username(session)
                
                user = User(
                    user_id=user_id,
                    username=username,
                    wechat_openid=openid,
                    user_type="registered",
                    is_verified=True,
                    registered_at=datetime.now(),
                    current_background_image_id="default",  # 设置默认背景图片
                )
                session.add(user)
                await session.flush()  # 先 flush 确保 user 有 ID，但不提交事务
                
                # 初始化背景图片
                await self._init_user_background_image(session, user_id)
                
                # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                await self._init_user_avatar(session, user_id, user_type="registered")
                await self._init_registered_user_chatbot_shop(session, user_id)
                
                await session.commit()
                await session.refresh(user)
            
            # 初始化 user_ui_states 的 current_avatar_id（注册用户优先使用注册赠送头像）
            await self.upsert_user_ui_state(
                user_id=user_id,
                current_avatar_id=REGISTERED_AVATAR_ID
            )
            
            logger.info(f"✅ 微信注册成功: user_id={user_id}, openid={openid}")
            return user.to_dict()
        except Exception as e:
            logger.error(f"❌ 微信登录/注册失败: {e}")
            raise
    
    async def login_with_qq(
        self,
        openid: str,
        username: str = "",
        unionid: Optional[str] = None,
        user_id: Optional[str] = None,
    ) -> Dict:
        """
        使用QQ OpenID登录/注册
        
        Args:
            openid: QQ OpenID
            username: 用户名（可选，仅新用户时使用）
            unionid: QQ UnionID（可选）
            user_id: 可选的现有用户ID，如果提供则更新该用户而不是创建新用户
        
        Returns:
            用户信息字典
        """
        try:
            async with db_manager.get_session() as session:
                # 优先用 openid 查询
                result = await session.execute(
                    select(User).where(User.qq_openid == openid)
                )
                existing_user = result.scalar_one_or_none()

                # 如果未命中且拿到了 unionid，则尝试用 unionid 查询
                if not existing_user and unionid:
                    union_result = await session.execute(
                        select(User).where(User.qq_unionid == unionid)
                    )
                    existing_user = union_result.scalar_one_or_none()
                    # 若命中用户还未绑定 openid，则补充绑定
                    if existing_user and not existing_user.qq_openid:
                        existing_user.qq_openid = openid
                        session.add(existing_user)
                        await session.commit()
                        await session.refresh(existing_user)
                        logger.info(
                            f"🔁 已根据 unionid={unionid} 绑定新的 QQ openid={openid}"
                        )

                if existing_user:
                    # 如果提供了 user_id，检查是否是同一个用户
                    if user_id and existing_user.user_id != user_id:
                        raise AuthError(QQ_ALREADY_BOUND)
                    
                    updated = False
                    if unionid and not existing_user.qq_unionid:
                        existing_user.qq_unionid = unionid
                        updated = True
                    if updated:
                        session.add(existing_user)
                        await session.commit()
                        await session.refresh(existing_user)
                    logger.info(
                        f"✅ QQ登录成功: user_id={existing_user.user_id}, "
                        f"openid={openid}, unionid={unionid}"
                    )
                    return existing_user.to_dict()
                
                # 如果提供了 user_id，使用现有用户
                final_user_id = None
                if user_id:
                    result = await session.execute(
                        select(User).where(User.user_id == user_id)
                    )
                    existing_user_by_id = result.scalar_one_or_none()
                    
                    if not existing_user_by_id:
                        raise AuthError(USER_ID_NOT_FOUND)
                    
                    if existing_user_by_id.user_type == "registered":
                        raise AuthError(ALREADY_REGISTERED)
                    
                    # 更新现有用户
                    if not username or username.strip() == "":
                        # 如果用户已有用户名，保留；否则生成新用户名
                        if not existing_user_by_id.username or existing_user_by_id.username.strip() == "":
                            username = await self._generate_temp_username(session)
                        else:
                            username = existing_user_by_id.username
                    
                    existing_user_by_id.qq_openid = openid
                    if unionid:
                        existing_user_by_id.qq_unionid = unionid
                    existing_user_by_id.username = username
                    existing_user_by_id.user_type = "registered"
                    existing_user_by_id.is_verified = True
                    existing_user_by_id.registered_at = datetime.now()
                    
                    user = existing_user_by_id
                    final_user_id = user_id
                    await session.flush()
                    
                    # 如果用户还没有背景图片，初始化背景图片
                    if not existing_user_by_id.current_background_image_id or existing_user_by_id.current_background_image_id == "default":
                        await self._init_user_background_image(session, user_id)
                    
                    # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                    await self._init_user_avatar(session, user_id, user_type="registered")
                    await self._init_registered_user_chatbot_shop(session, user_id)
                    
                    await session.commit()
                    await session.refresh(user)
                else:
                    # 新用户，创建账号
                    new_user_id = str(uuid.uuid4())
                    if not username or username.strip() == "":
                        username = await self._generate_temp_username(session)
                    
                    user = User(
                        user_id=new_user_id,
                        username=username,
                        qq_openid=openid,
                        qq_unionid=unionid,
                        user_type="registered",
                        is_verified=True,
                        registered_at=datetime.now(),
                        current_background_image_id="default",  # 设置默认背景图片
                    )
                    session.add(user)
                    await session.flush()  # 先 flush 确保 user 有 ID，但不提交事务
                    
                    final_user_id = new_user_id
                    
                    # 初始化背景图片
                    await self._init_user_background_image(session, new_user_id)
                    
                    # 初始化头像（注册用户：默认头像 + 注册赠送头像）
                    await self._init_user_avatar(session, new_user_id, user_type="registered")
                    await self._init_registered_user_chatbot_shop(session, new_user_id)
                    
                    await session.commit()
                    await session.refresh(user)
            
            # 初始化 user_ui_states 的 current_avatar_id（注册用户优先使用注册赠送头像）
            final_user_id = final_user_id if final_user_id else user.user_id
            await self.upsert_user_ui_state(
                user_id=final_user_id,
                current_avatar_id=REGISTERED_AVATAR_ID
            )
            
            logger.info(
                f"✅ QQ注册成功: user_id={final_user_id}, openid={openid}, unionid={unionid}"
            )
            return user.to_dict()
        except Exception as e:
            logger.error(f"❌ QQ登录/注册失败: {e}")
            raise
    
    async def login_with_phone(
        self,
        phone: str,
        area_code: Optional[str] = None,
        preferred_language: Optional[str] = None,
    ) -> Optional[Dict]:
        """
        使用手机号登录
        
        Args:
            phone: 手机号
        
        Returns:
            用户信息字典，如果不存在返回None
        """
        try:
            lang = preferred_language if preferred_language in ("zh", "en") else None
            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(User).where(
                        User.phone == phone,
                        *( [User.area_code == area_code] if area_code else [] )
                    )
                )
                user = result.scalar_one_or_none()
                
                # 兼容旧数据：未存区号时再用空区号回退查询
                if not user and area_code:
                    fallback_result = await session.execute(
                        select(User).where(User.phone == phone, User.area_code.is_(None))
                    )
                    user = fallback_result.scalar_one_or_none()
            
            if user:
                if lang is not None:
                    try:
                        async with db_manager.get_session() as session:
                            await session.execute(
                                update(User)
                                .where(User.user_id == user.user_id)
                                .values(preferred_language=lang)
                            )
                            await session.commit()
                    except Exception as update_error:  # noqa: BLE001
                        logger.warning(
                            f"⚠️ 更新用户首选语言失败: user_id={user.user_id}, lang={lang}, error={update_error}"
                        )
                logger.info(f"✅ 手机号登录成功: user_id={user.user_id}, area_code={area_code}, phone={phone}")
                return user.to_dict()
            return None
        except Exception as e:
            logger.error(f"❌ 手机号登录失败: {e}")
            return None
    
    async def login_with_email(self, email: str, preferred_language: Optional[str] = None) -> Optional[Dict]:
        """
        使用邮箱登录
        
        Args:
            email: 邮箱
        
        Returns:
            用户信息字典，如果不存在返回None
        """
        try:
            lang = preferred_language if preferred_language in ("zh", "en") else None
            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(User).where(User.email == email)
                )
                user = result.scalar_one_or_none()
            
            if user:
                if lang is not None:
                    try:
                        async with db_manager.get_session() as session:
                            await session.execute(
                                update(User)
                                .where(User.user_id == user.user_id)
                                .values(preferred_language=lang)
                            )
                            await session.commit()
                    except Exception as update_error:  # noqa: BLE001
                        logger.warning(
                            f"⚠️ 更新用户首选语言失败: user_id={user.user_id}, lang={lang}, error={update_error}"
                        )
                logger.info(f"✅ 邮箱登录成功: user_id={user.user_id}, email={email}")
                return user.to_dict()
            return None
        except Exception as e:
            logger.error(f"❌ 邮箱登录失败: {e}")
            return None

    async def verify_and_auth(
        self,
        auth_type: str,
        code: str,
        email: Optional[str] = None,
        phone: Optional[str] = None,
        area_code: Optional[str] = None,
        preferred_language: Optional[str] = None,
    ) -> Tuple[Dict, bool]:
        """
        验证码校验通过后统一登录或注册（V2.0）。
        auth_type: "email" | "phone"
        返回 (user_dict, is_new_user)
        """
        if auth_type == "email":
            send_type = "email"
        else:
            send_type = "international_sms" if (area_code and area_code.strip() != "+86") else "domestic_sms"
        if auth_type == "email":
            if not email:
                raise AuthError(EMAIL_REQUIRED)
        else:
            if not phone:
                raise AuthError(PHONE_REQUIRED)
        if not code or not code.strip():
            raise AuthError(CODE_REQUIRED)

        async with db_manager.get_session() as session:
            if auth_type == "email":
                result = await session.execute(
                    select(VerificationCode)
                    .where(
                        VerificationCode.email == email,
                        VerificationCode.send_type == send_type,
                        VerificationCode.code == code.strip(),
                        VerificationCode.is_valid == True,
                    )
                    .order_by(VerificationCode.created_at.desc())
                    .limit(1)
                )
            else:
                result = await session.execute(
                    select(VerificationCode)
                    .where(
                        VerificationCode.phone == phone,
                        VerificationCode.send_type == send_type,
                        VerificationCode.code == code.strip(),
                        VerificationCode.is_valid == True,
                    )
                    .order_by(VerificationCode.created_at.desc())
                    .limit(1)
                )
            verification_code = result.scalar_one_or_none()
            if not verification_code:
                logger.warning(f"❌ 验证码无效: type={auth_type}, email={email}, phone={phone}")
                raise AuthError(CODE_INVALID)
            verification_code.is_valid = False
            await session.commit()

        if auth_type == "email":
            user = await self.login_with_email(email, preferred_language=preferred_language)
            if user:
                logger.info(f"✅ verify_and_auth 邮箱登录: user_id={user['user_id']}")
                return (user, False)
            user = await self.register_with_email(email, preferred_language=preferred_language)
            logger.info(f"✅ verify_and_auth 邮箱注册: user_id={user['user_id']}")
            return (user, True)
        else:
            user = await self.login_with_phone(phone, area_code, preferred_language=preferred_language)
            if user:
                logger.info(f"✅ verify_and_auth 手机登录: user_id={user['user_id']}")
                return (user, False)
            user = await self.register_with_phone(phone, area_code=area_code, preferred_language=preferred_language)
            logger.info(f"✅ verify_and_auth 手机注册: user_id={user['user_id']}")
            return (user, True)

    async def get_community_choice(self, user_id: str) -> Optional[Dict]:
        """获取用户「退出早期共创社区」选项（V2.0），用于登录页回显"""
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(UserCommunityChoice).where(UserCommunityChoice.user_id == user_id)
            )
            row = result.scalar_one_or_none()
            return row.to_dict() if row else None

    async def upsert_community_choice(self, user_id: str, opted_out: bool) -> None:
        """更新用户「退出早期共创社区」选项（V2.0）"""
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(UserCommunityChoice).where(UserCommunityChoice.user_id == user_id)
            )
            row = result.scalar_one_or_none()
            if row:
                row.opted_out = opted_out
            else:
                session.add(UserCommunityChoice(user_id=user_id, opted_out=opted_out))
            await session.commit()

    async def get_early_community(self, user_id: str) -> Optional[Dict]:
        """获取用户早期共创状态（V2.0）"""
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(UserEarlyCommunity).where(UserEarlyCommunity.user_id == user_id)
            )
            row = result.scalar_one_or_none()
            return row.to_dict() if row else None

    async def submit_early_community(self, user_id: str, email: Optional[str] = None) -> Dict:
        """提交早期共创：标记弹窗已展示；若传 email 则写入并标记已加入（V2.0）"""
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(UserEarlyCommunity).where(UserEarlyCommunity.user_id == user_id)
            )
            row = result.scalar_one_or_none()
            now = datetime.now()
            if row:
                row.popup_shown = True
                if email and email.strip():
                    row.email = email.strip()
                    row.joined_at = now
            else:
                session.add(
                    UserEarlyCommunity(
                        user_id=user_id,
                        email=email.strip() if email and email.strip() else None,
                        popup_shown=True,
                        joined_at=now if (email and email.strip()) else None,
                    )
                )
            await session.commit()
            result2 = await session.execute(
                select(UserEarlyCommunity).where(UserEarlyCommunity.user_id == user_id)
            )
            r = result2.scalar_one_or_none()
            return r.to_dict() if r else {}

    async def check_user_exists_by_qq_openid(self, openid: str, unionid: Optional[str] = None) -> bool:
        """
        检查QQ OpenID对应的用户是否存在
        
        Args:
            openid: QQ OpenID
            unionid: QQ UnionID（可选）
        
        Returns:
            用户是否存在
        """
        try:
            async with db_manager.get_session() as session:
                # 优先用 openid 查询
                result = await session.execute(
                    select(User).where(User.qq_openid == openid)
                )
                existing_user = result.scalar_one_or_none()
                logger.debug(f"🔍 [检查QQ用户] openid查询结果 - openid={openid[:16]}..., found={existing_user is not None}, user_id={existing_user.user_id if existing_user else None}")
                
                # 如果未命中且拿到了 unionid，则尝试用 unionid 查询
                if not existing_user and unionid:
                    union_result = await session.execute(
                        select(User).where(User.qq_unionid == unionid)
                    )
                    existing_user = union_result.scalar_one_or_none()
                    logger.debug(f"🔍 [检查QQ用户] unionid查询结果 - unionid={unionid[:16] if unionid else None}..., found={existing_user is not None}, user_id={existing_user.user_id if existing_user else None}")
                
                exists = existing_user is not None
                logger.info(f"🔍 [检查QQ用户] 最终结果 - openid={openid[:16]}..., unionid={unionid[:16] if unionid else None}..., exists={exists}")
                return exists
        except Exception as e:
            logger.error(f"❌ 检查QQ用户是否存在失败: {e}")
            return False


# 全局用户服务实例
user_service = UserService()