import mimetypes from typing import List, Optional, Set from fastapi import APIRouter, Depends, HTTPException, Request, status, Query from sqlalchemy.orm import Session from starlette.datastructures import UploadFile as StarletteUploadFile from app.api.bx.export_common import ( EXPORT_MAX_ROWS, ExportColumn, ExportValidationError, BxExportBodyBase, dump_export_row, respond_xlsx, ) from app.core.bx_auth import BxUserContext, get_bx_current_user from app.core.tenant_context import get_tenant_no from app.core.database import get_db from app.schemas.bx.common import ApiResponse, PagedData from app.schemas.bx.invoice import InvoiceOut, LedgerOut, WalletSummaryOut from app.services.bx.invoice import InvoiceService router = APIRouter(tags=["发票夹"]) @router.get("/invoice-wallets/mine/summary", response_model=ApiResponse[WalletSummaryOut]) def get_mine_wallet_summary( userId: Optional[str] = None, db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): data = InvoiceService.get_mine_summary(db, user=user, owner_user_id=userId) return ApiResponse(data=WalletSummaryOut.model_validate(data)) @router.get("/invoice-wallets", response_model=ApiResponse[PagedData[WalletSummaryOut]]) def list_invoice_wallets( page: int = 1, pageSize: int = Query(10, ge=1, le=500), keyword: str = "", db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): rows, total = InvoiceService.list_wallets( db, user=user, page=page, page_size=pageSize, keyword=keyword ) out_rows = [WalletSummaryOut.model_validate(InvoiceService._summary_to_dict(r)) for r in rows] return ApiResponse(data=PagedData(rows=out_rows, total=total, page=page, page_size=pageSize)) WALLET_EXPORT_COLUMNS = [ ExportColumn(title="员工", field="ownerUserName"), ExportColumn(title="总额（元）", field="totalAmount"), ExportColumn(title="已报销占用（元）", field="usedAmount"), ExportColumn(title="余额（元）", field="balance"), ExportColumn(title="发票张数", field="invoiceCount"), ] class WalletExportBody(BxExportBodyBase): keyword: str = "" @router.post("/invoice-wallets/export") def export_invoice_wallets( body: WalletExportBody, db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): try: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) rows, _ = InvoiceService.list_wallets( db, user=user, page=1, page_size=EXPORT_MAX_ROWS, keyword=body.keyword or "" ) outs = [WalletSummaryOut.model_validate(InvoiceService._summary_to_dict(r)) for r in rows] if body.ids: id_set = {str(i) for i in body.ids} # wallet may key by ownerUserId outs = [o for o in outs if str(getattr(o, "owner_user_id", None) or getattr(o, "id", "")) in id_set or str(o.model_dump(by_alias=True).get("ownerUserId") or o.model_dump(by_alias=True).get("id") or "") in id_set] export_rows = [dump_export_row(o, status_fields=()) for o in outs] return respond_xlsx( db, tenant_no=tenant_no, sheet_title="发票夹", columns=WALLET_EXPORT_COLUMNS, rows=export_rows, filename_stem="发票夹汇总", requested_columns=body.columns, ) except ExportValidationError as e: raise HTTPException(status_code=400, detail=str(e)) from e @router.get("/invoices", response_model=ApiResponse[PagedData[InvoiceOut]]) def list_invoices( page: int = 1, pageSize: int = Query(10, ge=1, le=500), userId: Optional[str] = None, db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): tenant_no = InvoiceService._tenant_no(db, user) rows, total = InvoiceService.list_invoices( db, user=user, owner_user_id=userId, page=page, page_size=pageSize ) out_rows = InvoiceService.out_rows_with_attachments(db, tenant_no, list(rows)) return ApiResponse( data=PagedData( rows=[InvoiceOut.model_validate(r) for r in out_rows], total=total, page=page, page_size=pageSize, ) ) @router.post("/invoices", response_model=ApiResponse[InvoiceOut]) async def create_invoice( request: Request, db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): """一次上传一条发票：多文件挂同一档案，金额为总额，只写一条额度流水。 使用 request.form().getlist，兼容只传 1 个文件时 FastAPI List[UploadFile] 的 422。 """ form = await request.form() amount_raw = form.get("amount") if amount_raw is None or str(amount_raw).strip() == "": raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请填写金额") try: amount = float(amount_raw) except (TypeError, ValueError) as exc: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="金额格式不正确" ) from exc remark = str(form.get("remark") or "") target_raw = form.get("targetUserId") target_user_id = str(target_raw).strip() if target_raw not in (None, "") else None upload_files: List[StarletteUploadFile] = [] seen_ids: Set[int] = set() for key in ("files", "file"): for item in form.getlist(key): if not isinstance(item, StarletteUploadFile): continue obj_id = id(item) if obj_id in seen_ids: continue seen_ids.add(obj_id) upload_files.append(item) if not upload_files: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择文件") file_payload = [] for uf in upload_files: if not uf.filename: continue data = await uf.read() content_type = ( uf.content_type or mimetypes.guess_type(uf.filename or "")[0] or "application/octet-stream" ) file_payload.append((data, uf.filename or "unknown", content_type)) if not file_payload: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择文件") try: invoice = InvoiceService.create( db, user=user, amount=amount, files=file_payload, remark=remark, target_user_id=target_user_id, ) db.commit() db.refresh(invoice) except HTTPException: db.rollback() raise except Exception as exc: db.rollback() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="上传发票失败" ) from exc tenant_no = InvoiceService._tenant_no(db, user) out = InvoiceService.out_rows_with_attachments(db, tenant_no, [invoice])[0] return ApiResponse(data=InvoiceOut.model_validate(out)) @router.delete("/invoices/{invoice_id}", response_model=ApiResponse[bool]) def delete_invoice( invoice_id: str, db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): try: success = InvoiceService.delete(db, user=user, invoice_id=invoice_id) db.commit() except HTTPException: db.rollback() raise return ApiResponse(data=success) @router.get("/invoice-ledgers", response_model=ApiResponse[List[LedgerOut]]) def list_invoice_ledgers( userId: Optional[str] = None, db: Session = Depends(get_db), user: BxUserContext = Depends(get_bx_current_user), ): rows = InvoiceService.list_ledgers(db, user=user, owner_user_id=userId) return ApiResponse(data=[LedgerOut.model_validate(r) for r in rows])