import logging import sys logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('app_debug.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) from flask import Flask, render_template, request, jsonify, session from services.deepseek_service import deepseek_service from config import Config import uuid import datetime app = Flask(__name__) app.secret_key = "interviewer-secret-key-change-in-production" # 设置session过期时间为1小时 app.permanent_session_lifetime = datetime.timedelta(hours=1) # 内存数据库存储面试数据，避免session过期导致的"面试不存在"错误 interviews_db = {} INTERVIEW_PHASES = { "intro": "自我介绍", "professional": "专业能力", "scenario": "情景假设", "career": "职业规划", "closing": "面试结束" } QUESTION_COUNTS = { "intro": 2, "professional": 4, "scenario": 2, "career": 1 } @app.route("/") def index(): return render_template("index.html") @app.route("/api/chat", methods=["POST"]) def chat(): data = request.json user_input = data.get("message", "").strip() system_type = data.get("system_type", "general_assistant") conversation_key = data.get("conversation_key", "default") if not user_input: return jsonify({"error": "请输入内容"}), 400 if conversation_key not in session: session[conversation_key] = [] conversation_history = session[conversation_key] try: result = deepseek_service.chat( user_input=user_input, conversation_history=conversation_history, system_type=system_type ) conversation_history.append({"role": "user", "content": user_input}) conversation_history.append(result) session[conversation_key] = conversation_history[-20:] return jsonify({"response": result["content"]}) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/resume/optimize", methods=["POST"]) def optimize_resume(): data = request.json resume_content = data.get("resume_content", "").strip() target_position = data.get("target_position", "").strip() if not resume_content: return jsonify({"error": "请提供简历内容"}), 400 try: result = deepseek_service.optimize_resume(resume_content, target_position) return jsonify(result) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/interview/start", methods=["POST"]) def start_interview(): data = request.json job_position = data.get("job_position", "").strip() difficulty = data.get("difficulty", "intermediate") if not job_position: return jsonify({"error": "请选择目标岗位"}), 400 interview_id = str(uuid.uuid4()) interview_data = { "job_position": job_position, "difficulty": difficulty, "current_phase": "intro", "question_count": 0, "conversation_history": [], "is_active": True } session[f"interview_{interview_id}"] = interview_data interviews_db[interview_id] = interview_data first_question = deepseek_service.generate_interview_question( job_position=job_position, difficulty=difficulty, phase="intro" ) session[f"interview_{interview_id}"]["conversation_history"].append({ "role": "assistant", "content": f"你好！我是面试官，现在开始针对{job_position}岗位的面试。\n\n{first_question}" }) return jsonify({ "interview_id": interview_id, "job_position": job_position, "difficulty": difficulty, "question": first_question, "phase": "intro" }) @app.route("/api/interview/answer", methods=["POST"]) def answer_question(): data = request.json interview_id = data.get("interview_id", "").strip() user_answer = data.get("answer", "").strip() request_feedback = data.get("request_feedback", False) if not interview_id: return jsonify({"error": "无效的面试ID"}), 400 # 首先从内存数据库查找面试数据 if interview_id not in interviews_db: # 如果内存数据库中没有，再检查session interview_key = f"interview_{interview_id}" if interview_key not in session: return jsonify({"error": "面试不存在或已结束"}), 400 # 如果session中有，同步到内存数据库 interviews_db[interview_id] = session[interview_key] interview_data = interviews_db[interview_id] if not interview_data["is_active"]: return jsonify({"error": "面试已结束"}), 400 if not user_answer: return jsonify({"error": "请输入你的回答"}), 400 conversation_history = interview_data["conversation_history"] if request_feedback: last_question = "" logger.debug(f"查找最后一个问题，对话历史长度：{len(conversation_history)}") for msg in reversed(conversation_history): logger.debug(f"检查消息：角色={msg['role']}, 内容={msg['content'][:50]}...") if msg["role"] == "assistant" and ("？" in msg["content"] or "?" in msg["content"]): last_question = msg["content"] logger.debug(f"找到最后一个问题：{last_question[:50]}...") break if last_question: try: feedback = deepseek_service.chat_with_feedback( user_input=last_question, user_answer=user_answer, conversation_history=conversation_history[:-1] ) conversation_history.append({"role": "user", "content": user_answer}) conversation_history.append(feedback) return jsonify({ "feedback": feedback["content"], "ended": False }) except Exception as e: logger.error(f"生成反馈失败：{str(e)}", exc_info=True) return jsonify({"error": f"生成反馈失败：{str(e)}"}), 500 else: logger.warning("没有找到最后一个问题") return jsonify({"error": "没有找到相关问题"}), 400 conversation_history.append({"role": "user", "content": user_answer}) interview_data["question_count"] += 1 current_phase = interview_data["current_phase"] phase_order = ["intro", "professional", "scenario", "career", "closing"] current_index = phase_order.index(current_phase) questions_in_phase = QUESTION_COUNTS.get(current_phase, 1) next_question = None if interview_data["question_count"] >= questions_in_phase: if current_index < len(phase_order) - 1: next_phase = phase_order[current_index + 1] interview_data["current_phase"] = next_phase interview_data["question_count"] = 0 if next_phase == "closing": conversation_history.append({ "role": "assistant", "content": "面试结束！感谢你的参与。点击下方按钮获取本次面试的详细反馈。" }) interview_data["is_active"] = False return jsonify({ "question": None, "feedback": "面试结束", "ended": True, "conversation_history": conversation_history[-10:] }) else: phase_names = { "intro": "自我介绍", "professional": "专业能力", "scenario": "情景假设", "career": "职业规划" } try: next_question = deepseek_service.generate_interview_question( job_position=interview_data["job_position"], difficulty=interview_data["difficulty"], conversation_history=conversation_history, phase=next_phase ) conversation_history.append({ "role": "assistant", "content": f"（{phase_names.get(next_phase, next_phase)}阶段）\n\n{next_question}" }) except Exception as e: return jsonify({"error": f"生成问题失败：{str(e)}"}), 500 else: try: next_question = deepseek_service.generate_interview_question( job_position=interview_data["job_position"], difficulty=interview_data["difficulty"], conversation_history=conversation_history, phase=current_phase ) conversation_history.append({"role": "assistant", "content": next_question}) except Exception as e: return jsonify({"error": f"生成问题失败：{str(e)}"}), 500 # 同时更新内存数据库和session中的数据 interviews_db[interview_id] = interview_data session[f"interview_{interview_id}"] = interview_data return jsonify({ "question": next_question, "feedback": None, "ended": False, "phase": interview_data["current_phase"] }) @app.route("/api/interview/feedback", methods=["POST"]) def get_interview_feedback(): logger.info("接收到面试反馈请求") try: # 首先确保能够获取请求数据 if not request.is_json: logger.warning("请求数据不是JSON格式") return jsonify({"error": "请求数据必须是JSON格式"}), 400 data = request.json logger.debug(f"请求数据：{data}") # 检查必要参数 interview_id = data.get("interview_id", "").strip() conversation_history = data.get("conversation_history", []) if not interview_id: logger.warning("面试ID无效") return jsonify({"error": "无效的面试ID"}), 400 if not conversation_history: logger.warning("没有提供面试对话历史") return jsonify({"error": "没有提供面试对话历史"}), 400 # 获取岗位信息 job_position = "" if interview_id in interviews_db: job_position = interviews_db[interview_id].get("job_position", "") logger.debug(f"岗位信息：{job_position}") logger.debug(f"对话历史长度：{len(conversation_history)}") # 构建系统提示 system_prompt = """作为一位专业的面试评估专家，请对整场面试进行全面评估。 请提供： 1. 整体表现评分（0-100分）和评级（优秀/良好/一般/需改进） 2. 各轮回答的详细分析（针对每个问题和回答给出具体评价） 3. strengths（优势）- 列出至少3点 4. areas_for_improvement（需要改进的方面）- 列出至少3点 5. 具体的准备建议（针对改进点给出可操作的建议） 请用中文回复，格式清晰、结构化，避免过于笼统的描述。""" # 构建对话文本 conversation_text = "\n\n".join([ f"{'面试官' if msg['role'] == 'assistant' else '候选人'}：{msg['content']}" for msg in conversation_history ]) # 构建用户提示 if job_position: user_prompt = f"请分析以下针对{job_position}岗位的面试对话并给出综合反馈：\n\n{conversation_text}" else: user_prompt = f"请分析以下面试对话并给出综合反馈：\n\n{conversation_text}" # 构建完整的消息 messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] logger.debug("准备调用DeepSeek API生成反馈") logger.debug(f"API请求消息数量：{len(messages)}") # 调用DeepSeek API response = deepseek_service._call_api(messages) logger.debug("DeepSeek API调用成功") # 处理API响应 if not response or "choices" not in response or not response["choices"]: logger.error("API响应格式错误：缺少choices字段") return jsonify({"error": "生成反馈失败：API响应格式错误"}), 500 feedback = response["choices"][0]["message"]["content"] logger.info("反馈生成成功") logger.debug(f"生成的反馈内容长度：{len(feedback)}字符") logger.debug(f"生成的反馈内容开头：{feedback[:100]}...") return jsonify({"feedback": feedback}) except Exception as e: logger.error(f"生成面试反馈失败：{str(e)}", exc_info=True) # 返回更具体的错误信息 return jsonify({"error": f"生成面试反馈失败：{str(e)}", "details": str(type(e).__name__)}), 500 if __name__ == "__main__": app.run( host=Config.APP_HOST, port=Config.APP_PORT, debug=False )