""" 预置问题问答服务 根据任务状态生成预置问题，并基于真实数据回答用户问题 """ import json import logging from typing import List from app.core.config import get_settings from app.services.analysis.cost_calculator import CostCalculatorService logger = logging.getLogger(__name__) settings = get_settings() # 预置问题模板 PRESET_QUESTIONS = { "CREATED": [ "本月需要对账哪些文件？", "如何开始一个新的对账任务？", ], "FILES_UPLOADED": [ "本月还缺哪些文件？", "文件上传后下一步做什么？", ], "PARSING": [ "文件解析需要多长时间？", "解析过程中发现了什么问题？", ], "WAITING_MAPPING_CONFIRM": [ "哪些字段需要人工确认？", "AI 识别的准确率如何？", "低置信度字段有哪些？", ], "MAPPING_CONFIRMED": [ "字段映射确认后下一步做什么？", "可以开始对账了吗？", ], "RECONCILING": [ "对账进度如何？", "对账过程中发现了什么异常？", ], "COMPLETED": [ "本次对账发现了多少异常？", "哪些异常最严重？", "本月人工成本是多少？", "为什么本月人工成本上涨？", "哪些部门变化最大？", "现在可以生成金蝶凭证吗？", ], "FAILED": [ "对账失败的原因是什么？", "如何修复错误？", ], } # 问题到数据查询的映射 QUESTION_KEYWORDS = { "异常": "exceptions", "成本": "cost", "上涨": "cost", "下降": "cost", "部门": "department", "凭证": "voucher", "文件": "files", "字段": "fields", "置信度": "fields", } class QAService: """预置问题问答服务""" def __init__(self, db): self.db = db self.cost_calculator = CostCalculatorService(db) async def get_suggested_questions(self, task_id: int, context: str = "") -> List[str]: """ 根据当前任务状态生成合适的预置问题 Args: task_id: 任务ID context: 上下文信息（可选） Returns: 预置问题列表（5-8个） """ from app.models.reconciliation_task import ReconciliationTask task = await self.db.get(ReconciliationTask, task_id) if not task: return PRESET_QUESTIONS.get("CREATED", []) status = task.status questions = PRESET_QUESTIONS.get(status, []).copy() # 如果 context 指定了特定场景，追加相关问题 if context == "cost_analysis": cost_questions = [ "本月人工成本是多少？", "为什么本月人工成本上涨？", "哪些部门变化最大？", "社保和公积金成本占比如何？", "新增员工对成本的影响有多大？", ] questions = cost_questions # 确保至少5个问题 if len(questions) < 5: questions.extend(PRESET_QUESTIONS.get("COMPLETED", [])[: 5 - len(questions)]) return questions[:8] async def answer_preset_question(self, task_id: int, question: str) -> str: """ 回答预置问题 Args: task_id: 任务ID question: 用户问题 Returns: 包含数据点的简洁答案 """ # 根据问题关键词决定数据来源 data_type = self._classify_question(question) if data_type == "cost": return await self._answer_cost_question(task_id, question) elif data_type == "exceptions": return await self._answer_exception_question(task_id, question) elif data_type == "department": return await self._answer_department_question(task_id, question) elif data_type == "files": return await self._answer_files_question(task_id, question) elif data_type == "fields": return await self._answer_fields_question(task_id, question) elif data_type == "voucher": return "凭证生成功能即将上线，请先完成对账和异常处理。" else: return await self._answer_with_ai(task_id, question) def _classify_question(self, question: str) -> str: """根据问题关键词分类""" for keyword, data_type in QUESTION_KEYWORDS.items(): if keyword in question: return data_type return "general" async def _answer_cost_question(self, task_id: int, question: str) -> str: """回答成本相关问题""" summary = await self.cost_calculator.calculate_total_cost(task_id) if "上涨" in question or "增加" in question or "变化" in question: changes = await self.cost_calculator.analyze_cost_changes(task_id, task_id) new_cost = changes.get("new_employee_cost", 0) left_cost = changes.get("left_employee_saving", 0) adj_cost = changes.get("adjustment_cost", 0) return ( f"本月人工成本总额 {summary.total_cost:.2f} 元，" f"其中工资 {summary.salary_cost:.2f} 元、社保 {summary.social_security_cost:.2f} 元、公积金 {summary.fund_cost:.2f} 元。" f"新增员工增加成本 {new_cost:.2f} 元，离职员工减少 {left_cost:.2f} 元，薪资调整净变化 {adj_cost:.2f} 元。" ) return ( f"本月人工成本总额 {summary.total_cost:.2f} 元，" f"共 {summary.employee_count} 人。" f"其中工资成本 {summary.salary_cost:.2f} 元，" f"社保成本 {summary.social_security_cost:.2f} 元，" f"公积金成本 {summary.fund_cost:.2f} 元。" ) async def _answer_department_question(self, task_id: int, question: str) -> str: """回答部门相关问题""" dept_costs = await self.cost_calculator.calculate_by_department(task_id) if not dept_costs: return "暂无部门成本数据。" sorted_depts = sorted(dept_costs, key=lambda d: d.total_cost, reverse=True) top_dept = sorted_depts[0] return ( f"共 {len(sorted_depts)} 个部门，" f"成本最高的是 {top_dept.department}（{top_dept.total_cost:.2f} 元，{top_dept.employee_count} 人），" f"其次是 {sorted_depts[1].department if len(sorted_depts) > 1 else '无'}。" ) async def _answer_exception_question(self, task_id: int, question: str) -> str: """回答异常相关问题""" from app.models.exception_item import ExceptionItem from sqlalchemy import select, func count_result = await self.db.execute( select(func.count(ExceptionItem.id)).where(ExceptionItem.task_id == task_id) ) total = count_result.scalar() or 0 high_result = await self.db.execute( select(func.count(ExceptionItem.id)).where( ExceptionItem.task_id == task_id, ExceptionItem.severity == "HIGH", ) ) high_count = high_result.scalar() or 0 if "严重" in question: return f"本次对账共发现 {total} 个异常，其中高严重程度 {high_count} 个，建议优先处理。" return f"本次对账共发现 {total} 个异常，其中高严重程度 {high_count} 个。" async def _answer_files_question(self, task_id: int, question: str) -> str: """回答文件相关问题""" from app.models.uploaded_file import UploadedFile from sqlalchemy import select result = await self.db.execute( select(UploadedFile).where(UploadedFile.task_id == task_id) ) files = result.scalars().all() if not files: return "本月尚未上传任何文件，请上传工资表、社保表和个税表。" file_types = [f.file_type for f in files] missing = [] if "工资表" not in file_types: missing.append("工资表") if "社保表" not in file_types: missing.append("社保表") if "个税表" not in file_types: missing.append("个税表") if missing: return f"已上传 {len(files)} 个文件，还缺少：{'、'.join(missing)}。" return f"已上传 {len(files)} 个文件（工资表、社保表、个税表），可以进入下一步。" async def _answer_fields_question(self, task_id: int, question: str) -> str: """回答字段相关问题""" from app.models.field_mapping import FieldMapping from sqlalchemy import select, func low_result = await self.db.execute( select(func.count(FieldMapping.id)).where( FieldMapping.confidence < 0.7, FieldMapping.is_skipped == False, ) ) low_count = low_result.scalar() or 0 total_result = await self.db.execute( select(func.count(FieldMapping.id)) ) total = total_result.scalar() or 0 if "低置信度" in question or "确认" in question: return f"共有 {total} 个字段需要映射，其中 {low_count} 个低置信度字段需要人工确认。" return f"AI 共识别 {total} 个字段，其中 {low_count} 个低置信度字段需要人工确认。" async def _answer_with_ai(self, task_id: int, question: str) -> str: """使用 AI 回答通用问题""" try: summary = await self.cost_calculator.calculate_total_cost(task_id) context = f"任务ID: {task_id}, 总成本: {summary.total_cost}, 员工数: {summary.employee_count}" prompt = f"基于以下数据回答问题：\n数据：{context}\n问题：{question}\n要求：简洁2-3句话，包含数字。" from app.services.ai.cost_analyzer import CostAnalyzerService analyzer = CostAnalyzerService() result = await analyzer._call_llm(prompt) return result.strip() except Exception as e: logger.error(f"AI 回答失败: {e}") return "暂无法回答该问题，请稍后重试。"