main.py 7.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274
  1. """
  2. 主应用入口
  3. """
  4. import logging
  5. from typing import Optional
  6. from fastapi import FastAPI, WebSocket, HTTPException
  7. from fastapi.middleware.cors import CORSMiddleware
  8. from pydantic import BaseModel
  9. from contextlib import asynccontextmanager
  10. from config import settings, get_dashscope_domain
  11. from websocket_handler import WebSocketHandler
  12. from database import init_database, close_database
  13. from services.redis_cache_service import RedisCacheService
  14. from services.user_service import user_service
  15. # 配置日志
  16. logging.basicConfig(
  17. level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO),
  18. format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
  19. datefmt='%Y-%m-%d %H:%M:%S'
  20. )
  21. logger = logging.getLogger(__name__)
  22. @asynccontextmanager
  23. async def lifespan(app: FastAPI):
  24. """应用生命周期管理"""
  25. # 启动时初始化数据库(包含MySQL和Redis,自动启动健康检查任务)
  26. try:
  27. await init_database()
  28. # 获取Redis客户端(已配置健康检查),用于验证连通性
  29. from database import db_manager
  30. await db_manager.get_redis()
  31. # 创建Redis缓存服务,按需获取最新连接
  32. app.state.redis_cache = RedisCacheService(db_manager.get_redis)
  33. except Exception as e:
  34. logger.error(f"❌ 数据库初始化失败: {e}")
  35. app.state.redis_cache = None
  36. # 🔥 启动HTTP连接保活任务(保持连接活跃,避免长时间不交互后响应变慢)
  37. try:
  38. from http_pool import http_pool
  39. from ai_service import AIService
  40. # 创建AIService实例用于保活(共享连接池)
  41. ai_service = AIService()
  42. http_pool.start_keepalive(ai_service=ai_service)
  43. except Exception as e:
  44. logger.warning(f"⚠️ HTTP连接保活任务启动失败: {e}")
  45. yield
  46. # 关闭时清理资源
  47. await close_database()
  48. # 🔥 关闭HTTP连接池(类似SQLAlchemy的Engine.dispose())
  49. from http_pool import http_pool
  50. http_pool.close()
  51. app = FastAPI(
  52. title="Tolan Backend API",
  53. lifespan=lifespan
  54. )
  55. # 配置CORS
  56. app.add_middleware(
  57. CORSMiddleware,
  58. allow_origins=["*"], # 生产环境应该限制具体域名
  59. allow_credentials=True,
  60. allow_methods=["*"],
  61. allow_headers=["*"],
  62. )
  63. @app.get("/")
  64. async def root():
  65. """健康检查"""
  66. return {"status": "ok", "message": "Tolan Backend API"}
  67. class CreateUserRequest(BaseModel):
  68. """创建用户请求"""
  69. username: str = ""
  70. bot_name: Optional[str] = None
  71. class UpdateUsernameRequest(BaseModel):
  72. """更新用户名请求"""
  73. username: str
  74. class UpdateBotNameRequest(BaseModel):
  75. """更新机器人名字请求"""
  76. bot_name: str
  77. @app.post("/api/users/create")
  78. async def create_user(request: CreateUserRequest):
  79. """
  80. 创建游客用户
  81. Returns:
  82. {
  83. "user_id": "uuid",
  84. "username": "",
  85. "user_type": "guest",
  86. "bot_name": "机器人名字(可选)"
  87. }
  88. """
  89. try:
  90. user = await user_service.create_guest_user(
  91. username=request.username,
  92. bot_name=request.bot_name
  93. )
  94. return {
  95. "success": True,
  96. "data": user
  97. }
  98. except Exception as e:
  99. raise HTTPException(status_code=500, detail=f"创建用户失败: {str(e)}")
  100. @app.get("/api/users/{user_id}")
  101. async def get_user(user_id: str):
  102. """
  103. 获取用户信息
  104. Args:
  105. user_id: 用户ID
  106. Returns:
  107. {
  108. "success": true,
  109. "data": {
  110. "user_id": "uuid",
  111. "username": "",
  112. "user_type": "guest",
  113. "bot_name": "机器人名字(可选)",
  114. ...
  115. }
  116. }
  117. """
  118. try:
  119. user = await user_service.get_user(user_id)
  120. if user:
  121. return {
  122. "success": True,
  123. "data": user
  124. }
  125. else:
  126. raise HTTPException(status_code=404, detail="用户不存在")
  127. except HTTPException:
  128. raise
  129. except Exception as e:
  130. raise HTTPException(status_code=500, detail=f"获取用户信息失败: {str(e)}")
  131. @app.put("/api/users/{user_id}/username")
  132. async def update_username(user_id: str, request: UpdateUsernameRequest):
  133. """
  134. 更新用户名
  135. Args:
  136. user_id: 用户ID
  137. request: 更新请求(包含username)
  138. Returns:
  139. {
  140. "success": true,
  141. "message": "更新成功"
  142. }
  143. """
  144. try:
  145. success = await user_service.update_username(user_id, request.username)
  146. if success:
  147. return {
  148. "success": True,
  149. "message": "用户名更新成功"
  150. }
  151. else:
  152. raise HTTPException(status_code=500, detail="更新用户名失败")
  153. except Exception as e:
  154. raise HTTPException(status_code=500, detail=f"更新用户名失败: {str(e)}")
  155. @app.put("/api/users/{user_id}/bot-name")
  156. async def update_bot_name(user_id: str, request: UpdateBotNameRequest):
  157. """
  158. 更新用户给机器人取的名字
  159. Args:
  160. user_id: 用户ID
  161. request: 更新请求(包含bot_name)
  162. Returns:
  163. {
  164. "success": true,
  165. "message": "更新成功"
  166. }
  167. """
  168. try:
  169. success = await user_service.update_bot_name(user_id, request.bot_name)
  170. if success:
  171. return {
  172. "success": True,
  173. "message": "机器人名字更新成功"
  174. }
  175. else:
  176. raise HTTPException(status_code=500, detail="更新机器人名字失败")
  177. except Exception as e:
  178. raise HTTPException(status_code=500, detail=f"更新机器人名字失败: {str(e)}")
  179. @app.websocket("/ws")
  180. async def websocket_endpoint(websocket: WebSocket):
  181. """WebSocket端点"""
  182. # 从查询参数获取user_id(tolan_user_id)
  183. user_id = websocket.query_params.get("user_id")
  184. # 记录连接信息
  185. import logging
  186. logger = logging.getLogger(__name__)
  187. logger.info(f"🔌 WebSocket连接: user_id={user_id}, query_params={dict(websocket.query_params)}")
  188. # 🔥 传入Redis缓存服务和user_id
  189. redis_cache = getattr(app.state, 'redis_cache', None)
  190. handler = WebSocketHandler(websocket, user_id=user_id, redis_cache=redis_cache)
  191. await handler.handle_connection()
  192. if __name__ == "__main__":
  193. import uvicorn
  194. import os
  195. import socket
  196. from pathlib import Path
  197. # 🔥 DNS预解析(减少首次连接延迟)
  198. dashscope_domain = get_dashscope_domain()
  199. try:
  200. socket.getaddrinfo(dashscope_domain, 443)
  201. except Exception as e:
  202. pass
  203. # SSL证书路径配置
  204. # 优先使用环境变量配置的证书路径(生产环境)
  205. ssl_keyfile = settings.SSL_KEYFILE
  206. ssl_certfile = settings.SSL_CERTFILE
  207. # 如果环境变量未配置,使用本地开发证书
  208. if not ssl_keyfile or not ssl_certfile:
  209. ssl_dir = Path(__file__).parent.parent / "ssl"
  210. ssl_keyfile = str(ssl_dir / "key.pem")
  211. ssl_certfile = str(ssl_dir / "cert.pem")
  212. # 检查证书是否存在
  213. if ssl_keyfile and ssl_certfile:
  214. if not os.path.exists(ssl_keyfile) or not os.path.exists(ssl_certfile):
  215. ssl_keyfile = None
  216. ssl_certfile = None
  217. uvicorn.run(
  218. "main:app",
  219. host=settings.host,
  220. port=settings.port,
  221. reload=settings.DEBUG,
  222. ssl_keyfile=ssl_keyfile,
  223. ssl_certfile=ssl_certfile,
  224. )