""" 成本分析计算服务 计算人工成本总额、部门拆分、费用科目拆分、环比变化等 """ 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 []