Tolan/backend/main.py

"""
主应用入口
"""
from __future__ import annotations

import json
import logging
import urllib.parse
import hashlib
import uuid
import math
import random
from decimal import Decimal
import base64
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any, Literal
from fastapi import FastAPI, WebSocket, HTTPException, Request, Query, Header, Depends
from fastapi.exceptions import RequestValidationError
from fastapi.responses import RedirectResponse, PlainTextResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request as StarletteRequest
from starlette.responses import Response
from pydantic import BaseModel, Field, field_validator, ValidationError as PydanticValidationError
from contextlib import asynccontextmanager
import httpx
from xml.etree import ElementTree as ET
from sqlalchemy import text, select, update
from sqlalchemy.orm import selectinload
from config import settings, get_dashscope_domain
from websocket_handler import WebSocketHandler
from database import init_database, close_database, db_manager
from services.redis_cache_service import RedisCacheService
from services.user_service import user_service, AuthError
from ai_service import ai_service
from error_codes import (
    UNAUTHORIZED_USER_MISMATCH,
    SESSION_NOT_INITIALIZED,
    SESSION_EXPIRED,
    USER_NOT_FOUND,
    USER_NOT_FOUND_REGISTER_FIRST,
    CREATE_USER_FAILED,
    INIT_SESSION_FAILED,
    GET_USER_INFO_FAILED,
    UPDATE_USERNAME_FAILED,
    USER_NOT_FOUND_OR_UPDATE_FAILED,
    UPDATE_BG_IMAGE_FAILED,
    UPDATE_BOT_NAME_FAILED,
    UPDATE_BOT_NAME_TIMEOUT,
    UPDATE_PREFERRED_LANGUAGE_FAILED,
    GET_ENERGY_STONES_FAILED,
    GET_AVATARS_FAILED,
    GET_BG_CONFIG_FAILED,
    GET_USER_BG_IMAGES_FAILED,
    CLAIM_SHARE_REWARD_FAILED,
    FEEDBACK_CONTENT_EMPTY,
    SUBMIT_FEEDBACK_FAILED,
    REGISTER_FAILED,
    LOGIN_FAILED,
    PHONE_REQUIRED,
    EMAIL_REQUIRED,
    SEND_TYPE_INVALID,
    SEND_CODE_FAILED,
    CODE_REQUIRED,
    VERIFY_CODE_FAILED,
    TYPE_MUST_BE_EMAIL_OR_PHONE,
    GET_UI_STATE_FAILED,
    UPDATE_UI_STATE_FAILED,
    WECHAT_OAUTH_NOT_CONFIGURED,
    QQ_OAUTH_NOT_CONFIGURED,
    WECHAT_MP_NOT_CONFIGURED,
    WECHAT_JSAPI_CONFIG_FAILED,
    GET_CONFIG_FAILED,
    SERVER_ERROR,
    INSUFFICIENT_CONVERSATION_ENERGY,
    PURCHASE_LIMIT_REACHED,
    SHOP_ITEM_NOT_FOUND,
    CART_QUANTITY_LIMIT_EXCEEDED,
)
from services.wechat_jsapi_service import WechatJSAPIService
from services.energy_stone_service import energy_stone_service
from services.scenario_topic_service import scenario_topic_service
from models_db import VerificationCode, UserAvatar, UserBackgroundImage, Feedback, BackgroundImage, User
from session_store import SessionStore
from shop_item_image_utils import scene_image_name_from_shop_row

# 配置日志
logging.basicConfig(
    level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# 🔇 过滤第三方库的 DEBUG 日志（避免刷屏）
# 这些库的 DEBUG 日志包含 HTTP/2 协议细节、HPACK 压缩等底层信息，对生产环境无意义
third_party_loggers = [
    'openai',
    'openai._base_client',
    'httpcore',
    'httpcore.http2',
    'httpx',
    'hpack',
    'hpack.hpack',
    'hpack.table',
]
for logger_name in third_party_loggers:
    logging.getLogger(logger_name).setLevel(logging.INFO)

logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动时初始化数据库（包含MySQL和Redis，自动启动健康检查任务）
    try:
        await init_database()
        
        # 获取Redis客户端（已配置健康检查），用于验证连通性
        from database import db_manager
        await db_manager.get_redis()
        
        # 创建Redis缓存服务，按需获取最新连接
        app.state.redis_cache = RedisCacheService(db_manager.get_redis)
        
    except Exception as e:
        logger.error(f"❌ 数据库初始化失败: {e}")
        app.state.redis_cache = None
    
    # 🔥 启动HTTP连接保活任务（保持连接活跃，避免长时间不交互后响应变慢）
    try:
        from http_pool import http_pool
        from ai_service import AIService
        
        # 创建AIService实例用于保活（共享连接池）
        ai_service = AIService()
        http_pool.start_keepalive(ai_service=ai_service)
    except Exception as e:
        logger.warning(f"⚠️ HTTP连接保活任务启动失败: {e}")
    
    yield
    
    # 关闭时清理资源
    await close_database()
    
    # 🔥 关闭HTTP连接池（类似SQLAlchemy的Engine.dispose()）
    from http_pool import http_pool
    http_pool.close()


app = FastAPI(
    title="Lyrria Backend API",
    lifespan=lifespan
)
# 全局会话存储（单用户单活 token），使用 Redis 持久化
from database import db_manager  # 延迟导入以避免循环依赖问题
app.state.session_store = SessionStore(
    redis_getter=db_manager.get_redis,
    default_ttl_seconds=settings.SESSION_TOKEN_TTL_SECONDS,
)


class APIKeyMiddleware(BaseHTTPMiddleware):
    """API Key验证中间件"""
    
    # 不需要验证的路径（健康检查、WebSocket、微信/QQ回调等）
    EXEMPT_PATHS = [
        "/",
        "/docs",
        "/openapi.json",
        "/redoc",
        "/ws",
        # 微信 OAuth & 公众号回调
        "/api/auth/wechat/authorize",
        "/api/auth/wechat/callback",
        "/api/wechat/mp/verify",
        "/api/wechat/mp/message",
        # QQ OAuth & 回调
        "/api/auth/qq/authorize",
        "/api/qq/mp/verify",
        # QQ 分享配置（SDK 拉取）
        "/api/qq/share/config",
        # web-admin 免登录管理接口（按需求）
        "/api/admin/inbox",
        "/api/admin/users",
    ]
    
    async def dispatch(self, request: StarletteRequest, call_next):
        # 检查是否为豁免路径
        path = request.url.path
        if any(path.startswith(exempt) for exempt in self.EXEMPT_PATHS):
            return await call_next(request)
        
        # 如果未启用API Key验证，直接通过
        if not settings.API_KEY_ENABLED:
            return await call_next(request)
        
        # 检查API Key
        api_key = request.headers.get("X-Client-Key", "")
        expected_key = settings.API_CLIENT_KEY
        
        if not expected_key:
            logger.warning("⚠️ API_CLIENT_KEY未配置，跳过验证")
            return await call_next(request)
        
        if api_key != expected_key:
            logger.warning(f"❌ API Key验证失败: path={path}, provided_key={'***' if api_key else '(empty)'}")
            return Response(
                content='{"detail":"Invalid API Key"}',
                status_code=403,
                media_type="application/json"
            )
        
        return await call_next(request)


# 配置CORS - 使用配置中的允许来源
cors_origins = settings.CORS_ORIGINS if not settings.CORS_ALLOW_ALL else ["*"]
app.add_middleware(
    CORSMiddleware,
    allow_origins=cors_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 添加API Key验证中间件（在CORS之后）
app.add_middleware(APIKeyMiddleware)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(_request: Request, exc: RequestValidationError) -> JSONResponse:
    """422 时在服务端打印校验错误，便于排查（如 user_events 缺失字段）"""
    errors = exc.errors()
    for err in errors:
        logger.warning("422 校验失败: loc=%s msg=%s type=%s", err.get("loc"), err.get("msg"), err.get("type"))
    return JSONResponse(status_code=422, content={"detail": errors})


async def require_user_session_from_headers(
    request: Request,
    header_user_id: Optional[str] = Header(None, alias="X-User-Id"),
    token: Optional[str] = Header(None, alias="X-User-Token"),
) -> str:
    """仅根据 X-User-Id + X-User-Token 校验会话，返回 user_id（用于无 path user_id 的接口如 POST /api/user-events）"""
    if not header_user_id or not token:
        logger.warning(
            "require_user_session_from_headers: missing headers - path=%s",
            request.url.path,
        )
        raise HTTPException(status_code=401, detail={"code": SESSION_EXPIRED})
    session_store = getattr(request.app.state, "session_store", None)
    if not session_store:
        logger.error(
            "require_user_session_from_headers: SESSION_NOT_INITIALIZED - path=%s",
            request.url.path,
        )
        raise HTTPException(status_code=500, detail={"code": SESSION_NOT_INITIALIZED})
    is_valid = await session_store.validate(header_user_id, token, refresh=True)
    if not is_valid:
        logger.warning(
            "require_user_session_from_headers: SESSION_EXPIRED - path=%s, user_id=%s",
            request.url.path,
            header_user_id,
        )
        raise HTTPException(status_code=401, detail={"code": SESSION_EXPIRED})
    return header_user_id


async def require_user_session(
    request: Request,
    user_id: str,
    header_user_id: Optional[str] = Header(None, alias="X-User-Id"),
    token: Optional[str] = Header(None, alias="X-User-Token"),
) -> None:
    """
    基于 user_id + token 的简单 HTTP 鉴权，复用 WebSocket 的 session_store。
    仅用于需要用户身份的业务接口，保持单用户单活语义。
    """
    if not header_user_id or header_user_id != user_id:
        logger.warning(
            "require_user_session: UNAUTHORIZED_USER_MISMATCH - path=%s, param_user_id=%s, header_user_id=%s",
            request.url.path,
            user_id,
            header_user_id,
        )
        raise HTTPException(status_code=401, detail={"code": UNAUTHORIZED_USER_MISMATCH})

    session_store = getattr(request.app.state, "session_store", None)
    if not session_store:
        logger.error(
            "require_user_session: SESSION_NOT_INITIALIZED - path=%s, user_id=%s",
            request.url.path,
            header_user_id,
        )
        raise HTTPException(status_code=500, detail={"code": SESSION_NOT_INITIALIZED})

    if not token:
        logger.warning(
            "require_user_session: missing X-User-Token header - path=%s, user_id=%s",
            request.url.path,
            header_user_id,
        )
        raise HTTPException(status_code=401, detail={"code": SESSION_EXPIRED})

    is_valid = await session_store.validate(user_id, token, refresh=True)
    if not is_valid:
        try:
            entry = await session_store.get_current_token(user_id)
        except Exception as diag_exc:  # noqa: BLE001
            logger.warning(
                "require_user_session: validate failed and get_current_token raised - path=%s, user_id=%s, token_prefix=%s, error=%r",
                request.url.path,
                user_id,
                str(token)[:8],
                diag_exc,
            )
            raise HTTPException(status_code=401, detail={"code": SESSION_EXPIRED})

        if entry is None:
            logger.warning(
                "require_user_session: SESSION_EXPIRED - no active session entry - path=%s, user_id=%s, token_prefix=%s",
                request.url.path,
                user_id,
                str(token)[:8],
            )
        else:
            try:
                expires_iso = datetime.fromtimestamp(entry.expires_at).isoformat()
            except Exception:
                expires_iso = str(entry.expires_at)
            logger.warning(
                "require_user_session: SESSION_EXPIRED - token mismatch or expired - path=%s, user_id=%s, header_token_prefix=%s, stored_token_prefix=%s, stored_expires_at=%s, now=%s",
                request.url.path,
                user_id,
                str(token)[:8],
                str(entry.token)[:8],
                expires_iso,
                datetime.utcnow().isoformat(),
            )

        raise HTTPException(status_code=401, detail={"code": SESSION_EXPIRED})


async def optional_user_id(
    request: Request,
    header_user_id: Optional[str] = Header(None, alias="X-User-Id"),
    token: Optional[str] = Header(None, alias="X-User-Token"),
) -> Optional[str]:
    """可选鉴权：有且有效则返回 user_id，否则返回 None（用于页面访问等不强制登录的接口）"""
    if not header_user_id or not token:
        return None
    session_store = getattr(request.app.state, "session_store", None)
    if not session_store or not await session_store.validate(header_user_id, token, refresh=False):
        return None
    return header_user_id


class AdminCreateEnergyGiftInboxRequest(BaseModel):
    title_zh: str = Field(..., min_length=1, max_length=200)
    title_en: str = Field(..., min_length=1, max_length=200)
    content_zh: str = Field(..., min_length=1, max_length=10000)
    content_en: str = Field(..., min_length=1, max_length=10000)
    energy_stone_amount: int = Field(..., gt=0, le=100000)
    user_ids: List[str] = Field(..., min_length=1)

    @field_validator("user_ids")
    @classmethod
    def validate_user_ids(cls, value: List[str]) -> List[str]:
        normalized = [str(v).strip() for v in value if str(v).strip()]
        if not normalized:
            raise ValueError("user_ids不能为空")
        return list(dict.fromkeys(normalized))


class AdminCreateNewsInboxRequest(BaseModel):
    title_zh: str = Field(..., min_length=1, max_length=200)
    title_en: str = Field(..., min_length=1, max_length=200)
    content_zh: str = Field(..., min_length=1, max_length=10000)
    content_en: str = Field(..., min_length=1, max_length=10000)


@app.get("/")
async def root():
    """健康检查"""
    return {"status": "ok", "message": "Lyrria Backend API"}


@app.post("/api/user-events")
async def create_user_event(
    http_request: Request,
    auth_user_id: str = Depends(require_user_session_from_headers),
) -> Dict[str, Any]:
    """
    前端上报单条用户事件

    - 主要用于页面进入、按钮点击等埋点
    - event_name 必须遵循 .cursorrules 中的“埋点事件命名规范”
    - 从 body 手动读取并校验，避免与依赖中的 Request 冲突导致 body 被当成 query
    """
    try:
        body = await http_request.json()
    except Exception as e:
        logger.warning("user_events 读取 body 失败: %s", e)
        raise HTTPException(status_code=422, detail=[{"type": "value_error", "loc": ("body",), "msg": "Invalid JSON body"}])
    try:
        payload = UserEventPayload.model_validate(body)
    except PydanticValidationError as val_err:
        logger.warning("user_events 校验失败: %s", val_err.errors())
        raise RequestValidationError(val_err.errors()) from val_err
    if payload.user_id != auth_user_id:
        raise HTTPException(status_code=401, detail={"code": UNAUTHORIZED_USER_MISMATCH})
    try:
        ok = await user_service.save_user_event(payload.model_dump())
        if not ok:
            logger.warning("user_events 落库返回 False: event_id=%s, event_name=%s", payload.event_id, payload.event_name)
            raise HTTPException(status_code=500, detail="failed_to_save_user_event")
        logger.info("user_events 已写入: event_name=%s, user_id=%s", payload.event_name, payload.user_id)
        return {"success": True}
    except HTTPException:
        raise
    except Exception as e:  # noqa: BLE001
        logger.error(f"❌ 保存用户事件失败: event_id={payload.event_id}, error={e}")
        raise HTTPException(status_code=500, detail="failed_to_save_user_event") from e


class CreateUserRequest(BaseModel):
    """创建用户请求"""
    username: str = Field("", description="用户名/昵称，可为空字符串")
    bot_name: Optional[str] = Field(None, description="机器人名字，可选")
    user_segment: str = Field("individual", description="用户分群：individual/school/internal等")
    org_id: Optional[str] = Field(None, description="组织ID(学校/机构用户)")
    is_test_user: bool = Field(False, description="是否测试账号")
    region: Optional[str] = Field(None, description="国家/地区代码，如 CN/US")


class UpdateUsernameRequest(BaseModel):
    """更新用户名请求"""
    username: str


class UpdateBotNameRequest(BaseModel):
    """更新机器人名字请求"""
    bot_name: str


class UpdatePreferredLanguageRequest(BaseModel):
    """更新用户界面语言偏好（zh/en）"""
    preferred_language: Literal["zh", "en"]


class UpdateBackgroundImageIdRequest(BaseModel):
    """更新背景图片ID请求"""
    background_image_id: str


class UpdateChatbotBackdropRequest(BaseModel):
    """切换 ChatBot 幕布请求"""
    backdrop_item_id: int


class ChatbotFurnitureActionRequest(BaseModel):
    """装备/卸下 ChatBot 家具请求"""
    item_id: int


class ChatbotShopPurchaseRequest(BaseModel):
    """ChatBot 商店购买家具/幕布"""
    item_id: int


class ChatbotShopCartAddRequest(BaseModel):
    """ChatBot 商店加入购物车（家具/幕布）"""
    item_id: int


class ChatbotShopCartRemoveRequest(BaseModel):
    """ChatBot 商店移除购物车商品（家具/幕布）"""

    item_id: int


class ChatbotShopCartPurchaseRequest(BaseModel):
    """ChatBot 商店购买购物车全部商品（家具/幕布）"""

    pass


def _floor_shop_price_energy(price: object) -> int:
    """与 energy_stone_service 一致：售价向下取整为整颗对话能量石。"""
    try:
        v = float(price or 0)
    except (TypeError, ValueError):
        return 0
    if v <= 0:
        return 0
    return int(math.floor(v))


class UserEventPayload(BaseModel):
    """前端上报用户事件 payload（单条）"""

    event_id: str = Field(..., description="事件ID，客户端或服务端生成，需全局唯一")
    user_id: str = Field(..., description="用户ID(关联 users.user_id)")
    event_name: str = Field(..., description="事件代码，如 page.chatbot__enter（遵循埋点命名规范）")
    event_label: Optional[str] = Field(None, description="事件中文名称/描述")
    event_time: datetime = Field(..., description="事件发生时间（UTC）")
    session_id: Optional[str] = Field(None, description="会话ID，同一次前台停留共享")
    platform: str = Field(..., description="平台：ios/android/web")
    app_version: Optional[str] = Field(None, description="应用版本号，如 v0.1.x")
    timezone_offset_minutes: Optional[int] = Field(
        None,
        description="用户本地相对UTC的分钟偏移量（分钟）",
    )
    properties: Optional[Dict[str, Any]] = Field(
        None,
        description="扩展属性（page、entry、button_id、duration_seconds 等）",
    )

    @field_validator("event_time", mode="before")
    @classmethod
    def parse_event_time(cls, v: Any) -> datetime:
        """接受 ISO 字符串（含 Z 后缀），避免 422"""
        if isinstance(v, datetime):
            return v
        if isinstance(v, str):
            s = v.strip().replace("Z", "+00:00")
            return datetime.fromisoformat(s)
        raise ValueError("event_time 需为 datetime 或 ISO 字符串")


class RegisterPhoneRequest(BaseModel):
    """手机号注册请求"""
    area_code: Optional[str] = None
    phone: str
    username: str = ""
    user_id: Optional[str] = None  # 可选的现有用户ID，如果提供则更新该用户而不是创建新用户
    preferred_language: Optional[str] = None  # "zh" | "en"


class RegisterEmailRequest(BaseModel):
    """邮箱注册请求"""
    email: str
    username: str = ""
    user_id: Optional[str] = None  # 可选的现有用户ID，如果提供则更新该用户而不是创建新用户
    preferred_language: Optional[str] = None  # "zh" | "en"


class LoginWechatRequest(BaseModel):
    """微信登录请求"""
    openid: str
    username: str = ""


class LoginQQRequest(BaseModel):
    """QQ登录请求"""
    openid: str
    username: str = ""


class LoginPhoneRequest(BaseModel):
    """手机号登录请求"""
    area_code: Optional[str] = None
    phone: str
    preferred_language: Optional[str] = None  # "zh" | "en"


class LoginEmailRequest(BaseModel):
    """邮箱登录请求"""
    email: str
    preferred_language: Optional[str] = None  # "zh" | "en"


class SendCodeRequest(BaseModel):
    """发送验证码请求"""
    area_code: Optional[str] = None  # 手机区号（如 +86）
    phone: Optional[str] = None
    email: Optional[str] = None
    send_type: str  # "domestic_sms", "international_sms", "email"


class VerifyCodeRequest(BaseModel):
    """验证验证码请求"""
    phone: Optional[str] = None
    email: Optional[str] = None
    code: str
    send_type: str  # "domestic_sms", "international_sms", "email"


class VerifyAndAuthRequest(BaseModel):
    """V2.0 验证码通过后统一登录/注册请求"""
    type: str  # "email" | "phone"
    code: str
    email: Optional[str] = None
    phone: Optional[str] = None
    area_code: Optional[str] = None
    preferred_language: Optional[str] = None  # "zh" | "en"


class CommunityChoiceRequest(BaseModel):
    """V2.0 退出早期共创社区选项"""
    opted_out: bool


class EarlyCommunityRequest(BaseModel):
    """V2.0 早期共创邮箱（可选）"""
    email: Optional[str] = None


class UserUIStateResponse(BaseModel):
    """用户前端可视化状态"""

    user_id: str
    current_page: Optional[str] = None
    visited_routes: List[str] = []
    has_shown_tap_to_chat: bool = False
    has_shown_cancel_tooltip: bool = False
    has_shown_send_tooltip: bool = False
    has_shown_progress_bar_tooltip: bool = False
    has_shown_mission_stone_modal: bool = False
    has_shown_conversation_stone_modal: bool = False
    has_clicked_mission_stone: bool = False
    has_clicked_conversation_stone: bool = False
    subtitle_on: bool = False
    avatar_c_unlocked: bool = False
    has_shared_avatar: bool = False
    current_avatar_id: Optional[str] = None
    sfx_enabled: bool = True
    bgm_enabled: bool = True


class PageVisitRequest(BaseModel):
    """页面访问上报"""
    path: str


class UpdateUserUIStateRequest(BaseModel):
    """部分更新用户前端可视化状态"""

    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
    has_shared_avatar: Optional[bool] = None
    current_avatar_id: Optional[str] = None
    subtitle_on: Optional[bool] = None
    sfx_enabled: Optional[bool] = None
    bgm_enabled: Optional[bool] = None


# 分享图片响应模型
class ShareImageItem(BaseModel):
    id: str
    url: str
    title: Optional[str] = None
    sort_order: Optional[int] = None
    is_active: Optional[bool] = None


SHARE_IMAGES: List[ShareImageItem] = [
    ShareImageItem(id="share-1", url="https://www.lyrria.com/img/share-1.png"),
    ShareImageItem(id="share-2", url="https://www.lyrria.com/img/share-2.png"),
]


class AvatarGifSet(BaseModel):
    default: str
    positive: str
    negative: str
    talking: str


async def _generate_session_token(user_id: str) -> Dict[str, Any]:
    """
    为指定用户生成会话 token（复用 init_session 的逻辑）
    
    Args:
        user_id: 用户ID
    
    Returns:
        包含 token 和 expires_at 的字典
    """
    session_store = getattr(app.state, "session_store", None)
    token = str(uuid.uuid4())
    expires_at = datetime.utcnow() + timedelta(seconds=settings.SESSION_TOKEN_TTL_SECONDS)
    if session_store:
        await session_store.set_current_token(user_id, token, settings.SESSION_TOKEN_TTL_SECONDS)
        # 主动踢掉旧连接
        await session_store.invalidate_connection(user_id)
    return {
        "token": token,
        "expires_at": expires_at.isoformat() + "Z",
    }


class SessionInitRequest(BaseModel):
    user_id: Optional[str] = None
    username: Optional[str] = ""


AVATAR_GIF_CONFIG = {
    "avatar_default": AvatarGifSet(
        default="https://www.lyrria.com/img/01-Default.gif",
        positive="https://www.lyrria.com/img/04-Positive-emotion.gif",
        negative="https://www.lyrria.com/img/05-Negative-emotion.gif",
        talking="https://www.lyrria.com/img/06-talking.gif",
    ),
    "avatar_registered_cap": AvatarGifSet(
        default="https://www.lyrria.com/img/01-Default-Registered.gif",
        positive="https://www.lyrria.com/img/04-Positive-emotion-Registered.gif",
        negative="https://www.lyrria.com/img/05-Negative-emotion-Registered.gif",
        talking="https://www.lyrria.com/img/06-talking-Registered.gif",
    ),
}


@app.post("/api/users/create")
async def create_user(request: CreateUserRequest):
    """
    创建游客用户
    
    Returns:
        {
            "user_id": "uuid",
            "username": "",
            "user_type": "guest",
            "bot_name": "机器人名字（可选）"
        }
    """
    try:
        user = await user_service.create_guest_user(
            username=request.username,
            bot_name=request.bot_name,
            user_segment=request.user_segment,
            org_id=request.org_id,
            is_test_user=request.is_test_user,
            region=request.region,
        )
        return {
            "success": True,
            "data": user
        }
    except Exception as e:
        logger.exception("创建用户失败")
        raise HTTPException(status_code=500, detail={"code": CREATE_USER_FAILED})


@app.post("/api/session/init")
async def init_session(request: SessionInitRequest):
    """
    初始化或刷新用户会话，返回 user_id + 会话 token（单用户单活）
    """
    try:
        session_store = getattr(app.state, "session_store", None)
        # 1. 获取或创建用户
        user_id = request.user_id
        user_payload: Optional[Dict[str, Any]] = None
        if not user_id:
            user_payload = await user_service.create_guest_user(
                username=request.username or ""
            )
            user_id = user_payload["user_id"]
        else:
            # 确认用户存在
            user_payload = await user_service.get_user(user_id)
            if not user_payload:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
        # 2. 签发新 token（覆盖旧 token，单活）
        token = str(uuid.uuid4())
        expires_at = datetime.utcnow() + timedelta(seconds=settings.SESSION_TOKEN_TTL_SECONDS)
        if session_store:
            await session_store.set_current_token(user_id, token, settings.SESSION_TOKEN_TTL_SECONDS)
            # 主动踢掉旧连接
            await session_store.invalidate_connection(user_id)
        data = {
            "user_id": user_id,
            "token": token,
            "expires_at": expires_at.isoformat() + "Z",
        }
        # 如果刚创建用户，透传部分信息，兼容前端缓存逻辑
        if user_payload:
            data.update(
                {
                    "username": user_payload.get("username"),
                    "user_level": user_payload.get("user_level"),
                    "current_background_image_id": user_payload.get(
                        "current_background_image_id"
                    ),
                    "current_background_image_metadata": user_payload.get(
                        "current_background_image_metadata"
                    ),
                }
            )
        return {"success": True, "data": data}
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("初始化会话失败")
        raise HTTPException(status_code=500, detail={"code": INIT_SESSION_FAILED})


@app.get("/api/users/{user_id}")
async def get_user(user_id: str, _: None = Depends(require_user_session)):
    """
    获取用户信息
    
    Args:
        user_id: 用户ID
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "username": "",
                "user_type": "guest",
                "bot_name": "机器人名字（可选）",
                ...
            }
        }
    """
    try:
        user = await user_service.get_user(user_id)
        if user:
            return {
                "success": True,
                "data": user
            }
        else:
            raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("获取用户信息失败")
        raise HTTPException(status_code=500, detail={"code": GET_USER_INFO_FAILED})


@app.get("/api/users/{user_id}/energy-stones/count")
async def get_energy_stone_counts(user_id: str, _: None = Depends(require_user_session)):
    """
    获取用户的能量石数量
    
    Args:
        user_id: 用户ID
    
    Returns:
        {
            "success": true,
            "data": {
                "mission_count": 0,
                "conversation_count": 0
            }
        }
    """
    try:
        if not energy_stone_service.db:
            await energy_stone_service.set_db_manager(db_manager)
        
        counts = await energy_stone_service.get_stone_counts(user_id)
        return {
            "success": True,
            "data": counts
        }
    except Exception as e:
        logger.exception("获取能量石数量失败")
        raise HTTPException(status_code=500, detail={"code": GET_ENERGY_STONES_FAILED})


@app.get("/api/users/{user_id}/journal-latest-topic")
async def get_journal_latest_topic(user_id: str, _: None = Depends(require_user_session)):
    """
    个人页「日记本」左侧图：用户最近在 ChatBot 触发特殊情景话题的最新一条记录。
    无记录时 data 为 null。
    """
    try:
        if not scenario_topic_service.db:
            await scenario_topic_service.set_db_manager(db_manager)
        latest = await scenario_topic_service.get_latest_log_for_journal(user_id)
        return {"success": True, "data": latest}
    except Exception:
        logger.exception("获取日记本最近情景话题失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


def _get_tutorial_intro_text(step: int) -> str | None:
    """根据教程步骤返回对应的固定英文文案。"""
    mapping: Dict[int, str] = {
        1: "Hi! Nice to meet you, my friend. Let me show you my home.",
        3: "I’ve added captions so we can understand each other even better!",
        5: "Tap the button and let's talk! How was your day",
        9: "You can see your Energy Stones here.",
        10: "You can also see our Bond Level here. It shows how close we are.",
        11: "That's all! let's start our first chat!",
    }
    return mapping.get(step)


@app.get("/api/tutorials/{step}/intro-audio")
async def get_tutorial_intro_audio(
    step: int,
    _: str = Depends(require_user_session_from_headers),
):
    """
    新手教程步骤：返回对应字幕文案的 TTS 音频。

    - 支持步骤：1 / 3 / 5 / 9 / 10 / 11
    - 文案与前端字幕保持一致（ChatBotStaticShell 中的英文文案）
    - 使用全局 ai_service 进行 TTS 合成，返回单段 Base64 PCM。
    """
    text = _get_tutorial_intro_text(step)
    if text is None:
        raise HTTPException(status_code=404, detail={"code": GET_CONFIG_FAILED})

    try:
        audio_chunks: list[bytes] = []
        async for chunk in ai_service.text_to_speech_stream(text, emotion="friendly"):
            audio_chunks.append(chunk)

        if not audio_chunks:
            raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})

        audio_bytes = b"".join(audio_chunks)
        audio_base64 = base64.b64encode(audio_bytes).decode("utf-8")

        return {
            "success": True,
            "data": {
                "audio": audio_base64,
                "sampleRate": 24000,
                "format": "pcm",
                "text": text,
            },
        }
    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("生成教程步骤 TTS 音频失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/avatars")
async def get_user_avatars(user_id: str, _: None = Depends(require_user_session)):
    """
    获取用户已拥有的头像列表
    
    Args:
        user_id: 用户ID
    
    Returns:
        {
            "success": true,
            "data": {
                "avatar_ids": ["avatar_default", "avatar_registered_cap", ...]
            }
        }
    """
    try:
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(UserAvatar.avatar_id).where(UserAvatar.user_id == user_id)
            )
            avatar_ids = [row[0] for row in result.fetchall()]
            
            return {
                "success": True,
                "data": {
                    "avatar_ids": avatar_ids
                }
            }
    except Exception as e:
        logger.exception("获取用户头像信息失败")
        raise HTTPException(status_code=500, detail={"code": GET_AVATARS_FAILED})


@app.get("/api/background-images")
async def get_background_images():
    """
    获取所有背景图片配置列表
    
    Returns:
        {
            "success": true,
            "data": [
                {
                    "image_id": "default",
                    "image_name": "默认背景",
                    "image_url": "https://...",
                    "thumbnail_url": "https://...",
                    "unlock_type": "default",
                    "display_order": 0,
                    "is_active": true,
                    ...
                },
                ...
            ]
        }
    """
    try:
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(BackgroundImage).where(BackgroundImage.is_active == True).order_by(BackgroundImage.display_order)
            )
            background_images = result.scalars().all()
            
            return {
                "success": True,
                "data": [bg_image.to_dict() for bg_image in background_images]
            }
    except Exception as e:
        logger.error(f"获取背景图片配置失败: {e}")
        logger.exception("获取背景图片配置失败")
        raise HTTPException(status_code=500, detail={"code": GET_BG_CONFIG_FAILED})


@app.get("/api/users/{user_id}/background-images")
async def get_user_background_images(user_id: str, _: None = Depends(require_user_session)):
    """
    获取用户已拥有的背景图片列表
    
    Args:
        user_id: 用户ID
    
    Returns:
        {
            "success": true,
            "data": {
                "image_ids": ["default", "happynewyear", ...]
            }
        }
    """
    try:
        async with db_manager.get_session() as session:
            result = await session.execute(
                select(UserBackgroundImage.image_id).where(UserBackgroundImage.user_id == user_id)
            )
            image_ids = [row[0] for row in result.fetchall()]
            
            return {
                "success": True,
                "data": {
                    "image_ids": image_ids
                }
            }
    except Exception as e:
        logger.exception("获取用户背景图片信息失败")
        raise HTTPException(status_code=500, detail={"code": GET_USER_BG_IMAGES_FAILED})


@app.get("/api/users/{user_id}/chatbot-layout")
async def get_chatbot_layout(user_id: str, _: None = Depends(require_user_session)):
    """
    获取用户 /chatbot 顶部装饰区布局（背景幕布 + 5x4 家具网格）
    返回给前端用于动态渲染。
    """
    try:
        async with db_manager.get_session() as session:
            # 兼容：chatbot_layouts / user_chatbot_backdrops 表里的 user_id 可能是 BIGINT(users.id)
            # 也可能是 VARCHAR(users.user_id)。运行时探测信息架构来决定映射方式。
            async def _get_user_id_is_bigint(table_name: str) -> bool:
                res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = :table
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE, "table": table_name},
                )
                row = res.mappings().first()
                data_type = (row or {}).get("DATA_TYPE") or ""
                return "bigint" in data_type.lower()

            layout_user_id_is_bigint = await _get_user_id_is_bigint("chatbot_layouts")
            backdrop_user_id_is_bigint = await _get_user_id_is_bigint("user_chatbot_backdrops")

            layout_user_id_where_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if layout_user_id_is_bigint
                else ":user_id"
            )
            backdrop_user_id_where_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if backdrop_user_id_is_bigint
                else ":user_id"
            )

            layout_row_res = await session.execute(
                text(
                    f"""
                    SELECT
                        id,
                        grid_rows,
                        grid_cols,
                        cell_width,
                        cell_height,
                        decor_area_width,
                        decor_area_height,
                        updated_at
                    FROM chatbot_layouts l
                    WHERE l.user_id = {layout_user_id_where_expr}
                    ORDER BY l.updated_at DESC, l.id DESC
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            layout_row = layout_row_res.mappings().first()

            backdrop_res = await session.execute(
                text(
                    f"""
                    SELECT
                        ub.z_index,
                        ub.backdrop_item_id,
                        COALESCE(NULLIF(TRIM(si.render_image_name), ''), si.image_name) AS image_name
                    FROM user_chatbot_backdrops ub
                    JOIN shop_items si
                      ON ub.backdrop_item_id = si.id
                    WHERE ub.user_id = {backdrop_user_id_where_expr}
                    ORDER BY ub.z_index DESC, ub.id DESC
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            backdrop_row = backdrop_res.mappings().first()
            current_backdrop_item_id = (
                int(backdrop_row["backdrop_item_id"])
                if backdrop_row and backdrop_row.get("backdrop_item_id") is not None
                else None
            )

            async def _table_exists(table_name: str) -> bool:
                res = 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 = res.mappings().first()
                return bool(row and int(row.get("c") or 0) > 0)

            items: list[dict[str, Any]] = []
            slots: list[dict[str, Any]] = []
            has_new_slot_tables = (
                await _table_exists("user_chatbot_slot_placements")
                and await _table_exists("chatbot_backdrop_slots")
            )
            if has_new_slot_tables and current_backdrop_item_id is not None:
                slots_res = await session.execute(
                    text(
                        """
                        SELECT
                            s.id,
                            s.slot_code,
                            s.placement_type,
                            s.start_row,
                            s.start_col,
                            s.size_rows,
                            s.size_cols,
                            COALESCE(s.z_index, 0) AS z_index,
                            s.render_mode,
                            s.offset_x,
                            s.offset_y,
                            s.pixel_width,
                            s.pixel_height
                        FROM chatbot_backdrop_slots s
                        WHERE s.backdrop_item_id = :backdrop_item_id
                          AND s.is_active = 1
                        ORDER BY COALESCE(s.z_index, 0) ASC, s.id ASC
                        """
                    ),
                    {"backdrop_item_id": current_backdrop_item_id},
                )
                slots = [dict(r) for r in slots_res.mappings().all()]

                slot_items_res = await session.execute(
                    text(
                        f"""
                        SELECT
                            usp.id,
                            usp.slot_code AS slot_code,
                            s.placement_type,
                            s.start_row,
                            s.start_col,
                            s.size_rows,
                            s.size_cols,
                            COALESCE(usp.z_index, s.z_index, si.default_z_index, 0) AS z_index,
                            usp.image_name_snapshot,
                            usp.item_id,
                            COALESCE(si.scene_anchor_type, 'none') AS scene_anchor_type,
                            s.render_mode,
                            s.offset_x,
                            s.offset_y,
                            s.pixel_width,
                            s.pixel_height
                        FROM user_chatbot_slot_placements usp
                        JOIN chatbot_backdrop_slots s
                          ON s.backdrop_item_id = usp.backdrop_item_id
                         AND s.slot_code = usp.slot_code
                         AND s.is_active = 1
                        LEFT JOIN shop_items si ON si.id = usp.item_id
                        WHERE usp.user_id = {layout_user_id_where_expr}
                          AND usp.backdrop_item_id = :backdrop_item_id
                        ORDER BY COALESCE(usp.z_index, s.z_index, si.default_z_index, 0) ASC, usp.id ASC
                        """
                    ),
                    {"user_id": user_id, "backdrop_item_id": current_backdrop_item_id},
                )
                items = [dict(r) for r in slot_items_res.mappings().all()]

            # 兼容旧模型（灰度期间）：新槽位数据为空时回退旧的 chatbot_layout_items
            if not items and layout_row and layout_row.get("id") is not None:
                layout_id = layout_row["id"]
                items_res = await session.execute(
                    text(
                        """
                        SELECT
                            li.id,
                            li.placement_type,
                            li.start_row,
                            li.start_col,
                            li.size_rows,
                            li.size_cols,
                            li.z_index,
                            li.image_name_snapshot,
                            li.item_id,
                            COALESCE(si.scene_anchor_type, 'none') AS scene_anchor_type,
                            'grid' AS render_mode,
                            0 AS offset_x,
                            0 AS offset_y,
                            NULL AS pixel_width,
                            NULL AS pixel_height
                        FROM chatbot_layout_items li
                        LEFT JOIN shop_items si ON si.id = li.item_id
                        WHERE li.layout_id = :layout_id
                        ORDER BY li.z_index ASC, li.id ASC
                        """
                    ),
                    {"layout_id": layout_id},
                )
                items = [dict(r) for r in items_res.mappings().all()]

            data: dict[str, Any] = {
                "grid_rows": int(layout_row.get("grid_rows")) if layout_row else 5,
                "grid_cols": int(layout_row.get("grid_cols")) if layout_row else 4,
                "cell_width": int(layout_row.get("cell_width")) if layout_row else 96,
                "cell_height": int(layout_row.get("cell_height")) if layout_row else 96,
                "decor_area_width": int(layout_row.get("decor_area_width")) if layout_row else 393,
                "decor_area_height": int(layout_row.get("decor_area_height")) if layout_row else 482,
                "backdrop": None,
                "items": items,
                "slots": slots,
            }

            if backdrop_row and backdrop_row.get("image_name"):
                data["backdrop"] = {
                    "image_name": backdrop_row["image_name"],
                    "z_index": int(backdrop_row.get("z_index") or 0),
                }

            return {"success": True, "data": data}
    except Exception as exc:  # noqa: BLE001
        logger.exception("获取用户 /chatbot 布局失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


async def _mysql_user_id_column_is_bigint(session, table_name: str) -> bool:
    res = await session.execute(
        text(
            """
            SELECT DATA_TYPE
            FROM information_schema.columns
            WHERE table_schema = :db
              AND table_name = :table
              AND column_name = 'user_id'
            """
        ),
        {"db": settings.MYSQL_DATABASE, "table": table_name},
    )
    row = res.mappings().first()
    data_type = (row or {}).get("DATA_TYPE") or ""
    return "bigint" in str(data_type).lower()


async def _chatbot_in_use_item_ids(session, user_id: str) -> set[int]:
    """当前幕布 + 槽位/布局中已装备的 shop_items.id 集合。"""
    layout_user_id_expr = (
        "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
        if await _mysql_user_id_column_is_bigint(session, "chatbot_layouts")
        else ":user_id"
    )
    backdrop_user_id_expr = (
        "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
        if await _mysql_user_id_column_is_bigint(session, "user_chatbot_backdrops")
        else ":user_id"
    )

    current_backdrop_res = await session.execute(
        text(
            f"""
            SELECT ub.backdrop_item_id
            FROM user_chatbot_backdrops ub
            WHERE ub.user_id = {backdrop_user_id_expr}
            ORDER BY ub.updated_at DESC, ub.id DESC
            LIMIT 1
            """
        ),
        {"user_id": user_id},
    )
    current_backdrop_row = current_backdrop_res.mappings().first()
    current_backdrop_item_id = (
        int(current_backdrop_row["backdrop_item_id"])
        if current_backdrop_row and current_backdrop_row.get("backdrop_item_id") is not None
        else None
    )

    in_use_item_ids: set[int] = set()
    if current_backdrop_item_id is not None:
        in_use_item_ids.add(current_backdrop_item_id)

    slot_table_res = await session.execute(
        text(
            """
            SELECT COUNT(*) AS c
            FROM information_schema.tables
            WHERE table_schema = :db
              AND table_name = 'user_chatbot_slot_placements'
            """
        ),
        {"db": settings.MYSQL_DATABASE},
    )
    has_slot_table = int((slot_table_res.mappings().first() or {}).get("c") or 0) > 0
    if has_slot_table and current_backdrop_item_id is not None:
        slot_items_res = await session.execute(
            text(
                f"""
                SELECT DISTINCT usp.item_id
                FROM user_chatbot_slot_placements usp
                WHERE usp.user_id = {layout_user_id_expr}
                  AND usp.backdrop_item_id = :backdrop_item_id
                  AND usp.item_id IS NOT NULL
                """
            ),
            {"user_id": user_id, "backdrop_item_id": current_backdrop_item_id},
        )
        for row in slot_items_res.mappings().all():
            iid = row.get("item_id")
            if iid is not None:
                in_use_item_ids.add(int(iid))
    else:
        latest_layout_res = await session.execute(
            text(
                f"""
                SELECT l.id
                FROM chatbot_layouts l
                WHERE l.user_id = {layout_user_id_expr}
                ORDER BY l.updated_at DESC, l.id DESC
                LIMIT 1
                """
            ),
            {"user_id": user_id},
        )
        latest_layout_row = latest_layout_res.mappings().first()
        if latest_layout_row and latest_layout_row.get("id") is not None:
            layout_item_res = await session.execute(
                text(
                    """
                    SELECT DISTINCT li.item_id
                    FROM chatbot_layout_items li
                    WHERE li.layout_id = :layout_id
                      AND li.item_id IS NOT NULL
                    """
                ),
                {"layout_id": int(latest_layout_row["id"])},
            )
            for row in layout_item_res.mappings().all():
                iid = row.get("item_id")
                if iid is not None:
                    in_use_item_ids.add(int(iid))

    return in_use_item_ids


@app.get("/api/users/{user_id}/chatbot-owned-items")
async def get_chatbot_owned_items(user_id: str, _: None = Depends(require_user_session)):
    """获取用户已拥有的 ChatBot 商品（家具 + 幕布），用于前端背包展示。"""
    try:
        async with db_manager.get_session() as session:
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _mysql_user_id_column_is_bigint(session, "user_assets")
                else ":user_id"
            )
            in_use_item_ids = await _chatbot_in_use_item_ids(session, user_id)

            owned_res = await session.execute(
                text(
                    f"""
                    SELECT
                        si.id,
                        si.item_code,
                        si.name,
                        si.name_zh,
                        si.name_en,
                        si.has_special_scene_trigger,
                        si.image_name,
                        si.render_image_name,
                        si.item_type,
                        si.biz_type,
                        si.sort_order
                    FROM user_assets ua
                    JOIN shop_items si
                      ON si.id = ua.item_id
                    WHERE ua.user_id = {assets_user_id_expr}
                      AND ua.owned_qty > 0
                      AND si.is_active = 1
                      AND si.item_type IN ('furniture', 'backdrop')
                    ORDER BY si.sort_order ASC, si.id ASC
                    """
                ),
                {"user_id": user_id},
            )
            rows = owned_res.mappings().all()
            return {
                "success": True,
                "data": [
                    {
                        "item_id": int(r["id"]),
                        "item_code": r.get("item_code"),
                        "name": r.get("name"),
                        "name_zh": r.get("name_zh"),
                        "name_en": r.get("name_en"),
                        "has_special_scene_trigger": int(r.get("has_special_scene_trigger") or 0),
                        "image_name": r.get("image_name"),
                        "render_image_name": r.get("render_image_name"),
                        "item_type": r.get("item_type"),
                        "biz_type": r.get("biz_type"),
                        "is_in_use": int(r["id"]) in in_use_item_ids,
                    }
                    for r in rows
                ],
            }
    except Exception as exc:  # noqa: BLE001
        logger.exception("获取用户已购商品失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/chatbot-shop-catalog")
async def get_chatbot_shop_catalog(user_id: str, _: None = Depends(require_user_session)):
    """上架中的家具 + 幕布全库（含是否已拥有、是否当前使用中），用于 Lumi 商店等场景。"""
    try:
        async with db_manager.get_session() as session:
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _mysql_user_id_column_is_bigint(session, "user_assets")
                else ":user_id"
            )
            in_use_item_ids = await _chatbot_in_use_item_ids(session, user_id)

            catalog_res = await session.execute(
                text(
                    f"""
                    SELECT
                        si.id,
                        si.item_code,
                        si.name,
                        si.name_zh,
                        si.name_en,
                        si.has_special_scene_trigger,
                        si.image_name,
                        si.render_image_name,
                        si.item_type,
                        si.biz_type,
                        si.sort_order,
                        si.price_energy,
                        si.max_per_user,
                        si.repeat_purchase_mode,
                        COALESCE(ua.owned_qty, 0) AS owned_qty,
                        CASE
                            WHEN COALESCE(ua.owned_qty, 0) > 0 THEN 1
                            ELSE 0
                        END AS is_owned
                    FROM shop_items si
                    LEFT JOIN user_assets ua
                      ON ua.item_id = si.id
                     AND ua.user_id = {assets_user_id_expr}
                    WHERE si.is_active = 1
                      AND si.item_type IN ('furniture', 'backdrop')
                    ORDER BY si.sort_order ASC, si.id ASC
                    """
                ),
                {"user_id": user_id},
            )
            rows = catalog_res.mappings().all()
            return {
                "success": True,
                "data": [
                    {
                        "item_id": int(r["id"]),
                        "item_code": r.get("item_code"),
                        "name": r.get("name"),
                        "name_zh": r.get("name_zh"),
                        "name_en": r.get("name_en"),
                        "has_special_scene_trigger": int(r.get("has_special_scene_trigger") or 0),
                        "image_name": r.get("image_name"),
                        "render_image_name": r.get("render_image_name"),
                        "item_type": r.get("item_type"),
                        "biz_type": r.get("biz_type"),
                        "is_in_use": int(r["id"]) in in_use_item_ids,
                        "is_owned": bool(int(r.get("is_owned") or 0)),
                        "price_energy": float(r.get("price_energy") or 0),
                        "max_per_user": (
                            int(r["max_per_user"])
                            if r.get("max_per_user") is not None
                            else None
                        ),
                        "owned_qty": int(r.get("owned_qty") or 0),
                        "repeat_purchase_mode": r.get("repeat_purchase_mode") or "single",
                    }
                    for r in rows
                ],
            }
    except Exception as exc:  # noqa: BLE001
        logger.exception("获取 ChatBot 商品库失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/chatbot-shop-cart")
async def get_chatbot_shop_cart(user_id: str, _: None = Depends(require_user_session)):
    """用户 ChatBot 商店购物车（家具/幕布）。"""
    try:
        async with db_manager.get_session() as session:
            user_row = await session.execute(
                text("SELECT id FROM users WHERE user_id = :user_id LIMIT 1"),
                {"user_id": user_id},
            )
            ur = user_row.mappings().first()
            if not ur:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            internal_uid = int(ur["id"])

            cart_res = await session.execute(
                text(
                    """
                    SELECT id FROM shopping_carts
                    WHERE user_id = :uid
                    ORDER BY id ASC
                    LIMIT 1
                    """
                ),
                {"uid": internal_uid, "user_id": user_id},
            )
            cart_row = cart_res.mappings().first()
            if not cart_row:
                return {"success": True, "data": {"items": [], "cart_total_quantity": 0}}

            cart_id = int(cart_row["id"])
            items_res = await session.execute(
                text(
                    """
                    SELECT item_id, quantity, unit_price_energy, unit_price_crystal
                    FROM shopping_cart_items
                    WHERE cart_id = :cid
                    ORDER BY id ASC
                    """
                ),
                {"cid": cart_id},
            )
            rows = items_res.mappings().all()
            items = [
                {
                    "item_id": int(r["item_id"]),
                    "quantity": int(r.get("quantity") or 0),
                    "unit_price_energy": float(r.get("unit_price_energy") or 0),
                    "unit_price_crystal": float(r.get("unit_price_crystal") or 0),
                }
                for r in rows
            ]
            total_q = sum(i["quantity"] for i in items)
            return {"success": True, "data": {"items": items, "cart_total_quantity": total_q}}
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("获取 ChatBot 购物车失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/chatbot-shop-cart/add")
async def add_chatbot_shop_cart_item(
    user_id: str,
    request: ChatbotShopCartAddRequest,
    _: None = Depends(require_user_session),
):
    """上架家具/幕布加入购物车；数量与已购合计不得超过 shop_items.max_per_user（若有限）。"""
    try:
        async with db_manager.get_session() as session:
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _mysql_user_id_column_is_bigint(session, "user_assets")
                else ":user_id"
            )

            user_row = await session.execute(
                text("SELECT id FROM users WHERE user_id = :user_id LIMIT 1"),
                {"user_id": user_id},
            )
            ur = user_row.mappings().first()
            if not ur:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            internal_uid = int(ur["id"])

            item_res = await session.execute(
                text(
                    """
                    SELECT
                        id,
                        price_energy,
                        price_crystal,
                        max_per_user,
                        repeat_purchase_mode,
                        is_active,
                        item_type
                    FROM shop_items
                    WHERE id = :item_id
                    LIMIT 1
                    """
                ),
                {"item_id": request.item_id},
            )
            item_row = item_res.mappings().first()
            if not item_row or int(item_row.get("is_active") or 0) != 1:
                raise HTTPException(status_code=404, detail={"code": SHOP_ITEM_NOT_FOUND})
            if str(item_row.get("item_type") or "") not in ("furniture", "backdrop"):
                raise HTTPException(status_code=404, detail={"code": SHOP_ITEM_NOT_FOUND})

            owned_res = await session.execute(
                text(
                    f"""
                    SELECT COALESCE(ua.owned_qty, 0) AS owned_qty
                    FROM shop_items si
                    LEFT JOIN user_assets ua
                      ON ua.item_id = si.id
                     AND ua.user_id = {assets_user_id_expr}
                    WHERE si.id = :item_id
                    LIMIT 1
                    """
                ),
                {"user_id": user_id, "item_id": request.item_id},
            )
            owned_qty = int((owned_res.mappings().first() or {}).get("owned_qty") or 0)

            repeat_mode = str(item_row.get("repeat_purchase_mode") or "single")
            if repeat_mode == "single" and owned_qty >= 1:
                raise HTTPException(status_code=400, detail={"code": PURCHASE_LIMIT_REACHED})

            cart_res = await session.execute(
                text(
                    """
                    SELECT id FROM shopping_carts
                    WHERE user_id = :uid
                    ORDER BY id ASC
                    LIMIT 1
                    """
                ),
                {"uid": internal_uid},
            )
            cart_row = cart_res.mappings().first()
            if not cart_row:
                await session.execute(
                    text("INSERT INTO shopping_carts (user_id) VALUES (:uid)"),
                    {"uid": internal_uid},
                )
                oid_ins = await session.execute(text("SELECT LAST_INSERT_ID() AS cid"))
                cart_id = int(oid_ins.scalar_one() or 0)
            else:
                cart_id = int(cart_row["id"])

            line_res = await session.execute(
                text(
                    """
                    SELECT id, quantity FROM shopping_cart_items
                    WHERE cart_id = :cid AND item_id = :item_id
                    LIMIT 1
                    """
                ),
                {"cid": cart_id, "item_id": request.item_id},
            )
            line = line_res.mappings().first()
            current_cart_qty = int(line.get("quantity") or 0) if line else 0

            new_cart_qty = current_cart_qty + 1
            max_per = item_row.get("max_per_user")
            max_per_int = int(max_per) if max_per is not None else None
            if max_per_int is not None and owned_qty + new_cart_qty > max_per_int:
                raise HTTPException(status_code=400, detail={"code": CART_QUANTITY_LIMIT_EXCEEDED})

            try:
                pe = Decimal(str(item_row.get("price_energy") or 0))
                pc = Decimal(str(item_row.get("price_crystal") or 0))
            except (ArithmeticError, ValueError, TypeError):
                pe = Decimal("0")
                pc = Decimal("0")

            if line:
                await session.execute(
                    text(
                        """
                        UPDATE shopping_cart_items
                        SET quantity = :q, updated_at = NOW()
                        WHERE id = :id
                        """
                    ),
                    {"q": new_cart_qty, "id": int(line["id"])},
                )
            else:
                await session.execute(
                    text(
                        """
                        INSERT INTO shopping_cart_items
                          (cart_id, item_id, quantity, unit_price_energy, unit_price_crystal)
                        VALUES
                          (:cid, :iid, 1, :pe, :pc)
                        """
                    ),
                    {"cid": cart_id, "iid": int(request.item_id), "pe": pe, "pc": pc},
                )

            sum_res = await session.execute(
                text(
                    """
                    SELECT COALESCE(SUM(quantity), 0) AS tq
                    FROM shopping_cart_items
                    WHERE cart_id = :cid
                    """
                ),
                {"cid": cart_id},
            )
            cart_total = int(sum_res.scalar_one() or 0)

            await session.commit()
            return {
                "success": True,
                "data": {
                    "item_id": int(request.item_id),
                    "item_quantity": new_cart_qty,
                    "cart_total_quantity": cart_total,
                },
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("ChatBot 购物车加购失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/chatbot-shop-cart/remove")
async def remove_chatbot_shop_cart_item(
    user_id: str,
    request: ChatbotShopCartRemoveRequest,
    _: None = Depends(require_user_session),
):
    """移除购物车中的家具/幕布商品（删除该 item 的购物车行）"""
    try:
        async with db_manager.get_session() as session:
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _mysql_user_id_column_is_bigint(session, "user_assets")
                else ":user_id"
            )

            user_row = await session.execute(
                text("SELECT id FROM users WHERE user_id = :user_id LIMIT 1"),
                {"user_id": user_id},
            )
            ur = user_row.mappings().first()
            if not ur:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            internal_uid = int(ur["id"])

            await session.execute(
                text(
                    """
                    DELETE sci
                    FROM shopping_cart_items sci
                    INNER JOIN shopping_carts sc ON sc.id = sci.cart_id
                    WHERE sc.user_id = :uid
                      AND sci.item_id = :item_id
                    """
                ),
                {"uid": internal_uid, "item_id": int(request.item_id)},
            )

            sum_res = await session.execute(
                text(
                    """
                    SELECT COALESCE(SUM(sci.quantity), 0) AS tq
                    FROM shopping_cart_items sci
                    INNER JOIN shopping_carts sc ON sc.id = sci.cart_id
                    WHERE sc.user_id = :uid
                    """
                ),
                {"uid": internal_uid},
            )
            cart_total = int(sum_res.scalar_one() or 0)

            await session.commit()
            return {
                "success": True,
                "data": {"item_id": int(request.item_id), "cart_total_quantity": cart_total},
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("ChatBot 移除购物车失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/chatbot-shop-cart/purchase")
async def purchase_chatbot_shop_cart_item(
    user_id: str,
    request: ChatbotShopCartPurchaseRequest,
    _: None = Depends(require_user_session),
):
    """购买购物车中的所有家具/幕布商品（一次性校验与扣费）"""
    try:
        async with db_manager.get_session() as session:
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _mysql_user_id_column_is_bigint(session, "user_assets")
                else ":user_id"
            )

            user_row = await session.execute(
                text("SELECT id FROM users WHERE user_id = :user_id LIMIT 1"),
                {"user_id": user_id},
            )
            ur = user_row.mappings().first()
            if not ur:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            internal_uid = int(ur["id"])

            cart_res = await session.execute(
                text(
                    """
                    SELECT id FROM shopping_carts
                    WHERE user_id = :uid
                    ORDER BY id ASC
                    LIMIT 1
                    """
                ),
                {"uid": internal_uid},
            )
            cart_row = cart_res.mappings().first()
            if not cart_row:
                return {"success": True, "data": {"total_quantity": 0, "total_cost_energy": 0}}
            cart_id = int(cart_row["id"])

            cart_lines_res = await session.execute(
                text(
                    f"""
                    SELECT
                        sci.item_id,
                        sci.quantity AS cart_qty,
                        si.price_energy,
                        si.price_crystal,
                        si.max_per_user,
                        si.repeat_purchase_mode,
                        si.is_active,
                        si.item_type,
                        si.name,
                        si.image_name,
                        COALESCE(ua.owned_qty, 0) AS owned_qty
                    FROM shopping_cart_items sci
                    INNER JOIN shopping_carts sc ON sc.id = sci.cart_id
                    INNER JOIN shop_items si ON si.id = sci.item_id
                    LEFT JOIN user_assets ua
                      ON ua.item_id = si.id
                     AND ua.user_id = {assets_user_id_expr}
                    WHERE sc.user_id = :uid
                    ORDER BY sci.id ASC
                    """
                ),
                {"uid": internal_uid, "user_id": user_id},
            )
            cart_lines = cart_lines_res.mappings().all()
            if not cart_lines:
                return {"success": True, "data": {"total_quantity": 0, "total_cost_energy": 0}}

            # 预校验：限购/重复模式 + 总价校验（在写入前完成，避免部分成功）
            total_qty = 0
            total_cost_energy = 0
            for r in cart_lines:
                item_id = int(r["item_id"])
                cart_qty = int(r.get("cart_qty") or 0)
                if cart_qty <= 0:
                    continue

                if str(r.get("is_active") or 0) != "1":
                    raise HTTPException(status_code=404, detail={"code": SHOP_ITEM_NOT_FOUND})
                if str(r.get("item_type") or "") not in ("furniture", "backdrop"):
                    raise HTTPException(status_code=404, detail={"code": SHOP_ITEM_NOT_FOUND})

                owned_qty = int(r.get("owned_qty") or 0)
                max_per = r.get("max_per_user")
                max_per_int = int(max_per) if max_per is not None else None
                if max_per_int is not None and owned_qty + cart_qty > max_per_int:
                    raise HTTPException(status_code=400, detail={"code": CART_QUANTITY_LIMIT_EXCEEDED})

                repeat_mode = str(r.get("repeat_purchase_mode") or "single")
                # repeat_purchase_mode=single：用户购买后即无法再购买
                if repeat_mode == "single" and owned_qty + cart_qty > 1:
                    raise HTTPException(status_code=400, detail={"code": PURCHASE_LIMIT_REACHED})

                unit_cost_energy = _floor_shop_price_energy(r.get("price_energy"))
                total_qty += cart_qty
                total_cost_energy += unit_cost_energy * cart_qty

            # 校验余额并扣费
            bal_res = await session.execute(
                text(
                    """
                    SELECT id, conversation_energy_stone_total
                    FROM users
                    WHERE user_id = :user_id
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            bal_row = bal_res.mappings().first()
            if not bal_row:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            balance = int(bal_row.get("conversation_energy_stone_total") or 0)
            if total_cost_energy > balance:
                raise HTTPException(status_code=400, detail={"code": INSUFFICIENT_CONVERSATION_ENERGY})

            upd_res = await session.execute(
                text(
                    """
                    UPDATE users
                    SET conversation_energy_stone_total = conversation_energy_stone_total - :cost
                    WHERE user_id = :user_id
                      AND conversation_energy_stone_total >= :cost
                    """
                ),
                {"cost": total_cost_energy, "user_id": user_id},
            )
            if (getattr(upd_res, "rowcount", None) or 0) < 1:
                raise HTTPException(status_code=400, detail={"code": INSUFFICIENT_CONVERSATION_ENERGY})

            # 写入订单 + 更新拥有量 + 清空购物车（仍在同一事务里）
            await session.execute(
                text(
                    """
                    INSERT INTO purchase_orders
                      (user_id, cart_id, status, total_quantity, total_cost_energy, total_cost_crystal, completed_at)
                    VALUES
                      (:uid, :cart_id, 'completed', :total_quantity, :tce, 0.00, NOW())
                    """
                ),
                {
                    "uid": internal_uid,
                    "cart_id": cart_id,
                    "total_quantity": total_qty,
                    "tce": Decimal(str(total_cost_energy)),
                },
            )
            oid_row = await session.execute(text("SELECT LAST_INSERT_ID() AS oid"))
            order_id_raw = oid_row.scalar_one()
            order_id = int(order_id_raw) if order_id_raw is not None else 0
            if order_id <= 0:
                raise RuntimeError("purchase_orders insert did not return id")

            # 更新用户资产：逐条处理，避免不同 item 存在差异导致 SQL 复杂化
            for r in cart_lines:
                item_id = int(r["item_id"])
                cart_qty = int(r.get("cart_qty") or 0)
                if cart_qty <= 0:
                    continue

                owned_qty = int(r.get("owned_qty") or 0)
                max_per = r.get("max_per_user")
                max_per_int = int(max_per) if max_per is not None else None
                # owned_qty 已在预校验阶段确认；这里不再重复校验，保证业务逻辑一致

                unit_cost_energy = _floor_shop_price_energy(r.get("price_energy"))
                unit_pe = Decimal(str(unit_cost_energy))
                try:
                    unit_pc = Decimal(str(r.get("price_crystal")) if r.get("price_crystal") is not None else "0")
                except (ArithmeticError, ValueError, TypeError):
                    unit_pc = Decimal("0")

                chk = await session.execute(
                    text(
                        f"""
                        SELECT id FROM user_assets
                        WHERE user_id = {assets_user_id_expr} AND item_id = :item_id
                        LIMIT 1
                        """
                    ),
                    {"user_id": user_id, "item_id": item_id},
                )
                existing_uaid = chk.scalar_one_or_none()
                if existing_uaid is not None:
                    await session.execute(
                        text(
                            """
                            UPDATE user_assets
                            SET owned_qty = owned_qty + :q,
                                last_acquired_at = NOW(),
                                updated_at = NOW()
                            WHERE id = :id
                            """
                        ),
                        {"q": cart_qty, "id": int(existing_uaid)},
                    )
                else:
                    await session.execute(
                        text(
                            f"""
                            INSERT INTO user_assets
                              (user_id, item_id, owned_qty, used_qty, last_acquired_at, created_at, updated_at)
                            VALUES
                              ({assets_user_id_expr}, :item_id, :q, 0, NOW(), NOW(), NOW())
                            """
                        ),
                        {"user_id": user_id, "item_id": item_id, "q": cart_qty},
                    )

                item_name_snap = str(r.get("name") or "")[:100]
                image_name_snap = str(r.get("image_name") or "")[:255]

                await session.execute(
                    text(
                        """
                        INSERT INTO purchase_order_items
                          (order_id, item_id, quantity, unit_price_energy, unit_price_crystal,
                           cost_energy, cost_crystal, item_name_snapshot, image_name_snapshot)
                        VALUES
                          (:order_id, :item_id, :qty, :upe, :upc, :ce, 0.00, :name_snap, :img_snap)
                        """
                    ),
                    {
                        "order_id": order_id,
                        "item_id": item_id,
                        "qty": cart_qty,
                        "upe": unit_pe,
                        "upc": unit_pc,
                        "ce": unit_pe * cart_qty,
                        "name_snap": item_name_snap,
                        "img_snap": image_name_snap,
                    },
                )

            # 清空购物车（该 cart_id 下的所有行）
            await session.execute(
                text(
                    """
                    DELETE sci
                    FROM shopping_cart_items sci
                    INNER JOIN shopping_carts sc ON sc.id = sci.cart_id
                    WHERE sc.user_id = :uid
                    """
                ),
                {"uid": internal_uid},
            )

            await session.commit()
            return {
                "success": True,
                "data": {"total_quantity": total_qty, "total_cost_energy": total_cost_energy},
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("ChatBot 购物车购买失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/chatbot-shop-purchase")
async def purchase_chatbot_shop_item(
    user_id: str,
    request: ChatbotShopPurchaseRequest,
    _: None = Depends(require_user_session),
):
    """使用对话能量石购买上架中的家具/幕布。"""
    try:
        async with db_manager.get_session() as session:
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _mysql_user_id_column_is_bigint(session, "user_assets")
                else ":user_id"
            )

            item_res = await session.execute(
                text(
                    """
                    SELECT
                        id,
                        price_energy,
                        price_crystal,
                        name,
                        image_name,
                        max_per_user,
                        repeat_purchase_mode,
                        is_active,
                        item_type
                    FROM shop_items
                    WHERE id = :item_id
                    LIMIT 1
                    """
                ),
                {"item_id": request.item_id},
            )
            item_row = item_res.mappings().first()
            if not item_row or int(item_row.get("is_active") or 0) != 1:
                raise HTTPException(status_code=404, detail={"code": SHOP_ITEM_NOT_FOUND})
            if str(item_row.get("item_type") or "") not in ("furniture", "backdrop"):
                raise HTTPException(status_code=404, detail={"code": SHOP_ITEM_NOT_FOUND})

            owned_res = await session.execute(
                text(
                    f"""
                    SELECT COALESCE(ua.owned_qty, 0) AS owned_qty
                    FROM shop_items si
                    LEFT JOIN user_assets ua
                      ON ua.item_id = si.id
                     AND ua.user_id = {assets_user_id_expr}
                    WHERE si.id = :item_id
                    LIMIT 1
                    """
                ),
                {"user_id": user_id, "item_id": request.item_id},
            )
            owned_row = owned_res.mappings().first()
            owned_qty = int((owned_row or {}).get("owned_qty") or 0)

            max_per = item_row.get("max_per_user")
            max_per_int = int(max_per) if max_per is not None else None
            if max_per_int is not None and owned_qty >= max_per_int:
                raise HTTPException(status_code=400, detail={"code": PURCHASE_LIMIT_REACHED})

            repeat_mode = str(item_row.get("repeat_purchase_mode") or "single")
            if repeat_mode == "single" and owned_qty >= 1:
                raise HTTPException(status_code=400, detail={"code": PURCHASE_LIMIT_REACHED})

            cost = _floor_shop_price_energy(item_row.get("price_energy"))

            bal_res = await session.execute(
                text(
                    """
                    SELECT id, conversation_energy_stone_total
                    FROM users
                    WHERE user_id = :user_id
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            bal_row = bal_res.mappings().first()
            if not bal_row:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            balance = int(bal_row.get("conversation_energy_stone_total") or 0)
            if cost > balance:
                raise HTTPException(
                    status_code=400,
                    detail={"code": INSUFFICIENT_CONVERSATION_ENERGY},
                )

            upd_res = await session.execute(
                text(
                    """
                    UPDATE users
                    SET conversation_energy_stone_total = conversation_energy_stone_total - :cost
                    WHERE user_id = :user_id
                      AND conversation_energy_stone_total >= :cost
                    """
                ),
                {"cost": cost, "user_id": user_id},
            )
            if (getattr(upd_res, "rowcount", None) or 0) < 1:
                raise HTTPException(
                    status_code=400,
                    detail={"code": INSUFFICIENT_CONVERSATION_ENERGY},
                )

            chk = await session.execute(
                text(
                    f"""
                    SELECT id FROM user_assets
                    WHERE user_id = {assets_user_id_expr} AND item_id = :item_id
                    LIMIT 1
                    """
                ),
                {"user_id": user_id, "item_id": request.item_id},
            )
            existing_uaid = chk.scalar_one_or_none()
            if existing_uaid is not None:
                await session.execute(
                    text(
                        """
                        UPDATE user_assets
                        SET owned_qty = owned_qty + 1,
                            last_acquired_at = NOW(),
                            updated_at = NOW()
                        WHERE id = :id
                        """
                    ),
                    {"id": int(existing_uaid)},
                )
            else:
                await session.execute(
                    text(
                        f"""
                        INSERT INTO user_assets
                          (user_id, item_id, owned_qty, used_qty, last_acquired_at, created_at, updated_at)
                        VALUES
                          ({assets_user_id_expr}, :item_id, 1, 0, NOW(), NOW(), NOW())
                        """
                    ),
                    {"user_id": user_id, "item_id": request.item_id},
                )

            internal_uid = int(bal_row["id"])
            # 直购成功：购物车中若有同商品则减 1 件，避免与已购数量叠加超过上限
            await session.execute(
                text(
                    """
                    UPDATE shopping_cart_items sci
                    INNER JOIN shopping_carts sc ON sc.id = sci.cart_id
                    SET sci.quantity = sci.quantity - 1,
                        sci.updated_at = NOW()
                    WHERE sc.user_id = :uid
                      AND sci.item_id = :item_id
                      AND sci.quantity >= 1
                    """
                ),
                {"uid": internal_uid, "item_id": int(request.item_id)},
            )
            await session.execute(
                text(
                    """
                    DELETE sci FROM shopping_cart_items sci
                    INNER JOIN shopping_carts sc ON sc.id = sci.cart_id
                    WHERE sc.user_id = :uid
                      AND sci.item_id = :item_id
                      AND sci.quantity = 0
                    """
                ),
                {"uid": internal_uid, "item_id": int(request.item_id)},
            )

            # 购买订单主表/明细（弹窗直购无购物车，cart_id 用 0 占位）
            item_name_snap = str(item_row.get("name") or "")[:100]
            image_name_snap = str(item_row.get("image_name") or "")[:255]
            try:
                pc_snap = item_row.get("price_crystal")
                unit_pc = Decimal(str(pc_snap)) if pc_snap is not None else Decimal("0")
            except (ArithmeticError, ValueError, TypeError):
                unit_pc = Decimal("0")
            unit_pe = Decimal(str(cost))

            await session.execute(
                text(
                    """
                    INSERT INTO purchase_orders
                      (user_id, cart_id, status, total_quantity, total_cost_energy, total_cost_crystal, completed_at)
                    VALUES
                      (:uid, 0, 'completed', 1, :tce, 0.00, NOW())
                    """
                ),
                {"uid": internal_uid, "tce": unit_pe},
            )
            oid_row = await session.execute(text("SELECT LAST_INSERT_ID() AS oid"))
            order_id_raw = oid_row.scalar_one()
            order_id = int(order_id_raw) if order_id_raw is not None else 0
            if order_id <= 0:
                raise RuntimeError("purchase_orders insert did not return id")

            await session.execute(
                text(
                    """
                    INSERT INTO purchase_order_items
                      (order_id, item_id, quantity, unit_price_energy, unit_price_crystal,
                       cost_energy, cost_crystal, item_name_snapshot, image_name_snapshot)
                    VALUES
                      (:order_id, :item_id, 1, :upe, :upc, :ce, 0.00, :name_snap, :img_snap)
                    """
                ),
                {
                    "order_id": order_id,
                    "item_id": int(request.item_id),
                    "upe": unit_pe,
                    "upc": unit_pc,
                    "ce": unit_pe,
                    "name_snap": item_name_snap,
                    "img_snap": image_name_snap,
                },
            )

            await session.commit()
            return {
                "success": True,
                "data": {
                    "item_id": int(request.item_id),
                    "cost_energy": cost,
                    "order_id": order_id,
                },
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("ChatBot 商店购买失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.put("/api/users/{user_id}/chatbot-backdrop")
async def update_chatbot_backdrop(
    user_id: str,
    request: UpdateChatbotBackdropRequest,
    _: None = Depends(require_user_session),
):
    """切换用户 ChatBot 当前幕布。"""
    try:
        async with db_manager.get_session() as session:
            async def _get_user_id_is_bigint(table_name: str) -> bool:
                res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = :table
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE, "table": table_name},
                )
                row = res.mappings().first()
                return "bigint" in str((row or {}).get("DATA_TYPE") or "").lower()

            backdrop_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_backdrops")
                else ":user_id"
            )
            placement_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_slot_placements")
                else ":user_id"
            )
            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_assets")
                else ":user_id"
            )

            owned_res = await session.execute(
                text(
                    f"""
                    SELECT
                        ua.id AS user_asset_id,
                        si.id AS item_id,
                        si.item_type,
                        COALESCE(si.default_z_index, 0) AS default_z_index
                    FROM user_assets ua
                    JOIN shop_items si ON si.id = ua.item_id
                    WHERE ua.user_id = {assets_user_id_expr}
                      AND ua.item_id = :item_id
                      AND ua.owned_qty > 0
                      AND si.is_active = 1
                    LIMIT 1
                    """
                ),
                {"user_id": user_id, "item_id": request.backdrop_item_id},
            )
            owned_row = owned_res.mappings().first()
            if not owned_row:
                raise HTTPException(status_code=400, detail={"code": "ITEM_NOT_OWNED"})
            if str(owned_row.get("item_type") or "") != "backdrop":
                raise HTTPException(status_code=400, detail={"code": "ITEM_NOT_BACKDROP"})

            target_backdrop_item_id = int(request.backdrop_item_id)
            current_backdrop_res = await session.execute(
                text(
                    f"""
                    SELECT backdrop_item_id
                    FROM user_chatbot_backdrops
                    WHERE user_id = {backdrop_user_id_expr}
                    ORDER BY updated_at DESC, id DESC
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            current_backdrop_row = current_backdrop_res.mappings().first()
            source_backdrop_item_id = (
                int(current_backdrop_row["backdrop_item_id"])
                if current_backdrop_row and current_backdrop_row.get("backdrop_item_id") is not None
                else None
            )

            await session.execute(
                text(
                    f"""
                    DELETE FROM user_chatbot_backdrops
                    WHERE user_id = {backdrop_user_id_expr}
                    """
                ),
                {"user_id": user_id},
            )
            await session.execute(
                text(
                    f"""
                    INSERT INTO user_chatbot_backdrops
                      (user_id, backdrop_item_id, z_index, created_at, updated_at)
                    VALUES
                      ({backdrop_user_id_expr}, :backdrop_item_id, :z_index, NOW(), NOW())
                    """
                ),
                {
                    "user_id": user_id,
                    "backdrop_item_id": target_backdrop_item_id,
                    "z_index": int(owned_row.get("default_z_index") or 0),
                },
            )

            slot_table_exists_res = await session.execute(
                text(
                    """
                    SELECT COUNT(*) AS c
                    FROM information_schema.tables
                    WHERE table_schema = :db
                      AND table_name = 'user_chatbot_slot_placements'
                    """
                ),
                {"db": settings.MYSQL_DATABASE},
            )
            slot_table_exists = int((slot_table_exists_res.mappings().first() or {}).get("c") or 0) > 0

            # 幕布切换时：先清空目标幕布已有摆放，再尝试按槽位复制来源幕布的有效摆放。
            if slot_table_exists:
                await session.execute(
                    text(
                        f"""
                        DELETE FROM user_chatbot_slot_placements
                        WHERE user_id = {placement_user_id_expr}
                          AND backdrop_item_id = :target_backdrop_item_id
                        """
                    ),
                    {"user_id": user_id, "target_backdrop_item_id": target_backdrop_item_id},
                )

                if source_backdrop_item_id is not None and source_backdrop_item_id != target_backdrop_item_id:
                    await session.execute(
                        text(
                            f"""
                            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)
                            SELECT
                              src.user_id,
                              :target_backdrop_item_id AS backdrop_item_id,
                              src.slot_code,
                              src.user_asset_id,
                              src.item_id,
                              src.z_index,
                              src.image_name_snapshot,
                              NOW(),
                              NOW()
                            FROM user_chatbot_slot_placements src
                            JOIN chatbot_backdrop_slots target_slot
                              ON target_slot.backdrop_item_id = :target_backdrop_item_id
                             AND target_slot.slot_code = src.slot_code
                             AND target_slot.is_active = 1
                            WHERE src.user_id = {placement_user_id_expr}
                              AND src.backdrop_item_id = :source_backdrop_item_id
                            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),
                              placed_at = VALUES(placed_at),
                              updated_at = NOW()
                            """
                        ),
                        {
                            "user_id": user_id,
                            "source_backdrop_item_id": source_backdrop_item_id,
                            "target_backdrop_item_id": target_backdrop_item_id,
                        },
                    )

            await session.commit()
            return {
                "success": True,
                "data": {"backdrop_item_id": target_backdrop_item_id},
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("切换 ChatBot 幕布失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/chatbot-furniture/equip-preview")
async def get_chatbot_furniture_equip_preview(
    user_id: str,
    item_id: int = Query(..., ge=1),
    _: None = Depends(require_user_session),
):
    """
    装备前预览：解析与 POST equip 相同的槽位规则，返回槽位几何与场景图文件名。
    无有效槽位或数据不全时 data 为 null（前端不展示预览即可）。
    """
    try:
        async with db_manager.get_session() as session:
            async def _get_user_id_is_bigint(table_name: str) -> bool:
                res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = :table
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE, "table": table_name},
                )
                row = res.mappings().first()
                return "bigint" in str((row or {}).get("DATA_TYPE") or "").lower()

            backdrop_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_backdrops")
                else ":user_id"
            )

            item_res = await session.execute(
                text(
                    """
                    SELECT
                        si.id AS item_id,
                        si.item_type,
                        si.image_name,
                        si.render_image_name,
                        COALESCE(si.default_z_index, 0) AS default_z_index
                    FROM shop_items si
                    WHERE si.id = :item_id
                      AND si.is_active = 1
                    LIMIT 1
                    """
                ),
                {"item_id": item_id},
            )
            item_row = item_res.mappings().first()
            if not item_row or str(item_row.get("item_type") or "") != "furniture":
                return {"success": True, "data": None}

            rules_table_res = await session.execute(
                text(
                    """
                    SELECT COUNT(*) AS c
                    FROM information_schema.tables
                    WHERE table_schema = :db
                      AND table_name = 'shop_item_slot_rules'
                    """
                ),
                {"db": settings.MYSQL_DATABASE},
            )
            if int((rules_table_res.mappings().first() or {}).get("c") or 0) == 0:
                return {"success": True, "data": None}

            current_backdrop_res = await session.execute(
                text(
                    f"""
                    SELECT backdrop_item_id
                    FROM user_chatbot_backdrops
                    WHERE user_id = {backdrop_user_id_expr}
                    ORDER BY updated_at DESC, id DESC
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            current_backdrop_row = current_backdrop_res.mappings().first()
            if not current_backdrop_row or current_backdrop_row.get("backdrop_item_id") is None:
                return {"success": True, "data": None}
            backdrop_item_id = int(current_backdrop_row["backdrop_item_id"])

            target_slot_res = await session.execute(
                text(
                    """
                    SELECT
                        r.slot_code,
                        r.id AS rule_id
                    FROM shop_item_slot_rules r
                    JOIN chatbot_backdrop_slots s
                      ON s.backdrop_item_id = r.backdrop_item_id
                     AND s.slot_code = r.slot_code
                     AND s.is_active = 1
                    WHERE r.item_id = :item_id
                      AND r.backdrop_item_id = :backdrop_item_id
                      AND r.is_active = 1
                    ORDER BY COALESCE(r.is_default, 0) DESC, COALESCE(r.priority, 9999) ASC, r.id ASC
                    LIMIT 1
                    """
                ),
                {"item_id": item_id, "backdrop_item_id": backdrop_item_id},
            )
            target_slot_row = target_slot_res.mappings().first()
            if not target_slot_row or not target_slot_row.get("slot_code"):
                return {"success": True, "data": None}
            slot_code = str(target_slot_row["slot_code"])

            slot_geo_res = await session.execute(
                text(
                    """
                    SELECT
                        s.id,
                        s.slot_code,
                        s.placement_type,
                        s.start_row,
                        s.start_col,
                        s.size_rows,
                        s.size_cols,
                        COALESCE(s.z_index, 0) AS z_index,
                        s.render_mode,
                        s.offset_x,
                        s.offset_y,
                        s.pixel_width,
                        s.pixel_height
                    FROM chatbot_backdrop_slots s
                    WHERE s.backdrop_item_id = :backdrop_item_id
                      AND s.slot_code = :slot_code
                      AND s.is_active = 1
                    LIMIT 1
                    """
                ),
                {"backdrop_item_id": backdrop_item_id, "slot_code": slot_code},
            )
            slot_row = slot_geo_res.mappings().first()
            if not slot_row:
                return {"success": True, "data": None}

            preview_image_name = scene_image_name_from_shop_row(dict(item_row))
            item_z_index = int(item_row.get("default_z_index") or 0)
            slot_payload = {
                "id": int(slot_row["id"]),
                "slot_code": str(slot_row.get("slot_code") or ""),
                "placement_type": str(slot_row.get("placement_type") or "floor"),
                "start_row": int(slot_row.get("start_row") or 1),
                "start_col": int(slot_row.get("start_col") or 1),
                "size_rows": int(slot_row.get("size_rows") or 1),
                "size_cols": int(slot_row.get("size_cols") or 1),
                # 与 POST /chatbot-furniture/equip 保持一致：渲染层级取商品 default_z_index
                "z_index": item_z_index,
                "render_mode": slot_row.get("render_mode"),
                "offset_x": slot_row.get("offset_x"),
                "offset_y": slot_row.get("offset_y"),
                "pixel_width": slot_row.get("pixel_width"),
                "pixel_height": slot_row.get("pixel_height"),
            }
            return {
                "success": True,
                "data": {
                    "slot": slot_payload,
                    "preview_image_name": preview_image_name,
                },
            }
    except Exception as exc:  # noqa: BLE001
        logger.exception("获取家具装备预览失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/chatbot-furniture/equip")
async def equip_chatbot_furniture(
    user_id: str,
    request: ChatbotFurnitureActionRequest,
    _: None = Depends(require_user_session),
):
    """装备家具：同槽位已有元素则替换，否则新增。"""
    try:
        async with db_manager.get_session() as session:
            async def _get_user_id_is_bigint(table_name: str) -> bool:
                res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = :table
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE, "table": table_name},
                )
                row = res.mappings().first()
                return "bigint" in str((row or {}).get("DATA_TYPE") or "").lower()

            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_assets")
                else ":user_id"
            )
            backdrop_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_backdrops")
                else ":user_id"
            )
            placement_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_slot_placements")
                else ":user_id"
            )

            owned_res = await session.execute(
                text(
                    f"""
                    SELECT
                        ua.id AS user_asset_id,
                        si.id AS item_id,
                        si.item_type,
                        si.image_name,
                        si.render_image_name,
                        COALESCE(si.default_z_index, 0) AS default_z_index
                    FROM user_assets ua
                    JOIN shop_items si ON si.id = ua.item_id
                    WHERE ua.user_id = {assets_user_id_expr}
                      AND ua.item_id = :item_id
                      AND ua.owned_qty > 0
                      AND si.is_active = 1
                    LIMIT 1
                    """
                ),
                {"user_id": user_id, "item_id": request.item_id},
            )
            owned_row = owned_res.mappings().first()
            if not owned_row:
                raise HTTPException(status_code=400, detail={"code": "ITEM_NOT_OWNED"})
            if str(owned_row.get("item_type") or "") != "furniture":
                raise HTTPException(status_code=400, detail={"code": "ITEM_NOT_FURNITURE"})

            current_backdrop_res = await session.execute(
                text(
                    f"""
                    SELECT backdrop_item_id
                    FROM user_chatbot_backdrops
                    WHERE user_id = {backdrop_user_id_expr}
                    ORDER BY updated_at DESC, id DESC
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            current_backdrop_row = current_backdrop_res.mappings().first()
            if not current_backdrop_row or current_backdrop_row.get("backdrop_item_id") is None:
                raise HTTPException(status_code=400, detail={"code": "BACKDROP_NOT_SET"})
            backdrop_item_id = int(current_backdrop_row["backdrop_item_id"])

            target_slot_res = await session.execute(
                text(
                    """
                    SELECT
                        r.slot_code,
                        r.id AS rule_id
                    FROM shop_item_slot_rules r
                    JOIN chatbot_backdrop_slots s
                      ON s.backdrop_item_id = r.backdrop_item_id
                     AND s.slot_code = r.slot_code
                     AND s.is_active = 1
                    WHERE r.item_id = :item_id
                      AND r.backdrop_item_id = :backdrop_item_id
                      AND r.is_active = 1
                    ORDER BY COALESCE(r.is_default, 0) DESC, COALESCE(r.priority, 9999) ASC, r.id ASC
                    LIMIT 1
                    """
                ),
                {"item_id": request.item_id, "backdrop_item_id": backdrop_item_id},
            )
            target_slot_row = target_slot_res.mappings().first()
            if not target_slot_row or not target_slot_row.get("slot_code"):
                raise HTTPException(status_code=400, detail={"code": "NO_VALID_SLOT_FOR_ITEM"})
            slot_code = str(target_slot_row["slot_code"])

            dependency_table_res = await session.execute(
                text(
                    """
                    SELECT COUNT(*) AS c
                    FROM information_schema.tables
                    WHERE table_schema = :db
                      AND table_name = 'chatbot_slot_dependencies'
                    """
                ),
                {"db": settings.MYSQL_DATABASE},
            )
            has_dependency_table = int((dependency_table_res.mappings().first() or {}).get("c") or 0) > 0

            if has_dependency_table:
                user_lang_res = await session.execute(
                    select(User.preferred_language).where(User.user_id == user_id).limit(1)
                )
                preferred_language = user_lang_res.scalar_one_or_none()
                is_zh = str(preferred_language or "zh").lower().startswith("zh")
                dependency_res = await session.execute(
                    text(
                        """
                        SELECT
                            dep.requires_slot_code,
                            dep.requires_biz_type,
                            cur.slot_name_zh AS current_slot_name_zh,
                            cur.slot_name_en AS current_slot_name_en,
                            req.slot_name_zh AS requires_slot_name_zh,
                            req.slot_name_en AS requires_slot_name_en
                        FROM chatbot_slot_dependencies
                        dep
                        LEFT JOIN chatbot_backdrop_slots cur
                          ON cur.backdrop_item_id = dep.backdrop_item_id
                         AND cur.slot_code = dep.slot_code
                        LEFT JOIN chatbot_backdrop_slots req
                          ON req.backdrop_item_id = dep.backdrop_item_id
                         AND req.slot_code = dep.requires_slot_code
                        WHERE dep.backdrop_item_id = :backdrop_item_id
                          AND dep.slot_code = :slot_code
                          AND dep.is_active = 1
                        ORDER BY COALESCE(dep.priority, 9999) ASC, dep.id ASC
                        """
                    ),
                    {"backdrop_item_id": backdrop_item_id, "slot_code": slot_code},
                )
                dependency_rows = dependency_res.mappings().all()
                for dep in dependency_rows:
                    requires_slot_code = str(dep.get("requires_slot_code") or "").strip()
                    requires_biz_type = dep.get("requires_biz_type")
                    if not requires_slot_code:
                        continue

                    required_placement_res = await session.execute(
                        text(
                            f"""
                            SELECT
                                usp.item_id,
                                si.biz_type
                            FROM user_chatbot_slot_placements usp
                            LEFT JOIN shop_items si ON si.id = usp.item_id
                            WHERE usp.user_id = {placement_user_id_expr}
                              AND usp.backdrop_item_id = :backdrop_item_id
                              AND usp.slot_code = :slot_code
                            LIMIT 1
                            """
                        ),
                        {
                            "user_id": user_id,
                            "backdrop_item_id": backdrop_item_id,
                            "slot_code": requires_slot_code,
                        },
                    )
                    required_placement = required_placement_res.mappings().first()
                    if not required_placement:
                        current_slot_name = (
                            str(dep.get("current_slot_name_zh") or slot_code)
                            if is_zh
                            else str(dep.get("current_slot_name_en") or slot_code)
                        )
                        requires_slot_name = (
                            str(dep.get("requires_slot_name_zh") or requires_slot_code)
                            if is_zh
                            else str(dep.get("requires_slot_name_en") or requires_slot_code)
                        )
                        detail_message = (
                            f"请先在「{requires_slot_name}」放置支撑家具后，再放置「{current_slot_name}」。"
                            if is_zh
                            else f"Please place a support item in \"{requires_slot_name}\" before placing items in \"{current_slot_name}\"."
                        )
                        raise HTTPException(
                            status_code=400,
                            detail={"code": "SLOT_DEPENDENCY_NOT_SATISFIED", "message": detail_message},
                        )

                    if requires_biz_type:
                        required_biz_type = str(required_placement.get("biz_type") or "")
                        if required_biz_type != str(requires_biz_type):
                            current_slot_name = (
                                str(dep.get("current_slot_name_zh") or slot_code)
                                if is_zh
                                else str(dep.get("current_slot_name_en") or slot_code)
                            )
                            requires_slot_name = (
                                str(dep.get("requires_slot_name_zh") or requires_slot_code)
                                if is_zh
                                else str(dep.get("requires_slot_name_en") or requires_slot_code)
                            )
                            required_type_name = (
                                "桌类家具"
                                if str(requires_biz_type) == "table_cabinet" and is_zh
                                else "table item"
                                if str(requires_biz_type) == "table_cabinet"
                                else str(requires_biz_type)
                            )
                            detail_message = (
                                f"请先在「{requires_slot_name}」放置{required_type_name}后，再放置「{current_slot_name}」。"
                                if is_zh
                                else f"Please place a {required_type_name} in \"{requires_slot_name}\" before placing items in \"{current_slot_name}\"."
                            )
                            raise HTTPException(
                                status_code=400,
                                detail={"code": "SLOT_DEPENDENCY_NOT_SATISFIED", "message": detail_message},
                            )

            await session.execute(
                text(
                    f"""
                    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
                      ({placement_user_id_expr}, :backdrop_item_id, :slot_code, :user_asset_id, :item_id, :z_index, :image_name_snapshot, 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()
                    """
                ),
                {
                    "user_id": user_id,
                    "backdrop_item_id": backdrop_item_id,
                    "slot_code": slot_code,
                    "user_asset_id": int(owned_row.get("user_asset_id") or 0),
                    "item_id": int(request.item_id),
                    "z_index": int(owned_row.get("default_z_index") or 0),
                    "image_name_snapshot": scene_image_name_from_shop_row(owned_row),
                },
            )

            await session.commit()
            return {
                "success": True,
                "data": {
                    "item_id": int(request.item_id),
                    "slot_code": slot_code,
                    "backdrop_item_id": backdrop_item_id,
                },
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("装备 ChatBot 家具失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/chatbot-furniture/unequip")
async def unequip_chatbot_furniture(
    user_id: str,
    request: ChatbotFurnitureActionRequest,
    _: None = Depends(require_user_session),
):
    """卸下家具：移除当前幕布下该 item 的槽位摆放记录。"""
    try:
        async with db_manager.get_session() as session:
            async def _get_user_id_is_bigint(table_name: str) -> bool:
                res = await session.execute(
                    text(
                        """
                        SELECT DATA_TYPE
                        FROM information_schema.columns
                        WHERE table_schema = :db
                          AND table_name = :table
                          AND column_name = 'user_id'
                        """
                    ),
                    {"db": settings.MYSQL_DATABASE, "table": table_name},
                )
                row = res.mappings().first()
                return "bigint" in str((row or {}).get("DATA_TYPE") or "").lower()

            assets_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_assets")
                else ":user_id"
            )
            backdrop_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_backdrops")
                else ":user_id"
            )
            placement_user_id_expr = (
                "(SELECT id FROM users WHERE user_id = :user_id LIMIT 1)"
                if await _get_user_id_is_bigint("user_chatbot_slot_placements")
                else ":user_id"
            )

            owned_res = await session.execute(
                text(
                    f"""
                    SELECT
                        si.id AS item_id,
                        si.item_type
                    FROM user_assets ua
                    JOIN shop_items si ON si.id = ua.item_id
                    WHERE ua.user_id = {assets_user_id_expr}
                      AND ua.item_id = :item_id
                      AND ua.owned_qty > 0
                      AND si.is_active = 1
                    LIMIT 1
                    """
                ),
                {"user_id": user_id, "item_id": request.item_id},
            )
            owned_row = owned_res.mappings().first()
            if not owned_row:
                raise HTTPException(status_code=400, detail={"code": "ITEM_NOT_OWNED"})
            if str(owned_row.get("item_type") or "") != "furniture":
                raise HTTPException(status_code=400, detail={"code": "ITEM_NOT_FURNITURE"})

            current_backdrop_res = await session.execute(
                text(
                    f"""
                    SELECT backdrop_item_id
                    FROM user_chatbot_backdrops
                    WHERE user_id = {backdrop_user_id_expr}
                    ORDER BY updated_at DESC, id DESC
                    LIMIT 1
                    """
                ),
                {"user_id": user_id},
            )
            current_backdrop_row = current_backdrop_res.mappings().first()
            if not current_backdrop_row or current_backdrop_row.get("backdrop_item_id") is None:
                raise HTTPException(status_code=400, detail={"code": "BACKDROP_NOT_SET"})
            backdrop_item_id = int(current_backdrop_row["backdrop_item_id"])

            dependency_table_res = await session.execute(
                text(
                    """
                    SELECT COUNT(*) AS c
                    FROM information_schema.tables
                    WHERE table_schema = :db
                      AND table_name = 'chatbot_slot_dependencies'
                    """
                ),
                {"db": settings.MYSQL_DATABASE},
            )
            has_dependency_table = int((dependency_table_res.mappings().first() or {}).get("c") or 0) > 0

            user_lang_res = await session.execute(
                select(User.preferred_language).where(User.user_id == user_id).limit(1)
            )
            preferred_language = user_lang_res.scalar_one_or_none()
            is_zh = str(preferred_language or "zh").lower().startswith("zh")

            cascade_removed_count = 0
            cascade_removed_item_ids: list[int] = []
            cascade_notice: Optional[str] = None
            if has_dependency_table:
                cascade_notice_meta_res = await session.execute(
                    text(
                        f"""
                        SELECT
                          dep.slot_code AS child_slot_code,
                          dep.requires_slot_code AS base_slot_code,
                          child_slot.slot_name_zh AS child_slot_name_zh,
                          child_slot.slot_name_en AS child_slot_name_en,
                          base_slot.slot_name_zh AS base_slot_name_zh,
                          base_slot.slot_name_en AS base_slot_name_en
                        FROM user_chatbot_slot_placements child
                        JOIN chatbot_slot_dependencies dep
                          ON dep.backdrop_item_id = child.backdrop_item_id
                         AND dep.slot_code = child.slot_code
                         AND dep.is_active = 1
                        JOIN user_chatbot_slot_placements base
                          ON base.user_id = child.user_id
                         AND base.backdrop_item_id = child.backdrop_item_id
                         AND base.slot_code = dep.requires_slot_code
                        LEFT JOIN chatbot_backdrop_slots child_slot
                          ON child_slot.backdrop_item_id = dep.backdrop_item_id
                         AND child_slot.slot_code = dep.slot_code
                        LEFT JOIN chatbot_backdrop_slots base_slot
                          ON base_slot.backdrop_item_id = dep.backdrop_item_id
                         AND base_slot.slot_code = dep.requires_slot_code
                        WHERE child.user_id = {placement_user_id_expr}
                          AND child.backdrop_item_id = :backdrop_item_id
                          AND base.item_id = :item_id
                          AND child.item_id <> :item_id
                        LIMIT 1
                        """
                    ),
                    {
                        "user_id": user_id,
                        "backdrop_item_id": backdrop_item_id,
                        "item_id": int(request.item_id),
                    },
                )
                cascade_notice_meta = cascade_notice_meta_res.mappings().first()

                cascade_item_res = await session.execute(
                    text(
                        f"""
                        SELECT DISTINCT child.item_id
                        FROM user_chatbot_slot_placements child
                        JOIN chatbot_slot_dependencies dep
                          ON dep.backdrop_item_id = child.backdrop_item_id
                         AND dep.slot_code = child.slot_code
                         AND dep.is_active = 1
                        JOIN user_chatbot_slot_placements base
                          ON base.user_id = child.user_id
                         AND base.backdrop_item_id = child.backdrop_item_id
                         AND base.slot_code = dep.requires_slot_code
                        WHERE child.user_id = {placement_user_id_expr}
                          AND child.backdrop_item_id = :backdrop_item_id
                          AND base.item_id = :item_id
                          AND child.item_id <> :item_id
                        """
                    ),
                    {
                        "user_id": user_id,
                        "backdrop_item_id": backdrop_item_id,
                        "item_id": int(request.item_id),
                    },
                )
                cascade_removed_item_ids = [
                    int(row["item_id"])
                    for row in cascade_item_res.mappings().all()
                    if row.get("item_id") is not None
                ]

                cascade_delete_res = await session.execute(
                    text(
                        f"""
                        DELETE child
                        FROM user_chatbot_slot_placements child
                        JOIN chatbot_slot_dependencies dep
                          ON dep.backdrop_item_id = child.backdrop_item_id
                         AND dep.slot_code = child.slot_code
                         AND dep.is_active = 1
                        JOIN user_chatbot_slot_placements base
                          ON base.user_id = child.user_id
                         AND base.backdrop_item_id = child.backdrop_item_id
                         AND base.slot_code = dep.requires_slot_code
                        WHERE child.user_id = {placement_user_id_expr}
                          AND child.backdrop_item_id = :backdrop_item_id
                          AND base.item_id = :item_id
                          AND child.item_id <> :item_id
                        """
                    ),
                    {
                        "user_id": user_id,
                        "backdrop_item_id": backdrop_item_id,
                        "item_id": int(request.item_id),
                    },
                )
                cascade_removed_count = int(cascade_delete_res.rowcount or 0)
                if cascade_removed_count > 0 and cascade_notice_meta:
                    base_name = (
                        str(cascade_notice_meta.get("base_slot_name_zh") or cascade_notice_meta.get("base_slot_code") or "")
                        if is_zh
                        else str(cascade_notice_meta.get("base_slot_name_en") or cascade_notice_meta.get("base_slot_code") or "")
                    )
                    child_name = (
                        str(cascade_notice_meta.get("child_slot_name_zh") or cascade_notice_meta.get("child_slot_code") or "")
                        if is_zh
                        else str(cascade_notice_meta.get("child_slot_name_en") or cascade_notice_meta.get("child_slot_code") or "")
                    )
                    cascade_notice = (
                        f"你卸下了「{base_name}」的支撑家具，「{child_name}」上的关联物品已自动卸下。"
                        if is_zh
                        else f"You removed the support item in \"{base_name}\", so linked items in \"{child_name}\" were auto-unequipped."
                    )

            delete_res = await session.execute(
                text(
                    f"""
                    DELETE FROM user_chatbot_slot_placements
                    WHERE user_id = {placement_user_id_expr}
                      AND backdrop_item_id = :backdrop_item_id
                      AND item_id = :item_id
                    """
                ),
                {
                    "user_id": user_id,
                    "backdrop_item_id": backdrop_item_id,
                    "item_id": int(request.item_id),
                },
            )
            removed_count = int(delete_res.rowcount or 0)
            await session.commit()

            return {
                "success": True,
                "data": {
                    "item_id": int(request.item_id),
                    "removed_count": removed_count,
                    "backdrop_item_id": backdrop_item_id,
                    "cascade_removed_count": cascade_removed_count,
                    "cascade_removed_item_ids": cascade_removed_item_ids,
                    "cascade_notice": cascade_notice,
                },
            }
    except HTTPException:
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception("卸下 ChatBot 家具失败: %s", exc)
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/claim-share-reward")
async def claim_share_reward(user_id: str, _: None = Depends(require_user_session)):
    """
    领取分享奖励
    - 给用户添加新年背景图片（happynewyear）
    - 设置用户的当前背景图片为新年的背景
    
    Args:
        user_id: 用户ID
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "xxx",
                "background_image_id": "happynewyear",
                "message": "分享奖励已领取"
            }
        }
    """
    try:
        from database import db_manager
        from models_db import User, UserBackgroundImage
        from sqlalchemy import select, update
        
        async with db_manager.get_session() as session:
            # 1. 检查用户是否存在
            result = await session.execute(
                select(User).where(User.user_id == user_id)
            )
            user = result.scalar_one_or_none()
            if not user:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            
            # 2. 检查用户是否已经拥有新年背景
            bg_result = await session.execute(
                select(UserBackgroundImage).where(
                    UserBackgroundImage.user_id == user_id,
                    UserBackgroundImage.image_id == "happynewyear"
                )
            )
            existing_bg = bg_result.scalar_one_or_none()
            
            # 3. 如果还没有，添加到用户背景图片列表
            if not existing_bg:
                user_bg_image = UserBackgroundImage(
                    user_id=user_id,
                    image_id="happynewyear",
                    obtained_type="gift",
                    obtained_source="share_reward"
                )
                session.add(user_bg_image)
                logger.info(f"✅ 为用户添加新年背景图片: user_id={user_id}")
            else:
                logger.info(f"⏭️ 用户已拥有新年背景图片: user_id={user_id}")
            
            # 4. 更新用户的当前背景图片
            await session.execute(
                update(User)
                .where(User.user_id == user_id)
                .values(current_background_image_id="happynewyear")
            )
            
            # 5. 更新用户的 has_shared_avatar 状态
            from models_db import UserUIState
            ui_state_result = await session.execute(
                select(UserUIState).where(UserUIState.user_id == user_id)
            )
            ui_state = ui_state_result.scalar_one_or_none()
            if ui_state:
                ui_state.has_shared_avatar = True
                session.add(ui_state)
                logger.info(f"✅ 更新用户分享头像状态: user_id={user_id}, has_shared_avatar=True")
            else:
                # 如果不存在 UI 状态记录，创建一个新的
                new_ui_state = UserUIState(
                    user_id=user_id,
                    has_shared_avatar=True
                )
                session.add(new_ui_state)
                logger.info(f"✅ 创建用户UI状态并设置分享头像状态: user_id={user_id}, has_shared_avatar=True")
            
            # 6. 获取背景图片的 metadata（用于前端设置透明度等）
            bg_metadata = None
            try:
                from models_db import BackgroundImage
                bg_info_result = await session.execute(
                    select(BackgroundImage).where(BackgroundImage.image_id == "happynewyear")
                )
                bg_info = bg_info_result.scalar_one_or_none()
                if bg_info and bg_info.metadata_json:
                    bg_metadata = bg_info.metadata_json
            except Exception as e:
                logger.warning(f"⚠️ 获取背景图片metadata失败: {e}")
            
            await session.commit()
            
            logger.info(f"✅ 用户领取分享奖励成功: user_id={user_id}, background_image_id=happynewyear")
            
            return {
                "success": True,
                "data": {
                    "user_id": user_id,
                    "background_image_id": "happynewyear",
                    "background_image_metadata": bg_metadata,
                    "message": "分享奖励已领取"
                }
            }
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("领取分享奖励失败")
        raise HTTPException(status_code=500, detail={"code": CLAIM_SHARE_REWARD_FAILED})


@app.get("/api/avatar-gifs")
async def get_avatar_gifs() -> Dict[str, Any]:
    """
    获取头像 GIF 配置
    - 优先读取数据库 avatar_assets 表（按 avatar_id、state 分组）
    - 字段：avatar_id, state, image_url
    - 失败/无数据时返回静态兜底配置
    """
    try:
        from database import db_manager

        async with db_manager.get_session() as session:
            result = await session.execute(
                text(
                    """
                    SELECT avatar_id, state, image_url
                    FROM avatar_assets
                    WHERE state IN ('default', 'positive', 'negative', 'talking')
                    ORDER BY avatar_id, state
                    """
                )
            )
            rows = result.mappings().all()

        if rows:
            # 按 avatar_id 分组，转换为期望的格式
            data: Dict[str, Dict[str, str]] = {}
            for row in rows:
                avatar_id = row.get("avatar_id")
                state = row.get("state")
                image_url = row.get("image_url")
                
                if not avatar_id or not state or not image_url:
                    continue
                
                if avatar_id not in data:
                    data[avatar_id] = {}
                
                data[avatar_id][state] = image_url
            
            if data:
                return {"success": True, "data": data}

        logger.warning("avatar_assets 表无记录，返回兜底静态配置")
    except Exception as exc:
        logger.error(f"获取头像 GIF 配置失败，使用兜底: {exc}")

    return {"success": True, "data": {k: v.model_dump() for k, v in AVATAR_GIF_CONFIG.items()}}


@app.get("/api/share-images")
async def get_share_images() -> Dict[str, Any]:
    """
    获取分享图片列表
    - 优先从数据库 share_images 表读取（is_active=1，按 sort_order、id 排序）
    - 失败或无数据时返回兜底静态列表
    """
    try:
        from database import db_manager

        async with db_manager.get_session() as session:
            result = await session.execute(
                text(
                    """
                    SELECT id, code, url, title, sort_order, is_active
                    FROM share_images
                    WHERE is_active = 1
                    ORDER BY sort_order ASC, id ASC
                    """
                )
            )
            rows = result.mappings().all()

        if rows:
            data = [
                {
                    "id": str(row.get("code") or row.get("id")),
                    "url": row.get("url"),
                    "title": row.get("title"),
                    "sort_order": row.get("sort_order"),
                    "is_active": bool(row.get("is_active")) if row.get("is_active") is not None else None,
                }
                for row in rows
            ]
            return {"success": True, "data": data}

        logger.warning("share_images 表无记录，返回兜底静态列表")
    except Exception as exc:
        logger.error(f"获取分享图片失败，使用兜底: {exc}")

    return {"success": True, "data": [item.model_dump() for item in SHARE_IMAGES]}


@app.put("/api/users/{user_id}/username")
async def update_username(user_id: str, request: UpdateUsernameRequest, _: None = Depends(require_user_session)):
    """
    更新用户名
    
    Args:
        user_id: 用户ID
        request: 更新请求（包含username）
    
    Returns:
        {
            "success": true,
            "message": "更新成功"
        }
    """
    try:
        success = await user_service.update_username(user_id, request.username)
        if success:
            return {
                "success": True,
                "message": "用户名更新成功"
            }
        else:
            raise HTTPException(status_code=500, detail={"code": UPDATE_USERNAME_FAILED})
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("更新用户名失败")
        raise HTTPException(status_code=500, detail={"code": UPDATE_USERNAME_FAILED})


@app.put("/api/users/{user_id}/background-image-id")
async def update_background_image_id(user_id: str, request: UpdateBackgroundImageIdRequest, _: None = Depends(require_user_session)):
    """
    更新用户当前使用的背景图片ID
    
    Args:
        user_id: 用户ID
        request: 更新请求（包含background_image_id）
    
    Returns:
        {
            "success": true,
            "message": "更新成功"
        }
    """
    try:
        success = await user_service.update_current_background_image_id(
            user_id, request.background_image_id
        )
        if not success:
            raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND_OR_UPDATE_FAILED})
        return {"success": True, "message": "更新成功"}
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("更新背景图片ID失败")
        raise HTTPException(status_code=500, detail={"code": UPDATE_BG_IMAGE_FAILED}) from e


@app.put("/api/users/{user_id}/bot-name")
async def update_bot_name(user_id: str, request: UpdateBotNameRequest, _: None = Depends(require_user_session)):
    """
    更新用户给机器人取的名字
    
    Args:
        user_id: 用户ID
        request: 更新请求（包含bot_name）
    
    Returns:
        {
            "success": true,
            "message": "更新成功"
        }
    """
    import asyncio
    import time
    
    start_time = time.time()
    
    try:
        # 🔥 添加超时保护（5秒），避免长时间等待
        success = await asyncio.wait_for(
            user_service.update_bot_name(user_id, request.bot_name),
            timeout=5.0
        )
        
        duration = (time.time() - start_time) * 1000
        
        if success:
            return {
                "success": True,
                "message": "机器人名字更新成功"
            }
        else:
            raise HTTPException(status_code=500, detail={"code": UPDATE_BOT_NAME_FAILED})
    except asyncio.TimeoutError:
        duration = (time.time() - start_time) * 1000
        logger.error(f"更新机器人名字超时: user_id={user_id}, 耗时={duration:.1f}ms")
        raise HTTPException(status_code=504, detail={"code": UPDATE_BOT_NAME_TIMEOUT})
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("更新机器人名字异常")
        raise HTTPException(status_code=500, detail={"code": UPDATE_BOT_NAME_FAILED})


@app.put("/api/users/{user_id}/preferred-language")
async def update_preferred_language(
    user_id: str,
    request: UpdatePreferredLanguageRequest,
    _: None = Depends(require_user_session),
):
    """更新用户首选界面语言（写入 users.preferred_language）"""
    try:
        ok = await user_service.update_preferred_language(user_id, request.preferred_language)
        if not ok:
            raise HTTPException(status_code=500, detail={"code": UPDATE_PREFERRED_LANGUAGE_FAILED})
        return {"success": True, "message": "语言偏好已更新"}
    except HTTPException:
        raise
    except Exception:
        logger.exception("更新首选语言失败")
        raise HTTPException(status_code=500, detail={"code": UPDATE_PREFERRED_LANGUAGE_FAILED})


class SubmitFeedbackRequest(BaseModel):
    """提交反馈请求"""
    content: str


@app.post("/api/users/{user_id}/feedback")
async def submit_feedback(user_id: str, request: SubmitFeedbackRequest, _: None = Depends(require_user_session)):
    """
    提交用户反馈
    
    Args:
        user_id: 用户ID
        request: 反馈请求（包含content）
    
    Returns:
        {
            "success": true,
            "data": {
                "feedback_id": "uuid",
                "user_id": "uuid",
                "content": "反馈内容",
                "status": "pending",
                "created_at": "2024-01-01T00:00:00"
            }
        }
    """
    try:
        # 验证反馈内容
        if not request.content or not request.content.strip():
            raise HTTPException(status_code=400, detail={"code": FEEDBACK_CONTENT_EMPTY})
        
        # 验证用户是否存在
        async with db_manager.get_session() as session:
            from models_db import User
            user_result = await session.execute(
                select(User).where(User.user_id == user_id)
            )
            user = user_result.scalar_one_or_none()
            if not user:
                raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND})
            
            # 创建反馈记录
            feedback_id = str(uuid.uuid4())
            feedback = Feedback(
                feedback_id=feedback_id,
                user_id=user_id,
                content=request.content.strip(),
                status="pending"
            )
            session.add(feedback)
            await session.commit()
            await session.refresh(feedback)
            
            logger.info(f"✅ 用户反馈提交成功: user_id={user_id}, feedback_id={feedback_id}")
            
            return {
                "success": True,
                "data": feedback.to_dict()
            }
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("提交反馈失败")
        raise HTTPException(status_code=500, detail={"code": SUBMIT_FEEDBACK_FAILED})


@app.post("/api/auth/register/phone")
async def register_with_phone(request: RegisterPhoneRequest):
    """
    手机号注册
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "username": "",
                "phone": "",
                "user_type": "registered",
                "token": "会话token",
                "expires_at": "过期时间",
                ...
            }
        }
    """
    try:
        user = await user_service.register_with_phone(
            area_code=request.area_code,
            phone=request.phone,
            username=request.username,
            user_id=request.user_id,
            preferred_language=request.preferred_language,
        )
        # 生成会话 token（注册成功后自动登录）
        session_data = await _generate_session_token(user["user_id"])
        # 将 token 添加到返回数据中
        user.update(session_data)
        return {
            "success": True,
            "data": user
        }
    except AuthError as e:
        raise HTTPException(status_code=400, detail={"code": e.code})
    except Exception as e:
        logger.exception("手机号注册失败")
        raise HTTPException(status_code=500, detail={"code": REGISTER_FAILED})


@app.post("/api/auth/register/email")
async def register_with_email(request: RegisterEmailRequest):
    """
    邮箱注册
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "username": "",
                "email": "",
                "user_type": "registered",
                "token": "会话token",
                "expires_at": "过期时间",
                ...
            }
        }
    """
    try:
        user = await user_service.register_with_email(
            email=request.email,
            username=request.username,
            user_id=request.user_id,
            preferred_language=request.preferred_language,
        )
        # 生成会话 token（注册成功后自动登录）
        session_data = await _generate_session_token(user["user_id"])
        # 将 token 添加到返回数据中
        user.update(session_data)
        return {
            "success": True,
            "data": user
        }
    except AuthError as e:
        raise HTTPException(status_code=400, detail={"code": e.code})
    except Exception as e:
        logger.exception("邮箱注册失败")
        raise HTTPException(status_code=500, detail={"code": REGISTER_FAILED})


@app.post("/api/auth/login/wechat")
async def login_with_wechat(request: LoginWechatRequest):
    """
    微信登录/注册
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "username": "",
                "wechat_openid": "",
                "user_type": "registered",
                "token": "会话token",
                "expires_at": "过期时间",
                ...
            }
        }
    """
    try:
        user = await user_service.login_with_wechat(
            openid=request.openid,
            username=request.username
        )
        # 生成会话 token
        session_data = await _generate_session_token(user["user_id"])
        # 将 token 添加到返回数据中
        user.update(session_data)
        return {
            "success": True,
            "data": user
        }
    except Exception as e:
        logger.exception("微信登录失败")
        raise HTTPException(status_code=500, detail={"code": LOGIN_FAILED})


@app.post("/api/auth/login/qq")
async def login_with_qq(request: LoginQQRequest):
    """
    QQ登录/注册
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "username": "",
                "qq_openid": "",
                "user_type": "registered",
                "token": "会话token",
                "expires_at": "过期时间",
                ...
            }
        }
    """
    try:
        user = await user_service.login_with_qq(
            openid=request.openid,
            username=request.username
        )
        # 生成会话 token
        session_data = await _generate_session_token(user["user_id"])
        # 将 token 添加到返回数据中
        user.update(session_data)
        return {
            "success": True,
            "data": user
        }
    except Exception as e:
        logger.exception("QQ登录失败")
        raise HTTPException(status_code=500, detail={"code": LOGIN_FAILED})


@app.post("/api/auth/login/phone")
async def login_with_phone(request: LoginPhoneRequest):
    """
    手机号登录
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "phone": "",
                "token": "会话token",
                "expires_at": "过期时间",
                ...
            }
        }
    """
    try:
        user = await user_service.login_with_phone(
            phone=request.phone,
            area_code=request.area_code,
            preferred_language=request.preferred_language,
        )
        if user:
            # 生成会话 token
            session_data = await _generate_session_token(user["user_id"])
            # 将 token 添加到返回数据中
            user.update(session_data)
            return {
                "success": True,
                "data": user
            }
        else:
            raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND_REGISTER_FIRST})
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("手机号登录失败")
        raise HTTPException(status_code=500, detail={"code": LOGIN_FAILED})


@app.post("/api/auth/login/email")
async def login_with_email(request: LoginEmailRequest):
    """
    邮箱登录
    
    Returns:
        {
            "success": true,
            "data": {
                "user_id": "uuid",
                "email": "",
                "token": "会话token",
                "expires_at": "过期时间",
                ...
            }
        }
    """
    try:
        user = await user_service.login_with_email(
            request.email,
            preferred_language=request.preferred_language,
        )
        if user:
            # 生成会话 token
            session_data = await _generate_session_token(user["user_id"])
            # 将 token 添加到返回数据中
            user.update(session_data)
            return {
                "success": True,
                "data": user
            }
        else:
            raise HTTPException(status_code=404, detail={"code": USER_NOT_FOUND_REGISTER_FIRST})
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("邮箱登录失败")
        raise HTTPException(status_code=500, detail={"code": LOGIN_FAILED})


@app.post("/api/auth/send-code")
async def send_code(request: SendCodeRequest):
    """
    生成并保存验证码
    
    当用户点击 send code 按钮时调用此接口，生成临时验证码并保存到数据库
    
    Args:
        request: 发送验证码请求
            - phone: 手机号（发送类型为短信时必填）
            - email: 邮箱（发送类型为邮箱时必填）
            - send_type: 发送类型 ("domestic_sms", "international_sms", "email")
    
    Returns:
        {
            "success": true,
            "data": {
                "code_id": "验证码ID",
                "message": "验证码已生成，等待发送"
            }
        }
    """
    try:
        # 验证参数
        if request.send_type in ["domestic_sms", "international_sms"]:
            if not request.phone:
                raise HTTPException(status_code=400, detail={"code": PHONE_REQUIRED})
        elif request.send_type == "email":
            if not request.email:
                raise HTTPException(status_code=400, detail={"code": EMAIL_REQUIRED})
        else:
            raise HTTPException(status_code=400, detail={"code": SEND_TYPE_INVALID})
        
        # 生成4位数字验证码
        code = str(random.randint(1000, 9999))
        code_id = str(uuid.uuid4())
        
        # 保存到数据库
        async with db_manager.get_session() as session:
            # 🔥 将同一手机号/邮箱的旧验证码标记为无效
            # 确保同一时间只有一个有效的验证码
            if request.send_type in ["domestic_sms", "international_sms"]:
                # 将同一手机号的旧验证码标记为无效
                await session.execute(
                    update(VerificationCode)
                    .where(
                        VerificationCode.phone == request.phone,
                        VerificationCode.send_type == request.send_type,
                        VerificationCode.is_valid == True
                    )
                    .values(is_valid=False)
                )
            elif request.send_type == "email":
                # 将同一邮箱的旧验证码标记为无效
                await session.execute(
                    update(VerificationCode)
                    .where(
                        VerificationCode.email == request.email,
                        VerificationCode.send_type == request.send_type,
                        VerificationCode.is_valid == True
                    )
                    .values(is_valid=False)
                )
            
            # 创建新的验证码
            verification_code = VerificationCode(
                code_id=code_id,
                area_code=request.area_code,
                phone=request.phone,
                email=request.email,
                code=code,
                send_type=request.send_type,
                is_sent=False,
                is_valid=True,
            )
            session.add(verification_code)
            await session.commit()
        
        logger.info(f"✅ 验证码已生成: code_id={code_id}, send_type={request.send_type}, area_code={request.area_code}, phone={request.phone}, email={request.email}")
        
        return {
            "success": True,
            "data": {
                "code_id": code_id,
                "message": "验证码已生成，等待发送"
            }
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("生成验证码失败")
        raise HTTPException(status_code=500, detail={"code": SEND_CODE_FAILED})


@app.post("/api/auth/verify-code")
async def verify_code(request: VerifyCodeRequest):
    """
    验证验证码
    
    Args:
        request: 验证验证码请求
            - phone: 手机号（发送类型为短信时必填）
            - email: 邮箱（发送类型为邮箱时必填）
            - code: 验证码
            - send_type: 发送类型 ("domestic_sms", "international_sms", "email")
    
    Returns:
        {
            "success": true,
            "data": {
                "valid": true/false,
                "message": "验证成功" 或 "验证码错误"
            }
        }
    """
    try:
        # 验证参数
        if request.send_type in ["domestic_sms", "international_sms"]:
            if not request.phone:
                raise HTTPException(status_code=400, detail={"code": PHONE_REQUIRED})
        elif request.send_type == "email":
            if not request.email:
                raise HTTPException(status_code=400, detail={"code": EMAIL_REQUIRED})
        else:
            raise HTTPException(status_code=400, detail={"code": SEND_TYPE_INVALID})
        
        if not request.code:
            raise HTTPException(status_code=400, detail={"code": CODE_REQUIRED})
        
        # 查询有效的验证码
        async with db_manager.get_session() as session:
            if request.send_type in ["domestic_sms", "international_sms"]:
                result = await session.execute(
                    select(VerificationCode)
                    .where(
                        VerificationCode.phone == request.phone,
                        VerificationCode.send_type == request.send_type,
                        VerificationCode.code == request.code,
                        VerificationCode.is_valid == True
                    )
                    .order_by(VerificationCode.created_at.desc())
                    .limit(1)
                )
            else:
                result = await session.execute(
                    select(VerificationCode)
                    .where(
                        VerificationCode.email == request.email,
                        VerificationCode.send_type == request.send_type,
                        VerificationCode.code == request.code,
                        VerificationCode.is_valid == True
                    )
                    .order_by(VerificationCode.created_at.desc())
                    .limit(1)
                )
            
            verification_code = result.scalar_one_or_none()
            
            if verification_code:
                # 验证成功，将验证码标记为无效（一次性使用）
                verification_code.is_valid = False
                await session.commit()
                
                logger.info(f"✅ 验证码验证成功: phone={request.phone}, email={request.email}, send_type={request.send_type}")
                
                return {
                    "success": True,
                    "data": {
                        "valid": True,
                        "message": "验证成功"
                    }
                }
            else:
                logger.warning(f"❌ 验证码验证失败: phone={request.phone}, email={request.email}, code={request.code}")
                
                return {
                    "success": True,
                    "data": {
                        "valid": False,
                        "message": "验证码错误或已失效"
                    }
                }
                
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("验证验证码失败")
        raise HTTPException(status_code=500, detail={"code": VERIFY_CODE_FAILED})


@app.post("/api/auth/verify-and-auth")
async def verify_and_auth(request: VerifyAndAuthRequest):
    """
    V2.0：验证码通过后统一登录或注册。
    返回 { success, data: { user_id, token, expires_at, is_new_user, ... } }
    """
    if request.type not in ("email", "phone"):
        raise HTTPException(status_code=400, detail={"code": TYPE_MUST_BE_EMAIL_OR_PHONE})
    try:
        user, is_new_user = await user_service.verify_and_auth(
            auth_type=request.type,
            code=request.code,
            email=request.email,
            phone=request.phone,
            area_code=request.area_code,
            preferred_language=request.preferred_language,
        )
        session_data = await _generate_session_token(user["user_id"])
        user.update(session_data)
        user["is_new_user"] = is_new_user
        return {"success": True, "data": user}
    except AuthError as e:
        raise HTTPException(status_code=400, detail={"code": e.code})
    except Exception as e:
        logger.exception("verify-and-auth 失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/community-choice")
async def get_community_choice(user_id: str, _: None = Depends(require_user_session)) -> Dict[str, Any]:
    """V2.0：获取用户「退出早期共创社区」选项（登录页回显）"""
    try:
        data = await user_service.get_community_choice(user_id)
        return {"success": True, "data": data or {"opted_out": False}}
    except Exception as e:
        logger.exception("获取 community-choice 失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.put("/api/users/{user_id}/community-choice")
async def update_community_choice(
    user_id: str,
    request: CommunityChoiceRequest,
    _: None = Depends(require_user_session),
):
    """V2.0：更新用户「退出早期共创社区」选项"""
    try:
        await user_service.upsert_community_choice(user_id, request.opted_out)
        return {"success": True}
    except Exception as e:
        logger.exception("更新 community-choice 失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/early-community")
async def get_early_community(user_id: str, _: None = Depends(require_user_session)) -> Dict[str, Any]:
    """V2.0：获取用户早期共创状态（是否已展示弹窗、是否已留邮箱）"""
    try:
        data = await user_service.get_early_community(user_id)
        return {"success": True, "data": data or {}}
    except Exception as e:
        logger.exception("获取 early-community 失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/early-community")
async def submit_early_community(
    user_id: str,
    request: EarlyCommunityRequest,
    _: None = Depends(require_user_session),
):
    """V2.0：提交早期共创（标记弹窗已展示，可选留邮箱）"""
    try:
        data = await user_service.submit_early_community(user_id, request.email)
        return {"success": True, "data": data}
    except Exception as e:
        logger.exception("提交 early-community 失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/ui-state")
async def get_user_ui_state(user_id: str, _: None = Depends(require_user_session)) -> Dict[str, Any]:
    """
    获取用户前端可视化状态（页面、路由、提示状态）
    """
    try:
        state = await user_service.get_user_ui_state(user_id)
        if not state:
            # 没有记录时返回默认值
            default_state = UserUIStateResponse(
                user_id=user_id,
                current_page=None,
                visited_routes=[],
            )
            return {"success": True, "data": default_state.dict()}
        return {"success": True, "data": state}
    except Exception as e:  # noqa: BLE001
        logger.exception("获取用户UI状态失败")
        raise HTTPException(status_code=500, detail={"code": GET_UI_STATE_FAILED}) from e


@app.post("/api/page-visit")
async def report_page_visit(
    request: PageVisitRequest,
    user_id: Optional[str] = Depends(optional_user_id),
) -> Dict[str, Any]:
    """记录一次页面访问（任意界面打开时上报，未登录也可上报）"""
    path = (request.path or "").strip() or "/"
    await user_service.log_page_visit(user_id, path)
    return {"success": True}


@app.put("/api/users/{user_id}/ui-state")
async def update_user_ui_state(
    user_id: str,
    request: UpdateUserUIStateRequest,
    _: None = Depends(require_user_session),
) -> Dict[str, Any]:
    """
    更新用户前端可视化状态（部分字段）
    """
    try:
        updated = await user_service.upsert_user_ui_state(
            user_id=user_id,
            current_page=request.current_page,
            visited_routes=request.visited_routes,
            has_shown_tap_to_chat=request.has_shown_tap_to_chat,
            has_shown_cancel_tooltip=request.has_shown_cancel_tooltip,
            has_shown_send_tooltip=request.has_shown_send_tooltip,
            has_shown_progress_bar_tooltip=request.has_shown_progress_bar_tooltip,
            has_shown_mission_stone_modal=request.has_shown_mission_stone_modal,
            has_shown_conversation_stone_modal=request.has_shown_conversation_stone_modal,
            has_clicked_mission_stone=request.has_clicked_mission_stone,
            has_clicked_conversation_stone=request.has_clicked_conversation_stone,
            has_shown_first_shop_tooltip=request.has_shown_first_shop_tooltip,
            has_clicked_shop_in_settings=request.has_clicked_shop_in_settings,
            avatar_c_unlocked=request.avatar_c_unlocked,
            has_shared_avatar=request.has_shared_avatar,
            current_avatar_id=request.current_avatar_id,
            subtitle_on=request.subtitle_on,
            sfx_enabled=request.sfx_enabled,
            bgm_enabled=request.bgm_enabled,
        )
        if not updated:
            raise HTTPException(status_code=500, detail={"code": UPDATE_UI_STATE_FAILED})
        return {"success": True, "data": updated}
    except HTTPException:
        raise
    except Exception as e:  # noqa: BLE001
        logger.exception("更新用户UI状态失败")
        raise HTTPException(status_code=500, detail={"code": UPDATE_UI_STATE_FAILED}) from e


@app.get("/api/users/{user_id}/intimacy")
async def get_user_intimacy(user_id: str, _: None = Depends(require_user_session)) -> Dict[str, Any]:
    """获取用户亲密度：heart_count 0-3，days_since_last_visit 距上次访问天数"""
    try:
        data = await user_service.get_intimacy(user_id)
        return {"success": True, "data": data}
    except Exception as e:  # noqa: BLE001
        logger.exception("获取亲密度失败")
        raise HTTPException(status_code=500, detail={"code": "GET_INTIMACY_FAILED"}) from e


@app.post("/api/users/{user_id}/intimacy/light-up")
async def light_up_intimacy(user_id: str, _: None = Depends(require_user_session)) -> Dict[str, Any]:
    """点亮一颗心，最多 3 颗"""
    try:
        data = await user_service.light_up_intimacy(user_id)
        return {"success": True, "data": data}
    except Exception as e:  # noqa: BLE001
        logger.exception("点亮亲密度失败")
        raise HTTPException(status_code=500, detail={"code": "LIGHT_UP_INTIMACY_FAILED"}) from e


@app.get("/api/users/{user_id}/intimacy/status")
async def get_user_intimacy_status(user_id: str, _: None = Depends(require_user_session)) -> Dict[str, Any]:
    """
    获取用户当天是否可以点亮亲密度。
    返回:
    {
        "success": true,
        "data": {
            "heart_count": int,           # 当前心数 0-3
            "can_light_up_today": bool,  # 今天是否还能点亮一次
        }
    }
    """
    try:
        data = await user_service.get_intimacy_status(user_id)
        return {"success": True, "data": data}
    except Exception as e:  # noqa: BLE001
        logger.exception("获取亲密度状态失败")
        raise HTTPException(status_code=500, detail={"code": "GET_INTIMACY_STATUS_FAILED"}) from e


# ==================== OAuth 授权端点 ====================

@app.get("/api/auth/wechat/authorize")
async def wechat_authorize(request: Request):
    """
    获取微信授权URL
    
    根据微信官方文档：https://developers.weixin.qq.com/doc/oplatform/Website_App/WeChat_Login/Wechat_Login.html
    - 使用 qrconnect 端点进行网站应用授权
    - scope 必须为 snsapi_login
    - redirect_uri 需要 urlEncode 处理
    - state 用于防止 CSRF 攻击
    
    Returns:
        重定向到微信授权页面
    """
    if not settings.WECHAT_OPEN_APP_ID or not settings.WECHAT_REDIRECT_URI:
        raise HTTPException(status_code=500, detail={"code": WECHAT_OAUTH_NOT_CONFIGURED})
    
    # 生成state参数（用于防止CSRF攻击）
    import secrets
    state = secrets.token_urlsafe(32)
    
    # 将 state 存储到 Redis（如果可用），用于回调时验证
    # TTL 设置为 10 分钟（code 的有效期也是 10 分钟）
    redis_cache = getattr(request.app.state, 'redis_cache', None)
    if redis_cache:
        try:
            redis_client = await redis_cache._get_redis()
            await redis_client.setex(f"oauth:wechat:state:{state}", 600, "1")  # 10分钟过期
        except Exception as e:
            logger.warning(f"存储 state 到 Redis 失败，将跳过验证: {e}")
    
    # 构建微信授权URL（使用开放平台的AppID）
    # redirect_uri 会自动通过 urlencode 进行编码
    params = {
        "appid": settings.WECHAT_OPEN_APP_ID,  # 使用开放平台的AppID
        "redirect_uri": settings.WECHAT_REDIRECT_URI,  # urlencode 会自动编码
        "response_type": "code",
        "scope": "snsapi_login",  # 网站应用必须使用 snsapi_login
        "state": state,
    }
    
    # 根据官方文档格式：https://open.weixin.qq.com/connect/qrconnect?appid=APPID&redirect_uri=REDIRECT_URI&response_type=code&scope=SCOPE&state=STATE#wechat_redirect
    auth_url = f"https://open.weixin.qq.com/connect/qrconnect?{urllib.parse.urlencode(params)}#wechat_redirect"
    
    return RedirectResponse(url=auth_url)


@app.get("/api/auth/wechat/callback")
async def wechat_callback(
    request: Request,
    code: str = Query(..., description="微信授权码"),
    state: str = Query(None, description="状态参数")
):
    """
    微信OAuth回调处理
    
    根据微信官方文档流程：
    1. 验证 state 参数（防止 CSRF 攻击）
    2. 通过 code 换取 access_token
    3. 使用 openid 进行登录/注册
    
    Args:
        code: 微信返回的授权码（有效期10分钟，只能使用一次）
        state: 状态参数（用于验证请求合法性）
    
    Returns:
        重定向到前端页面（带token）
    """
    try:
        if not settings.WECHAT_OPEN_APP_ID or not settings.WECHAT_OPEN_APP_SECRET:
            raise HTTPException(status_code=500, detail={"code": WECHAT_OAUTH_NOT_CONFIGURED})
        
        # 1. 验证 state 参数（防止 CSRF 攻击）
        if state:
            redis_cache = getattr(request.app.state, 'redis_cache', None)
            if redis_cache:
                try:
                    redis_client = await redis_cache._get_redis()
                    state_key = f"oauth:wechat:state:{state}"
                    exists = await redis_client.exists(state_key)
                    if not exists:
                        logger.warning(f"State 验证失败: {state} (可能已过期或被篡改)")
                        return RedirectResponse(
                            url=f"{settings.FRONTEND_URL}/register?error=wechat_auth_failed&reason=invalid_state"
                        )
                    # 验证成功后删除 state（一次性使用）
                    await redis_client.delete(state_key)
                except Exception as e:
                    logger.warning(f"State 验证异常: {e}，继续处理")
        
        # 2. 通过code获取access_token（使用开放平台的AppID和AppSecret）
        # 根据官方文档：https://api.weixin.qq.com/sns/oauth2/access_token
        token_url = "https://api.weixin.qq.com/sns/oauth2/access_token"
        token_params = {
            "appid": settings.WECHAT_OPEN_APP_ID,  # 使用开放平台的AppID
            "secret": settings.WECHAT_OPEN_APP_SECRET,  # 使用开放平台的AppSecret
            "code": code,
            "grant_type": "authorization_code",
        }
        
        async with httpx.AsyncClient() as client:
            token_response = await client.get(token_url, params=token_params)
            token_data = token_response.json()
        
        if "errcode" in token_data:
            logger.error(f"微信获取access_token失败: {token_data}")
            return RedirectResponse(
                url=f"{settings.FRONTEND_URL}/register?error=wechat_auth_failed"
            )
        
        access_token = token_data.get("access_token")
        openid = token_data.get("openid")
        # refresh_token 可用于刷新 access_token，当前实现暂不保存
        
        if not access_token or not openid:
            logger.error(f"微信返回数据不完整: {token_data}")
            return RedirectResponse(
                url=f"{settings.FRONTEND_URL}/register?error=wechat_auth_failed"
            )
        
        # 3. 使用openid登录/注册
        # 注意：当前使用 snsapi_login scope，只能获取 openid
        # 如需获取用户昵称、头像等信息，需要：
        # - 将 scope 改为 snsapi_login,snsapi_userinfo
        # - 调用 /sns/userinfo 接口获取用户信息
        user = await user_service.login_with_wechat(openid=openid)
        
        # 4. 重定向到前端，携带用户信息
        frontend_url = f"{settings.FRONTEND_URL}/register?oauth=wechat&user_id={user['user_id']}&username={urllib.parse.quote(user['username'])}"
        return RedirectResponse(url=frontend_url)
        
    except Exception as e:
        logger.error(f"微信OAuth回调处理失败: {e}")
        return RedirectResponse(
            url=f"{settings.FRONTEND_URL}/register?error=wechat_auth_failed"
        )


@app.get("/api/auth/qq/authorize")
async def qq_authorize(
    request: Request, 
    from_page: str = Query(None, alias="from", description="来源页面"),
    user_id: str = Query(None, description="可选的现有用户ID，如果提供则更新该用户而不是创建新用户")
):
    """
    获取QQ授权URL
    
    Args:
        from_page: 来源页面（如 first-login），用于回调时决定重定向位置
        user_id: 可选的现有用户ID，如果提供则更新该用户而不是创建新用户
    
    Returns:
        重定向到QQ授权页面
    """
    if not settings.QQ_APP_ID or not settings.QQ_REDIRECT_URI:
        raise HTTPException(status_code=500, detail={"code": QQ_OAUTH_NOT_CONFIGURED})
    
    # 生成state参数（用于防止CSRF攻击）
    import secrets
    state_token = secrets.token_urlsafe(32)
    
    # state只存储token，from_page和user_id存储在Redis中（避免token包含下划线导致解析错误）
    state = state_token
    
    # 将 state 存储到 Redis 以便回调校验（存储原始 token 和来源信息）
    redis_cache = getattr(request.app.state, 'redis_cache', None)
    state_stored = False
    state_key = f"oauth:qq:state:{state_token}"
    
    logger.info(f"📝 [QQ授权] 开始存储state - state_token={state_token}, from_page={from_page}, user_id={user_id}, state={state}, state_key={state_key}")
    
    if redis_cache:
        try:
            redis_client = await redis_cache._get_redis()
            
            # 🔥 检查Redis连接状态
            try:
                ping_result = await redis_client.ping()
                logger.debug(f"✅ [QQ授权] Redis连接正常: ping={ping_result}")
            except Exception as ping_err:
                logger.error(f"❌ [QQ授权] Redis ping失败: {ping_err}")
            
            # 🔥 检查Redis配置信息（DB、连接池等）
            try:
                redis_info = await redis_client.info('server')
                redis_db = redis_client.connection_pool.connection_kwargs.get('db', 0)
                logger.debug(f"🔍 [QQ授权] Redis配置 - DB={redis_db}, server_version={redis_info.get('redis_version', 'unknown')}")
            except Exception as info_err:
                logger.debug(f"⚠️ [QQ授权] 获取Redis信息失败: {info_err}")
            
            # 🔥 检查key是否已存在（防止重复）
            existing_value = await redis_client.get(state_key)
            if existing_value:
                logger.warning(f"⚠️ [QQ授权] state key已存在，将被覆盖: key={state_key}, old_value={existing_value[:50]}...")
            
            # 存储原始 token、来源信息和 user_id
            state_data = {"token": state_token}
            if from_page:
                state_data["from"] = from_page
            if user_id:
                state_data["user_id"] = user_id
            state_value = json.dumps(state_data)
            logger.debug(f"📦 [QQ授权] 准备存储 - key={state_key}, value={state_value}, TTL=1800s")
            
            # 🔥 增加TTL到1800秒（30分钟），避免授权流程过长导致过期
            await redis_client.setex(state_key, 1800, state_value)
            state_stored = True
            
            # 🔥 验证存储是否成功
            verify_value = await redis_client.get(state_key)
            if verify_value:
                # 🔥 检查TTL
                ttl = await redis_client.ttl(state_key)
                logger.info(f"✅ [QQ授权] QQ state 已存储并验证成功 - key={state_key}, token={state_token[:16]}..., TTL={ttl}s, value_length={len(verify_value)}")
            else:
                logger.error(f"❌ [QQ授权] QQ state 存储后验证失败 - key={state_key}, token={state_token[:16]}...")
                
                # 🔥 尝试查找所有相关的keys
                try:
                    pattern = f"oauth:qq:state:*"
                    keys = []
                    async for key in redis_client.scan_iter(match=pattern, count=100):
                        keys.append(key)
                    logger.debug(f"🔍 [QQ授权] Redis中存在的QQ state keys: {keys[:10]}... (共{len(keys)}个)")
                except Exception as scan_err:
                    logger.debug(f"🔍 [QQ授权] 扫描keys失败: {scan_err}")
        except Exception as e:
            logger.error(f"❌ [QQ授权] 存储 QQ state 失败 - key={state_key}, error={e}", exc_info=True)
    else:
        logger.warning("⚠️ [QQ授权] Redis缓存服务不可用，QQ state 验证将被跳过")
    
    # 确保回调地址格式正确（去除末尾斜杠，确保与QQ应用配置完全一致）
    redirect_uri = settings.QQ_REDIRECT_URI.rstrip('/')
    
    # 构建QQ授权URL
    # 注意：QQ开放平台要求redirect_uri必须与配置的回调域完全匹配
    params = {
        "response_type": "code",
        "client_id": settings.QQ_APP_ID,
        "redirect_uri": redirect_uri,  # 使用处理后的回调地址
        "state": state,
        "scope": "get_user_info",  # QQ授权范围
    }
    
    # URL编码参数（确保特殊字符正确编码）
    auth_url = f"https://graph.qq.com/oauth2.0/authorize?{urllib.parse.urlencode(params)}"
    
    # 调试日志：记录实际发送的授权URL（隐藏敏感信息）
    logger.info(f"🔍 QQ授权URL生成 - AppID: {settings.QQ_APP_ID[:8]}..., 回调地址: {redirect_uri}, 来源: {from_page}")
    
    return RedirectResponse(url=auth_url)


@app.get("/api/qq/mp/verify")
async def qq_callback(
    request: Request,
    code: str = Query(..., description="QQ授权码"),
    state: str = Query(None, description="状态参数")
):
    """
    QQ OAuth回调处理
    
    Args:
        code: QQ返回的授权码
        state: 状态参数
    
    Returns:
        重定向到前端页面（带token）
    """
    try:
        logger.info(f"📥 [QQ回调] 收到回调请求 - code={'***' if code else '(empty)'}, state={state}")
        
        # 0. 验证 state 防止 CSRF，并提取来源信息
        from_page = None
        user_id_from_state = None
        state_token = state  # state就是token，不再需要分割
        if state:
            logger.debug(f"🔍 [QQ回调] 开始解析state - state={state}, length={len(state)}")
            
            redis_cache = getattr(request.app.state, 'redis_cache', None)
            state_key = f"oauth:qq:state:{state_token}"
            logger.info(f"🔑 [QQ回调] 准备验证state - state_token={state_token[:16]}..., state_key={state_key}")
            
            if redis_cache:
                try:
                    redis_client = await redis_cache._get_redis()
                    
                    # 🔥 检查Redis连接状态
                    try:
                        ping_result = await redis_client.ping()
                        logger.debug(f"✅ [QQ回调] Redis连接正常: ping={ping_result}")
                    except Exception as ping_error:
                        logger.error(f"❌ [QQ回调] Redis ping失败: {ping_error}", exc_info=True)
                        # Redis连接失败时，跳过验证但记录警告（为了不影响用户体验）
                        logger.warning("⚠️ [QQ回调] Redis连接失败，跳过state验证（可能存在安全风险）")
                        # 继续处理，不阻止登录流程
                    else:
                        # 🔥 检查Redis配置信息（DB、连接池等）
                        try:
                            redis_db = redis_client.connection_pool.connection_kwargs.get('db', 0)
                            logger.debug(f"🔍 [QQ回调] Redis配置 - DB={redis_db}")
                        except Exception as info_err:
                            logger.debug(f"⚠️ [QQ回调] 获取Redis配置失败: {info_err}")
                        
                        # 🔥 尝试查找key
                        logger.debug(f"🔍 [QQ回调] 开始查找key: {state_key}")
                        state_data_str = await redis_client.get(state_key)
                        
                        if not state_data_str:
                            # 🔥 增加详细日志，帮助诊断问题
                            logger.warning(f"❌ [QQ回调] QQ state 验证失败 - state_token={state_token[:16]}..., state_key={state_key}")
                            
                            # 🔥 检查key的TTL（如果key存在但值为None，可能是其他问题）
                            try:
                                exists = await redis_client.exists(state_key)
                                logger.debug(f"🔍 [QQ回调] key存在性检查: exists={exists}")
                                if exists:
                                    ttl = await redis_client.ttl(state_key)
                                    logger.debug(f"🔍 [QQ回调] key存在但值为空，TTL={ttl}s")
                            except Exception as exists_err:
                                logger.debug(f"🔍 [QQ回调] 检查key存在性失败: {exists_err}")
                            
                            # 🔥 尝试查找所有相关的state key（用于调试）
                            try:
                                pattern = f"oauth:qq:state:*"
                                keys = []
                                async for key in redis_client.scan_iter(match=pattern, count=100):
                                    keys.append(key)
                                if keys:
                                    logger.warning(f"🔍 [QQ回调] Redis中存在其他QQ state keys (共{len(keys)}个): {keys[:10]}")
                                    # 🔥 检查这些keys的TTL
                                    for key in keys[:5]:
                                        try:
                                            ttl = await redis_client.ttl(key)
                                            value = await redis_client.get(key)
                                            logger.debug(f"🔍 [QQ回调] key详情 - {key}: TTL={ttl}s, value_length={len(value) if value else 0}")
                                        except Exception as key_err:
                                            logger.debug(f"🔍 [QQ回调] 获取key详情失败 {key}: {key_err}")
                                else:
                                    logger.warning(f"🔍 [QQ回调] Redis中不存在任何QQ state keys")
                            except Exception as scan_error:
                                logger.debug(f"🔍 [QQ回调] 扫描Redis keys失败: {scan_error}")
                            
                            return RedirectResponse(
                                url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed&reason=invalid_state"
                            )
                        
                        # 🔥 找到key，记录详细信息
                        ttl = await redis_client.ttl(state_key)
                        logger.info(f"✅ [QQ回调] 找到state key - key={state_key}, TTL={ttl}s, value_length={len(state_data_str)}")
                        
                        # 如果存储的是 JSON 格式，解析获取来源信息和user_id
                        try:
                            state_data = json.loads(state_data_str)
                            from_page = state_data.get("from") or from_page
                            user_id_from_state = state_data.get("user_id")
                            logger.info(f"✅ [QQ回调] QQ state 验证成功 - state_token={state_token[:16]}..., 来源: {from_page}, user_id: {user_id_from_state}, data={state_data}")
                        except (json.JSONDecodeError, TypeError) as json_err:
                            # 如果不是 JSON 格式，使用从 state 中解析的 from_page
                            logger.debug(f"✅ [QQ回调] QQ state 验证成功 (非JSON格式) - state_token={state_token[:16]}..., value={state_data_str[:50]}..., json_err={json_err}")
                            pass
                        
                        # 🔥 删除已使用的state key
                        deleted = await redis_client.delete(state_key)
                        logger.debug(f"🗑️ [QQ回调] 删除state key - key={state_key}, deleted={deleted}")
                        
                except Exception as state_error:
                    logger.error(f"❌ [QQ回调] QQ state 验证异常 - state_token={state_token[:16]}..., error={state_error}", exc_info=True)
                    # 🔥 验证异常时不阻止流程，但记录详细错误
            else:
                logger.warning("⚠️ [QQ回调] Redis缓存服务不可用，跳过state验证")
        
        if not settings.QQ_APP_ID or not settings.QQ_APP_KEY:
            raise HTTPException(status_code=500, detail={"code": QQ_OAUTH_NOT_CONFIGURED})
        
        # 小工具：解析 QQ JSONP 或 querystring 响应
        def parse_qq_payload(payload_text: str) -> dict:
            cleaned = (payload_text or "").strip()
            if not cleaned:
                return {}
            if cleaned.startswith("callback("):
                cleaned = cleaned.split("callback(", 1)[1]
                cleaned = cleaned.rsplit(");", 1)[0]
            try:
                return json.loads(cleaned)
            except json.JSONDecodeError:
                parsed = {}
                for part in cleaned.split("&"):
                    if "=" in part:
                        key, value = part.split("=", 1)
                        parsed[key] = value
                return parsed
        
        # 1. 通过code获取access_token
        # 确保回调地址格式正确（与授权时保持一致）
        redirect_uri = settings.QQ_REDIRECT_URI.rstrip('/')
        
        token_url = "https://graph.qq.com/oauth2.0/token"
        token_params = {
            "grant_type": "authorization_code",
            "client_id": settings.QQ_APP_ID,
            "client_secret": settings.QQ_APP_KEY,
            "code": code,
            "redirect_uri": redirect_uri,  # 使用处理后的回调地址
            "fmt": "json",
        }
        
        logger.info(f"🔍 QQ获取token - 回调地址: {redirect_uri}")
        
        async with httpx.AsyncClient() as client:
            token_response = await client.get(token_url, params=token_params)
            token_text = token_response.text
            try:
                token_data = token_response.json()
            except ValueError:
                token_data = parse_qq_payload(token_text)
        
        if "error" in token_data:
            logger.error(f"QQ获取access_token失败: {token_data}")
            return RedirectResponse(
                url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed&reason=token"
            )
        
        access_token = token_data.get("access_token")
        
        if not access_token:
            logger.error(f"QQ返回数据不完整: {token_data}")
            return RedirectResponse(
                url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed"
            )
        
        # 2. 通过access_token获取openid
        openid_url = "https://graph.qq.com/oauth2.0/me"
        openid_params = {
            "access_token": access_token,
            "unionid": 1,
            "fmt": "json",
        }
        
        async with httpx.AsyncClient() as client:
            openid_response = await client.get(openid_url, params=openid_params)
            openid_data = parse_qq_payload(openid_response.text)
        
        if "error" in openid_data:
            logger.error(f"QQ获取openid失败: {openid_data}")
            return RedirectResponse(
                url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed"
            )
        
        openid = openid_data.get("openid")
        unionid = openid_data.get("unionid")
        
        if not openid:
            logger.error(f"QQ返回openid为空: {openid_data}")
            return RedirectResponse(
                url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed"
            )
        
        # 2.1 可选：拉取用户信息用于填充昵称
        nickname = ""
        user_info = {}
        user_info_url = "https://graph.qq.com/user/get_user_info"
        user_info_params = {
            "access_token": access_token,
            "oauth_consumer_key": settings.QQ_APP_ID,
            "openid": openid,
            "fmt": "json",
        }
        try:
            async with httpx.AsyncClient() as client:
                user_info_resp = await client.get(
                    user_info_url,
                    params=user_info_params,
                )
                user_info = user_info_resp.json()
                if user_info.get("ret") == 0:
                    nickname = user_info.get("nickname", "") or ""
                else:
                    logger.warning(
                        f"QQ用户信息获取失败: {user_info}"
                    )
        except Exception as info_error:
            logger.warning(f"拉取QQ用户信息异常: {info_error}")
        
        # 3. 根据来源页面决定处理逻辑
        if from_page == "login":
            # 从 login 页面来的，需要先检查用户是否存在
            # 如果用户不存在，不允许自动注册，应该提示用户先注册
            logger.info(f"🔍 [QQ回调] 开始检查用户是否存在 - openid={openid[:16]}..., unionid={unionid[:16] if unionid else None}...")
            user_exists = await user_service.check_user_exists_by_qq_openid(openid=openid, unionid=unionid)
            logger.info(f"🔍 [QQ回调] 用户存在性检查结果 - user_exists={user_exists}, openid={openid[:16]}...")
            
            if not user_exists:
                # 用户不存在，重定向到 begin 页面并提示（和手机号登录保持一致）
                logger.info(f"⚠️ QQ用户未注册（来自login页面）: openid={openid[:16]}..., unionid={unionid[:16] if unionid else None}...")
                return RedirectResponse(
                    url=f"{settings.FRONTEND_URL}/begin?error=user_not_registered&oauth=qq"
                )
            
            # 用户存在，使用openid登录
            # 不传递QQ返回的nickname，避免覆盖用户原有用户名
            logger.info(f"✅ [QQ回调] 用户存在，开始登录 - openid={openid[:16]}...")
            user = await user_service.login_with_qq(
                openid=openid,
                username="",  # 不使用QQ返回的nickname
                unionid=unionid,
            )
            
            # 🔧 生成会话 token（前端需要 token 来调用 getUserInfo API）
            session_data = await _generate_session_token(user['user_id'])
            token = session_data['token']
            logger.info(f"✅ [QQ回调] 已生成会话 token - user_id={user['user_id']}, token={token[:16]}...")
            
            # 重定向到聊天界面（传递 token，避免前端调用 ensureSession 时使用错误的 token）
            logger.info(f"✅ [QQ回调] 登录成功，准备重定向到聊天界面 - user_id={user['user_id']}")
            frontend_url = f"{settings.FRONTEND_URL}/chatbot?oauth=qq&user_id={user['user_id']}&token={token}"
            return RedirectResponse(url=frontend_url)
        
        elif from_page == "first-login":
            # 从 first-login 页面来的，允许自动注册
            # 如果提供了 user_id，则更新现有用户；否则创建新用户
            # 不传递QQ返回的nickname，避免覆盖用户原有用户名
            user = await user_service.login_with_qq(
                openid=openid,
                username="",  # 不使用QQ返回的nickname
                unionid=unionid,
                user_id=user_id_from_state,
            )
            
            # 重定向到 first-login-reward（不传递username，避免使用QQ返回的nickname）
            frontend_url = f"{settings.FRONTEND_URL}/first-login-reward?oauth=qq&user_id={user['user_id']}"
            return RedirectResponse(url=frontend_url)
        
        else:
            # 其他来源（如 register 页面），允许自动注册
            # 使用openid登录/注册（login_with_qq 会自动创建新用户）
            # 不传递QQ返回的nickname，避免使用QQ昵称作为用户名
            user = await user_service.login_with_qq(
                openid=openid,
                username="",  # 不使用QQ返回的nickname
                unionid=unionid,
            )
            
            # 默认重定向到聊天界面（不传递username，由前端从数据库同步）
            frontend_url = f"{settings.FRONTEND_URL}/chatbot?oauth=qq&user_id={user['user_id']}"
            return RedirectResponse(url=frontend_url)
        
    except Exception as e:
        logger.error(f"QQ OAuth回调处理失败: {e}")
        return RedirectResponse(
            url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed"
        )


@app.get("/api/wechat/jsapi/config")
async def get_wechat_jsapi_config(
    url: str = Query(..., description="当前页面URL（用于生成签名）")
):
    """
    获取微信JS-SDK配置（包含签名）
    
    Args:
        url: 当前页面完整URL（前端传入，用于生成签名）
    
    Returns:
        {
            "success": true,
            "data": {
                "appId": "微信AppID",
                "timestamp": 1234567890,
                "nonceStr": "随机字符串",
                "signature": "SHA1签名"
            }
        }
    """
    try:
        if not settings.WECHAT_MP_APP_ID:
            logger.error("微信公众平台AppID未配置，请检查环境变量 WECHAT_MP_APP_ID")
            raise HTTPException(status_code=500, detail={"code": WECHAT_MP_NOT_CONFIGURED})
        
        # 记录接收到的URL（用于调试）
        logger.info(f"📥 收到微信JS-SDK配置请求: url={url[:100]}...")
        
        # 获取Redis客户端（用于缓存jsapi_ticket）
        from database import db_manager
        redis_getter = db_manager.get_redis
        
        # 创建微信JS-SDK服务
        jsapi_service = WechatJSAPIService(redis_getter=redis_getter)
        
        # 获取配置
        config = await jsapi_service.get_jsapi_config(url)
        
        if not config:
            logger.error("获取微信JS-SDK配置失败（返回None）")
            raise HTTPException(status_code=500, detail={"code": WECHAT_JSAPI_CONFIG_FAILED})
        
        logger.info(f"✅ 微信JS-SDK配置生成成功: appId={config.get('appId', 'N/A')[:10]}..., timestamp={config.get('timestamp')}")
        
        return {
            "success": True,
            "data": config
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取微信JS-SDK配置异常: {e}", exc_info=True)
        logger.exception("获取微信JS-SDK配置异常")
        raise HTTPException(status_code=500, detail={"code": GET_CONFIG_FAILED})


@app.get("/api/qq/share/config")
async def get_qq_share_config():
    """
    获取QQ分享配置（包含AppID和应用名称）
    
    Returns:
        {
            "success": true,
            "data": {
                "appId": "QQ AppID",
                "appName": "应用名称"
            }
        }
    """
    try:
        if not settings.QQ_APP_ID:
            logger.warning("QQ AppID未配置，返回空配置")
            return {
                "success": True,
                "data": {
                    "appId": "",
                    "appName": settings.QQ_APP_NAME
                }
            }
        
        logger.info(f"📥 收到QQ分享配置请求: appId={settings.QQ_APP_ID[:8]}..., appName={settings.QQ_APP_NAME}")
        
        return {
            "success": True,
            "data": {
                "appId": settings.QQ_APP_ID,
                "appName": settings.QQ_APP_NAME
            }
        }
    except Exception as e:
        logger.exception("获取QQ分享配置异常")
        raise HTTPException(status_code=500, detail={"code": GET_CONFIG_FAILED})


# ==================== 微信公众平台服务器配置 ====================

@app.get("/api/wechat/mp/verify")
async def wechat_mp_verify(
    signature: str = Query(..., description="微信加密签名"),
    timestamp: str = Query(..., description="时间戳"),
    nonce: str = Query(..., description="随机数"),
    echostr: str = Query(..., description="随机字符串")
):
    """
    微信公众平台服务器验证接口（GET）
    
    用于微信公众平台配置服务器URL时的验证
    
    Args:
        signature: 微信加密签名
        timestamp: 时间戳
        nonce: 随机数
        echostr: 随机字符串
    
    Returns:
        验证成功返回echostr，失败返回空字符串
    """
    if not settings.WECHAT_MP_TOKEN:
        logger.error("微信公众平台Token未配置")
        return PlainTextResponse("", status_code=500)
    
    try:
        # 1. 将token、timestamp、nonce按字典序排序
        token = settings.WECHAT_MP_TOKEN
        tmp_arr = [token, timestamp, nonce]
        tmp_arr.sort()
        
        # 2. 拼接字符串并SHA1加密
        tmp_str = ''.join(tmp_arr)
        tmp_str = hashlib.sha1(tmp_str.encode('utf-8')).hexdigest()
        
        # 3. 与signature比较
        if tmp_str == signature:
            logger.info("✅ 微信公众平台服务器验证成功")
            return PlainTextResponse(echostr)
        else:
            logger.warning(f"⚠️ 微信公众平台服务器验证失败: signature不匹配")
            return PlainTextResponse("", status_code=403)
    except Exception as e:
        logger.error(f"❌ 微信公众平台服务器验证异常: {e}")
        return PlainTextResponse("", status_code=500)


@app.post("/api/wechat/mp/message")
async def wechat_mp_message(request: Request):
    """
    微信公众平台消息推送接口（POST）
    
    接收微信公众平台推送的消息（文本、事件等）
    
    Returns:
        XML格式的响应消息
    """
    if not settings.WECHAT_MP_TOKEN:
        logger.error("微信公众平台Token未配置")
        return PlainTextResponse("", status_code=500)
    
    try:
        # 1. 获取请求参数（用于验证签名）
        signature = request.query_params.get("signature", "")
        timestamp = request.query_params.get("timestamp", "")
        nonce = request.query_params.get("nonce", "")
        
        # 2. 验证签名
        token = settings.WECHAT_MP_TOKEN
        tmp_arr = [token, timestamp, nonce]
        tmp_arr.sort()
        tmp_str = ''.join(tmp_arr)
        tmp_str = hashlib.sha1(tmp_str.encode('utf-8')).hexdigest()
        
        if tmp_str != signature:
            logger.warning(f"⚠️ 微信公众平台消息签名验证失败")
            return PlainTextResponse("", status_code=403)
        
        # 3. 读取XML消息体
        body = await request.body()
        xml_data = body.decode('utf-8')
        
        # 4. 解析XML
        root = ET.fromstring(xml_data)
        msg_type = root.find('MsgType').text if root.find('MsgType') is not None else ''
        from_user = root.find('FromUserName').text if root.find('FromUserName') is not None else ''
        to_user = root.find('ToUserName').text if root.find('ToUserName') is not None else ''
        
        logger.info(f"📨 收到微信公众平台消息: type={msg_type}, from={from_user}")
        
        # 5. 处理不同类型的消息
        if msg_type == 'text':
            # 文本消息
            content = root.find('Content').text if root.find('Content') is not None else ''
            logger.info(f"📝 文本消息内容: {content}")
            
            # 回复文本消息（示例：自动回复）
            reply_content = "感谢您的消息！"
            reply_xml = f"""<xml>
<ToUserName><![CDATA[{from_user}]]></ToUserName>
<FromUserName><![CDATA[{to_user}]]></FromUserName>
<CreateTime>{int(__import__('time').time())}</CreateTime>
<MsgType><![CDATA[text]]></MsgType>
<Content><![CDATA[{reply_content}]]></Content>
</xml>"""
            return PlainTextResponse(reply_xml, media_type="application/xml")
        
        elif msg_type == 'event':
            # 事件消息
            event = root.find('Event').text if root.find('Event') is not None else ''
            logger.info(f"🎯 事件消息: {event}")
            
            # 处理关注/取消关注事件
            if event == 'subscribe':
                reply_content = "欢迎关注！"
            elif event == 'unsubscribe':
                reply_content = "感谢您的关注！"
            else:
                reply_content = "收到事件消息"
            
            reply_xml = f"""<xml>
<ToUserName><![CDATA[{from_user}]]></ToUserName>
<FromUserName><![CDATA[{to_user}]]></FromUserName>
<CreateTime>{int(__import__('time').time())}</CreateTime>
<MsgType><![CDATA[text]]></MsgType>
<Content><![CDATA[{reply_content}]]></Content>
</xml>"""
            return PlainTextResponse(reply_xml, media_type="application/xml")
        
        else:
            # 其他类型消息，返回空响应
            logger.info(f"ℹ️ 未处理的消息类型: {msg_type}")
            return PlainTextResponse("success")
    
    except Exception as e:
        logger.error(f"❌ 处理微信公众平台消息异常: {e}")
        return PlainTextResponse("success")  # 返回success避免微信重试


@app.post("/api/admin/inbox/energy-gifts")
async def admin_create_energy_gift_inbox(payload: AdminCreateEnergyGiftInboxRequest):
    """创建能量石赠送站内信（指定用户）。"""
    user_ids = payload.user_ids
    now = datetime.utcnow()
    try:
        async with db_manager.get_session() as session:
            insert_message_sql = text(
                """
                INSERT INTO inbox_messages (
                    message_type, title, title_en, content, content_en, target_type, energy_stone_amount, created_by
                ) VALUES (
                    'energy_gift', :title_zh, :title_en, :content_zh, :content_en, 'users', :energy_stone_amount, 'web-admin'
                )
                """
            )
            result = await session.execute(
                insert_message_sql,
                {
                    "title_zh": payload.title_zh.strip(),
                    "title_en": payload.title_en.strip(),
                    "content_zh": payload.content_zh.strip(),
                    "content_en": payload.content_en.strip(),
                    "energy_stone_amount": payload.energy_stone_amount,
                },
            )
            message_id = int(result.lastrowid)

            insert_target_sql = text(
                """
                INSERT INTO inbox_message_targets (message_id, user_id, created_at)
                VALUES (:message_id, :user_id, :created_at)
                ON DUPLICATE KEY UPDATE created_at = VALUES(created_at)
                """
            )
            for uid in user_ids:
                await session.execute(
                    insert_target_sql,
                    {"message_id": message_id, "user_id": uid, "created_at": now},
                )
            await session.commit()

        return {"success": True, "data": {"message_id": message_id, "created_count": len(user_ids)}}
    except Exception:
        logger.exception("创建能量石站内信失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/admin/inbox/news")
async def admin_create_news_inbox(payload: AdminCreateNewsInboxRequest):
    """创建新闻通知站内信（创建即发送，全员）。"""
    try:
        async with db_manager.get_session() as session:
            insert_sql = text(
                """
                INSERT INTO inbox_messages (
                    message_type, title, title_en, content, content_en, target_type, energy_stone_amount, created_by
                ) VALUES (
                    'news_notice', :title_zh, :title_en, :content_zh, :content_en, 'all_users', 0, 'web-admin'
                )
                """
            )
            result = await session.execute(
                insert_sql,
                {
                    "title_zh": payload.title_zh.strip(),
                    "title_en": payload.title_en.strip(),
                    "content_zh": payload.content_zh.strip(),
                    "content_en": payload.content_en.strip(),
                },
            )
            await session.commit()
            return {"success": True, "data": {"message_id": int(result.lastrowid)}}
    except Exception:
        logger.exception("创建新闻站内信失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/admin/users")
async def admin_list_users(
    keyword: Optional[str] = Query(None, description="按 user_id/username/email/phone 模糊搜索"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
):
    """供 web-admin 选择用户列表使用。"""
    offset = (page - 1) * page_size
    kw = (keyword or "").strip()
    try:
        async with db_manager.get_session() as session:
            if kw:
                like_kw = f"%{kw}%"
                list_sql = text(
                    """
                    SELECT user_id, username, email, phone
                    FROM users
                    WHERE user_id LIKE :kw
                       OR username LIKE :kw
                       OR email LIKE :kw
                       OR phone LIKE :kw
                    ORDER BY id DESC
                    LIMIT :limit OFFSET :offset
                    """
                )
                count_sql = text(
                    """
                    SELECT COUNT(1)
                    FROM users
                    WHERE user_id LIKE :kw
                       OR username LIKE :kw
                       OR email LIKE :kw
                       OR phone LIKE :kw
                    """
                )
                params: Dict[str, Any] = {"kw": like_kw, "limit": page_size, "offset": offset}
            else:
                list_sql = text(
                    """
                    SELECT user_id, username, email, phone
                    FROM users
                    ORDER BY id DESC
                    LIMIT :limit OFFSET :offset
                    """
                )
                count_sql = text("SELECT COUNT(1) FROM users")
                params = {"limit": page_size, "offset": offset}

            rows = (await session.execute(list_sql, params)).mappings().all()
            total = int((await session.execute(count_sql, {"kw": f"%{kw}%"} if kw else {})).scalar_one())

            items = [
                {
                    "user_id": row["user_id"],
                    "username": row["username"],
                    "email": row["email"],
                    "phone": row["phone"],
                }
                for row in rows
            ]
            return {"success": True, "data": {"items": items, "total": total, "page": page, "page_size": page_size}}
    except Exception:
        logger.exception("查询管理端用户列表失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/admin/inbox/messages")
async def admin_list_inbox_messages(
    message_type: Optional[str] = Query(None, description="energy_gift/news_notice"),
    title_keyword: Optional[str] = Query(None, description="标题关键字"),
    start_date: Optional[str] = Query(None, description="开始日期，格式 YYYY-MM-DD"),
    end_date: Optional[str] = Query(None, description="结束日期，格式 YYYY-MM-DD"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
):
    """管理端站内信历史列表。"""
    if message_type and message_type not in {"energy_gift", "news_notice"}:
        raise HTTPException(status_code=400, detail="invalid message_type")

    offset = (page - 1) * page_size
    where_clauses: list[str] = []
    params: Dict[str, Any] = {"limit": page_size, "offset": offset}
    if message_type:
        where_clauses.append("m.message_type = :message_type")
        params["message_type"] = message_type
    if title_keyword and title_keyword.strip():
        where_clauses.append("(m.title LIKE :title_keyword OR COALESCE(m.title_en, '') LIKE :title_keyword)")
        params["title_keyword"] = f"%{title_keyword.strip()}%"
    if start_date and start_date.strip():
        where_clauses.append("DATE(m.created_at) >= :start_date")
        params["start_date"] = start_date.strip()
    if end_date and end_date.strip():
        where_clauses.append("DATE(m.created_at) <= :end_date")
        params["end_date"] = end_date.strip()
    where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
    try:
        async with db_manager.get_session() as session:
            list_sql = text(
                f"""
                SELECT
                    m.id,
                    m.message_type,
                    m.title,
                    COALESCE(m.title_en, '') AS title_en,
                    m.target_type,
                    m.energy_stone_amount,
                    m.created_at,
                    CASE
                        WHEN m.target_type = 'all_users' THEN (SELECT COUNT(1) FROM users)
                        ELSE COALESCE(t.target_count, 0)
                    END AS target_count,
                    COALESCE(s.read_count, 0) AS read_count,
                    COALESCE(s.claim_count, 0) AS claim_count
                FROM inbox_messages m
                LEFT JOIN (
                    SELECT message_id, COUNT(1) AS target_count
                    FROM inbox_message_targets
                    GROUP BY message_id
                ) t ON m.id = t.message_id
                LEFT JOIN (
                    SELECT
                        message_id,
                        SUM(CASE WHEN read_status = 1 THEN 1 ELSE 0 END) AS read_count,
                        SUM(CASE WHEN claim_status = 1 THEN 1 ELSE 0 END) AS claim_count
                    FROM inbox_user_message_states
                    GROUP BY message_id
                ) s ON m.id = s.message_id
                {where_sql}
                ORDER BY m.created_at DESC, m.id DESC
                LIMIT :limit OFFSET :offset
                """
            )
            count_sql = text(
                f"""
                SELECT COUNT(1)
                FROM inbox_messages m
                {where_sql}
                """
            )

            rows = (await session.execute(list_sql, params)).mappings().all()
            count_params = {k: v for k, v in params.items() if k not in {"limit", "offset"}}
            total = int((await session.execute(count_sql, count_params)).scalar_one())
            items = [
                {
                    "id": int(row["id"]),
                    "message_type": row["message_type"],
                    "title": row["title"],
                    "title_zh": row["title"],
                    "title_en": row["title_en"],
                    "target_type": row["target_type"],
                    "energy_stone_amount": int(row["energy_stone_amount"] or 0),
                    "target_count": int(row["target_count"] or 0),
                    "read_count": int(row["read_count"] or 0),
                    "claim_count": int(row["claim_count"] or 0),
                    "created_at": row["created_at"].isoformat() if row["created_at"] else None,
                }
                for row in rows
            ]
            return {"success": True, "data": {"items": items, "total": total, "page": page, "page_size": page_size}}
    except Exception:
        logger.exception("查询管理端站内信历史失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/admin/inbox/messages/{message_id}")
async def admin_get_inbox_message_detail(
    message_id: int,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
):
    """管理端站内信详情（含用户阅读/领取状态）。"""
    offset = (page - 1) * page_size
    try:
        async with db_manager.get_session() as session:
            message_sql = text(
                """
                SELECT id, message_type, title, COALESCE(title_en, '') AS title_en, content, COALESCE(content_en, '') AS content_en, target_type, energy_stone_amount, created_at
                FROM inbox_messages
                WHERE id = :message_id
                LIMIT 1
                """
            )
            message_row = (await session.execute(message_sql, {"message_id": message_id})).mappings().first()
            if not message_row:
                raise HTTPException(status_code=404, detail="message not found")

            summary_sql = text(
                """
                SELECT
                    COALESCE(SUM(CASE WHEN read_status = 1 THEN 1 ELSE 0 END), 0) AS read_count,
                    COALESCE(SUM(CASE WHEN claim_status = 1 THEN 1 ELSE 0 END), 0) AS claim_count
                FROM inbox_user_message_states
                WHERE message_id = :message_id
                """
            )
            summary_row = (await session.execute(summary_sql, {"message_id": message_id})).mappings().first()
            read_count = int(summary_row["read_count"] or 0) if summary_row else 0
            claim_count = int(summary_row["claim_count"] or 0) if summary_row else 0

            if message_row["target_type"] == "all_users":
                target_count = int((await session.execute(text("SELECT COUNT(1) FROM users"))).scalar_one())
                state_list_sql = text(
                    """
                    SELECT
                        s.user_id,
                        u.username,
                        u.phone,
                        u.email,
                        s.read_status,
                        s.claim_status,
                        s.deleted_status
                    FROM inbox_user_message_states s
                    LEFT JOIN users u ON s.user_id = u.user_id
                    WHERE s.message_id = :message_id
                    ORDER BY s.updated_at DESC, s.id DESC
                    LIMIT :limit OFFSET :offset
                    """
                )
                state_count_sql = text(
                    "SELECT COUNT(1) FROM inbox_user_message_states WHERE message_id = :message_id"
                )
                state_rows = (
                    await session.execute(
                        state_list_sql, {"message_id": message_id, "limit": page_size, "offset": offset}
                    )
                ).mappings().all()
                state_total = int((await session.execute(state_count_sql, {"message_id": message_id})).scalar_one())
            else:
                target_count_sql = text(
                    "SELECT COUNT(1) FROM inbox_message_targets WHERE message_id = :message_id"
                )
                target_count = int((await session.execute(target_count_sql, {"message_id": message_id})).scalar_one())
                state_list_sql = text(
                    """
                    SELECT
                        t.user_id,
                        u.username,
                        u.phone,
                        u.email,
                        COALESCE(s.read_status, 0) AS read_status,
                        COALESCE(s.claim_status, 0) AS claim_status,
                        COALESCE(s.deleted_status, 0) AS deleted_status
                    FROM inbox_message_targets t
                    LEFT JOIN users u ON t.user_id = u.user_id
                    LEFT JOIN inbox_user_message_states s
                        ON s.message_id = t.message_id AND s.user_id = t.user_id
                    WHERE t.message_id = :message_id
                    ORDER BY t.id DESC
                    LIMIT :limit OFFSET :offset
                    """
                )
                state_count_sql = text("SELECT COUNT(1) FROM inbox_message_targets WHERE message_id = :message_id")
                state_rows = (
                    await session.execute(
                        state_list_sql, {"message_id": message_id, "limit": page_size, "offset": offset}
                    )
                ).mappings().all()
                state_total = int((await session.execute(state_count_sql, {"message_id": message_id})).scalar_one())

            state_items = [
                {
                    "user_id": row["user_id"],
                    "username": row["username"],
                    "phone": row["phone"],
                    "email": row["email"],
                    "read_status": int(row["read_status"] or 0),
                    "claim_status": int(row["claim_status"] or 0),
                    "deleted_status": int(row["deleted_status"] or 0),
                }
                for row in state_rows
            ]

            return {
                "success": True,
                "data": {
                    "message": {
                        "id": int(message_row["id"]),
                        "message_type": message_row["message_type"],
                        "title": message_row["title"],
                        "title_zh": message_row["title"],
                        "title_en": message_row["title_en"],
                        "content": message_row["content"],
                        "content_zh": message_row["content"],
                        "content_en": message_row["content_en"],
                        "target_type": message_row["target_type"],
                        "energy_stone_amount": int(message_row["energy_stone_amount"] or 0),
                        "created_at": message_row["created_at"].isoformat() if message_row["created_at"] else None,
                        "target_count": target_count,
                        "read_count": read_count,
                        "claim_count": claim_count,
                    },
                    "states": {
                        "items": state_items,
                        "total": state_total,
                        "page": page,
                        "page_size": page_size,
                    },
                },
            }
    except HTTPException:
        raise
    except Exception:
        logger.exception("查询管理端站内信详情失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.delete("/api/admin/inbox/messages/{message_id}")
async def admin_delete_inbox_message(message_id: int):
    """管理端删除站内信（物理删除消息与关联状态）。"""
    try:
        async with db_manager.get_session() as session:
            exists_sql = text("SELECT id FROM inbox_messages WHERE id = :message_id LIMIT 1")
            exists_row = (await session.execute(exists_sql, {"message_id": message_id})).mappings().first()
            if not exists_row:
                raise HTTPException(status_code=404, detail="message not found")

            await session.execute(
                text("DELETE FROM inbox_user_message_states WHERE message_id = :message_id"),
                {"message_id": message_id},
            )
            await session.execute(
                text("DELETE FROM inbox_message_targets WHERE message_id = :message_id"),
                {"message_id": message_id},
            )
            await session.execute(
                text("DELETE FROM inbox_messages WHERE id = :message_id"),
                {"message_id": message_id},
            )
            await session.commit()
            return {"success": True}
    except HTTPException:
        raise
    except Exception:
        logger.exception("管理端删除站内信失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/inbox/messages")
async def get_user_inbox_messages(
    user_id: str,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    message_type: Optional[str] = Query(None, description="energy_gift/news_notice"),
    _: None = Depends(require_user_session),
):
    """查询用户站内信列表（仅返回当前用户未删除消息）。"""
    if message_type and message_type not in {"energy_gift", "news_notice"}:
        raise HTTPException(status_code=400, detail="invalid message_type")

    offset = (page - 1) * page_size
    try:
        async with db_manager.get_session() as session:
            where_type_clause = "AND m.message_type = :message_type" if message_type else ""
            list_sql = text(
                f"""
                SELECT
                    m.id,
                    m.message_type,
                    m.title,
                    COALESCE(m.title_en, '') AS title_en,
                    m.energy_stone_amount,
                    m.created_at,
                    COALESCE(s.read_status, 0) AS read_status,
                    COALESCE(s.claim_status, 0) AS claim_status
                FROM inbox_messages m
                LEFT JOIN inbox_message_targets t
                    ON m.id = t.message_id AND t.user_id = :user_id
                LEFT JOIN inbox_user_message_states s
                    ON m.id = s.message_id AND s.user_id = :user_id
                WHERE
                    (m.target_type = 'all_users' OR t.user_id IS NOT NULL)
                    AND COALESCE(s.deleted_status, 0) = 0
                    {where_type_clause}
                ORDER BY m.created_at DESC, m.id DESC
                LIMIT :limit OFFSET :offset
                """
            )
            count_sql = text(
                f"""
                SELECT COUNT(1) AS total
                FROM inbox_messages m
                LEFT JOIN inbox_message_targets t
                    ON m.id = t.message_id AND t.user_id = :user_id
                LEFT JOIN inbox_user_message_states s
                    ON m.id = s.message_id AND s.user_id = :user_id
                WHERE
                    (m.target_type = 'all_users' OR t.user_id IS NOT NULL)
                    AND COALESCE(s.deleted_status, 0) = 0
                    {where_type_clause}
                """
            )
            params: Dict[str, Any] = {"user_id": user_id, "limit": page_size, "offset": offset}
            if message_type:
                params["message_type"] = message_type

            rows = (await session.execute(list_sql, params)).mappings().all()
            total = int((await session.execute(count_sql, params)).scalar_one())

            items = []
            for row in rows:
                items.append(
                    {
                        "id": int(row["id"]),
                        "type": row["message_type"],
                        "title": row["title"],
                        "title_zh": row["title"],
                        "title_en": row["title_en"],
                        "energy_stone_amount": int(row["energy_stone_amount"] or 0),
                        "read_status": int(row["read_status"]),
                        "claim_status": int(row["claim_status"]),
                        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
                    }
                )
            return {"success": True, "data": {"items": items, "total": total, "page": page, "page_size": page_size}}
    except Exception:
        logger.exception("查询站内信列表失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.get("/api/users/{user_id}/inbox/messages/{message_id}")
async def get_user_inbox_message_detail(
    user_id: str,
    message_id: int,
    mark_read: bool = Query(True),
    _: None = Depends(require_user_session),
):
    """查询站内信详情，默认标记为已读。"""
    try:
        async with db_manager.get_session() as session:
            detail_sql = text(
                """
                SELECT
                    m.id,
                    m.message_type,
                    m.title,
                    COALESCE(m.title_en, '') AS title_en,
                    m.content,
                    COALESCE(m.content_en, '') AS content_en,
                    m.energy_stone_amount,
                    m.created_at,
                    COALESCE(s.read_status, 0) AS read_status,
                    COALESCE(s.deleted_status, 0) AS deleted_status,
                    COALESCE(s.claim_status, 0) AS claim_status
                FROM inbox_messages m
                LEFT JOIN inbox_message_targets t
                    ON m.id = t.message_id AND t.user_id = :user_id
                LEFT JOIN inbox_user_message_states s
                    ON m.id = s.message_id AND s.user_id = :user_id
                WHERE
                    m.id = :message_id
                    AND (m.target_type = 'all_users' OR t.user_id IS NOT NULL)
                LIMIT 1
                """
            )
            row = (await session.execute(detail_sql, {"user_id": user_id, "message_id": message_id})).mappings().first()
            if not row or int(row["deleted_status"]) == 1:
                raise HTTPException(status_code=404, detail="message not found")

            if mark_read and int(row["read_status"]) == 0:
                upsert_read_sql = text(
                    """
                    INSERT INTO inbox_user_message_states (
                        message_id, user_id, read_status, read_at
                    ) VALUES (
                        :message_id, :user_id, 1, NOW()
                    )
                    ON DUPLICATE KEY UPDATE
                        read_status = 1,
                        read_at = IFNULL(read_at, NOW())
                    """
                )
                await session.execute(upsert_read_sql, {"message_id": message_id, "user_id": user_id})
                await session.commit()

            return {
                "success": True,
                "data": {
                    "id": int(row["id"]),
                    "type": row["message_type"],
                    "title": row["title"],
                    "title_zh": row["title"],
                    "title_en": row["title_en"],
                    "content": row["content"],
                    "content_zh": row["content"],
                    "content_en": row["content_en"],
                    "energy_stone_amount": int(row["energy_stone_amount"] or 0),
                    "read_status": 1 if mark_read else int(row["read_status"]),
                    "claim_status": int(row["claim_status"]),
                    "created_at": row["created_at"].isoformat() if row["created_at"] else None,
                },
            }
    except HTTPException:
        raise
    except Exception:
        logger.exception("查询站内信详情失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/inbox/messages/{message_id}/read")
async def mark_user_inbox_message_read(
    user_id: str, message_id: int, _: None = Depends(require_user_session)
):
    """显式标记消息已读。"""
    try:
        async with db_manager.get_session() as session:
            upsert_sql = text(
                """
                INSERT INTO inbox_user_message_states (message_id, user_id, read_status, read_at)
                VALUES (:message_id, :user_id, 1, NOW())
                ON DUPLICATE KEY UPDATE
                    read_status = 1,
                    read_at = IFNULL(read_at, NOW())
                """
            )
            await session.execute(upsert_sql, {"message_id": message_id, "user_id": user_id})
            await session.commit()
            return {"success": True}
    except Exception:
        logger.exception("标记站内信已读失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.post("/api/users/{user_id}/inbox/messages/{message_id}/claim")
async def claim_user_inbox_energy_stones(
    user_id: str, message_id: int, _: None = Depends(require_user_session)
):
    """领取能量石站内信奖励（幂等：每条仅可领取一次）。"""
    try:
        async with db_manager.get_session() as session:
            message_sql = text(
                """
                SELECT
                    m.id,
                    m.message_type,
                    m.energy_stone_amount
                FROM inbox_messages m
                LEFT JOIN inbox_message_targets t
                    ON m.id = t.message_id AND t.user_id = :user_id
                WHERE
                    m.id = :message_id
                    AND (m.target_type = 'all_users' OR t.user_id IS NOT NULL)
                LIMIT 1
                """
            )
            message_row = (
                await session.execute(message_sql, {"user_id": user_id, "message_id": message_id})
            ).mappings().first()
            if not message_row:
                raise HTTPException(status_code=404, detail="message not found")
            if message_row["message_type"] != "energy_gift":
                raise HTTPException(status_code=400, detail="message is not energy gift")

            state_sql = text(
                """
                SELECT id, claim_status
                FROM inbox_user_message_states
                WHERE message_id = :message_id AND user_id = :user_id
                FOR UPDATE
                """
            )
            state_row = (
                await session.execute(state_sql, {"message_id": message_id, "user_id": user_id})
            ).mappings().first()
            if state_row and int(state_row["claim_status"]) == 1:
                raise HTTPException(status_code=409, detail="already claimed")

            amount = int(message_row["energy_stone_amount"] or 0)
            if amount <= 0:
                raise HTTPException(status_code=400, detail="invalid energy amount")

            update_user_sql = text(
                """
                UPDATE users
                SET conversation_energy_stone_total = conversation_energy_stone_total + :amount
                WHERE user_id = :user_id
                """
            )
            update_result = await session.execute(update_user_sql, {"amount": amount, "user_id": user_id})
            if int(update_result.rowcount or 0) <= 0:
                raise HTTPException(status_code=404, detail="user not found")

            if state_row:
                update_state_sql = text(
                    """
                    UPDATE inbox_user_message_states
                    SET claim_status = 1,
                        claimed_energy_stones = :amount,
                        claimed_at = NOW(),
                        read_status = 1,
                        read_at = IFNULL(read_at, NOW()),
                        updated_at = NOW()
                    WHERE message_id = :message_id AND user_id = :user_id
                    """
                )
                await session.execute(update_state_sql, {"amount": amount, "message_id": message_id, "user_id": user_id})
            else:
                insert_state_sql = text(
                    """
                    INSERT INTO inbox_user_message_states (
                        message_id, user_id, read_status, claim_status,
                        claimed_energy_stones, read_at, claimed_at
                    ) VALUES (
                        :message_id, :user_id, 1, 1,
                        :amount, NOW(), NOW()
                    )
                    """
                )
                await session.execute(insert_state_sql, {"message_id": message_id, "user_id": user_id, "amount": amount})

            total_sql = text("SELECT conversation_energy_stone_total FROM users WHERE user_id = :user_id LIMIT 1")
            total_count = int((await session.execute(total_sql, {"user_id": user_id})).scalar_one())
            await session.commit()
            return {"success": True, "data": {"claimed_energy_stones": amount, "user_total_energy_stones": total_count}}
    except HTTPException:
        raise
    except Exception:
        logger.exception("领取站内信能量石失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.delete("/api/users/{user_id}/inbox/messages/{message_id}")
async def delete_user_inbox_message(
    user_id: str, message_id: int, _: None = Depends(require_user_session)
):
    """删除新闻通知（用户维度软删除）。"""
    try:
        async with db_manager.get_session() as session:
            message_sql = text(
                """
                SELECT
                    m.id,
                    m.message_type
                FROM inbox_messages m
                LEFT JOIN inbox_message_targets t
                    ON m.id = t.message_id AND t.user_id = :user_id
                WHERE
                    m.id = :message_id
                    AND (m.target_type = 'all_users' OR t.user_id IS NOT NULL)
                LIMIT 1
                """
            )
            message_row = (
                await session.execute(message_sql, {"user_id": user_id, "message_id": message_id})
            ).mappings().first()
            if not message_row:
                raise HTTPException(status_code=404, detail="message not found")
            if message_row["message_type"] != "news_notice":
                raise HTTPException(status_code=400, detail="energy gift message cannot be deleted")

            upsert_sql = text(
                """
                INSERT INTO inbox_user_message_states (
                    message_id, user_id, deleted_status, deleted_at
                ) VALUES (
                    :message_id, :user_id, 1, NOW()
                )
                ON DUPLICATE KEY UPDATE
                    deleted_status = 1,
                    deleted_at = NOW()
                """
            )
            await session.execute(upsert_sql, {"message_id": message_id, "user_id": user_id})
            await session.commit()
            return {"success": True}
    except HTTPException:
        raise
    except Exception:
        logger.exception("删除站内信失败")
        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket端点"""
    # 从查询参数获取user_id（Lyrria_user_id）与会话token
    user_id = websocket.query_params.get("user_id")
    token = websocket.query_params.get("token")
    
    # 记录连接信息
    import logging
    logger = logging.getLogger(__name__)
    logger.info(
        f"🔌 WebSocket连接: user_id={user_id}, query_params={dict(websocket.query_params)}"
    )
    
    # 🔥 传入Redis缓存服务、user_id、session_store
    redis_cache = getattr(app.state, 'redis_cache', None)
    session_store = getattr(app.state, "session_store", None)
    handler = WebSocketHandler(
        websocket,
        user_id=user_id,
        token=token,
        redis_cache=redis_cache,
        session_store=session_store,
    )
    await handler.handle_connection()


if __name__ == "__main__":
    import uvicorn
    import os
    import socket
    from pathlib import Path
    
    # 🔥 DNS预解析（减少首次连接延迟）
    dashscope_domain = get_dashscope_domain()
    try:
        socket.getaddrinfo(dashscope_domain, 443)
    except Exception as e:
        pass
    
    # SSL证书路径配置
    # 优先使用环境变量配置的证书路径（生产环境）
    ssl_keyfile = settings.SSL_KEYFILE
    ssl_certfile = settings.SSL_CERTFILE
    
    # 如果环境变量未配置，使用本地开发证书
    if not ssl_keyfile or not ssl_certfile:
        ssl_dir = Path(__file__).parent.parent / "ssl"
        ssl_keyfile = str(ssl_dir / "key.pem")
        ssl_certfile = str(ssl_dir / "cert.pem")
    
    # 检查证书是否存在
    if ssl_keyfile and ssl_certfile:
        if not os.path.exists(ssl_keyfile) or not os.path.exists(ssl_certfile):
            ssl_keyfile = None
            ssl_certfile = None
    
    uvicorn.run(
        "main:app",
        host=settings.host,
        port=settings.port,
        reload=settings.DEBUG,
        ssl_keyfile=ssl_keyfile,
        ssl_certfile=ssl_certfile,
    )