""" 成本分析 API 提供人工成本分析、部门拆分、费用科目拆分、环比变化等接口 """ import csv import io from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.tenant import get_current_company_id from app.services.analysis.cost_calculator import CostCalculatorService from app.services.ai.cost_analyzer import CostAnalyzerService router = APIRouter(prefix="/api/analysis", tags=["成本分析"]) class CostSummaryResponse(BaseModel): """成本汇总响应""" total_cost: float salary_cost: float social_security_cost: float fund_cost: float employee_count: int class DepartmentCostResponse(BaseModel): """部门成本响应""" department: str employee_count: int salary_cost: float social_security_cost: float fund_cost: float total_cost: float class ExpenseBreakdownResponse(BaseModel): """费用科目拆分响应""" expense_type: str amount: float class MonthOverMonthResponse(BaseModel): """环比变化响应""" total_cost: Dict[str, Any] salary_cost: Dict[str, Any] social_security_cost: Dict[str, Any] fund_cost: Dict[str, Any] employee_count: Dict[str, Any] class CostChangeAnalysisResponse(BaseModel): """成本变化分析响应""" new_employees: List[Dict[str, Any]] left_employees: List[Dict[str, Any]] salary_adjustments: List[Dict[str, Any]] new_employee_cost: float left_employee_saving: float adjustment_cost: float net_change: float class FullAnalysisResponse(BaseModel): """完整成本分析响应""" summary: CostSummaryResponse departments: List[DepartmentCostResponse] expenses: List[ExpenseBreakdownResponse] changes: Optional[Dict[str, Any]] = None ai_summary: Optional[str] = None @router.get("/labor-cost/{task_id}", response_model=FullAnalysisResponse) async def get_labor_cost_analysis( task_id: int, prev_task_id: Optional[int] = Query(None, description="上月任务ID，用于环比"), db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> FullAnalysisResponse: """ 获取人工成本分析 包含总额、部门拆分、费用科目拆分，可选环比 """ service = CostCalculatorService(db) summary = await service.calculate_total_cost(task_id) departments = await service.calculate_by_department(task_id) expenses = await service.calculate_by_expense_type(task_id) changes = None ai_summary = None if prev_task_id: changes = await service.calculate_month_over_month(task_id, prev_task_id) change_detail = await service.analyze_cost_changes(task_id, prev_task_id) # 尝试 AI 分析 try: analyzer = CostAnalyzerService() prev_summary = await service.calculate_total_cost(prev_task_id) ai_summary = await analyzer.analyze_cost_changes( summary.to_dict(), prev_summary.to_dict(), change_detail, ) except Exception: pass return FullAnalysisResponse( summary=CostSummaryResponse(**summary.to_dict()), departments=[ DepartmentCostResponse(**d.to_dict()) for d in departments ], expenses=[ ExpenseBreakdownResponse(expense_type=k, amount=v) for k, v in expenses.items() ], changes=changes, ai_summary=ai_summary, ) @router.get("/labor-cost/{task_id}/summary", response_model=CostSummaryResponse) async def get_cost_summary( task_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> CostSummaryResponse: """获取成本汇总""" service = CostCalculatorService(db) summary = await service.calculate_total_cost(task_id) return CostSummaryResponse(**summary.to_dict()) @router.get("/labor-cost/{task_id}/departments", response_model=List[DepartmentCostResponse]) async def get_department_costs( task_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> List[DepartmentCostResponse]: """获取部门成本拆分""" service = CostCalculatorService(db) departments = await service.calculate_by_department(task_id) return [DepartmentCostResponse(**d.to_dict()) for d in departments] @router.get("/labor-cost/{task_id}/expenses", response_model=List[ExpenseBreakdownResponse]) async def get_expense_breakdown( task_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> List[ExpenseBreakdownResponse]: """获取费用科目拆分""" service = CostCalculatorService(db) expenses = await service.calculate_by_expense_type(task_id) return [ ExpenseBreakdownResponse(expense_type=k, amount=v) for k, v in expenses.items() ] @router.get("/labor-cost/{task_id}/changes", response_model=CostChangeAnalysisResponse) async def get_cost_changes( task_id: int, prev_task_id: int = Query(..., description="上月任务ID"), db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> CostChangeAnalysisResponse: """获取成本变化分析""" service = CostCalculatorService(db) changes = await service.analyze_cost_changes(task_id, prev_task_id) return CostChangeAnalysisResponse(**changes) @router.get("/labor-cost/{task_id}/export") async def export_cost_analysis( task_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> StreamingResponse: """ 导出成本分析为 Excel 返回包含成本汇总、部门拆分、费用拆分的 Excel 文件 """ service = CostCalculatorService(db) summary = await service.calculate_total_cost(task_id) departments = await service.calculate_by_department(task_id) expenses = await service.calculate_by_expense_type(task_id) # 使用 openpyxl 生成 Excel from openpyxl import Workbook wb = Workbook() # Sheet1: 成本汇总 ws1 = wb.active ws1.title = "成本汇总" ws1.append(["项目", "金额（元）"]) ws1.append(["人工成本总额", summary.total_cost]) ws1.append(["工资成本", summary.salary_cost]) ws1.append(["社保成本（公司部分）", summary.social_security_cost]) ws1.append(["公积金成本（公司部分）", summary.fund_cost]) ws1.append(["员工人数", summary.employee_count]) # Sheet2: 部门拆分 ws2 = wb.create_sheet("部门拆分") ws2.append(["部门", "人数", "工资成本", "社保成本", "公积金成本", "合计"]) for dept in departments: ws2.append([ dept.department, dept.employee_count, dept.salary_cost, dept.social_security_cost, dept.fund_cost, dept.total_cost, ]) # Sheet3: 费用科目拆分 ws3 = wb.create_sheet("费用科目拆分") ws3.append(["费用科目", "金额（元）"]) for expense_type, amount in expenses.items(): ws3.append([expense_type, amount]) # 输出到内存 output = io.BytesIO() wb.save(output) output.seek(0) filename = f"cost_analysis_{task_id}.xlsx" return StreamingResponse( output, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, )