s2f/backend/app/services/voucher/kingdee_exporter.py

"""
金蝶格式导出服务

将凭证导出为金蝶K3/星空可导入的格式
"""

import csv
import io
import logging
from datetime import datetime
from typing import List, Optional

from fastapi.responses import StreamingResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.voucher import Voucher

logger = logging.getLogger(__name__)


class KingdeeExporterService:
    """金蝶格式导出服务"""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def export_voucher(
        self,
        voucher_id: int,
        format_type: str = "csv",
    ) -> StreamingResponse:
        """
        导出凭证为金蝶格式

        Args:
            voucher_id: 凭证ID
            format_type: 导出格式 (csv/excel)

        Returns:
            StreamingResponse
        """
        voucher = await self.db.get(Voucher, voucher_id)
        if not voucher:
            raise ValueError(f"凭证不存在: {voucher_id}")

        if format_type == "excel":
            return self._export_excel(voucher)
        else:
            return self._export_csv(voucher)

    async def export_task_vouchers(
        self,
        task_id: int,
        format_type: str = "csv",
    ) -> StreamingResponse:
        """
        导出任务的所有凭证

        Args:
            task_id: 任务ID
            format_type: 导出格式

        Returns:
            StreamingResponse
        """
        result = await self.db.execute(
            select(Voucher).where(Voucher.task_id == task_id)
        )
        vouchers = list(result.scalars().all())

        if not vouchers:
            raise ValueError(f"任务 {task_id} 无凭证")

        if len(vouchers) == 1:
            return await self.export_voucher(vouchers[0].id, format_type)

        # 多凭证导出
        if format_type == "excel":
            return self._export_multiple_excel(vouchers)
        else:
            return self._export_multiple_csv(vouchers)

    def _export_csv(self, voucher: Voucher) -> StreamingResponse:
        """
        金蝶K3 CSV导入格式

        格式: 凭证日期, 凭证号, 摘要, 科目代码, 科目名称, 借方金额, 贷方金额, 制单人
        """
        output = io.StringIO()
        writer = csv.writer(output)

        # 金蝶K3导入格式头
        writer.writerow([
            "凭证日期", "凭证字", "凭证号", "摘要",
            "科目代码", "科目名称",
            "借方金额", "贷方金额",
            "制单人", "审核人",
        ])

        entries = voucher.entries or []
        for i, entry in enumerate(entries):
            writer.writerow([
                voucher.voucher_date,
                "记",
                voucher.voucher_number,
                entry.get("summary", voucher.summary),
                entry.get("account_code", ""),
                entry.get("account_name", ""),
                f'{entry.get("debit_amount", 0):.2f}',
                f'{entry.get("credit_amount", 0):.2f}',
                "AI助手",
                "",
            ])

        output.seek(0)

        # 转为 bytes
        content = output.getvalue().encode("utf-8-sig")  # BOM for Excel compatibility
        return StreamingResponse(
            io.BytesIO(content),
            media_type="text/csv",
            headers={
                "Content-Disposition": f'attachment; filename="voucher_{voucher.voucher_number}.csv"',
            },
        )

    def _export_excel(self, voucher: Voucher) -> StreamingResponse:
        """导出为 Excel 格式"""
        from openpyxl import Workbook

        wb = Workbook()
        ws = wb.active
        ws.title = "凭证"

        # 标题行
        ws.append(["凭证编号", voucher.voucher_number])
        ws.append(["凭证日期", voucher.voucher_date])
        ws.append(["会计期间", voucher.period])
        ws.append(["摘要", voucher.summary])
        ws.append([])

        # 分录表头
        ws.append(["序号", "科目代码", "科目名称", "摘要", "借方金额", "贷方金额"])

        entries = voucher.entries or []
        for i, entry in enumerate(entries, 1):
            ws.append([
                i,
                entry.get("account_code", ""),
                entry.get("account_name", ""),
                entry.get("summary", ""),
                entry.get("debit_amount", 0),
                entry.get("credit_amount", 0),
            ])

        # 合计行
        ws.append([])
        ws.append(["", "", "", "合计", voucher.total_debit, voucher.total_credit])

        output = io.BytesIO()
        wb.save(output)
        output.seek(0)

        return StreamingResponse(
            output,
            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            headers={
                "Content-Disposition": f'attachment; filename="voucher_{voucher.voucher_number}.xlsx"',
            },
        )

    def _export_multiple_csv(self, vouchers: List[Voucher]) -> StreamingResponse:
        """多凭证 CSV 导出"""
        output = io.StringIO()
        writer = csv.writer(writer)

        writer.writerow([
            "凭证日期", "凭证字", "凭证号", "摘要",
            "科目代码", "科目名称",
            "借方金额", "贷方金额",
        ])

        for voucher in vouchers:
            entries = voucher.entries or []
            for entry in entries:
                writer.writerow([
                    voucher.voucher_date,
                    "记",
                    voucher.voucher_number,
                    entry.get("summary", voucher.summary),
                    entry.get("account_code", ""),
                    entry.get("account_name", ""),
                    f'{entry.get("debit_amount", 0):.2f}',
                    f'{entry.get("credit_amount", 0):.2f}',
                ])

        output.seek(0)
        content = output.getvalue().encode("utf-8-sig")
        return StreamingResponse(
            io.BytesIO(content),
            media_type="text/csv",
            headers={
                "Content-Disposition": f'attachment; filename="vouchers_batch.csv"',
            },
        )

    def _export_multiple_excel(self, vouchers: List[Voucher]) -> StreamingResponse:
        """多凭证 Excel 导出"""
        from openpyxl import Workbook

        wb = Workbook()
        ws = wb.active
        ws.title = "凭证汇总"

        ws.append([
            "凭证日期", "凭证号", "会计期间", "摘要",
            "科目代码", "科目名称",
            "借方金额", "贷方金额",
        ])

        for voucher in vouchers:
            entries = voucher.entries or []
            for entry in entries:
                ws.append([
                    voucher.voucher_date,
                    voucher.voucher_number,
                    voucher.period,
                    entry.get("summary", voucher.summary),
                    entry.get("account_code", ""),
                    entry.get("account_name", ""),
                    entry.get("debit_amount", 0),
                    entry.get("credit_amount", 0),
                ])

        output = io.BytesIO()
        wb.save(output)
        output.seek(0)

        return StreamingResponse(
            output,
            media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            headers={
                "Content-Disposition": f'attachment; filename="vouchers_batch.xlsx"',
            },
        )