| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672 |
- """
- 用户服务模块(SQLAlchemy ORM 版本)
- """
- import uuid
- import random
- import asyncio
- import logging
- from typing import Dict, Optional, Any
- from datetime import datetime
- from sqlalchemy import select, update, func, insert
- from sqlalchemy.ext.asyncio import AsyncSession
- from database import db_manager
- from models_db import User, UserUIState
- logger = logging.getLogger(__name__)
- class UserService:
- """用户管理服务"""
-
- async def _generate_temp_username(self, session: AsyncSession) -> str:
- """
- 生成临时用户名(格式:User + 7位随机数字)
- 确保用户名不重复
-
- Args:
- session: 数据库会话
-
- Returns:
- 临时用户名
- """
- max_attempts = 100 # 最大重试次数,避免无限循环
-
- for attempt in range(max_attempts):
- # 生成7位随机数字(1000000-9999999)
- random_number = random.randint(1000000, 9999999)
- temp_username = f"User{random_number}"
-
- # 检查用户名是否已存在
- result = await session.execute(
- select(User).where(User.username == temp_username)
- )
- existing_user = result.scalar_one_or_none()
-
- if not existing_user:
- logger.debug(f"生成临时用户名: {temp_username} (尝试次数: {attempt + 1})")
- return temp_username
-
- # 如果100次都重复,使用时间戳的后7位作为备选方案
- # 使用当前时间戳的毫秒数取模,确保是7位数字
- for fallback_attempt in range(10): # 备选方案也最多尝试10次
- timestamp_suffix = int(datetime.now().timestamp() * 1000) % 10000000
- # 确保至少是7位数字(如果不足7位,前面补0)
- fallback_number = str(timestamp_suffix).zfill(7)
- temp_username = f"User{fallback_number}"
-
- # 再次检查是否重复
- result = await session.execute(
- select(User).where(User.username == temp_username)
- )
- existing_user = result.scalar_one_or_none()
-
- if not existing_user:
- logger.warning(f"⚠️ 临时用户名生成达到最大重试次数,使用备选方案: {temp_username}")
- return temp_username
-
- # 如果还是重复,等待1毫秒后重试(确保时间戳变化)
- await asyncio.sleep(0.001)
-
- # 最后的备选方案:使用UUID的十六进制转数字(取后7位)
- uuid_hex = uuid.uuid4().hex
- final_number = str(int(uuid_hex[:8], 16) % 10000000).zfill(7)
- temp_username = f"User{final_number}"
- logger.error(f"❌ 临时用户名生成异常,使用最终备选方案: {temp_username}")
- return temp_username
-
- async def create_guest_user(self, username: str = "", bot_name: Optional[str] = None) -> Dict:
- """
- 创建游客用户
-
- Args:
- username: 用户名(默认为空字符串,为空时自动生成临时用户名)
- bot_name: 用户给机器人取的名字(可选)
-
- Returns:
- 用户信息字典(包含user_id)
- """
- try:
- # 生成唯一用户ID
- user_id = str(uuid.uuid4())
-
- # 使用 ORM 创建用户
- async with db_manager.get_session() as session:
- # 如果用户名为空,生成临时用户名
- if not username or username.strip() == "":
- username = await self._generate_temp_username(session)
- logger.info(f"🔧 自动生成临时用户名: {username}")
-
- user = User(
- user_id=user_id,
- username=username,
- user_type="guest",
- bot_name=bot_name,
- )
- session.add(user)
- await session.commit()
- await session.refresh(user)
-
- logger.info(f"✅ 创建游客用户成功: user_id={user_id}, username={username}, bot_name={bot_name}")
-
- return {
- "user_id": user_id,
- "username": username,
- "user_type": "guest",
- "bot_name": bot_name,
- "created_at": datetime.now().isoformat()
- }
-
- except Exception as e:
- logger.error(f"❌ 创建用户失败: {e}")
- raise
-
- async def get_user(self, user_id: str) -> Optional[Dict]:
- """
- 获取用户信息
-
- Args:
- user_id: 用户ID
-
- Returns:
- 用户信息字典,如果不存在返回None
- """
- try:
- async with db_manager.get_session() as session:
- result = await session.execute(
- select(User).where(User.user_id == user_id)
- )
- user = result.scalar_one_or_none()
-
- if user:
- return user.to_dict()
- return None
-
- except Exception as e:
- logger.error(f"❌ 获取用户信息失败: user_id={user_id}, error={e}")
- return None
-
- async def update_user_last_active(self, user_id: str):
- """
- 更新用户最后活跃时间
-
- Args:
- user_id: 用户ID
- """
- try:
- async with db_manager.get_session() as session:
- await session.execute(
- update(User)
- .where(User.user_id == user_id)
- .values(last_active_at=func.now())
- )
- await session.commit()
- except Exception as e:
- logger.warning(f"⚠️ 更新用户活跃时间失败: user_id={user_id}, error={e}")
-
- async def update_username(self, user_id: str, username: str) -> bool:
- """
- 更新用户名
-
- Args:
- user_id: 用户ID
- username: 用户名
-
- Returns:
- 是否更新成功
- """
- try:
- async with db_manager.get_session() as session:
- result = await session.execute(
- update(User)
- .where(User.user_id == user_id)
- .values(username=username)
- )
- await session.commit()
-
- if result.rowcount > 0:
- logger.info(f"✅ 更新用户名成功: user_id={user_id}, username={username}")
- return True
- return False
- except Exception as e:
- logger.error(f"❌ 更新用户名失败: user_id={user_id}, error={e}")
- return False
-
- async def update_bot_name(self, user_id: str, bot_name: str) -> bool:
- """
- 更新用户给机器人取的名字
-
- Args:
- user_id: 用户ID
- bot_name: 机器人名字
-
- Returns:
- 是否更新成功
- """
- import time
- start_time = time.time()
-
- try:
- async with db_manager.get_session() as session:
- # 🔥 先查询当前值,避免不必要的更新
- query_start = time.time()
- result = await session.execute(
- select(User.bot_name).where(User.user_id == user_id)
- )
- current_bot_name = result.scalar_one_or_none()
- query_duration = (time.time() - query_start) * 1000
-
- # 如果值相同,直接返回(避免不必要的 UPDATE)
- if current_bot_name == bot_name:
- total_duration = (time.time() - start_time) * 1000
- logger.info(
- f"⏭️ 机器人名字未变化,跳过更新: user_id={user_id}, "
- f"bot_name={bot_name}, 耗时={total_duration:.1f}ms"
- )
- return True
-
- # 执行更新
- update_start = time.time()
- update_result = await session.execute(
- update(User)
- .where(User.user_id == user_id)
- .values(bot_name=bot_name)
- )
- await session.commit()
- update_duration = (time.time() - update_start) * 1000
- total_duration = (time.time() - start_time) * 1000
-
- if update_result.rowcount > 0:
- logger.info(
- f"✅ 更新机器人名字成功: user_id={user_id}, "
- f"bot_name={bot_name}, "
- f"查询耗时={query_duration:.1f}ms, "
- f"更新耗时={update_duration:.1f}ms, "
- f"总耗时={total_duration:.1f}ms"
- )
- return True
- else:
- logger.warning(
- f"⚠️ 更新机器人名字失败: 未找到用户 user_id={user_id}, "
- f"总耗时={total_duration:.1f}ms"
- )
- return False
- except asyncio.TimeoutError:
- total_duration = (time.time() - start_time) * 1000
- logger.error(
- f"❌ 更新机器人名字超时: user_id={user_id}, "
- f"耗时={total_duration:.1f}ms"
- )
- return False
- except Exception as e:
- total_duration = (time.time() - start_time) * 1000
- logger.error(
- f"❌ 更新机器人名字失败: user_id={user_id}, "
- f"error={e}, 耗时={total_duration:.1f}ms"
- )
- return False
-
- async def update_speech_duration(self, user_id: str, duration_ms: int) -> Optional[int]:
- """
- 更新用户累积说话时长
-
- Args:
- user_id: 用户ID
- duration_ms: 本次说话时长(毫秒)
-
- Returns:
- 是否更新成功
- """
- try:
- if duration_ms <= 0:
- return None # 时长无效,跳过更新
- async with db_manager.get_session() as session:
- result = await session.execute(
- select(User.total_speech_duration_ms)
- .where(User.user_id == user_id)
- .with_for_update()
- )
- current_total = result.scalar_one_or_none() or 0
- new_total = current_total + duration_ms
- await session.execute(
- update(User)
- .where(User.user_id == user_id)
- .values(total_speech_duration_ms=new_total)
- )
- await session.commit()
- logger.info(
- f"✅ 更新用户说话时长成功: user_id={user_id}, "
- f"duration_ms={duration_ms}, total_ms={new_total}"
- )
- return new_total
- except Exception as e:
- logger.error(f"❌ 更新用户说话时长失败: user_id={user_id}, error={e}")
- return None
- async def get_user_ui_state(self, user_id: str) -> Optional[Dict[str, Any]]:
- """
- 获取用户前端可视化状态
- """
- try:
- async with db_manager.get_session() as session:
- result = await session.execute(
- select(UserUIState).where(UserUIState.user_id == user_id)
- )
- state = result.scalar_one_or_none()
- if state is None:
- return None
- return state.to_dict()
- except Exception as e: # noqa: BLE001
- logger.error(f"❌ 获取用户UI状态失败: user_id={user_id}, error={e}")
- return None
- async def upsert_user_ui_state(
- self,
- user_id: str,
- *,
- current_page: Optional[str] = None,
- visited_routes: Optional[list[str]] = None,
- has_shown_tap_to_chat: Optional[bool] = None,
- has_shown_cancel_tooltip: Optional[bool] = None,
- has_shown_send_tooltip: Optional[bool] = None,
- has_shown_progress_bar_tooltip: Optional[bool] = None,
- avatar_c_unlocked: Optional[bool] = None,
- ) -> Optional[Dict[str, Any]]:
- """
- 创建或更新用户前端可视化状态(部分字段更新)
- """
- try:
- async with db_manager.get_session() as session:
- result = await session.execute(
- select(UserUIState).where(UserUIState.user_id == user_id).with_for_update()
- )
- state: Optional[UserUIState] = result.scalar_one_or_none()
- if state is None:
- # 新建记录,仅写入非空字段
- data: Dict[str, Any] = {"user_id": user_id}
- if current_page is not None:
- data["current_page"] = current_page
- if visited_routes is not None:
- data["visited_routes"] = visited_routes
- if has_shown_tap_to_chat is not None:
- data["has_shown_tap_to_chat"] = has_shown_tap_to_chat
- if has_shown_cancel_tooltip is not None:
- data["has_shown_cancel_tooltip"] = has_shown_cancel_tooltip
- if has_shown_send_tooltip is not None:
- data["has_shown_send_tooltip"] = has_shown_send_tooltip
- if has_shown_progress_bar_tooltip is not None:
- data["has_shown_progress_bar_tooltip"] = has_shown_progress_bar_tooltip
- if avatar_c_unlocked is not None:
- data["avatar_c_unlocked"] = avatar_c_unlocked
- stmt = insert(UserUIState).values(**data)
- await session.execute(stmt)
- else:
- # 更新现有记录(仅覆盖非空字段)
- if current_page is not None:
- state.current_page = current_page
- if visited_routes is not None:
- state.visited_routes = visited_routes
- if has_shown_tap_to_chat is not None:
- state.has_shown_tap_to_chat = has_shown_tap_to_chat
- if has_shown_cancel_tooltip is not None:
- state.has_shown_cancel_tooltip = has_shown_cancel_tooltip
- if has_shown_send_tooltip is not None:
- state.has_shown_send_tooltip = has_shown_send_tooltip
- if has_shown_progress_bar_tooltip is not None:
- state.has_shown_progress_bar_tooltip = has_shown_progress_bar_tooltip
- if avatar_c_unlocked is not None:
- state.avatar_c_unlocked = avatar_c_unlocked
- session.add(state)
- await session.commit()
- # 重新读取最新状态
- result = await session.execute(
- select(UserUIState).where(UserUIState.user_id == user_id)
- )
- latest_state = result.scalar_one_or_none()
- if latest_state is None:
- return None
- return latest_state.to_dict()
- except Exception as e: # noqa: BLE001
- logger.error(f"❌ 更新用户UI状态失败: user_id={user_id}, error={e}")
- return None
-
- async def register_with_phone(self, phone: str, username: str = "") -> Dict:
- """
- 使用手机号注册
-
- Args:
- phone: 手机号
- username: 用户名(可选)
-
- Returns:
- 用户信息字典
- """
- try:
- async with db_manager.get_session() as session:
- # 检查手机号是否已注册
- result = await session.execute(
- select(User).where(User.phone == phone)
- )
- existing_user = result.scalar_one_or_none()
-
- if existing_user:
- raise ValueError("该手机号已被注册")
-
- # 生成用户ID和用户名
- user_id = str(uuid.uuid4())
- if not username or username.strip() == "":
- username = await self._generate_temp_username(session)
-
- user = User(
- user_id=user_id,
- username=username,
- phone=phone,
- user_type="registered",
- is_verified=True,
- registered_at=datetime.now(),
- )
- session.add(user)
- await session.commit()
- await session.refresh(user)
-
- logger.info(f"✅ 手机号注册成功: user_id={user_id}, phone={phone}")
- return user.to_dict()
- except Exception as e:
- logger.error(f"❌ 手机号注册失败: {e}")
- raise
-
- async def register_with_email(self, email: str, username: str = "") -> Dict:
- """
- 使用邮箱注册
-
- Args:
- email: 邮箱
- username: 用户名(可选)
-
- Returns:
- 用户信息字典
- """
- try:
- async with db_manager.get_session() as session:
- # 检查邮箱是否已注册
- result = await session.execute(
- select(User).where(User.email == email)
- )
- existing_user = result.scalar_one_or_none()
-
- if existing_user:
- raise ValueError("该邮箱已被注册")
-
- # 生成用户ID和用户名
- user_id = str(uuid.uuid4())
- if not username or username.strip() == "":
- username = await self._generate_temp_username(session)
-
- user = User(
- user_id=user_id,
- username=username,
- email=email,
- user_type="registered",
- is_verified=True,
- registered_at=datetime.now(),
- )
- session.add(user)
- await session.commit()
- await session.refresh(user)
-
- logger.info(f"✅ 邮箱注册成功: user_id={user_id}, email={email}")
- return user.to_dict()
- except Exception as e:
- logger.error(f"❌ 邮箱注册失败: {e}")
- raise
-
- async def login_with_wechat(self, openid: str, username: str = "") -> Dict:
- """
- 使用微信OpenID登录/注册
-
- Args:
- openid: 微信OpenID
- username: 用户名(可选,仅新用户时使用)
-
- Returns:
- 用户信息字典
- """
- try:
- async with db_manager.get_session() as session:
- # 检查是否已存在
- result = await session.execute(
- select(User).where(User.wechat_openid == openid)
- )
- existing_user = result.scalar_one_or_none()
-
- if existing_user:
- # 已存在,直接返回
- logger.info(f"✅ 微信登录成功: user_id={existing_user.user_id}, openid={openid}")
- return existing_user.to_dict()
-
- # 新用户,创建账号
- user_id = str(uuid.uuid4())
- if not username or username.strip() == "":
- username = await self._generate_temp_username(session)
-
- user = User(
- user_id=user_id,
- username=username,
- wechat_openid=openid,
- user_type="registered",
- is_verified=True,
- registered_at=datetime.now(),
- )
- session.add(user)
- await session.commit()
- await session.refresh(user)
-
- logger.info(f"✅ 微信注册成功: user_id={user_id}, openid={openid}")
- return user.to_dict()
- except Exception as e:
- logger.error(f"❌ 微信登录/注册失败: {e}")
- raise
-
- async def login_with_qq(
- self,
- openid: str,
- username: str = "",
- unionid: Optional[str] = None,
- ) -> Dict:
- """
- 使用QQ OpenID登录/注册
-
- Args:
- openid: QQ OpenID
- username: 用户名(可选,仅新用户时使用)
-
- Returns:
- 用户信息字典
- """
- try:
- async with db_manager.get_session() as session:
- # 优先用 openid 查询
- result = await session.execute(
- select(User).where(User.qq_openid == openid)
- )
- existing_user = result.scalar_one_or_none()
- # 如果未命中且拿到了 unionid,则尝试用 unionid 查询
- if not existing_user and unionid:
- union_result = await session.execute(
- select(User).where(User.qq_unionid == unionid)
- )
- existing_user = union_result.scalar_one_or_none()
- # 若命中用户还未绑定 openid,则补充绑定
- if existing_user and not existing_user.qq_openid:
- existing_user.qq_openid = openid
- session.add(existing_user)
- await session.commit()
- await session.refresh(existing_user)
- logger.info(
- f"🔁 已根据 unionid={unionid} 绑定新的 QQ openid={openid}"
- )
- if existing_user:
- updated = False
- if unionid and not existing_user.qq_unionid:
- existing_user.qq_unionid = unionid
- updated = True
- if updated:
- session.add(existing_user)
- await session.commit()
- await session.refresh(existing_user)
- logger.info(
- f"✅ QQ登录成功: user_id={existing_user.user_id}, "
- f"openid={openid}, unionid={unionid}"
- )
- return existing_user.to_dict()
-
- # 新用户,创建账号
- user_id = str(uuid.uuid4())
- if not username or username.strip() == "":
- username = await self._generate_temp_username(session)
-
- user = User(
- user_id=user_id,
- username=username,
- qq_openid=openid,
- qq_unionid=unionid,
- user_type="registered",
- is_verified=True,
- registered_at=datetime.now(),
- )
- session.add(user)
- await session.commit()
- await session.refresh(user)
-
- logger.info(
- f"✅ QQ注册成功: user_id={user_id}, openid={openid}, unionid={unionid}"
- )
- return user.to_dict()
- except Exception as e:
- logger.error(f"❌ QQ登录/注册失败: {e}")
- raise
-
- async def login_with_phone(self, phone: str) -> Optional[Dict]:
- """
- 使用手机号登录
-
- Args:
- phone: 手机号
-
- Returns:
- 用户信息字典,如果不存在返回None
- """
- try:
- async with db_manager.get_session() as session:
- result = await session.execute(
- select(User).where(User.phone == phone)
- )
- user = result.scalar_one_or_none()
-
- if user:
- logger.info(f"✅ 手机号登录成功: user_id={user.user_id}, phone={phone}")
- return user.to_dict()
- return None
- except Exception as e:
- logger.error(f"❌ 手机号登录失败: {e}")
- return None
-
- async def login_with_email(self, email: str) -> Optional[Dict]:
- """
- 使用邮箱登录
-
- Args:
- email: 邮箱
-
- Returns:
- 用户信息字典,如果不存在返回None
- """
- try:
- async with db_manager.get_session() as session:
- result = await session.execute(
- select(User).where(User.email == email)
- )
- user = result.scalar_one_or_none()
-
- if user:
- logger.info(f"✅ 邮箱登录成功: user_id={user.user_id}, email={email}")
- return user.to_dict()
- return None
- except Exception as e:
- logger.error(f"❌ 邮箱登录失败: {e}")
- return None
- # 全局用户服务实例
- user_service = UserService()
|