""" 凭证 API 提供凭证生成、查询、确认和金蝶导出接口 """ 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 import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.tenant import get_current_company_id from app.models.voucher import Voucher, VoucherStatus from app.models.account_mapping import AccountMapping from app.services.voucher.generator import VoucherGeneratorService from app.services.voucher.kingdee_exporter import KingdeeExporterService from app.services.voucher.template import VoucherTemplate router = APIRouter(prefix="/api/vouchers", tags=["凭证管理"]) class VoucherEntryResponse(BaseModel): """凭证分录响应""" account_code: str account_name: str debit_amount: float credit_amount: float summary: str department: str = "" class VoucherResponse(BaseModel): """凭证响应""" id: int voucher_number: str voucher_date: str period: str summary: str entries: List[Dict[str, Any]] total_debit: float total_credit: float status: str confirmed_by: Optional[int] = None confirmed_at: Optional[str] = None created_at: str updated_at: str class GenerateVoucherRequest(BaseModel): """生成凭证请求""" task_id: int period: str = "" class ConfirmVoucherRequest(BaseModel): """确认凭证请求""" user_id: int class AccountMappingResponse(BaseModel): """科目映射响应""" id: int standard_field: str debit_account: str debit_account_name: str credit_account: str credit_account_name: str cost_center: Optional[str] = None is_active: bool class AccountMappingRequest(BaseModel): """科目映射请求""" standard_field: str debit_account: str debit_account_name: str credit_account: str credit_account_name: str cost_center: Optional[str] = None @router.post("/generate", response_model=VoucherResponse) async def generate_voucher( request: GenerateVoucherRequest, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> VoucherResponse: """生成会计凭证""" service = VoucherGeneratorService(db) # 检查是否已有凭证 existing = await service.get_voucher(request.task_id) if existing: return _to_response(existing) voucher = await service.generate_voucher( task_id=request.task_id, company_id=company_id, period=request.period, ) return _to_response(voucher) @router.get("/task/{task_id}", response_model=Optional[VoucherResponse]) async def get_task_voucher( task_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> Optional[VoucherResponse]: """获取任务的凭证""" service = VoucherGeneratorService(db) voucher = await service.get_voucher(task_id) if not voucher: return None return _to_response(voucher) @router.get("/{voucher_id}", response_model=VoucherResponse) async def get_voucher( voucher_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> VoucherResponse: """获取凭证详情""" voucher = await db.get(Voucher, voucher_id) if not voucher: raise HTTPException(status_code=404, detail="凭证不存在") return _to_response(voucher) @router.post("/{voucher_id}/confirm", response_model=VoucherResponse) async def confirm_voucher( voucher_id: int, request: ConfirmVoucherRequest, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> VoucherResponse: """确认凭证""" service = VoucherGeneratorService(db) voucher = await service.confirm_voucher(voucher_id, request.user_id) if not voucher: raise HTTPException(status_code=404, detail="凭证不存在") return _to_response(voucher) @router.get("/task/{task_id}/export") async def export_voucher( task_id: int, format: str = Query("csv", description="导出格式: csv 或 excel"), db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> StreamingResponse: """导出凭证为金蝶格式""" service = KingdeeExporterService(db) try: return await service.export_task_vouchers(task_id, format) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) # ===== 科目映射管理 ===== @router.get("/account-mappings/list", response_model=List[AccountMappingResponse]) async def list_account_mappings( db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> List[AccountMappingResponse]: """获取科目映射列表""" result = await db.execute( select(AccountMapping).where(AccountMapping.company_id == company_id) ) mappings = result.scalars().all() return [AccountMappingResponse(**m.__dict__) for m in mappings] @router.post("/account-mappings", response_model=AccountMappingResponse) async def create_account_mapping( request: AccountMappingRequest, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> AccountMappingResponse: """创建科目映射""" mapping = AccountMapping( company_id=company_id, standard_field=request.standard_field, debit_account=request.debit_account, debit_account_name=request.debit_account_name, credit_account=request.credit_account, credit_account_name=request.credit_account_name, cost_center=request.cost_center, is_active=True, ) db.add(mapping) await db.commit() await db.refresh(mapping) return AccountMappingResponse(**mapping.__dict__) @router.put("/account-mappings/{mapping_id}", response_model=AccountMappingResponse) async def update_account_mapping( mapping_id: int, request: AccountMappingRequest, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> AccountMappingResponse: """更新科目映射""" mapping = await db.get(AccountMapping, mapping_id) if not mapping: raise HTTPException(status_code=404, detail="科目映射不存在") mapping.standard_field = request.standard_field mapping.debit_account = request.debit_account mapping.debit_account_name = request.debit_account_name mapping.credit_account = request.credit_account mapping.credit_account_name = request.credit_account_name mapping.cost_center = request.cost_center await db.commit() await db.refresh(mapping) return AccountMappingResponse(**mapping.__dict__) @router.delete("/account-mappings/{mapping_id}") async def delete_account_mapping( mapping_id: int, db: AsyncSession = Depends(get_db), company_id: int = Depends(get_current_company_id), ) -> Dict[str, str]: """删除科目映射""" mapping = await db.get(AccountMapping, mapping_id) if not mapping: raise HTTPException(status_code=404, detail="科目映射不存在") await db.delete(mapping) await db.commit() return {"message": "已删除"} @router.get("/templates/list") async def get_voucher_templates() -> Dict[str, Any]: """获取默认凭证模板""" return VoucherTemplate.get_all_templates() def _to_response(voucher: Voucher) -> VoucherResponse: """转换凭证模型为响应""" return VoucherResponse( id=voucher.id, voucher_number=voucher.voucher_number, voucher_date=voucher.voucher_date, period=voucher.period, summary=voucher.summary, entries=voucher.entries or [], total_debit=voucher.total_debit, total_credit=voucher.total_credit, status=voucher.status, confirmed_by=voucher.confirmed_by, confirmed_at=voucher.confirmed_at.isoformat() if voucher.confirmed_at else None, created_at=voucher.created_at.isoformat() if voucher.created_at else "", updated_at=voucher.updated_at.isoformat() if voucher.updated_at else "", )