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
from sqlalchemy import select, update, func, insert
from sqlalchemy.ext.asyncio import AsyncSession
from database import db_manager
from models_db import User, UserUIState, UserBackgroundImage, BackgroundImage

logger = logging.getLogger(__name__)

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 _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) -> 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",
                    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 session.commit()
                await session.refresh(user)
            
            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_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_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,
        avatar_c_unlocked: Optional[bool] = None,
        has_shared_avatar: 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 avatar_c_unlocked is not None:
                        data["avatar_c_unlocked"] = avatar_c_unlocked
                    if has_shared_avatar is not None:
                        data["has_shared_avatar"] = has_shared_avatar

                    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 avatar_c_unlocked is not None:
                        state.avatar_c_unlocked = avatar_c_unlocked
                    if has_shared_avatar is not None:
                        state.has_shared_avatar = has_shared_avatar

                    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 register_with_phone(self, phone: str, username: str = "") -> Dict:
        """
        使用手机号注册
        
        Args:
            phone: 手机号
            username: 用户名（可选）
        
        Returns:
            用户信息字典
        """
        try:
            async with db_manager.get_session() as session:
                # 检查手机号是否已注册
                result = await session.execute(
                    select(User).where(User.phone == phone)
                )
                existing_user = result.scalar_one_or_none()
                
                if existing_user:
                    raise ValueError("该手机号已被注册")
                
                # 生成用户ID和用户名
                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,
                    phone=phone,
                    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 session.commit()
                await session.refresh(user)
            
            logger.info(f"✅ 手机号注册成功: user_id={user_id}, 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 = "") -> Dict:
        """
        使用邮箱注册
        
        Args:
            email: 邮箱
            username: 用户名（可选）
        
        Returns:
            用户信息字典
        """
        try:
            async with db_manager.get_session() as session:
                # 检查邮箱是否已注册
                result = await session.execute(
                    select(User).where(User.email == email)
                )
                existing_user = result.scalar_one_or_none()
                
                if existing_user:
                    raise ValueError("该邮箱已被注册")
                
                # 生成用户ID和用户名
                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,
                    email=email,
                    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 session.commit()
                await session.refresh(user)
            
            logger.info(f"✅ 邮箱注册成功: user_id={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 session.commit()
                await session.refresh(user)
            
            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,
    ) -> Dict:
        """
        使用QQ OpenID登录/注册
        
        Args:
            openid: QQ OpenID
            username: 用户名（可选，仅新用户时使用）
        
        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:
                    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 = str(uuid.uuid4())
                if not username or username.strip() == "":
                    username = await self._generate_temp_username(session)
                
                user = User(
                    user_id=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，但不提交事务
                
                # 初始化背景图片
                await self._init_user_background_image(session, user_id)
                
                await session.commit()
                await session.refresh(user)
            
            logger.info(
                f"✅ QQ注册成功: user_id={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) -> Optional[Dict]:
        """
        使用手机号登录
        
        Args:
            phone: 手机号
        
        Returns:
            用户信息字典，如果不存在返回None
        """
        try:
            async with db_manager.get_session() as session:
                result = await session.execute(
                    select(User).where(User.phone == phone)
                )
                user = result.scalar_one_or_none()
            
            if user:
                logger.info(f"✅ 手机号登录成功: user_id={user.user_id}, 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) -> Optional[Dict]:
        """
        使用邮箱登录
        
        Args:
            email: 邮箱
        
        Returns:
            用户信息字典，如果不存在返回None
        """
        try:
            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:
                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


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