selfrelease
5278190750
- 后端: 凭证生成引擎、金蝶导出器、凭证模板服务 - 后端: 成本分析服务、AI问答服务 - 后端: 科目映射CRUD API、分析API、QA API - 后端: 集成测试(认证/任务/凭证) 49个测试全部通过 - 前端: 凭证管理、成本分析、导出中心、知识库、系统设置页面 - 前端: AuthGuard认证守卫、Dashboard AI聊天功能 - 前端: Playwright E2E测试 16 passed, 1 skipped - 基础设施: Docker Compose、Nginx反向代理、.env.example - 文档: 用户手册、管理员手册、发布检查清单
357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""
|
|
成本分析计算服务
|
|
|
|
计算人工成本总额、部门拆分、费用科目拆分、环比变化等
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.cost_analysis import CostAnalysis
|
|
from app.models.reconciliation_task import ReconciliationTask
|
|
from app.services.data_cleaner import DataCleaner
|
|
|
|
|
|
class CostSummary:
|
|
"""成本汇总结果"""
|
|
|
|
def __init__(
|
|
self,
|
|
total_cost: float = 0.0,
|
|
salary_cost: float = 0.0,
|
|
social_security_cost: float = 0.0,
|
|
fund_cost: float = 0.0,
|
|
employee_count: int = 0,
|
|
):
|
|
self.total_cost = total_cost
|
|
self.salary_cost = salary_cost
|
|
self.social_security_cost = social_security_cost
|
|
self.fund_cost = fund_cost
|
|
self.employee_count = employee_count
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"total_cost": self.total_cost,
|
|
"salary_cost": self.salary_cost,
|
|
"social_security_cost": self.social_security_cost,
|
|
"fund_cost": self.fund_cost,
|
|
"employee_count": self.employee_count,
|
|
}
|
|
|
|
|
|
class DepartmentCost:
|
|
"""部门成本"""
|
|
|
|
def __init__(
|
|
self,
|
|
department: str,
|
|
employee_count: int = 0,
|
|
salary_cost: float = 0.0,
|
|
social_security_cost: float = 0.0,
|
|
fund_cost: float = 0.0,
|
|
):
|
|
self.department = department
|
|
self.employee_count = employee_count
|
|
self.salary_cost = salary_cost
|
|
self.social_security_cost = social_security_cost
|
|
self.fund_cost = fund_cost
|
|
|
|
@property
|
|
def total_cost(self) -> float:
|
|
return self.salary_cost + self.social_security_cost + self.fund_cost
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"department": self.department,
|
|
"employee_count": self.employee_count,
|
|
"salary_cost": self.salary_cost,
|
|
"social_security_cost": self.social_security_cost,
|
|
"fund_cost": self.fund_cost,
|
|
"total_cost": self.total_cost,
|
|
}
|
|
|
|
|
|
class CostCalculatorService:
|
|
"""成本分析计算服务"""
|
|
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def calculate_total_cost(self, task_id: int) -> CostSummary:
|
|
"""
|
|
计算人工成本总额
|
|
|
|
Args:
|
|
task_id: 对账任务ID
|
|
|
|
Returns:
|
|
成本汇总结果
|
|
"""
|
|
cleaned_data = await self._load_cleaned_data(task_id)
|
|
|
|
salary_cost = 0.0
|
|
social_security_cost = 0.0
|
|
fund_cost = 0.0
|
|
employee_count = 0
|
|
|
|
for record in cleaned_data:
|
|
salary_cost += float(record.get("应发工资", 0) or 0)
|
|
social_security_cost += float(record.get("养老保险(公司)", 0) or 0)
|
|
social_security_cost += float(record.get("医疗保险(公司)", 0) or 0)
|
|
social_security_cost += float(record.get("失业保险(公司)", 0) or 0)
|
|
fund_cost += float(record.get("公积金(公司)", 0) or 0)
|
|
employee_count += 1
|
|
|
|
total_cost = salary_cost + social_security_cost + fund_cost
|
|
|
|
return CostSummary(
|
|
total_cost=total_cost,
|
|
salary_cost=salary_cost,
|
|
social_security_cost=social_security_cost,
|
|
fund_cost=fund_cost,
|
|
employee_count=employee_count,
|
|
)
|
|
|
|
async def calculate_by_department(self, task_id: int) -> List[DepartmentCost]:
|
|
"""
|
|
按部门汇总人工成本
|
|
|
|
Args:
|
|
task_id: 对账任务ID
|
|
|
|
Returns:
|
|
部门成本列表
|
|
"""
|
|
cleaned_data = await self._load_cleaned_data(task_id)
|
|
|
|
dept_map: Dict[str, DepartmentCost] = {}
|
|
|
|
for record in cleaned_data:
|
|
department = record.get("部门", "未分配") or "未分配"
|
|
if department not in dept_map:
|
|
dept_map[department] = DepartmentCost(department=department)
|
|
|
|
dept = dept_map[department]
|
|
dept.employee_count += 1
|
|
dept.salary_cost += float(record.get("应发工资", 0) or 0)
|
|
dept.social_security_cost += float(record.get("养老保险(公司)", 0) or 0)
|
|
dept.social_security_cost += float(record.get("医疗保险(公司)", 0) or 0)
|
|
dept.social_security_cost += float(record.get("失业保险(公司)", 0) or 0)
|
|
dept.fund_cost += float(record.get("公积金(公司)", 0) or 0)
|
|
|
|
return list(dept_map.values())
|
|
|
|
async def calculate_by_expense_type(self, task_id: int) -> Dict[str, float]:
|
|
"""
|
|
按费用科目拆分人工成本
|
|
|
|
Args:
|
|
task_id: 对账任务ID
|
|
|
|
Returns:
|
|
费用科目 -> 金额 的映射
|
|
"""
|
|
dept_costs = await self.calculate_by_department(task_id)
|
|
|
|
expense_map: Dict[str, float] = {
|
|
"管理费用": 0.0,
|
|
"销售费用": 0.0,
|
|
"研发费用": 0.0,
|
|
"其他": 0.0,
|
|
}
|
|
|
|
dept_to_expense = {
|
|
"研发部": "研发费用",
|
|
"研发中心": "研发费用",
|
|
"技术部": "研发费用",
|
|
"销售部": "销售费用",
|
|
"市场部": "销售费用",
|
|
"商务部": "销售费用",
|
|
}
|
|
|
|
for dept in dept_costs:
|
|
expense_type = dept_to_expense.get(dept.department, "管理费用")
|
|
expense_map[expense_type] = expense_map.get(expense_type, 0.0) + dept.total_cost
|
|
|
|
return expense_map
|
|
|
|
async def calculate_month_over_month(
|
|
self, task_id: int, prev_task_id: int
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
计算环比变化
|
|
|
|
Args:
|
|
task_id: 当前任务ID
|
|
prev_task_id: 上月任务ID
|
|
|
|
Returns:
|
|
环比变化数据
|
|
"""
|
|
current = await self.calculate_total_cost(task_id)
|
|
previous = await self.calculate_total_cost(prev_task_id)
|
|
|
|
def _calc_change(curr: float, prev: float) -> Dict[str, Any]:
|
|
amount_change = curr - prev
|
|
ratio_change = (amount_change / prev * 100) if prev > 0 else 0.0
|
|
return {
|
|
"current": curr,
|
|
"previous": prev,
|
|
"amount_change": amount_change,
|
|
"ratio_change": round(ratio_change, 2),
|
|
}
|
|
|
|
return {
|
|
"total_cost": _calc_change(current.total_cost, previous.total_cost),
|
|
"salary_cost": _calc_change(current.salary_cost, previous.salary_cost),
|
|
"social_security_cost": _calc_change(
|
|
current.social_security_cost, previous.social_security_cost
|
|
),
|
|
"fund_cost": _calc_change(current.fund_cost, previous.fund_cost),
|
|
"employee_count": {
|
|
"current": current.employee_count,
|
|
"previous": previous.employee_count,
|
|
"amount_change": current.employee_count - previous.employee_count,
|
|
},
|
|
}
|
|
|
|
async def analyze_cost_changes(
|
|
self, curr_task_id: int, prev_task_id: int
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
分析成本变化原因
|
|
|
|
Args:
|
|
curr_task_id: 当前任务ID
|
|
prev_task_id: 上月任务ID
|
|
|
|
Returns:
|
|
变化分析结果，包含新增/离职员工影响、薪资调整影响等
|
|
"""
|
|
curr_data = await self._load_cleaned_data(curr_task_id)
|
|
prev_data = await self._load_cleaned_data(prev_task_id)
|
|
|
|
curr_map = {r.get("员工姓名", ""): r for r in curr_data if r.get("员工姓名")}
|
|
prev_map = {r.get("员工姓名", ""): r for r in prev_data if r.get("员工姓名")}
|
|
|
|
new_employees = []
|
|
left_employees = []
|
|
salary_adjustments = []
|
|
|
|
for name, record in curr_map.items():
|
|
if name not in prev_map:
|
|
new_employees.append({
|
|
"name": name,
|
|
"salary": float(record.get("应发工资", 0) or 0),
|
|
})
|
|
else:
|
|
prev_salary = float(prev_map[name].get("应发工资", 0) or 0)
|
|
curr_salary = float(record.get("应发工资", 0) or 0)
|
|
if abs(curr_salary - prev_salary) > 0.01:
|
|
salary_adjustments.append({
|
|
"name": name,
|
|
"previous": prev_salary,
|
|
"current": curr_salary,
|
|
"change": curr_salary - prev_salary,
|
|
})
|
|
|
|
for name, record in prev_map.items():
|
|
if name not in curr_map:
|
|
left_employees.append({
|
|
"name": name,
|
|
"salary": float(record.get("应发工资", 0) or 0),
|
|
})
|
|
|
|
new_cost = sum(e["salary"] for e in new_employees)
|
|
left_cost = sum(e["salary"] for e in left_employees)
|
|
adjustment_cost = sum(a["change"] for a in salary_adjustments)
|
|
|
|
return {
|
|
"new_employees": new_employees,
|
|
"left_employees": left_employees,
|
|
"salary_adjustments": salary_adjustments,
|
|
"new_employee_cost": new_cost,
|
|
"left_employee_saving": left_cost,
|
|
"adjustment_cost": adjustment_cost,
|
|
"net_change": new_cost - left_cost + adjustment_cost,
|
|
}
|
|
|
|
async def save_analysis(
|
|
self,
|
|
task_id: int,
|
|
company_id: int,
|
|
summary: CostSummary,
|
|
department_breakdown: Optional[List[DepartmentCost]] = None,
|
|
expense_breakdown: Optional[Dict[str, float]] = None,
|
|
month_over_month: Optional[Dict[str, Any]] = None,
|
|
ai_summary: Optional[str] = None,
|
|
) -> CostAnalysis:
|
|
"""
|
|
保存成本分析结果到数据库
|
|
|
|
Args:
|
|
task_id: 任务ID
|
|
company_id: 企业ID
|
|
summary: 成本汇总
|
|
department_breakdown: 部门拆分
|
|
expense_breakdown: 费用科目拆分
|
|
month_over_month: 环比变化
|
|
ai_summary: AI 摘要
|
|
|
|
Returns:
|
|
创建的成本分析记录
|
|
"""
|
|
analysis = CostAnalysis(
|
|
task_id=task_id,
|
|
company_id=company_id,
|
|
total_cost=summary.total_cost,
|
|
salary_cost=summary.salary_cost,
|
|
social_security_cost=summary.social_security_cost,
|
|
fund_cost=summary.fund_cost,
|
|
department_breakdown=(
|
|
[d.to_dict() for d in department_breakdown] if department_breakdown else None
|
|
),
|
|
expense_breakdown=expense_breakdown,
|
|
month_over_month=month_over_month,
|
|
ai_summary=ai_summary,
|
|
)
|
|
self.db.add(analysis)
|
|
await self.db.commit()
|
|
await self.db.refresh(analysis)
|
|
return analysis
|
|
|
|
async def get_analysis(self, task_id: int) -> Optional[CostAnalysis]:
|
|
"""获取任务的成本分析结果"""
|
|
result = await self.db.execute(
|
|
select(CostAnalysis)
|
|
.where(CostAnalysis.task_id == task_id)
|
|
.order_by(CostAnalysis.created_at.desc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def _load_cleaned_data(self, task_id: int) -> List[Dict[str, Any]]:
|
|
"""
|
|
加载清洗后的数据
|
|
|
|
Args:
|
|
task_id: 任务ID
|
|
|
|
Returns:
|
|
清洗后的数据列表
|
|
"""
|
|
task = await self.db.get(ReconciliationTask, task_id)
|
|
if not task:
|
|
return []
|
|
|
|
if task.reconciliation_result and isinstance(task.reconciliation_result, dict):
|
|
records = task.reconciliation_result.get("records", [])
|
|
if records:
|
|
return records
|
|
|
|
return []
|