""" 导出服务 导出对账结果为 Excel 和金蝶凭证格式 """ import io from datetime import datetime from typing import Any, Dict, List, Optional from sqlalchemy.ext.asyncio import AsyncSession from app.models.reconciliation_task import ReconciliationTask from app.models.exception_item import ExceptionItem try: from openpyxl import Workbook from openpyxl.styles import Font, Alignment, Border, Side, PatternFill from openpyxl.utils import get_column_letter OPENPYXL_AVAILABLE = True except ImportError: OPENPYXL_AVAILABLE = False class ExportService: """导出服务""" def __init__(self, db: AsyncSession): self.db = db async def export_task_result(self, task_id: int) -> bytes: """ 导出任务对账结果 Args: task_id: 任务ID Returns: Excel 文件二进制数据 """ if not OPENPYXL_AVAILABLE: raise ImportError("openpyxl 未安装") # 获取任务信息 task = await self.db.get(ReconciliationTask, task_id) if not task: raise ValueError("任务不存在") # 获取异常列表 from sqlalchemy import select query = select(ExceptionItem).where(ExceptionItem.task_id == task_id) result = await self.db.execute(query) exceptions = result.scalars().all() # 创建工作簿 wb = Workbook() # Sheet 1: 概览 ws_summary = wb.active ws_summary.title = "概览" self._fill_summary_sheet(ws_summary, task, exceptions) # Sheet 2: 异常列表 ws_exceptions = wb.create_sheet("异常列表") self._fill_exceptions_sheet(ws_exceptions, exceptions) # Sheet 3: 异常类型统计 ws_stats = wb.create_sheet("异常统计") self._fill_stats_sheet(ws_stats, exceptions) # 保存到字节流 output = io.BytesIO() wb.save(output) output.seek(0) return output.read() def _fill_summary_sheet( self, ws, task: ReconciliationTask, exceptions: List[ExceptionItem], ): """填充概览Sheet""" # 标题样式 title_font = Font(bold=True, size=14) header_font = Font(bold=True) header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") header_font_white = Font(bold=True, color="FFFFFF") thin_border = Border( left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"), ) # 标题 ws["A1"] = f"对账结果汇总 - {task.name}" ws["A1"].font = title_font ws.merge_cells("A1:D1") # 基本信息 ws["A3"] = "任务名称" ws["B3"] = task.name ws["A4"] = "对账期间" ws["B4"] = task.period ws["A5"] = "创建时间" ws["B5"] = task.created_at.strftime("%Y-%m-%d %H:%M:%S") if task.created_at else "" ws["A6"] = "完成时间" ws["B6"] = task.completed_at.strftime("%Y-%m-%d %H:%M:%S") if task.completed_at else "" # 统计数据 ws["A8"] = "统计数据" ws["A8"].font = header_font ws["A9"] = "总人数" ws["B9"] = task.total_employees ws["A10"] = "匹配人数" ws["B10"] = task.matched_count ws["A11"] = "异常数量" ws["B11"] = task.exception_count ws["A12"] = "匹配率" ws["B12"] = f"{(task.matched_count / task.total_employees * 100):.1f}%" if task.total_employees > 0 else "0%" # 设置列宽 ws.column_dimensions["A"].width = 15 ws.column_dimensions["B"].width = 30 def _fill_exceptions_sheet(self, ws, exceptions: List[ExceptionItem]): """填充异常列表Sheet""" # 样式 header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") header_font = Font(bold=True, color="FFFFFF") thin_border = Border( left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"), ) # 表头 headers = [ "异常ID", "员工ID", "员工姓名", "异常类型", "严重程度", "状态", "描述", "工资", "社保", "个税", "银行", "差异金额", "解决方案", "处理人", "创建时间" ] for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.fill = header_fill cell.font = header_font cell.border = thin_border cell.alignment = Alignment(horizontal="center") # 数据 for row_idx, exc in enumerate(exceptions, 2): ws.cell(row=row_idx, column=1, value=exc.id).border = thin_border ws.cell(row=row_idx, column=2, value=exc.employee_id).border = thin_border ws.cell(row=row_idx, column=3, value=exc.employee_name).border = thin_border ws.cell(row=row_idx, column=4, value=exc.exception_type).border = thin_border ws.cell(row=row_idx, column=5, value=exc.severity).border = thin_border ws.cell(row=row_idx, column=6, value=exc.status).border = thin_border ws.cell(row=row_idx, column=7, value=exc.description).border = thin_border # 金额（保留两位小数） ws.cell(row=row_idx, column=8, value=exc.salary_amount).border = thin_border ws.cell(row=row_idx, column=9, value=exc.social_security_amount).border = thin_border ws.cell(row=row_idx, column=10, value=exc.tax_amount).border = thin_border ws.cell(row=row_idx, column=11, value=exc.bank_amount).border = thin_border ws.cell(row=row_idx, column=12, value=exc.difference_amount).border = thin_border ws.cell(row=row_idx, column=13, value=exc.resolution).border = thin_border ws.cell(row=row_idx, column=14, value=exc.handler).border = thin_border ws.cell(row=row_idx, column=15, value=exc.created_at.strftime("%Y-%m-%d %H:%M:%S") if exc.created_at else "").border = thin_border # 设置列宽 col_widths = [10, 15, 15, 20, 10, 10, 40, 12, 12, 12, 12, 12, 20, 15, 20] for col, width in enumerate(col_widths, 1): ws.column_dimensions[get_column_letter(col)].width = width def _fill_stats_sheet(self, ws, exceptions: List[ExceptionItem]): """填充统计Sheet""" from collections import Counter header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") header_font = Font(bold=True, color="FFFFFF") thin_border = Border( left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"), ) # 按类型统计 ws["A1"] = "按异常类型统计" ws["A1"].font = Font(bold=True) ws["A2"] = "类型" ws["B2"] = "数量" ws["A2"].fill = header_fill ws["B2"].fill = header_fill ws["A2"].font = header_font ws["B2"].font = header_font ws["A2"].border = thin_border ws["B2"].border = thin_border type_counts = Counter(e.exception_type for e in exceptions) for row_idx, (exc_type, count) in enumerate(type_counts.items(), 3): ws.cell(row=row_idx, column=1, value=exc_type).border = thin_border ws.cell(row=row_idx, column=2, value=count).border = thin_border # 按严重程度统计 start_row = len(type_counts) + 5 ws.cell(row=start_row, column=1, value="按严重程度统计").font = Font(bold=True) ws.cell(row=start_row + 1, column=1, value="严重程度") ws.cell(row=start_row + 1, column=2, value="数量") ws.cell(row=start_row + 1, column=1).fill = header_fill ws.cell(row=start_row + 1, column=2).fill = header_fill ws.cell(row=start_row + 1, column=1).font = header_font ws.cell(row=start_row + 1, column=2).font = header_font severity_counts = Counter(e.severity for e in exceptions) for row_idx, (severity, count) in enumerate(severity_counts.items(), start_row + 2): ws.cell(row=row_idx, column=1, value=severity).border = thin_border ws.cell(row=row_idx, column=2, value=count).border = thin_border # 按状态统计 start_row = start_row + len(severity_counts) + 3 ws.cell(row=start_row, column=1, value="按状态统计").font = Font(bold=True) ws.cell(row=start_row + 1, column=1, value="状态") ws.cell(row=start_row + 1, column=2, value="数量") ws.cell(row=start_row + 1, column=1).fill = header_fill ws.cell(row=start_row + 1, column=2).fill = header_fill ws.cell(row=start_row + 1, column=1).font = header_font ws.cell(row=start_row + 1, column=2).font = header_font status_counts = Counter(e.status for e in exceptions) for row_idx, (status, count) in enumerate(status_counts.items(), start_row + 2): ws.cell(row=row_idx, column=1, value=status).border = thin_border ws.cell(row=row_idx, column=2, value=count).border = thin_border async def export_kingdee_voucher(self, task_id: int) -> bytes: """ 导出金蝶凭证 Args: task_id: 任务ID Returns: Excel 文件二进制数据 """ if not OPENPYXL_AVAILABLE: raise ImportError("openpyxl 未安装") # 获取任务信息 task = await self.db.get(ReconciliationTask, task_id) if not task: raise ValueError("任务不存在") # 获取已解决的异常（排除已忽略的） from sqlalchemy import select query = select(ExceptionItem).where( ExceptionItem.task_id == task_id, ExceptionItem.status == "resolved" ) result = await self.db.execute(query) resolved_exceptions = result.scalars().all() # 创建工作簿 wb = Workbook() ws = wb.active ws.title = "凭证数据" # 样式 header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") header_font = Font(bold=True, color="FFFFFF") thin_border = Border( left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin"), ) # 金蝶凭证格式表头 headers = [ "凭证字", "凭证号", "凭证日期", "附单据数", "摘要", "科目代码", "科目名称", "借方金额", "贷方金额", "币种", "汇率" ] for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.fill = header_fill cell.font = header_font cell.border = thin_border cell.alignment = Alignment(horizontal="center") # 填充数据（根据金蝶凭证格式要求） row_idx = 2 for exc in resolved_exceptions: # 生成凭证摘要 summary = f"工资对账调整 - {exc.employee_name} ({exc.employee_id})" # 如果有差异金额，生成借贷分录 if exc.difference_amount and exc.difference_amount != 0: # 借方分录 ws.cell(row=row_idx, column=1, value="记").border = thin_border ws.cell(row=row_idx, column=2, value="").border = thin_border # 凭证号自动生成 ws.cell(row=row_idx, column=3, value=task.period + "-01").border = thin_border ws.cell(row=row_idx, column=4, value=1).border = thin_border ws.cell(row=row_idx, column=5, value=summary).border = thin_border ws.cell(row=row_idx, column=6, value="").border = thin_border # 科目代码 ws.cell(row=row_idx, column=7, value="待处理").border = thin_border # 科目名称 if exc.difference_amount > 0: ws.cell(row=row_idx, column=8, value=abs(exc.difference_amount)).border = thin_border ws.cell(row=row_idx, column=9, value=0).border = thin_border else: ws.cell(row=row_idx, column=8, value=0).border = thin_border ws.cell(row=row_idx, column=9, value=abs(exc.difference_amount)).border = thin_border ws.cell(row=row_idx, column=10, value="人民币").border = thin_border ws.cell(row=row_idx, column=11, value=1).border = thin_border row_idx += 1 # 设置列宽 col_widths = [10, 12, 15, 10, 40, 15, 20, 15, 15, 10, 10] for col, width in enumerate(col_widths, 1): ws.column_dimensions[get_column_letter(col)].width = width # 保存到字节流 output = io.BytesIO() wb.save(output) output.seek(0) return output.read() # 便捷函数 async def export_task_result(db: AsyncSession, task_id: int) -> bytes: """导出任务对账结果""" service = ExportService(db) return await service.export_task_result(task_id) async def export_kingdee_voucher(db: AsyncSession, task_id: int) -> bytes: """导出金蝶凭证""" service = ExportService(db) return await service.export_kingdee_voucher(task_id)