selfrelease
5278190750
- 后端: 凭证生成引擎、金蝶导出器、凭证模板服务 - 后端: 成本分析服务、AI问答服务 - 后端: 科目映射CRUD API、分析API、QA API - 后端: 集成测试(认证/任务/凭证) 49个测试全部通过 - 前端: 凭证管理、成本分析、导出中心、知识库、系统设置页面 - 前端: AuthGuard认证守卫、Dashboard AI聊天功能 - 前端: Playwright E2E测试 16 passed, 1 skipped - 基础设施: Docker Compose、Nginx反向代理、.env.example - 文档: 用户手册、管理员手册、发布检查清单
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""
|
||
预置问题问答服务
|
||
|
||
根据任务状态生成预置问题，并基于真实数据回答用户问题
|
||
"""
|
||
|
||
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 "暂无法回答该问题，请稍后重试。"
|