| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187 |
- """
- 主应用入口
- """
- import json
- import logging
- import urllib.parse
- import hashlib
- from typing import Optional, List, Dict, Any
- from fastapi import FastAPI, WebSocket, HTTPException, Request, Query
- from fastapi.responses import RedirectResponse, PlainTextResponse
- from fastapi.middleware.cors import CORSMiddleware
- from pydantic import BaseModel
- from contextlib import asynccontextmanager
- import httpx
- from xml.etree import ElementTree as ET
- from config import settings, get_dashscope_domain
- from websocket_handler import WebSocketHandler
- from database import init_database, close_database
- from services.redis_cache_service import RedisCacheService
- from services.user_service import user_service
- from services.wechat_jsapi_service import WechatJSAPIService
- # 配置日志
- 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'
- )
- 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
- )
- # 配置CORS
- app.add_middleware(
- CORSMiddleware,
- allow_origins=["*"], # 生产环境应该限制具体域名
- allow_credentials=True,
- allow_methods=["*"],
- allow_headers=["*"],
- )
- @app.get("/")
- async def root():
- """健康检查"""
- return {"status": "ok", "message": "Lyrria Backend API"}
- class CreateUserRequest(BaseModel):
- """创建用户请求"""
- username: str = ""
- bot_name: Optional[str] = None
- class UpdateUsernameRequest(BaseModel):
- """更新用户名请求"""
- username: str
- class UpdateBotNameRequest(BaseModel):
- """更新机器人名字请求"""
- bot_name: str
- class RegisterPhoneRequest(BaseModel):
- """手机号注册请求"""
- phone: str
- username: str = ""
- class RegisterEmailRequest(BaseModel):
- """邮箱注册请求"""
- email: str
- username: str = ""
- class LoginWechatRequest(BaseModel):
- """微信登录请求"""
- openid: str
- username: str = ""
- class LoginQQRequest(BaseModel):
- """QQ登录请求"""
- openid: str
- username: str = ""
- class LoginPhoneRequest(BaseModel):
- """手机号登录请求"""
- phone: str
- class LoginEmailRequest(BaseModel):
- """邮箱登录请求"""
- email: str
- 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
- avatar_c_unlocked: bool = False
- 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
- avatar_c_unlocked: Optional[bool] = None
- @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
- )
- return {
- "success": True,
- "data": user
- }
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"创建用户失败: {str(e)}")
- @app.get("/api/users/{user_id}")
- async def get_user(user_id: str):
- """
- 获取用户信息
-
- 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="用户不存在")
- except HTTPException:
- raise
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"获取用户信息失败: {str(e)}")
- @app.put("/api/users/{user_id}/username")
- async def update_username(user_id: str, request: UpdateUsernameRequest):
- """
- 更新用户名
-
- 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="更新用户名失败")
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"更新用户名失败: {str(e)}")
- @app.put("/api/users/{user_id}/bot-name")
- async def update_bot_name(user_id: str, request: UpdateBotNameRequest):
- """
- 更新用户给机器人取的名字
-
- 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=f"更新机器人名字失败(耗时: {duration:.1f}ms)"
- )
- 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=f"更新机器人名字超时(耗时: {duration:.1f}ms),请稍后重试"
- )
- except HTTPException:
- raise
- except Exception as e:
- duration = (time.time() - start_time) * 1000
- logger.error(f"❌ 更新机器人名字异常: user_id={user_id}, error={e}, 耗时={duration:.1f}ms")
- raise HTTPException(
- status_code=500,
- detail=f"更新机器人名字失败: {str(e)}(耗时: {duration:.1f}ms)"
- )
- @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",
- ...
- }
- }
- """
- try:
- user = await user_service.register_with_phone(
- phone=request.phone,
- username=request.username
- )
- return {
- "success": True,
- "data": user
- }
- except ValueError as e:
- raise HTTPException(status_code=400, detail=str(e))
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"注册失败: {str(e)}")
- @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",
- ...
- }
- }
- """
- try:
- user = await user_service.register_with_email(
- email=request.email,
- username=request.username
- )
- return {
- "success": True,
- "data": user
- }
- except ValueError as e:
- raise HTTPException(status_code=400, detail=str(e))
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"注册失败: {str(e)}")
- @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",
- ...
- }
- }
- """
- try:
- user = await user_service.login_with_wechat(
- openid=request.openid,
- username=request.username
- )
- return {
- "success": True,
- "data": user
- }
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"登录失败: {str(e)}")
- @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",
- ...
- }
- }
- """
- try:
- user = await user_service.login_with_qq(
- openid=request.openid,
- username=request.username
- )
- return {
- "success": True,
- "data": user
- }
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"登录失败: {str(e)}")
- @app.post("/api/auth/login/phone")
- async def login_with_phone(request: LoginPhoneRequest):
- """
- 手机号登录
-
- Returns:
- {
- "success": true,
- "data": {
- "user_id": "uuid",
- "phone": "",
- ...
- }
- }
- """
- try:
- user = await user_service.login_with_phone(request.phone)
- if user:
- return {
- "success": True,
- "data": user
- }
- else:
- raise HTTPException(status_code=404, detail="用户不存在,请先注册")
- except HTTPException:
- raise
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"登录失败: {str(e)}")
- @app.post("/api/auth/login/email")
- async def login_with_email(request: LoginEmailRequest):
- """
- 邮箱登录
-
- Returns:
- {
- "success": true,
- "data": {
- "user_id": "uuid",
- "email": "",
- ...
- }
- }
- """
- try:
- user = await user_service.login_with_email(request.email)
- if user:
- return {
- "success": True,
- "data": user
- }
- else:
- raise HTTPException(status_code=404, detail="用户不存在,请先注册")
- except HTTPException:
- raise
- except Exception as e:
- raise HTTPException(status_code=500, detail=f"登录失败: {str(e)}")
- @app.get("/api/users/{user_id}/ui-state")
- async def get_user_ui_state(user_id: str) -> 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
- raise HTTPException(status_code=500, detail=f"获取用户UI状态失败: {str(e)}") from e
- @app.put("/api/users/{user_id}/ui-state")
- async def update_user_ui_state(
- user_id: str,
- request: UpdateUserUIStateRequest,
- ) -> 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,
- avatar_c_unlocked=request.avatar_c_unlocked,
- )
- if not updated:
- raise HTTPException(status_code=500, detail="更新用户UI状态失败")
- return {"success": True, "data": updated}
- except HTTPException:
- raise
- except Exception as e: # noqa: BLE001
- raise HTTPException(status_code=500, detail=f"更新用户UI状态失败: {str(e)}") 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="微信OAuth未配置,请检查环境变量 WECHAT_OPEN_APP_ID 和 WECHAT_REDIRECT_URI")
-
- # 生成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="微信OAuth未配置,请检查环境变量 WECHAT_OPEN_APP_ID 和 WECHAT_OPEN_APP_SECRET")
-
- # 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):
- """
- 获取QQ授权URL
-
- Returns:
- 重定向到QQ授权页面
- """
- if not settings.QQ_APP_ID or not settings.QQ_REDIRECT_URI:
- raise HTTPException(status_code=500, detail="QQ OAuth未配置,请检查环境变量")
-
- # 生成state参数(用于防止CSRF攻击)
- import secrets
- state = secrets.token_urlsafe(32)
-
- # 将 state 存储到 Redis 以便回调校验
- 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:qq:state:{state}", 600, "1")
- except Exception as e:
- logger.warning(f"存储 QQ state 失败,将跳过校验: {e}")
-
- # 确保回调地址格式正确(去除末尾斜杠,确保与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}")
-
- 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:
- # 0. 验证 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:qq:state:{state}"
- exists = await redis_client.exists(state_key)
- if not exists:
- logger.warning(f"QQ state 验证失败: {state}")
- return RedirectResponse(
- url=f"{settings.FRONTEND_URL}/register?error=qq_auth_failed&reason=invalid_state"
- )
- await redis_client.delete(state_key)
- except Exception as state_error:
- logger.warning(f"QQ state 验证异常: {state_error},继续流程")
-
- if not settings.QQ_APP_ID or not settings.QQ_APP_KEY:
- raise HTTPException(status_code=500, detail="QQ OAuth未配置")
-
- # 小工具:解析 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. 使用openid登录/注册
- user = await user_service.login_with_qq(
- openid=openid,
- username=nickname,
- unionid=unionid,
- )
-
- # 4. 重定向到前端,携带用户信息
- frontend_url = f"{settings.FRONTEND_URL}/register?oauth=qq&user_id={user['user_id']}&username={urllib.parse.quote(user['username'])}"
- 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="微信公众平台AppID未配置,请检查环境变量 WECHAT_MP_APP_ID")
-
- # 记录接收到的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="获取微信JS-SDK配置失败")
-
- 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)
- raise HTTPException(status_code=500, detail=f"获取配置失败: {str(e)}")
- @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.error(f"获取QQ分享配置异常: {e}", exc_info=True)
- raise HTTPException(status_code=500, detail=f"获取配置失败: {str(e)}")
- # ==================== 微信公众平台服务器配置 ====================
- @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.websocket("/ws")
- async def websocket_endpoint(websocket: WebSocket):
- """WebSocket端点"""
- # 从查询参数获取user_id(Lyrria_user_id)
- user_id = websocket.query_params.get("user_id")
-
- # 记录连接信息
- import logging
- logger = logging.getLogger(__name__)
- logger.info(f"🔌 WebSocket连接: user_id={user_id}, query_params={dict(websocket.query_params)}")
-
- # 🔥 传入Redis缓存服务和user_id
- redis_cache = getattr(app.state, 'redis_cache', None)
- handler = WebSocketHandler(websocket, user_id=user_id, redis_cache=redis_cache)
- 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,
- )
|