""" 对账引擎 执行对账逻辑，协调各个规则 """ import time from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.models.exception_item import ExceptionItem, ExceptionStatus from app.services.reconciliation.rules import ( RuleContext, RuleResult, RuleFactory, ReconciliationRule, ) @dataclass class ReconciliationResult: """对账结果""" task_id: int period: str total_employees: int matched_employees: int exception_count: int # 统计 exceptions_by_type: Dict[str, int] = field(default_factory=dict) exceptions_by_severity: Dict[str, int] = field(default_factory=dict) # 详细结果 matched_employees_list: List[Dict[str, Any]] = field(default_factory=list) exception_items: List[Dict[str, Any]] = field(default_factory=list) # 性能指标 execution_time_ms: float = 0 rules_executed: int = 0 # 时间戳 completed_at: str = "" class ReconciliationEngine: """ 对账引擎 协调数据加载、规则执行、结果存储 """ def __init__( self, db: AsyncSession, company_id: int, task_id: int, period: str, enable_bank_rules: bool = False, ): self.db = db self.company_id = company_id self.task_id = task_id self.period = period self.enable_bank_rules = enable_bank_rules # 创建规则 self.rules = RuleFactory.create_all_rules(enable_bank_rules) # 数据存储 self.salary_data: Dict[str, Dict[str, Any]] = {} self.social_security_data: Dict[str, Dict[str, Any]] = {} self.tax_data: Dict[str, Dict[str, Any]] = {} self.bank_data: Optional[Dict[str, Dict[str, Any]]] = None # 结果 self.exceptions: List[ExceptionItem] = [] self.matched_count = 0 self.exception_count = 0 def load_data( self, salary_records: List[Dict[str, Any]], social_security_records: List[Dict[str, Any]], tax_records: List[Dict[str, Any]], bank_records: Optional[List[Dict[str, Any]]] = None, employee_id_field: str = "employee_id", employee_name_field: str = "employee_name", ): """ 加载数据 Args: salary_records: 工资表数据 social_security_records: 社保表数据 tax_records: 个税表数据 bank_records: 银行数据（可选） employee_id_field: 员工ID字段名 employee_name_field: 员工姓名字段名 """ # 加载工资数据 for record in salary_records: emp_id = str(record.get(employee_id_field, "")) if emp_id: self.salary_data[emp_id] = record # 加载社保数据 for record in social_security_records: emp_id = str(record.get(employee_id_field, "")) if emp_id: self.social_security_data[emp_id] = record # 加载个税数据 for record in tax_records: emp_id = str(record.get(employee_id_field, "")) if emp_id: self.tax_data[emp_id] = record # 加载银行数据 if bank_records: self.bank_data = {} for record in bank_records: emp_id = str(record.get(employee_id_field, "")) if emp_id: self.bank_data[emp_id] = record async def execute(self) -> ReconciliationResult: """ 执行对账 Returns: 对账结果 """ start_time = time.time() # 获取所有员工ID（合并三个表） all_employee_ids = set() all_employee_ids.update(self.salary_data.keys()) all_employee_ids.update(self.social_security_data.keys()) all_employee_ids.update(self.tax_data.keys()) if self.bank_data: all_employee_ids.update(self.bank_data.keys()) total_employees = len(all_employee_ids) matched_employees = 0 exception_items: List[Dict[str, Any]] = [] # 统计 exceptions_by_type: Dict[str, int] = {} exceptions_by_severity: Dict[str, int] = {} matched_list: List[Dict[str, Any]] = [] # 重置需要状态的规则 for rule in self.rules: if hasattr(rule, 'reset'): rule.reset() # 遍历每个员工 for employee_id in all_employee_ids: # 获取员工姓名 employee_name = ( self.salary_data.get(employee_id, {}).get("employee_name") or self.social_security_data.get(employee_id, {}).get("employee_name") or self.tax_data.get(employee_id, {}).get("employee_name") or employee_id ) # 构建规则上下文 context = RuleContext( company_id=self.company_id, task_id=self.task_id, period=self.period, employee_id=employee_id, employee_name=employee_name, salary_data=self.salary_data, social_security_data=self.social_security_data, tax_data=self.tax_data, bank_data=self.bank_data, ) # 执行所有规则 employee_exceptions = [] for rule in self.rules: if not rule.enabled: continue try: result = rule.check(context) if result.is_exception: employee_exceptions.append((rule, result)) except Exception as e: # 规则执行出错，记录但不中断 pass # 处理该员工的异常 if employee_exceptions: for rule, result in employee_exceptions: # 保存异常到数据库 exception_item = await self._save_exception(rule, result, context) self.exceptions.append(exception_item) exception_items.append({ "id": exception_item.id, "type": result.exception_type.value if result.exception_type else rule.exception_type.value, "severity": result.severity.value, "description": result.description, "employee_id": employee_id, "employee_name": employee_name, }) # 统计 type_key = result.exception_type.value if result.exception_type else rule.exception_type.value exceptions_by_type[type_key] = exceptions_by_type.get(type_key, 0) + 1 exceptions_by_severity[result.severity.value] = ( exceptions_by_severity.get(result.severity.value, 0) + 1 ) else: matched_employees += 1 matched_list.append({ "employee_id": employee_id, "employee_name": employee_name, }) # 更新任务统计 await self._update_task_stats(matched_employees, len(exception_items)) execution_time = (time.time() - start_time) * 1000 return ReconciliationResult( task_id=self.task_id, period=self.period, total_employees=total_employees, matched_employees=matched_employees, exception_count=len(exception_items), exceptions_by_type=exceptions_by_type, exceptions_by_severity=exceptions_by_severity, matched_employees_list=matched_list, exception_items=exception_items, execution_time_ms=execution_time, rules_executed=len([r for r in self.rules if r.enabled]), completed_at=datetime.utcnow().isoformat(), ) async def _save_exception( self, rule: ReconciliationRule, result: RuleResult, context: RuleContext, ) -> ExceptionItem: """保存异常到数据库""" exception_item = ExceptionItem( company_id=self.company_id, task_id=self.task_id, exception_type=result.exception_type.value if result.exception_type else rule.exception_type.value, severity=result.severity.value, status=ExceptionStatus.PENDING.value, employee_id=context.employee_id, employee_name=context.employee_name, description=result.description, detail=result.detail, salary_amount=result.salary_amount, social_security_amount=result.social_security_amount, tax_amount=result.tax_amount, bank_amount=result.bank_amount, difference_amount=result.difference_amount, suggested_action=result.suggested_action, ) self.db.add(exception_item) await self.db.commit() await self.db.refresh(exception_item) return exception_item async def _update_task_stats(self, matched: int, exceptions: int): """更新任务统计""" from app.models.reconciliation_task import ReconciliationTask task = await self.db.get(ReconciliationTask, self.task_id) if task: task.matched_count = matched task.exception_count = exceptions task.total_employees = matched + exceptions task.status = "COMPLETED" await self.db.commit() # 便捷函数 async def run_reconciliation( db: AsyncSession, company_id: int, task_id: int, period: str, salary_records: List[Dict[str, Any]], social_security_records: List[Dict[str, Any]], tax_records: List[Dict[str, Any]], bank_records: Optional[List[Dict[str, Any]]] = None, enable_bank_rules: bool = False, ) -> ReconciliationResult: """ 运行对账 Args: db: 数据库会话 company_id: 企业ID task_id: 任务ID period: 对账期间 salary_records: 工资表数据 social_security_records: 社保表数据 tax_records: 个税表数据 bank_records: 银行数据（可选） enable_bank_rules: 是否启用银行规则 Returns: 对账结果 """ engine = ReconciliationEngine( db=db, company_id=company_id, task_id=task_id, period=period, enable_bank_rules=enable_bank_rules, ) engine.load_data( salary_records=salary_records, social_security_records=social_security_records, tax_records=tax_records, bank_records=bank_records, ) return await engine.execute()