# ============================================================================= # 企微IT智能服务台 — 答题与积分服务 # ============================================================================= # 说明:排队等待期间的答题系统,包含双模式题目选择、积分更新、插队计算 # # 答题双模式(决策 D1): # 模式A — 诊断题(info_locked=false):与当前问题相关的选择题 # 答案附加到会话上下文,供坐席接单时参考 # 模式B — IT知识题(info_locked=true):纯教育性质,提升IT素养 # # 积分规则(决策 D2): # 答对 +10分,答错不扣分,跨会话累积 # 5级等级:0-99 IT小白 → 100-299 IT入门 → 300-599 IT达人 → 600-999 IT专家 → 1000+ IT大师 # # 插队规则(决策 C4): # queue_priority = min(答题数 // 3, 2) # 每答3题前移1位,上限2 # ============================================================================= import logging import random from typing import Any, Dict, List, Optional from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.models.conversation import Conversation from app.models.quiz import ( EmployeePoints, QuizAnswer, QuizQuestion, ) from app.utils.response import AppException logger = logging.getLogger(__name__) # 积分常量 POINTS_PER_CORRECT = 10 MAX_QUEUE_PRIORITY = 2 QUIZ_PER_PRIORITY = 3 # 每答3题前移1位 class QuizService: """答题与积分服务。""" # ====================================================================== # 获取下一道题 # ====================================================================== async def get_next_question( self, db: AsyncSession, employee_id: str, conversation: Optional[Conversation] = None, ) -> Dict[str, Any]: """获取下一道题(双模式自动选择)。 模式选择逻辑: - 如果有活跃会话且 info_locked=false → 诊断题(模式A) - 如果有活跃会话且 info_locked=true → IT知识题(模式B) - 如果无活跃会话 → IT知识题(模式B) Args: db: 数据库会话 employee_id: 员工ID conversation: 当前会话(可选,排队时传入) Returns: Dict: 题目数据 {question_id, type, category, question, options} """ # 判定模式 use_diagnostic = ( conversation is not None and conversation.status == "queued" and not conversation.info_locked ) if use_diagnostic: # 模式A:诊断题 — 根据问题类别匹配 question = await self._get_diagnostic_question(db, employee_id, conversation) else: # 模式B:IT知识题 question = await self._get_knowledge_question(db, employee_id) if not question: return { "has_question": False, "message": "暂无更多题目,请稍后再试", } return { "has_question": True, "question_id": question.id, "type": question.type, "category": question.category, "difficulty": question.difficulty, "question": question.question, "options": question.options, } async def _get_diagnostic_question( self, db: AsyncSession, employee_id: str, conversation: Conversation, ) -> Optional[QuizQuestion]: """获取诊断题(模式A)。 诊断题按问题类别匹配,排除已答过的题目。 如果没有匹配的诊断题,降级为IT知识题。 Args: db: 数据库会话 employee_id: 员工ID conversation: 当前会话 Returns: QuizQuestion 或 None """ # 查找已答过的题目ID(避免重复) answered_ids_subquery = ( select(QuizAnswer.question_id) .where( QuizAnswer.employee_id == employee_id, QuizAnswer.conversation_id == conversation.id, ) ) # 查找诊断题(按问题类别匹配) stmt = ( select(QuizQuestion) .where( QuizQuestion.type == "diagnostic", QuizQuestion.is_active == True, # noqa: E712 QuizQuestion.id.notin_(answered_ids_subquery), ) .order_by(func.random()) .limit(1) ) result = await db.execute(stmt) question = result.scalar_one_or_none() if question: return question # 降级:如果没有匹配的诊断题,使用IT知识题 logger.info("无诊断题可用,降级为IT知识题: employee=%s", employee_id) return await self._get_knowledge_question(db, employee_id) async def _get_knowledge_question( self, db: AsyncSession, employee_id: str, ) -> Optional[QuizQuestion]: """获取IT知识题(模式B)。 排除已答过的题目,随机选取。 Args: db: 数据库会话 employee_id: 员工ID Returns: QuizQuestion 或 None """ # 查找已答过的题目ID(跨会话排除,避免重复) answered_ids_subquery = ( select(QuizAnswer.question_id) .where(QuizAnswer.employee_id == employee_id) ) stmt = ( select(QuizQuestion) .where( QuizQuestion.type == "knowledge", QuizQuestion.is_active == True, # noqa: E712 QuizQuestion.id.notin_(answered_ids_subquery), ) .order_by(func.random()) .limit(1) ) result = await db.execute(stmt) question = result.scalar_one_or_none() if question: return question # 如果所有题都答完了,重置(允许重复) logger.info("所有题目已答完,重置题目池: employee=%s", employee_id) stmt_all = ( select(QuizQuestion) .where( QuizQuestion.type == "knowledge", QuizQuestion.is_active == True, # noqa: E712 ) .order_by(func.random()) .limit(1) ) result = await db.execute(stmt_all) return result.scalar_one_or_none() # ====================================================================== # 提交答案 # ====================================================================== async def submit_answer( self, db: AsyncSession, employee_id: str, question_id: str, selected_index: int, conversation: Optional[Conversation] = None, ) -> Dict[str, Any]: """提交答案,返回正误+积分变化+插队效果+下一题。 处理流程: 1. 查询题目,判定正误 2. 记录答题(quiz_answers) 3. 更新积分账户(employee_points) 4. 如果在排队中,更新 queue_priority 5. 获取下一道题 Args: db: 数据库会话 employee_id: 员工ID question_id: 题目ID selected_index: 员工选择的答案索引 conversation: 当前会话(可选) Returns: Dict: {is_correct, correct_index, explanation, points_earned, total_points, level, queue_priority_changed, next_question} """ # 1. 查询题目 result = await db.execute( select(QuizQuestion).where(QuizQuestion.id == question_id) ) question = result.scalar_one_or_none() if not question: raise AppException(code=1004, message="题目不存在") # 2. 判定正误 is_correct = selected_index == question.correct_index points_earned = POINTS_PER_CORRECT if is_correct else 0 # 3. 记录答题 answer = QuizAnswer( employee_id=employee_id, conversation_id=conversation.id if conversation else None, question_id=question_id, selected_index=selected_index, is_correct=is_correct, points_earned=points_earned, ) db.add(answer) # 4. 更新积分账户 points_info = await self._update_employee_points( db, employee_id, points_earned, is_correct ) # 5. 如果在排队中,更新 queue_priority queue_priority_changed = False old_priority = 0 new_priority = 0 if conversation and conversation.status == "queued": old_priority = conversation.queue_priority # 计算本次会话的答题总数 answered_count = await db.scalar( select(func.count(QuizAnswer.id)).where( QuizAnswer.employee_id == employee_id, QuizAnswer.conversation_id == conversation.id, ) ) answered_count = answered_count or 0 new_priority = min(answered_count // QUIZ_PER_PRIORITY, MAX_QUEUE_PRIORITY) conversation.queue_priority = new_priority conversation.updated_at = __import__("datetime").datetime.now() queue_priority_changed = new_priority > old_priority if queue_priority_changed: logger.info( "答题插队: employee=%s, answered=%d, priority %d→%d", employee_id, answered_count, old_priority, new_priority ) await db.commit() # 6. 获取下一道题 next_question = await self.get_next_question(db, employee_id, conversation) # 7. 如果是诊断题且答对,将答案文本附加到会话上下文 if (question.type == "diagnostic" and conversation and is_correct and question.options and 0 <= selected_index < len(question.options)): await self._append_to_conversation_context( db, conversation, question.question, question.options[selected_index] ) return { "is_correct": is_correct, "correct_index": question.correct_index, "explanation": question.explanation, "points_earned": points_earned, "total_points": points_info["total_points"], "level": points_info["level"], "answered_count": points_info["answered_count"], "correct_count": points_info["correct_count"], "queue_priority_changed": queue_priority_changed, "old_priority": old_priority, "new_priority": new_priority, "next_question": next_question, } # ====================================================================== # 积分管理 # ====================================================================== async def _update_employee_points( self, db: AsyncSession, employee_id: str, points_earned: int, is_correct: bool, ) -> Dict[str, Any]: """更新员工积分账户。 Args: db: 数据库会话 employee_id: 员工ID points_earned: 本次获得积分 is_correct: 是否答对 Returns: Dict: 更新后的积分信息 """ result = await db.execute( select(EmployeePoints).where(EmployeePoints.employee_id == employee_id) ) points = result.scalar_one_or_none() if points: # 更新现有记录 points.total_points += points_earned points.answered_count += 1 if is_correct: points.correct_count += 1 points.level = EmployeePoints.calculate_level(points.total_points) else: # 首次答题,创建记录 points = EmployeePoints( employee_id=employee_id, total_points=points_earned, answered_count=1, correct_count=1 if is_correct else 0, level=EmployeePoints.calculate_level(points_earned), ) db.add(points) await db.flush() return { "total_points": points.total_points, "level": points.level, "answered_count": points.answered_count, "correct_count": points.correct_count, } async def get_employee_points( self, db: AsyncSession, employee_id: str ) -> Dict[str, Any]: """获取员工积分信息。 Args: db: 数据库会话 employee_id: 员工ID Returns: Dict: 积分信息 """ result = await db.execute( select(EmployeePoints).where(EmployeePoints.employee_id == employee_id) ) points = result.scalar_one_or_none() if points: return { "total_points": points.total_points, "level": points.level, "answered_count": points.answered_count, "correct_count": points.correct_count, "accuracy": round(points.correct_count / max(points.answered_count, 1) * 100, 1), } else: return { "total_points": 0, "level": "IT小白", "answered_count": 0, "correct_count": 0, "accuracy": 0, } # ====================================================================== # 答题历史 # ====================================================================== async def get_quiz_history( self, db: AsyncSession, employee_id: str, page: int = 1, page_size: int = 20, ) -> Dict[str, Any]: """获取答题历史记录。 Args: db: 数据库会话 employee_id: 员工ID page: 页码 page_size: 每页数量 Returns: Dict: {total, items, points} """ # 统计总数 total = await db.scalar( select(func.count(QuizAnswer.id)).where( QuizAnswer.employee_id == employee_id ) ) total = total or 0 # 分页查询 offset = (page - 1) * page_size stmt = ( select(QuizAnswer, QuizQuestion) .join(QuizQuestion, QuizAnswer.question_id == QuizQuestion.id) .where(QuizAnswer.employee_id == employee_id) .order_by(QuizAnswer.created_at.desc()) .offset(offset) .limit(page_size) ) result = await db.execute(stmt) rows = result.all() items = [] for answer, question in rows: items.append({ "answer_id": answer.id, "question_text": question.question, "options": question.options, "selected_index": answer.selected_index, "correct_index": question.correct_index, "is_correct": answer.is_correct, "points_earned": answer.points_earned, "category": question.category, "type": question.type, "created_at": answer.created_at.isoformat() if answer.created_at else None, }) # 积分信息 points_info = await self.get_employee_points(db, employee_id) return { "total": total, "page": page, "page_size": page_size, "items": items, "points": points_info, } # ====================================================================== # 诊断题答案附加到会话上下文 # ====================================================================== async def _append_to_conversation_context( self, db: AsyncSession, conversation: Conversation, question_text: str, answer_text: str, ) -> None: """将诊断题答案附加到会话上下文(供坐席接单时参考)。 这相当于员工在排队期间做了自助信息补充。 答案通过 WS 推送给坐席端,坐席接单时能看到。 Args: db: 数据库会话 conversation: 当前会话 question_text: 题目文本 answer_text: 员工选择的答案文本 """ try: from app.services.ws_manager import manager as ws_manager # 通过WS推送给坐席端 ws_data = { "type": "quiz_context_collected", "data": { "conversation_id": conversation.id, "employee_id": conversation.employee_id, "question": question_text, "answer": answer_text, "timestamp": __import__("datetime").datetime.now().isoformat(), }, } # 推送给坐席端(如果有分配的坐席) if conversation.assigned_agent_id: await ws_manager.send_to_agent(conversation.assigned_agent_id, ws_data) logger.info( "诊断题答案附加到上下文: conv=%s, Q=%s, A=%s", conversation.id, question_text[:50], answer_text[:50] ) except Exception as e: logger.warning("WS推送诊断题答案失败: %s", e) # 单例 _quiz_service: Optional[QuizService] = None def get_quiz_service() -> QuizService: """获取 QuizService 单例。""" global _quiz_service if _quiz_service is None: _quiz_service = QuizService() return _quiz_service