import os import json import uuid import time import logging from typing import Dict, Any, Generator, Optional, List from flask_cors import CORS import requests from flask import Flask, request, jsonify, render_template, session, Response, stream_with_context, redirect from http import HTTPStatus from dotenv import load_dotenv, find_dotenv from dashscope import Application from datetime import datetime # 先加载环境变量，确保后续模块读取到 .env（支持不同工作目录） load_dotenv(find_dotenv(), override=False) # 导入认证模块 from auth_middleware import init_auth from auth_routes import auth_bp # 导入配置和工具 from config import config from utils import ( sanitize_input, validate_user_id, rate_limit, truncate_conversation_history, log_request_info, get_beijing_time, generate_session_id ) # 导入数据库模型 from models import init_database, db, User, Conversation, Message, get_or_create_user # 配置日志 logging.basicConfig( level=getattr(logging, config.log_level), format=config.log_format ) logger = logging.getLogger(__name__) # 创建Flask应用 app = Flask(__name__) CORS(app) # 启用CORS支持 # 配置session app.secret_key = config.secret_key app.config['PERMANENT_SESSION_LIFETIME'] = 7 * 24 * 60 * 60 # 7天 app.config['SESSION_COOKIE_SECURE'] = False # 开发环境设为False app.config['SESSION_COOKIE_HTTPONLY'] = True app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' # 配置数据库 app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_url() app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SQLALCHEMY_ENGINE_OPTIONS'] = { 'pool_pre_ping': True, 'pool_recycle': 300, 'pool_timeout': 20, 'max_overflow': 0 } # 初始化数据库 try: init_database(app) except Exception as e: logger.error(f"数据库初始化失败: {str(e)}", exc_info=True) raise # 初始化认证中间件 auth_middleware = init_auth(app) app.auth_middleware = auth_middleware # 注册认证蓝图 app.register_blueprint(auth_bp) # 存储用户当前对话ID的字典（保留用于兼容性） user_current_conversations: Dict[str, int] = {} @app.after_request def add_cache_control_headers(response): """ 防止首页HTML被中间层或浏览器长期缓存，避免出现"旧HTML引用新/旧静态资源"导致白屏。 静态资源保持默认缓存策略，仅对HTML页面禁缓存。 """ content_type = (response.content_type or "").lower() if "text/html" in content_type: response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response def save_message_to_database(user_id: int, conversation_id: int, role: str, content: str, thinking_content: str = None, message_order: int = 1) -> Message: """保存消息到数据库""" try: message = Message( conversation_id=conversation_id, role=role, content=content, thinking_content=thinking_content, content_type='text', message_order=message_order ) db.session.add(message) # 更新对话的消息计数和最后消息时间 conversation = Conversation.query.get(conversation_id) if conversation: conversation.message_count += 1 conversation.last_message_at = get_beijing_time() conversation.updated_at = get_beijing_time() # 一次性提交所有更改 db.session.commit() return message except Exception as e: db.session.rollback() logger.error(f"保存消息到数据库时发生错误: {str(e)}", exc_info=True) raise def get_or_create_conversation(user_id: int, title: str = None, conversation_id: int = None, force_new: bool = False) -> Conversation: """获取或创建对话""" try: # 如果强制创建新对话，直接创建 if force_new: if not title: title = f"对话 {get_beijing_time().strftime('%Y-%m-%d %H:%M')}" # 生成新的session_id session_id = generate_session_id(user_id) conversation = Conversation( user_id=user_id, title=title, conversation_type='chat', session_id=session_id, message_count=0 ) db.session.add(conversation) db.session.commit() # 更新用户的当前对话ID user_current_conversations[str(user_id)] = conversation.conversation_id return conversation # 如果指定了conversation_id，验证并返回该对话 if conversation_id: conversation = Conversation.query.filter_by( conversation_id=conversation_id, user_id=user_id, is_deleted=False ).first() if conversation: # 如果对话没有session_id，生成一个 if not conversation.session_id: conversation.session_id = generate_session_id(user_id, conversation_id) db.session.commit() # 更新用户的当前对话ID user_current_conversations[str(user_id)] = conversation.conversation_id return conversation # 检查用户是否有当前对话映射 current_conv_id = user_current_conversations.get(str(user_id)) if current_conv_id: conversation = Conversation.query.filter_by( conversation_id=current_conv_id, user_id=user_id, is_deleted=False ).first() if conversation: # 如果对话没有session_id，生成一个 if not conversation.session_id: conversation.session_id = generate_session_id(user_id, current_conv_id) db.session.commit() return conversation # 创建新对话 if not title: title = f"对话 {get_beijing_time().strftime('%Y-%m-%d %H:%M')}" # 生成新的session_id session_id = generate_session_id(user_id) conversation = Conversation( user_id=user_id, title=title, conversation_type='chat', session_id=session_id, message_count=0 ) db.session.add(conversation) db.session.commit() # 更新用户的当前对话ID user_current_conversations[str(user_id)] = conversation.conversation_id return conversation except Exception as e: db.session.rollback() logger.error(f"获取或创建对话时发生错误: {str(e)}", exc_info=True) raise def switch_to_conversation(user_id: int, conversation_id: int) -> Conversation: """切换到指定对话""" try: # 验证对话是否属于当前用户 conversation = Conversation.query.filter_by( conversation_id=conversation_id, user_id=user_id, is_deleted=False ).first() if not conversation: raise ValueError(f"对话 {conversation_id} 不存在或无权限访问") # 如果对话没有session_id，生成一个 if not conversation.session_id: conversation.session_id = generate_session_id(user_id, conversation_id) db.session.commit() # 更新用户的当前对话ID user_current_conversations[str(user_id)] = conversation.conversation_id return conversation except Exception as e: logger.error(f"切换对话时发生错误: {str(e)}", exc_info=True) raise def generate_conversation_title(user_message: str) -> str: """根据用户消息生成对话标题""" # 取用户消息的前50个字符作为标题 title = user_message.strip()[:50] if len(user_message) > 50: title += "..." return title @app.route('/') def index(): """渲染聊天界面""" # 检查是否是登录成功跳转 login_success = request.args.get('login_success') token = request.args.get('token') user_info_json = request.args.get('user_info') if login_success and token and user_info_json: # 登录成功，处理用户信息 try: import urllib.parse import json user_info = json.loads(urllib.parse.unquote(user_info_json)) # 保存到session session['auth_token'] = token session['user_info'] = user_info # 将用户信息传递给模板 return render_template('index.html', user_info=user_info, jwt_token=token) except Exception as e: logger.error(f'处理登录成功参数失败: {e}') # 如果处理失败，重定向到登录页面 return redirect('/login') else: # 正常访问，检查认证状态 # 优先从session获取 user_info = session.get('user_info', {}) jwt_token = session.get('auth_token', '') # 如果session中没有，尝试从localStorage获取（通过JavaScript） if not jwt_token: # 返回一个页面，让JavaScript检查localStorage return render_template('index.html', user_info={}, jwt_token='') # 将用户信息传递给模板 return render_template('index.html', user_info=user_info, jwt_token=jwt_token) @app.route('/login') def login(): """登录页面""" return render_template('login.html') @app.route('/login_error') def login_error(): """登录错误页面""" error_type = request.args.get('error', 'unknown') error_message = request.args.get('message', '') error_code = request.args.get('code', '') return render_template('login_error.html', error_type=error_type, error_message=error_message, error_code=error_code) @app.route('/api/chat', methods=['POST']) @auth_middleware.login_required @rate_limit(config.rate_limit_requests, config.rate_limit_window) def chat(): """处理聊天请求并调用阿里云百炼API (非流式响应)""" try: data = request.json or {} message = data.get('message', '') user_id = data.get('user_id', str(uuid.uuid4())) agent = data.get('agent') or 'kuaishou' conversation_id = data.get('conversation_id') # 新增：支持指定对话ID session_id = data.get('session_id') # 新增：支持指定session_id force_new = data.get('force_new', False) # 新增：强制创建新对话 # 输入验证 message, error = sanitize_input(message, config.max_message_length) if error: return jsonify({ 'success': False, 'error': error }), HTTPStatus.BAD_REQUEST user_id, error = validate_user_id(user_id) if error: return jsonify({ 'success': False, 'error': error }), HTTPStatus.BAD_REQUEST # 从认证中间件获取用户信息 auth_user_info = getattr(request, 'user_info', {}) auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID # 选择对应智能体 APP_ID chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent) if not chosen_app_id: logger.error(f"未找到对应智能体APP_ID，agent={agent}") return jsonify({ 'success': False, 'error': f"服务器未配置该智能体的 APP_ID: {agent}" }), HTTPStatus.INTERNAL_SERVER_ERROR # 记录请求信息 log_request_info(message, user_id, 'POST') # 获取或创建对话 conversation = get_or_create_conversation( auth_user_id, generate_conversation_title(message), conversation_id, force_new ) # 优先使用前端传来的session_id，否则使用对话的session_id api_session_id = session_id or conversation.session_id # 准备API请求参数 headers = { "Authorization": f"Bearer {config.api_key}", "Content-Type": "application/json", "X-DashScope-SSE": "disable" # 禁用流式输出 } # 从数据库获取对话历史 messages_query = Message.query.filter_by( conversation_id=conversation.conversation_id, is_deleted=False ).order_by(Message.message_order.asc()).all() # 构建多轮对话历史 messages = [] for msg in messages_query: messages.append({"role": msg.role, "content": msg.content}) # 添加用户新消息到对话历史 messages.append({"role": "user", "content": message}) # 截断对话历史以控制内存使用 messages = truncate_conversation_history(messages, config.max_conversation_history) payload = { "model": "application", "parameters": {"app_id": chosen_app_id}, "input": {"messages": messages} # 发送完整的对话历史 } # 如果有会话ID，添加到请求参数中 if api_session_id: payload["parameters"]["session_id"] = api_session_id # 发送API请求 response = requests.post( config.api_endpoint, headers=headers, json=payload, timeout=60 ) # 转换为JSON response_json = response.json() if response.status_code != HTTPStatus.OK: return jsonify({ 'success': False, 'error': f"错误: {response.status_code} - {response_json.get('message', '未知错误')}", 'request_id': response_json.get('request_id', ''), 'agent': agent }), HTTPStatus.INTERNAL_SERVER_ERROR # 解析返回结果 output = response_json.get('output', {}) choices = output.get('choices', [{}]) text = choices[0].get('message', {}).get('content', '') if choices else '' new_session_id = output.get('session_id', '') thoughts_content = output.get('thoughts', []) # 更新对话的session_id（如果API返回了新的session_id且与当前不同） if new_session_id and new_session_id != api_session_id: conversation.session_id = new_session_id db.session.commit() elif not api_session_id and new_session_id: # 如果对话原本没有session_id，现在有了，更新数据库 conversation.session_id = new_session_id db.session.commit() # 保存到数据库 try: # 获取当前对话的消息数量，用于设置消息顺序 current_message_count = Message.query.filter_by( conversation_id=conversation.conversation_id, is_deleted=False ).count() # 保存用户消息 save_message_to_database( auth_user_id, conversation.conversation_id, 'user', message, message_order=current_message_count + 1 ) # 保存AI回复 if text: thinking_text = '\n'.join(thoughts_content) if thoughts_content else None save_message_to_database( auth_user_id, conversation.conversation_id, 'assistant', text, thinking_text, message_order=current_message_count + 2 ) # 更新对话标题（如果是第一条消息） if conversation.message_count <= 2: # 用户消息 + AI回复 new_title = generate_conversation_title(message) # 检查是否是默认标题（"新对话"或"对话 时间"格式） is_default_title = ( conversation.title == "新对话" or conversation.title == f"对话 {conversation.created_at.strftime('%Y-%m-%d %H:%M')}" ) if is_default_title: conversation.title = new_title db.session.commit() except Exception as db_error: logger.error(f"保存对话到数据库时发生错误: {str(db_error)}", exc_info=True) # 不影响主要功能，继续返回结果 return jsonify({ 'success': True, 'message': text, 'user_id': user_id, 'session_id': new_session_id or api_session_id, 'thoughts': thoughts_content, 'agent': agent, 'conversation_id': conversation.conversation_id }) except requests.exceptions.RequestException as e: logger.error(f"API请求异常: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"API请求异常: {str(e)}" }), HTTPStatus.SERVICE_UNAVAILABLE except Exception as e: logger.error(f"API调用错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/chat/stream', methods=['GET', 'POST']) @auth_middleware.login_required @rate_limit(config.rate_limit_requests, config.rate_limit_window) def chat_stream(): """处理聊天请求并以流式方式返回响应""" try: # 从请求中获取数据 if request.method == 'GET': message = request.args.get('message', '') user_id = request.args.get('user_id', str(uuid.uuid4())) agent = request.args.get('agent') or 'kuaishou' conversation_id = request.args.get('conversation_id') # 支持指定对话ID else: # POST data = request.json or {} message = data.get('message', '') user_id = data.get('user_id', str(uuid.uuid4())) agent = data.get('agent') or 'kuaishou' conversation_id = data.get('conversation_id') # 支持指定对话ID session_id = data.get('session_id') # 支持指定session_id force_new = data.get('force_new', False) # 强制创建新对话 # 输入验证 message, error = sanitize_input(message, config.max_message_length) if error: return Response( f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n", content_type='text/event-stream' ) user_id, error = validate_user_id(user_id) if error: return Response( f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n", content_type='text/event-stream' ) # 选择对应智能体 APP_ID chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent) if not chosen_app_id: logger.error(f"未找到对应智能体APP_ID，agent={agent}") return Response( f"data: {json.dumps({'success': False, 'error': f'服务器未配置该智能体的 APP_ID: {agent}', 'done': True})}\n\n", content_type='text/event-stream' ) # 记录请求信息 log_request_info(message, user_id, request.method) # 从认证中间件获取用户信息 auth_user_info = getattr(request, 'user_info', {}) auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID # 获取或创建对话 conversation = get_or_create_conversation( auth_user_id, generate_conversation_title(message), conversation_id, force_new ) # 优先使用前端传来的session_id，否则使用对话的session_id api_session_id = session_id or conversation.session_id # 从数据库获取对话历史 messages_query = Message.query.filter_by( conversation_id=conversation.conversation_id, is_deleted=False ).order_by(Message.message_order.asc()).all() # 构建多轮对话历史 messages = [] for msg in messages_query: messages.append({"role": msg.role, "content": msg.content}) # 添加用户新消息到对话历史 messages.append({"role": "user", "content": message}) # 截断对话历史以控制内存使用 messages = truncate_conversation_history(messages, config.max_conversation_history) # 立即保存用户消息到数据库（在流式响应开始前） try: # 获取当前对话的消息数量，用于设置消息顺序 current_message_count = Message.query.filter_by( conversation_id=conversation.conversation_id, is_deleted=False ).count() # 立即保存用户消息 user_message = save_message_to_database( auth_user_id, conversation.conversation_id, 'user', message, message_order=current_message_count + 1 ) except Exception as db_error: logger.error(f"保存用户消息到数据库时发生错误: {str(db_error)}", exc_info=True) return Response( f"data: {json.dumps({'success': False, 'error': f'保存用户消息失败: {str(db_error)}', 'done': True})}\n\n", content_type='text/event-stream' ) # 定义流式响应生成器 def generate() -> Generator[str, None, None]: try: # 发送初始响应，包含用户消息ID，防止浏览器超时 initial_data = { 'success': True, 'chunk': '', 'done': False, 'agent': agent, 'user_message_id': user_message.message_id if user_message else None } yield f"data: {json.dumps(initial_data)}\n\n" try: # 使用DashScope SDK直接调用 - 多轮对话模式 params = { "api_key": config.api_key, "app_id": chosen_app_id, "messages": messages, # 发送完整的对话历史 "stream": True, "incremental_output": True, "has_thoughts": True } # 使用对话的session_id if api_session_id: params["session_id"] = api_session_id response = Application.call(**params) new_session_id: Optional[str] = None accumulated_text: str = "" thoughts_content: List[str] = [] # 处理流式响应 for chunk in response: # 检查客户端是否断开连接 if request.environ.get('werkzeug.server.shutdown') or not request: break if chunk.status_code != HTTPStatus.OK: yield f"data: {json.dumps({'success': False, 'error': f'错误: {chunk.status_code} - {chunk.message}', 'request_id': chunk.request_id, 'done': True, 'agent': agent})}\n\n" return # 处理思考过程 if hasattr(chunk.output, 'thoughts') and chunk.output.thoughts: for it in chunk.output.thoughts: if it.action_type == 'reasoning': content = str(it.thought) if not isinstance(it.thought, str) else it.thought thoughts_content.append(content) yield f"data: {json.dumps({'success': True, 'thought': content, 'done': False, 'agent': agent})}\n\n" # 处理回答内容 if hasattr(chunk.output, 'text') and chunk.output.text: current_text = str(chunk.output.text) accumulated_text += current_text yield f"data: {json.dumps({'success': True, 'chunk': current_text, 'done': False, 'agent': agent})}\n\n" # 检查是否有会话ID if hasattr(chunk.output, 'session_id') and chunk.output.session_id: new_session_id = chunk.output.session_id # 处理结束 if new_session_id and new_session_id != api_session_id: conversation.session_id = new_session_id db.session.commit() elif not api_session_id and new_session_id: # 如果对话原本没有session_id，现在有了，更新数据库 conversation.session_id = new_session_id db.session.commit() # 保存AI回复到数据库 try: if accumulated_text: thinking_text = '\n'.join(thoughts_content) if thoughts_content else None ai_message = save_message_to_database( auth_user_id, conversation.conversation_id, 'assistant', accumulated_text, thinking_text, message_order=current_message_count + 2 ) # 更新对话标题（如果是第一条消息） if conversation.message_count <= 2: # 用户消息 + AI回复 new_title = generate_conversation_title(message) # 检查是否是默认标题（"新对话"或"对话 时间"格式） is_default_title = ( conversation.title == "新对话" or conversation.title == f"对话 {conversation.created_at.strftime('%Y-%m-%d %H:%M')}" ) if is_default_title: conversation.title = new_title db.session.commit() except Exception as db_error: logger.error(f"保存AI回复到数据库时发生错误: {str(db_error)}", exc_info=True) # 不影响主要功能，继续返回结果 # 发送完成标记 final_data = { 'success': True, 'done': True, 'thoughts': thoughts_content, 'agent': agent, 'session_id': new_session_id or api_session_id, 'user_id': user_id, 'conversation_id': conversation.conversation_id } yield f"data: {json.dumps(final_data)}\n\n" except Exception as e: logger.error(f"DashScope SDK 调用错误: {str(e)}", exc_info=True) yield f"data: {json.dumps({'success': False, 'error': f'DashScope API调用错误: {str(e)}', 'done': True, 'agent': agent})}\n\n" except Exception as e: logger.error(f"流式API调用错误: {str(e)}", exc_info=True) yield f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True, 'agent': agent})}\n\n" # 创建并返回流式响应 response = Response(stream_with_context(generate()), content_type='text/event-stream') response.headers['Cache-Control'] = 'no-cache' response.headers['X-Accel-Buffering'] = 'no' return response except Exception as e: logger.error(f"请求处理错误: {str(e)}", exc_info=True) response = Response( f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True})}\n\n", content_type='text/event-stream' ) response.headers['Cache-Control'] = 'no-cache' return response @app.route('/api/reset', methods=['POST']) @auth_middleware.login_required @rate_limit(10, 60) def reset_session(): """重置用户会话""" try: data = request.json or {} user_id = data.get('user_id') if not user_id: return jsonify({ 'success': True, 'message': '没有提供用户ID，但重置请求已处理' }) # 验证用户ID格式 user_id, error = validate_user_id(user_id) if error: return jsonify({ 'success': False, 'error': error }), HTTPStatus.BAD_REQUEST # 从认证中间件获取用户信息 auth_user_info = getattr(request, 'user_info', {}) auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID # 清除数据库中的当前对话ID（如果有认证用户信息） if auth_user_id and str(auth_user_id) in user_current_conversations: del user_current_conversations[str(auth_user_id)] return jsonify({ 'success': True, 'message': '会话已重置' }) except Exception as e: logger.error(f"重置会话时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/status', methods=['GET']) @auth_middleware.login_required def get_status(): """获取应用状态信息""" try: active_users = len(user_current_conversations) total_conversations = Conversation.query.filter_by(is_deleted=False).count() return jsonify({ 'success': True, 'status': 'running', 'active_users': active_users, 'total_conversations': total_conversations, 'app_version': '2.0.0', 'python_version': '3.12', 'config': { 'max_message_length': config.max_message_length, 'max_conversation_history': config.max_conversation_history, 'rate_limit_requests': config.rate_limit_requests, 'rate_limit_window': config.rate_limit_window } }) except Exception as e: logger.error(f"获取状态时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR # ==================== 历史记录相关API ==================== @app.route('/api/conversations', methods=['GET']) @auth_middleware.login_required def get_conversations(): """获取用户的历史对话列表，按时间分组""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: logger.warning('获取历史记录失败 - 用户ID为空') return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 获取用户对象 user = User.query.filter_by(user_id=user_id, is_deleted=False).first() if not user: return jsonify({ 'success': False, 'error': '用户不存在' }), HTTPStatus.NOT_FOUND # 获取用户的对话列表，按更新时间倒序 conversations = Conversation.query.filter_by( user_id=user_id, is_deleted=False ).order_by(Conversation.updated_at.desc()).all() # 按时间分组 from datetime import datetime, timedelta now = get_beijing_time() today = now.date() yesterday = today - timedelta(days=1) week_ago = today - timedelta(days=7) month_ago = today - timedelta(days=30) grouped_conversations = { 'today': [], 'yesterday': [], 'week': [], 'month': [] } for conv in conversations: conv_date = conv.updated_at.date() conv_data = { 'conversation_id': conv.conversation_id, 'title': conv.title, 'message_count': conv.message_count, 'session_id': conv.session_id, 'last_message_at': conv.last_message_at.isoformat() if conv.last_message_at else None, 'created_at': conv.created_at.isoformat(), 'updated_at': conv.updated_at.isoformat() } if conv_date == today: grouped_conversations['today'].append(conv_data) elif conv_date == yesterday: grouped_conversations['yesterday'].append(conv_data) elif conv_date >= week_ago: grouped_conversations['week'].append(conv_data) elif conv_date >= month_ago: grouped_conversations['month'].append(conv_data) return jsonify({ 'success': True, 'conversations': grouped_conversations }) except Exception as e: logger.error(f"获取对话列表时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations/', methods=['GET']) @auth_middleware.login_required def get_conversation_messages(conversation_id): """获取指定对话的所有消息""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 验证对话是否属于当前用户 conversation = Conversation.query.filter_by( conversation_id=conversation_id, user_id=user_id, is_deleted=False ).first() if not conversation: return jsonify({ 'success': False, 'error': '对话不存在或无权限访问' }), HTTPStatus.NOT_FOUND # 获取对话的所有消息，按顺序排列 messages = Message.query.filter_by( conversation_id=conversation_id, is_deleted=False ).order_by(Message.message_order.asc()).all() messages_data = [] for msg in messages: message_data = { 'message_id': msg.message_id, 'role': msg.role, 'content': msg.content, 'thinking_content': msg.thinking_content, 'content_type': msg.content_type, 'message_order': msg.message_order, 'created_at': msg.created_at.isoformat() } messages_data.append(message_data) return jsonify({ 'success': True, 'conversation': { 'conversation_id': conversation.conversation_id, 'title': conversation.title, 'conversation_type': conversation.conversation_type, 'session_id': conversation.session_id, 'message_count': conversation.message_count, 'created_at': conversation.created_at.isoformat(), 'updated_at': conversation.updated_at.isoformat() }, 'messages': messages_data }) except Exception as e: logger.error(f"获取对话消息时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations//rename', methods=['POST']) @auth_middleware.login_required def rename_conversation(conversation_id): """重命名对话""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED data = request.json or {} new_title = data.get('title', '').strip() if not new_title: return jsonify({ 'success': False, 'error': '新标题不能为空' }), HTTPStatus.BAD_REQUEST if len(new_title) > 500: return jsonify({ 'success': False, 'error': '标题长度不能超过500个字符' }), HTTPStatus.BAD_REQUEST # 验证对话是否属于当前用户 conversation = Conversation.query.filter_by( conversation_id=conversation_id, user_id=user_id, is_deleted=False ).first() if not conversation: return jsonify({ 'success': False, 'error': '对话不存在或无权限访问' }), HTTPStatus.NOT_FOUND # 更新标题 conversation.title = new_title conversation.updated_at = get_beijing_time() db.session.commit() return jsonify({ 'success': True, 'message': '对话标题已更新', 'conversation': { 'conversation_id': conversation.conversation_id, 'title': conversation.title, 'updated_at': conversation.updated_at.isoformat() } }) except Exception as e: db.session.rollback() logger.error(f"重命名对话时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations/', methods=['DELETE']) @auth_middleware.login_required def delete_conversation(conversation_id): """删除对话（软删除）""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 验证对话是否属于当前用户 conversation = Conversation.query.filter_by( conversation_id=conversation_id, user_id=user_id, is_deleted=False ).first() if not conversation: return jsonify({ 'success': False, 'error': '对话不存在或无权限访问' }), HTTPStatus.NOT_FOUND # 软删除对话 conversation.is_deleted = True conversation.deleted_at = get_beijing_time() conversation.updated_at = get_beijing_time() # 同时软删除该对话的所有消息 Message.query.filter_by( conversation_id=conversation_id, is_deleted=False ).update({ 'is_deleted': True, 'deleted_at': get_beijing_time() }) db.session.commit() return jsonify({ 'success': True, 'message': '对话已删除' }) except Exception as e: db.session.rollback() logger.error(f"删除对话时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations/new', methods=['POST']) @auth_middleware.login_required def create_new_conversation(): """创建新对话""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 获取用户对象 user = User.query.filter_by(user_id=user_id, is_deleted=False).first() if not user: return jsonify({ 'success': False, 'error': '用户不存在' }), HTTPStatus.NOT_FOUND data = request.json or {} title = data.get('title', '新对话').strip() if not title: title = '新对话' if len(title) > 500: return jsonify({ 'success': False, 'error': '标题长度不能超过500个字符' }), HTTPStatus.BAD_REQUEST # 生成新的session_id session_id = generate_session_id(user_id) # 创建新对话 conversation = Conversation( user_id=user_id, title=title, conversation_type='chat', session_id=session_id, message_count=0 ) db.session.add(conversation) db.session.commit() # 更新用户的当前对话ID user_current_conversations[str(user_id)] = conversation.conversation_id return jsonify({ 'success': True, 'message': '新对话已创建', 'conversation': { 'conversation_id': conversation.conversation_id, 'title': conversation.title, 'created_at': conversation.created_at.isoformat(), 'updated_at': conversation.updated_at.isoformat() } }) except Exception as e: db.session.rollback() logger.error(f"创建新对话时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations/clear-current', methods=['POST']) @auth_middleware.login_required def clear_current_conversation(): """清除用户的当前对话映射，强制下次创建新对话""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 清除用户的当前对话映射 if str(user_id) in user_current_conversations: del user_current_conversations[str(user_id)] return jsonify({ 'success': True, 'message': '当前对话映射已清除' }) except Exception as e: logger.error(f"清除当前对话映射时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations//switch', methods=['POST']) @auth_middleware.login_required def switch_conversation(conversation_id): """切换到指定对话""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 切换到指定对话 conversation = switch_to_conversation(user_id, conversation_id) return jsonify({ 'success': True, 'message': '对话切换成功', 'conversation': { 'conversation_id': conversation.conversation_id, 'title': conversation.title, 'session_id': conversation.session_id, 'message_count': conversation.message_count, 'last_message_at': conversation.last_message_at.isoformat() if conversation.last_message_at else None, 'created_at': conversation.created_at.isoformat(), 'updated_at': conversation.updated_at.isoformat() } }) except ValueError as e: return jsonify({ 'success': False, 'error': str(e) }), HTTPStatus.NOT_FOUND except Exception as e: logger.error(f"切换对话时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/conversations//status', methods=['GET']) @auth_middleware.login_required def get_conversation_status(conversation_id): """获取对话的实时状态信息""" try: # 从认证中间件获取用户信息 user_info = getattr(request, 'user_info', {}) user_id = user_info.get('db_user_id') # 使用数据库用户ID if not user_id: return jsonify({ 'success': False, 'error': '用户信息获取失败' }), HTTPStatus.UNAUTHORIZED # 验证对话是否属于当前用户 conversation = Conversation.query.filter_by( conversation_id=conversation_id, user_id=user_id, is_deleted=False ).first() if not conversation: return jsonify({ 'success': False, 'error': '对话不存在或无权限访问' }), HTTPStatus.NOT_FOUND # 获取对话的最新消息 latest_message = Message.query.filter_by( conversation_id=conversation_id, is_deleted=False ).order_by(Message.message_order.desc()).first() status_info = { 'conversation_id': conversation_id, 'title': conversation.title, 'message_count': conversation.message_count, 'last_message_at': conversation.last_message_at.isoformat() if conversation.last_message_at else None, 'updated_at': conversation.updated_at.isoformat(), 'has_latest_message': latest_message is not None, 'latest_message_role': latest_message.role if latest_message else None, 'latest_message_preview': latest_message.content[:100] + '...' if latest_message and len(latest_message.content) > 100 else latest_message.content if latest_message else None } return jsonify({ 'success': True, 'status': status_info }) except Exception as e: logger.error(f"获取对话状态时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR @app.route('/api/messages//edit', methods=['POST']) @auth_middleware.login_required @rate_limit(config.rate_limit_requests, config.rate_limit_window) def edit_message(message_id): """编辑用户消息并重新生成AI回复（流式响应）""" try: # 从请求中获取数据 data = request.json or {} new_content = data.get('content', '') agent = data.get('agent') or 'kuaishou' # 输入验证 new_content, error = sanitize_input(new_content, config.max_message_length) if error: # 流式响应中的错误返回 return Response( f"data: {json.dumps({'success': False, 'error': error, 'done': True})}\n\n", content_type='text/event-stream' ) # 从认证中间件获取用户信息 auth_user_info = getattr(request, 'user_info', {}) auth_user_id = auth_user_info.get('db_user_id') # 使用数据库用户ID if not auth_user_id: return Response( f"data: {json.dumps({'success': False, 'error': '用户信息获取失败', 'done': True})}\n\n", content_type='text/event-stream' ) # 获取要编辑的消息 message = Message.query.filter_by( message_id=message_id, is_deleted=False ).first() if not message: return Response( f"data: {json.dumps({'success': False, 'error': '消息不存在', 'done': True})}\n\n", content_type='text/event-stream' ) # 验证消息是否是用户消息 if message.role != 'user': return Response( f"data: {json.dumps({'success': False, 'error': '只能编辑用户消息', 'done': True})}\n\n", content_type='text/event-stream' ) # 获取对话信息 conversation = Conversation.query.filter_by( conversation_id=message.conversation_id, user_id=auth_user_id, is_deleted=False ).first() if not conversation: return Response( f"data: {json.dumps({'success': False, 'error': '对话不存在或无权限访问', 'done': True})}\n\n", content_type='text/event-stream' ) # 记录请求信息 log_request_info(new_content, str(auth_user_id), 'POST') # 选择对应智能体 APP_ID chosen_app_id, appid_env_key = config.get_app_id_for_agent_with_source(agent) if not chosen_app_id: logger.error(f"未找到对应智能体APP_ID，agent={agent}") return Response( f"data: {json.dumps({'success': False, 'error': f'服务器未配置该智能体的 APP_ID: {agent}', 'done': True})}\n\n", content_type='text/event-stream' ) # 使用对话的session_id api_session_id = conversation.session_id # ===== 立即软删除当前消息及之后的所有消息（在流式响应开始前） ===== try: # 计算将被删除的消息数（包括当前消息及之后的所有消息） deleted_count = Message.query.filter_by( conversation_id=conversation.conversation_id, is_deleted=False ).filter( Message.message_order >= message.message_order ).count() # 软删除当前消息及之后的所有消息（包括原来的AI回复等） Message.query.filter_by( conversation_id=conversation.conversation_id, is_deleted=False ).filter( Message.message_order >= message.message_order ).update({ 'is_deleted': True, 'deleted_at': get_beijing_time() }, synchronize_session=False) # 更新消息计数：减去被删除的消息数 conversation.message_count -= deleted_count db.session.commit() # 保存编辑后的用户消息（使用相同的message_order） new_user_message = save_message_to_database( auth_user_id, conversation.conversation_id, 'user', new_content, None, message_order=message.message_order ) except Exception as db_error: db.session.rollback() logger.error(f"软删除消息或保存编辑消息时发生错误: {str(db_error)}", exc_info=True) return Response( f"data: {json.dumps({'success': False, 'error': f'更新消息失败: {str(db_error)}', 'done': True})}\n\n", content_type='text/event-stream' ) # 获取该消息之前的所有消息（不包括已删除的消息，现在数据库中已删除的消息不会出现在查询结果中） previous_messages = Message.query.filter_by( conversation_id=message.conversation_id, is_deleted=False ).filter( Message.message_order < message.message_order ).order_by(Message.message_order.asc()).all() # 构建对话历史（使用编辑后的消息） messages = [] for msg in previous_messages: messages.append({"role": msg.role, "content": msg.content}) # 添加编辑后的用户消息（刚保存的） messages.append({"role": "user", "content": new_content}) # 截断对话历史以控制内存使用 messages = truncate_conversation_history(messages, config.max_conversation_history) # 定义流式响应生成器 def generate() -> Generator[str, None, None]: try: # 发送初始响应 initial_data = { 'success': True, 'chunk': '', 'done': False, 'agent': agent, 'message_id': message_id } yield f"data: {json.dumps(initial_data)}\n\n" try: # 使用DashScope SDK直接调用 - 多轮对话模式 params = { "api_key": config.api_key, "app_id": chosen_app_id, "messages": messages, # 发送完整的对话历史 "stream": True, "incremental_output": True, "has_thoughts": True } # 使用对话的session_id if api_session_id: params["session_id"] = api_session_id response = Application.call(**params) new_session_id: Optional[str] = None accumulated_text: str = "" thoughts_content: List[str] = [] # 处理流式响应 for chunk in response: # 检查客户端是否断开连接 if request.environ.get('werkzeug.server.shutdown') or not request: break if chunk.status_code != HTTPStatus.OK: yield f"data: {json.dumps({'success': False, 'error': f'错误: {chunk.status_code} - {chunk.message}', 'request_id': chunk.request_id, 'done': True, 'agent': agent})}\n\n" return # 处理思考过程 if hasattr(chunk.output, 'thoughts') and chunk.output.thoughts: for it in chunk.output.thoughts: if it.action_type == 'reasoning': content = str(it.thought) if not isinstance(it.thought, str) else it.thought thoughts_content.append(content) yield f"data: {json.dumps({'success': True, 'thought': content, 'done': False, 'agent': agent})}\n\n" # 处理回答内容 if hasattr(chunk.output, 'text') and chunk.output.text: current_text = str(chunk.output.text) accumulated_text += current_text yield f"data: {json.dumps({'success': True, 'chunk': current_text, 'done': False, 'agent': agent})}\n\n" # 检查是否有会话ID if hasattr(chunk.output, 'session_id') and chunk.output.session_id: new_session_id = chunk.output.session_id # 处理结束，保存AI回复到数据库 try: # 更新session_id（如果需要） if new_session_id and new_session_id != api_session_id: conversation.session_id = new_session_id db.session.commit() # 保存新的AI回复（如果有） if accumulated_text: thinking_text = '\n'.join(thoughts_content) if thoughts_content else None ai_message = save_message_to_database( auth_user_id, conversation.conversation_id, 'assistant', accumulated_text, thinking_text, message_order=message.message_order + 1 ) # save_message_to_database 已经提交，message_count 已更新 else: # 如果没有AI回复，更新对话的更新时间 conversation.updated_at = get_beijing_time() conversation.last_message_at = get_beijing_time() db.session.commit() except Exception as db_error: db.session.rollback() logger.error(f"保存AI回复到数据库时发生错误: {str(db_error)}", exc_info=True) # 不影响主要功能，继续返回结果 # 发送完成标记 final_data = { 'success': True, 'done': True, 'thoughts': thoughts_content, 'agent': agent, 'session_id': new_session_id or api_session_id, 'message_id': message_id, 'conversation_id': conversation.conversation_id } yield f"data: {json.dumps(final_data)}\n\n" except Exception as e: logger.error(f"DashScope SDK 调用错误: {str(e)}", exc_info=True) yield f"data: {json.dumps({'success': False, 'error': f'DashScope API调用错误: {str(e)}', 'done': True, 'agent': agent})}\n\n" except Exception as e: logger.error(f"流式API调用错误: {str(e)}", exc_info=True) yield f"data: {json.dumps({'success': False, 'error': f'发生错误: {str(e)}', 'done': True, 'agent': agent})}\n\n" # 创建并返回流式响应 response = Response(stream_with_context(generate()), content_type='text/event-stream') response.headers['Cache-Control'] = 'no-cache' response.headers['X-Accel-Buffering'] = 'no' return response except Exception as e: logger.error(f"编辑消息时发生错误: {str(e)}", exc_info=True) return jsonify({ 'success': False, 'error': f"发生错误: {str(e)}" }), HTTPStatus.INTERNAL_SERVER_ERROR # 健康检查端点 @app.route('/health', methods=['GET']) def health_check(): """健康检查端点""" return jsonify({ 'status': 'ok', 'app_version': '2.0.0', 'python_version': '3.12', 'timestamp': time.time() }) if __name__ == '__main__': app.run(host='0.0.0.0', port=config.port, debug=config.debug)