130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
import mimetypes
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.bx_auth import BxUserContext, get_bx_current_user
|
|
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(
|
|
db: Session = Depends(get_db),
|
|
user: BxUserContext = Depends(get_bx_current_user),
|
|
):
|
|
data = InvoiceService.get_mine_summary(db, user=user)
|
|
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 = 10,
|
|
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))
|
|
|
|
|
|
@router.get("/invoices", response_model=ApiResponse[PagedData[InvoiceOut]])
|
|
def list_invoices(
|
|
page: int = 1,
|
|
pageSize: int = 10,
|
|
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])
|
|
def create_invoice(
|
|
file: UploadFile = File(...),
|
|
amount: float = Form(...),
|
|
remark: str = Form(""),
|
|
targetUserId: Optional[str] = Form(None),
|
|
db: Session = Depends(get_db),
|
|
user: BxUserContext = Depends(get_bx_current_user),
|
|
):
|
|
if not file or not file.filename:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择文件")
|
|
file_data = file.file.read()
|
|
content_type = (
|
|
file.content_type
|
|
or mimetypes.guess_type(file.filename or "")[0]
|
|
or "application/octet-stream"
|
|
)
|
|
try:
|
|
invoice = InvoiceService.create(
|
|
db,
|
|
user=user,
|
|
amount=amount,
|
|
file_data=file_data,
|
|
filename=file.filename or "unknown",
|
|
content_type=content_type,
|
|
remark=remark,
|
|
target_user_id=targetUserId,
|
|
)
|
|
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])
|