""" 凭证生成引擎 根据对账数据和科目映射生成会计凭证 """ import logging from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.account_mapping import AccountMapping from app.models.voucher import Voucher, VoucherStatus from app.models.reconciliation_task import ReconciliationTask from app.services.voucher.template import VoucherTemplate, DEFAULT_ACCOUNT_TEMPLATES logger = logging.getLogger(__name__) class VoucherEntry: """凭证分录""" def __init__( self, account_code: str, account_name: str, debit_amount: float = 0.0, credit_amount: float = 0.0, summary: str = "", department: str = "", ): self.account_code = account_code self.account_name = account_name self.debit_amount = debit_amount self.credit_amount = credit_amount self.summary = summary self.department = department def to_dict(self) -> Dict[str, Any]: return { "account_code": self.account_code, "account_name": self.account_name, "debit_amount": self.debit_amount, "credit_amount": self.credit_amount, "summary": self.summary, "department": self.department, } class VoucherGeneratorService: """凭证生成引擎""" def __init__(self, db: AsyncSession): self.db = db async def generate_voucher( self, task_id: int, company_id: int, period: str = "", ) -> Voucher: """ 根据对账任务数据生成会计凭证 Args: task_id: 对账任务ID company_id: 企业ID period: 会计期间（如 2024-01），为空则从任务中获取 Returns: 生成的凭证对象 """ # 1. 加载对账数据 task = await self.db.get(ReconciliationTask, task_id) if not task: raise ValueError(f"任务不存在: {task_id}") if not period: period = task.period # 2. 获取科目映射 mappings = await self._get_account_mappings(company_id) # 3. 加载清洗后的数据 records = await self._load_cleaned_data(task_id) # 4. 按字段汇总金额 field_totals = self._aggregate_by_field(records) # 5. 生成凭证分录 entries = self._generate_entries(field_totals, mappings) # 6. 计算合计 total_debit = sum(e.debit_amount for e in entries) total_credit = sum(e.credit_amount for e in entries) # 7. 生成凭证编号 voucher_number = f"PAY-{period.replace('-', '')}-{task_id:04d}" # 8. 创建凭证 voucher = Voucher( company_id=company_id, task_id=task_id, voucher_number=voucher_number, voucher_date=datetime.now().strftime("%Y-%m-%d"), period=period, summary=f"{period} 工资薪酬凭证", entries=[e.to_dict() for e in entries], total_debit=total_debit, total_credit=total_credit, status=VoucherStatus.DRAFT, ) self.db.add(voucher) await self.db.commit() await self.db.refresh(voucher) logger.info(f"凭证已生成: {voucher.voucher_number}, 借方合计={total_debit}, 贷方合计={total_credit}") return voucher async def get_voucher(self, task_id: int) -> Optional[Voucher]: """获取任务的凭证""" result = await self.db.execute( select(Voucher) .where(Voucher.task_id == task_id) .order_by(Voucher.created_at.desc()) .limit(1) ) return result.scalar_one_or_none() async def confirm_voucher(self, voucher_id: int, user_id: int) -> Optional[Voucher]: """确认凭证""" voucher = await self.db.get(Voucher, voucher_id) if not voucher: return None voucher.status = VoucherStatus.CONFIRMED voucher.confirmed_by = user_id voucher.confirmed_at = datetime.utcnow() await self.db.commit() await self.db.refresh(voucher) return voucher async def _get_account_mappings(self, company_id: int) -> Dict[str, Dict]: """ 获取企业的科目映射，不存在则使用默认模板 Args: company_id: 企业ID Returns: 标准字段 -> 科目映射配置 """ result = await self.db.execute( select(AccountMapping).where( AccountMapping.company_id == company_id, AccountMapping.is_active == True, ) ) db_mappings = result.scalars().all() # 先用默认模板 mappings = dict(DEFAULT_ACCOUNT_TEMPLATES) # 用数据库中的映射覆盖 for m in db_mappings: mappings[m.standard_field] = { "debit_account": m.debit_account, "debit_account_name": m.debit_account_name, "credit_account": m.credit_account, "credit_account_name": m.credit_account_name, "cost_center": m.cost_center or "", } return mappings async def _load_cleaned_data(self, task_id: int) -> List[Dict[str, Any]]: """加载清洗后的数据""" 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 [] def _aggregate_by_field(self, records: List[Dict[str, Any]]) -> Dict[str, float]: """ 按标准字段汇总金额 Args: records: 清洗后的数据列表 Returns: 标准字段 -> 总金额 """ totals: Dict[str, float] = {} cost_fields = [ "基本工资", "奖金", "补贴", "加班费", "应发工资", "养老保险(公司)", "医疗保险(公司)", "失业保险(公司)", "公积金(公司)", "养老保险", "医疗保险", "失业保险", "公积金", "应缴个税", "实发工资", ] for record in records: for field in cost_fields: value = float(record.get(field, 0) or 0) if value != 0: totals[field] = totals.get(field, 0.0) + value return totals def _generate_entries( self, field_totals: Dict[str, float], mappings: Dict[str, Dict], ) -> List[VoucherEntry]: """ 根据字段汇总和科目映射生成凭证分录 Args: field_totals: 字段金额汇总 mappings: 科目映射 Returns: 凭证分录列表 """ entries: List[VoucherEntry] = [] for field, amount in field_totals.items(): if abs(amount) < 0.01: continue template = mappings.get(field) if not template: logger.warning(f"字段 '{field}' 无科目映射，跳过") continue # 借方分录 entries.append(VoucherEntry( account_code=template["debit_account"], account_name=template["debit_account_name"], debit_amount=amount, summary=f"{field}", )) # 贷方分录 entries.append(VoucherEntry( account_code=template["credit_account"], account_name=template["credit_account_name"], credit_amount=amount, summary=f"{field}", )) # 合并相同科目的分录 entries = self._merge_entries(entries) return entries def _merge_entries(self, entries: List[VoucherEntry]) -> List[VoucherEntry]: """合并相同科目的分录""" merged: Dict[Tuple[str, str], VoucherEntry] = {} for entry in entries: key = (entry.account_code, "debit" if entry.debit_amount > 0 else "credit") if key in merged: if entry.debit_amount > 0: merged[key].debit_amount += entry.debit_amount else: merged[key].credit_amount += entry.credit_amount merged[key].summary += f", {entry.summary}" else: merged[key] = VoucherEntry( account_code=entry.account_code, account_name=entry.account_name, debit_amount=entry.debit_amount, credit_amount=entry.credit_amount, summary=entry.summary, ) return list(merged.values())