user_service.py 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672
  1. """
  2. 用户服务模块(SQLAlchemy ORM 版本)
  3. """
  4. import uuid
  5. import random
  6. import asyncio
  7. import logging
  8. from typing import Dict, Optional, Any
  9. from datetime import datetime
  10. from sqlalchemy import select, update, func, insert
  11. from sqlalchemy.ext.asyncio import AsyncSession
  12. from database import db_manager
  13. from models_db import User, UserUIState
  14. logger = logging.getLogger(__name__)
  15. class UserService:
  16. """用户管理服务"""
  17. async def _generate_temp_username(self, session: AsyncSession) -> str:
  18. """
  19. 生成临时用户名(格式:User + 7位随机数字)
  20. 确保用户名不重复
  21. Args:
  22. session: 数据库会话
  23. Returns:
  24. 临时用户名
  25. """
  26. max_attempts = 100 # 最大重试次数,避免无限循环
  27. for attempt in range(max_attempts):
  28. # 生成7位随机数字(1000000-9999999)
  29. random_number = random.randint(1000000, 9999999)
  30. temp_username = f"User{random_number}"
  31. # 检查用户名是否已存在
  32. result = await session.execute(
  33. select(User).where(User.username == temp_username)
  34. )
  35. existing_user = result.scalar_one_or_none()
  36. if not existing_user:
  37. logger.debug(f"生成临时用户名: {temp_username} (尝试次数: {attempt + 1})")
  38. return temp_username
  39. # 如果100次都重复,使用时间戳的后7位作为备选方案
  40. # 使用当前时间戳的毫秒数取模,确保是7位数字
  41. for fallback_attempt in range(10): # 备选方案也最多尝试10次
  42. timestamp_suffix = int(datetime.now().timestamp() * 1000) % 10000000
  43. # 确保至少是7位数字(如果不足7位,前面补0)
  44. fallback_number = str(timestamp_suffix).zfill(7)
  45. temp_username = f"User{fallback_number}"
  46. # 再次检查是否重复
  47. result = await session.execute(
  48. select(User).where(User.username == temp_username)
  49. )
  50. existing_user = result.scalar_one_or_none()
  51. if not existing_user:
  52. logger.warning(f"⚠️ 临时用户名生成达到最大重试次数,使用备选方案: {temp_username}")
  53. return temp_username
  54. # 如果还是重复,等待1毫秒后重试(确保时间戳变化)
  55. await asyncio.sleep(0.001)
  56. # 最后的备选方案:使用UUID的十六进制转数字(取后7位)
  57. uuid_hex = uuid.uuid4().hex
  58. final_number = str(int(uuid_hex[:8], 16) % 10000000).zfill(7)
  59. temp_username = f"User{final_number}"
  60. logger.error(f"❌ 临时用户名生成异常,使用最终备选方案: {temp_username}")
  61. return temp_username
  62. async def create_guest_user(self, username: str = "", bot_name: Optional[str] = None) -> Dict:
  63. """
  64. 创建游客用户
  65. Args:
  66. username: 用户名(默认为空字符串,为空时自动生成临时用户名)
  67. bot_name: 用户给机器人取的名字(可选)
  68. Returns:
  69. 用户信息字典(包含user_id)
  70. """
  71. try:
  72. # 生成唯一用户ID
  73. user_id = str(uuid.uuid4())
  74. # 使用 ORM 创建用户
  75. async with db_manager.get_session() as session:
  76. # 如果用户名为空,生成临时用户名
  77. if not username or username.strip() == "":
  78. username = await self._generate_temp_username(session)
  79. logger.info(f"🔧 自动生成临时用户名: {username}")
  80. user = User(
  81. user_id=user_id,
  82. username=username,
  83. user_type="guest",
  84. bot_name=bot_name,
  85. )
  86. session.add(user)
  87. await session.commit()
  88. await session.refresh(user)
  89. logger.info(f"✅ 创建游客用户成功: user_id={user_id}, username={username}, bot_name={bot_name}")
  90. return {
  91. "user_id": user_id,
  92. "username": username,
  93. "user_type": "guest",
  94. "bot_name": bot_name,
  95. "created_at": datetime.now().isoformat()
  96. }
  97. except Exception as e:
  98. logger.error(f"❌ 创建用户失败: {e}")
  99. raise
  100. async def get_user(self, user_id: str) -> Optional[Dict]:
  101. """
  102. 获取用户信息
  103. Args:
  104. user_id: 用户ID
  105. Returns:
  106. 用户信息字典,如果不存在返回None
  107. """
  108. try:
  109. async with db_manager.get_session() as session:
  110. result = await session.execute(
  111. select(User).where(User.user_id == user_id)
  112. )
  113. user = result.scalar_one_or_none()
  114. if user:
  115. return user.to_dict()
  116. return None
  117. except Exception as e:
  118. logger.error(f"❌ 获取用户信息失败: user_id={user_id}, error={e}")
  119. return None
  120. async def update_user_last_active(self, user_id: str):
  121. """
  122. 更新用户最后活跃时间
  123. Args:
  124. user_id: 用户ID
  125. """
  126. try:
  127. async with db_manager.get_session() as session:
  128. await session.execute(
  129. update(User)
  130. .where(User.user_id == user_id)
  131. .values(last_active_at=func.now())
  132. )
  133. await session.commit()
  134. except Exception as e:
  135. logger.warning(f"⚠️ 更新用户活跃时间失败: user_id={user_id}, error={e}")
  136. async def update_username(self, user_id: str, username: str) -> bool:
  137. """
  138. 更新用户名
  139. Args:
  140. user_id: 用户ID
  141. username: 用户名
  142. Returns:
  143. 是否更新成功
  144. """
  145. try:
  146. async with db_manager.get_session() as session:
  147. result = await session.execute(
  148. update(User)
  149. .where(User.user_id == user_id)
  150. .values(username=username)
  151. )
  152. await session.commit()
  153. if result.rowcount > 0:
  154. logger.info(f"✅ 更新用户名成功: user_id={user_id}, username={username}")
  155. return True
  156. return False
  157. except Exception as e:
  158. logger.error(f"❌ 更新用户名失败: user_id={user_id}, error={e}")
  159. return False
  160. async def update_bot_name(self, user_id: str, bot_name: str) -> bool:
  161. """
  162. 更新用户给机器人取的名字
  163. Args:
  164. user_id: 用户ID
  165. bot_name: 机器人名字
  166. Returns:
  167. 是否更新成功
  168. """
  169. import time
  170. start_time = time.time()
  171. try:
  172. async with db_manager.get_session() as session:
  173. # 🔥 先查询当前值,避免不必要的更新
  174. query_start = time.time()
  175. result = await session.execute(
  176. select(User.bot_name).where(User.user_id == user_id)
  177. )
  178. current_bot_name = result.scalar_one_or_none()
  179. query_duration = (time.time() - query_start) * 1000
  180. # 如果值相同,直接返回(避免不必要的 UPDATE)
  181. if current_bot_name == bot_name:
  182. total_duration = (time.time() - start_time) * 1000
  183. logger.info(
  184. f"⏭️ 机器人名字未变化,跳过更新: user_id={user_id}, "
  185. f"bot_name={bot_name}, 耗时={total_duration:.1f}ms"
  186. )
  187. return True
  188. # 执行更新
  189. update_start = time.time()
  190. update_result = await session.execute(
  191. update(User)
  192. .where(User.user_id == user_id)
  193. .values(bot_name=bot_name)
  194. )
  195. await session.commit()
  196. update_duration = (time.time() - update_start) * 1000
  197. total_duration = (time.time() - start_time) * 1000
  198. if update_result.rowcount > 0:
  199. logger.info(
  200. f"✅ 更新机器人名字成功: user_id={user_id}, "
  201. f"bot_name={bot_name}, "
  202. f"查询耗时={query_duration:.1f}ms, "
  203. f"更新耗时={update_duration:.1f}ms, "
  204. f"总耗时={total_duration:.1f}ms"
  205. )
  206. return True
  207. else:
  208. logger.warning(
  209. f"⚠️ 更新机器人名字失败: 未找到用户 user_id={user_id}, "
  210. f"总耗时={total_duration:.1f}ms"
  211. )
  212. return False
  213. except asyncio.TimeoutError:
  214. total_duration = (time.time() - start_time) * 1000
  215. logger.error(
  216. f"❌ 更新机器人名字超时: user_id={user_id}, "
  217. f"耗时={total_duration:.1f}ms"
  218. )
  219. return False
  220. except Exception as e:
  221. total_duration = (time.time() - start_time) * 1000
  222. logger.error(
  223. f"❌ 更新机器人名字失败: user_id={user_id}, "
  224. f"error={e}, 耗时={total_duration:.1f}ms"
  225. )
  226. return False
  227. async def update_speech_duration(self, user_id: str, duration_ms: int) -> Optional[int]:
  228. """
  229. 更新用户累积说话时长
  230. Args:
  231. user_id: 用户ID
  232. duration_ms: 本次说话时长(毫秒)
  233. Returns:
  234. 是否更新成功
  235. """
  236. try:
  237. if duration_ms <= 0:
  238. return None # 时长无效,跳过更新
  239. async with db_manager.get_session() as session:
  240. result = await session.execute(
  241. select(User.total_speech_duration_ms)
  242. .where(User.user_id == user_id)
  243. .with_for_update()
  244. )
  245. current_total = result.scalar_one_or_none() or 0
  246. new_total = current_total + duration_ms
  247. await session.execute(
  248. update(User)
  249. .where(User.user_id == user_id)
  250. .values(total_speech_duration_ms=new_total)
  251. )
  252. await session.commit()
  253. logger.info(
  254. f"✅ 更新用户说话时长成功: user_id={user_id}, "
  255. f"duration_ms={duration_ms}, total_ms={new_total}"
  256. )
  257. return new_total
  258. except Exception as e:
  259. logger.error(f"❌ 更新用户说话时长失败: user_id={user_id}, error={e}")
  260. return None
  261. async def get_user_ui_state(self, user_id: str) -> Optional[Dict[str, Any]]:
  262. """
  263. 获取用户前端可视化状态
  264. """
  265. try:
  266. async with db_manager.get_session() as session:
  267. result = await session.execute(
  268. select(UserUIState).where(UserUIState.user_id == user_id)
  269. )
  270. state = result.scalar_one_or_none()
  271. if state is None:
  272. return None
  273. return state.to_dict()
  274. except Exception as e: # noqa: BLE001
  275. logger.error(f"❌ 获取用户UI状态失败: user_id={user_id}, error={e}")
  276. return None
  277. async def upsert_user_ui_state(
  278. self,
  279. user_id: str,
  280. *,
  281. current_page: Optional[str] = None,
  282. visited_routes: Optional[list[str]] = None,
  283. has_shown_tap_to_chat: Optional[bool] = None,
  284. has_shown_cancel_tooltip: Optional[bool] = None,
  285. has_shown_send_tooltip: Optional[bool] = None,
  286. has_shown_progress_bar_tooltip: Optional[bool] = None,
  287. avatar_c_unlocked: Optional[bool] = None,
  288. ) -> Optional[Dict[str, Any]]:
  289. """
  290. 创建或更新用户前端可视化状态(部分字段更新)
  291. """
  292. try:
  293. async with db_manager.get_session() as session:
  294. result = await session.execute(
  295. select(UserUIState).where(UserUIState.user_id == user_id).with_for_update()
  296. )
  297. state: Optional[UserUIState] = result.scalar_one_or_none()
  298. if state is None:
  299. # 新建记录,仅写入非空字段
  300. data: Dict[str, Any] = {"user_id": user_id}
  301. if current_page is not None:
  302. data["current_page"] = current_page
  303. if visited_routes is not None:
  304. data["visited_routes"] = visited_routes
  305. if has_shown_tap_to_chat is not None:
  306. data["has_shown_tap_to_chat"] = has_shown_tap_to_chat
  307. if has_shown_cancel_tooltip is not None:
  308. data["has_shown_cancel_tooltip"] = has_shown_cancel_tooltip
  309. if has_shown_send_tooltip is not None:
  310. data["has_shown_send_tooltip"] = has_shown_send_tooltip
  311. if has_shown_progress_bar_tooltip is not None:
  312. data["has_shown_progress_bar_tooltip"] = has_shown_progress_bar_tooltip
  313. if avatar_c_unlocked is not None:
  314. data["avatar_c_unlocked"] = avatar_c_unlocked
  315. stmt = insert(UserUIState).values(**data)
  316. await session.execute(stmt)
  317. else:
  318. # 更新现有记录(仅覆盖非空字段)
  319. if current_page is not None:
  320. state.current_page = current_page
  321. if visited_routes is not None:
  322. state.visited_routes = visited_routes
  323. if has_shown_tap_to_chat is not None:
  324. state.has_shown_tap_to_chat = has_shown_tap_to_chat
  325. if has_shown_cancel_tooltip is not None:
  326. state.has_shown_cancel_tooltip = has_shown_cancel_tooltip
  327. if has_shown_send_tooltip is not None:
  328. state.has_shown_send_tooltip = has_shown_send_tooltip
  329. if has_shown_progress_bar_tooltip is not None:
  330. state.has_shown_progress_bar_tooltip = has_shown_progress_bar_tooltip
  331. if avatar_c_unlocked is not None:
  332. state.avatar_c_unlocked = avatar_c_unlocked
  333. session.add(state)
  334. await session.commit()
  335. # 重新读取最新状态
  336. result = await session.execute(
  337. select(UserUIState).where(UserUIState.user_id == user_id)
  338. )
  339. latest_state = result.scalar_one_or_none()
  340. if latest_state is None:
  341. return None
  342. return latest_state.to_dict()
  343. except Exception as e: # noqa: BLE001
  344. logger.error(f"❌ 更新用户UI状态失败: user_id={user_id}, error={e}")
  345. return None
  346. async def register_with_phone(self, phone: str, username: str = "") -> Dict:
  347. """
  348. 使用手机号注册
  349. Args:
  350. phone: 手机号
  351. username: 用户名(可选)
  352. Returns:
  353. 用户信息字典
  354. """
  355. try:
  356. async with db_manager.get_session() as session:
  357. # 检查手机号是否已注册
  358. result = await session.execute(
  359. select(User).where(User.phone == phone)
  360. )
  361. existing_user = result.scalar_one_or_none()
  362. if existing_user:
  363. raise ValueError("该手机号已被注册")
  364. # 生成用户ID和用户名
  365. user_id = str(uuid.uuid4())
  366. if not username or username.strip() == "":
  367. username = await self._generate_temp_username(session)
  368. user = User(
  369. user_id=user_id,
  370. username=username,
  371. phone=phone,
  372. user_type="registered",
  373. is_verified=True,
  374. registered_at=datetime.now(),
  375. )
  376. session.add(user)
  377. await session.commit()
  378. await session.refresh(user)
  379. logger.info(f"✅ 手机号注册成功: user_id={user_id}, phone={phone}")
  380. return user.to_dict()
  381. except Exception as e:
  382. logger.error(f"❌ 手机号注册失败: {e}")
  383. raise
  384. async def register_with_email(self, email: str, username: str = "") -> Dict:
  385. """
  386. 使用邮箱注册
  387. Args:
  388. email: 邮箱
  389. username: 用户名(可选)
  390. Returns:
  391. 用户信息字典
  392. """
  393. try:
  394. async with db_manager.get_session() as session:
  395. # 检查邮箱是否已注册
  396. result = await session.execute(
  397. select(User).where(User.email == email)
  398. )
  399. existing_user = result.scalar_one_or_none()
  400. if existing_user:
  401. raise ValueError("该邮箱已被注册")
  402. # 生成用户ID和用户名
  403. user_id = str(uuid.uuid4())
  404. if not username or username.strip() == "":
  405. username = await self._generate_temp_username(session)
  406. user = User(
  407. user_id=user_id,
  408. username=username,
  409. email=email,
  410. user_type="registered",
  411. is_verified=True,
  412. registered_at=datetime.now(),
  413. )
  414. session.add(user)
  415. await session.commit()
  416. await session.refresh(user)
  417. logger.info(f"✅ 邮箱注册成功: user_id={user_id}, email={email}")
  418. return user.to_dict()
  419. except Exception as e:
  420. logger.error(f"❌ 邮箱注册失败: {e}")
  421. raise
  422. async def login_with_wechat(self, openid: str, username: str = "") -> Dict:
  423. """
  424. 使用微信OpenID登录/注册
  425. Args:
  426. openid: 微信OpenID
  427. username: 用户名(可选,仅新用户时使用)
  428. Returns:
  429. 用户信息字典
  430. """
  431. try:
  432. async with db_manager.get_session() as session:
  433. # 检查是否已存在
  434. result = await session.execute(
  435. select(User).where(User.wechat_openid == openid)
  436. )
  437. existing_user = result.scalar_one_or_none()
  438. if existing_user:
  439. # 已存在,直接返回
  440. logger.info(f"✅ 微信登录成功: user_id={existing_user.user_id}, openid={openid}")
  441. return existing_user.to_dict()
  442. # 新用户,创建账号
  443. user_id = str(uuid.uuid4())
  444. if not username or username.strip() == "":
  445. username = await self._generate_temp_username(session)
  446. user = User(
  447. user_id=user_id,
  448. username=username,
  449. wechat_openid=openid,
  450. user_type="registered",
  451. is_verified=True,
  452. registered_at=datetime.now(),
  453. )
  454. session.add(user)
  455. await session.commit()
  456. await session.refresh(user)
  457. logger.info(f"✅ 微信注册成功: user_id={user_id}, openid={openid}")
  458. return user.to_dict()
  459. except Exception as e:
  460. logger.error(f"❌ 微信登录/注册失败: {e}")
  461. raise
  462. async def login_with_qq(
  463. self,
  464. openid: str,
  465. username: str = "",
  466. unionid: Optional[str] = None,
  467. ) -> Dict:
  468. """
  469. 使用QQ OpenID登录/注册
  470. Args:
  471. openid: QQ OpenID
  472. username: 用户名(可选,仅新用户时使用)
  473. Returns:
  474. 用户信息字典
  475. """
  476. try:
  477. async with db_manager.get_session() as session:
  478. # 优先用 openid 查询
  479. result = await session.execute(
  480. select(User).where(User.qq_openid == openid)
  481. )
  482. existing_user = result.scalar_one_or_none()
  483. # 如果未命中且拿到了 unionid,则尝试用 unionid 查询
  484. if not existing_user and unionid:
  485. union_result = await session.execute(
  486. select(User).where(User.qq_unionid == unionid)
  487. )
  488. existing_user = union_result.scalar_one_or_none()
  489. # 若命中用户还未绑定 openid,则补充绑定
  490. if existing_user and not existing_user.qq_openid:
  491. existing_user.qq_openid = openid
  492. session.add(existing_user)
  493. await session.commit()
  494. await session.refresh(existing_user)
  495. logger.info(
  496. f"🔁 已根据 unionid={unionid} 绑定新的 QQ openid={openid}"
  497. )
  498. if existing_user:
  499. updated = False
  500. if unionid and not existing_user.qq_unionid:
  501. existing_user.qq_unionid = unionid
  502. updated = True
  503. if updated:
  504. session.add(existing_user)
  505. await session.commit()
  506. await session.refresh(existing_user)
  507. logger.info(
  508. f"✅ QQ登录成功: user_id={existing_user.user_id}, "
  509. f"openid={openid}, unionid={unionid}"
  510. )
  511. return existing_user.to_dict()
  512. # 新用户,创建账号
  513. user_id = str(uuid.uuid4())
  514. if not username or username.strip() == "":
  515. username = await self._generate_temp_username(session)
  516. user = User(
  517. user_id=user_id,
  518. username=username,
  519. qq_openid=openid,
  520. qq_unionid=unionid,
  521. user_type="registered",
  522. is_verified=True,
  523. registered_at=datetime.now(),
  524. )
  525. session.add(user)
  526. await session.commit()
  527. await session.refresh(user)
  528. logger.info(
  529. f"✅ QQ注册成功: user_id={user_id}, openid={openid}, unionid={unionid}"
  530. )
  531. return user.to_dict()
  532. except Exception as e:
  533. logger.error(f"❌ QQ登录/注册失败: {e}")
  534. raise
  535. async def login_with_phone(self, phone: str) -> Optional[Dict]:
  536. """
  537. 使用手机号登录
  538. Args:
  539. phone: 手机号
  540. Returns:
  541. 用户信息字典,如果不存在返回None
  542. """
  543. try:
  544. async with db_manager.get_session() as session:
  545. result = await session.execute(
  546. select(User).where(User.phone == phone)
  547. )
  548. user = result.scalar_one_or_none()
  549. if user:
  550. logger.info(f"✅ 手机号登录成功: user_id={user.user_id}, phone={phone}")
  551. return user.to_dict()
  552. return None
  553. except Exception as e:
  554. logger.error(f"❌ 手机号登录失败: {e}")
  555. return None
  556. async def login_with_email(self, email: str) -> Optional[Dict]:
  557. """
  558. 使用邮箱登录
  559. Args:
  560. email: 邮箱
  561. Returns:
  562. 用户信息字典,如果不存在返回None
  563. """
  564. try:
  565. async with db_manager.get_session() as session:
  566. result = await session.execute(
  567. select(User).where(User.email == email)
  568. )
  569. user = result.scalar_one_or_none()
  570. if user:
  571. logger.info(f"✅ 邮箱登录成功: user_id={user.user_id}, email={email}")
  572. return user.to_dict()
  573. return None
  574. except Exception as e:
  575. logger.error(f"❌ 邮箱登录失败: {e}")
  576. return None
  577. # 全局用户服务实例
  578. user_service = UserService()