shanxia2.0/backend/app/api/bx/invoice.py

import mimetypes
from typing import Any, List, Optional, Set

from fastapi import APIRouter, Depends, HTTPException, Request, status, Query
from pydantic import ConfigDict
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
from app.services.bx.invoice_list_filters import (
    build_invoice_column_filters,
    build_ledger_column_filters,
    build_wallet_column_filters,
)
from app.services.bx.list_sort import order_invoice, order_ledger, parse_sort_params

router = APIRouter(tags=["发票夹"])


def _merge_query(request: Request, **known) -> dict:
    out = {k: v for k, v in known.items() if v is not None and v != ""}
    for k, v in request.query_params.multi_items():
        if k in ("page", "pageSize"):
            continue
        if k not in out and v not in (None, ""):
            out[k] = v
    return out


def _param_get(params: dict, *keys: str) -> Any:
    for key in keys:
        if key in params and params[key] not in (None, ""):
            return params[key]
    return None


def _wallet_filters_from_params(params: dict) -> list:
    return build_wallet_column_filters(
        keyword=str(_param_get(params, "keyword") or ""),
        owner_user_name=_param_get(params, "ownerUserName", "owner_user_name"),
        total_amount_op=_param_get(params, "totalAmountOp", "total_amount_op"),
        total_amount=_param_get(params, "totalAmount", "total_amount"),
        total_amount_min=_param_get(params, "totalAmountMin", "total_amount_min"),
        total_amount_max=_param_get(params, "totalAmountMax", "total_amount_max"),
        used_amount_op=_param_get(params, "usedAmountOp", "used_amount_op"),
        used_amount=_param_get(params, "usedAmount", "used_amount"),
        used_amount_min=_param_get(params, "usedAmountMin", "used_amount_min"),
        used_amount_max=_param_get(params, "usedAmountMax", "used_amount_max"),
        balance_op=_param_get(params, "balanceOp", "balance_op"),
        balance=_param_get(params, "balance"),
        balance_min=_param_get(params, "balanceMin", "balance_min"),
        balance_max=_param_get(params, "balanceMax", "balance_max"),
        invoice_count_op=_param_get(params, "invoiceCountOp", "invoice_count_op"),
        invoice_count=_param_get(params, "invoiceCount", "invoice_count"),
        invoice_count_min=_param_get(params, "invoiceCountMin", "invoice_count_min"),
        invoice_count_max=_param_get(params, "invoiceCountMax", "invoice_count_max"),
    )


def _invoice_filters_from_params(params: dict, *, tenant_no: str = "") -> list:
    return build_invoice_column_filters(
        tenant_no=tenant_no,
        keyword=str(_param_get(params, "keyword") or ""),
        amount_op=_param_get(params, "amountOp", "amount_op"),
        amount=_param_get(params, "amount"),
        amount_min=_param_get(params, "amountMin", "amount_min"),
        amount_max=_param_get(params, "amountMax", "amount_max"),
        remark=_param_get(params, "remark"),
        upload_by_name=_param_get(params, "uploadByName", "upload_by_name"),
        create_time_from=_param_get(params, "createTimeFrom", "create_time_from"),
        create_time_to=_param_get(params, "createTimeTo", "create_time_to"),
    )


def _ledger_filters_from_params(params: dict) -> list:
    return build_ledger_column_filters(
        keyword=str(_param_get(params, "keyword") or ""),
        entry_type=_param_get(params, "entryType", "entry_type"),
        entry_type_in=_param_get(params, "entryTypeIn", "entry_type_in"),
        amount_op=_param_get(params, "amountOp", "amount_op"),
        amount=_param_get(params, "amount"),
        amount_min=_param_get(params, "amountMin", "amount_min"),
        amount_max=_param_get(params, "amountMax", "amount_max"),
        remark=_param_get(params, "remark"),
        create_time_from=_param_get(params, "createTimeFrom", "create_time_from"),
        create_time_to=_param_get(params, "createTimeTo", "create_time_to"),
    )


def _body_to_params(body: BxExportBodyBase) -> dict:
    data = body.model_dump(by_alias=True, exclude_none=True)
    extra = getattr(body, "model_extra", None) or {}
    for key, value in extra.items():
        if key not in data and value not in (None, ""):
            data[key] = value
    return data


@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(
    request: Request,
    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),
):
    params = _merge_query(request, keyword=keyword or None)
    extra_filters = _wallet_filters_from_params(params)
    rows, total = InvoiceService.list_wallets(
        db,
        user=user,
        page=page,
        page_size=pageSize,
        keyword="",
        extra_filters=extra_filters or None,
    )
    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):
    model_config = ConfigDict(populate_by_name=True, extra="allow")

    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())
        params = _body_to_params(body)
        extra_filters = _wallet_filters_from_params(params)
        rows, _ = InvoiceService.list_wallets(
            db,
            user=user,
            page=1,
            page_size=EXPORT_MAX_ROWS,
            keyword="",
            extra_filters=extra_filters or None,
        )
        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(
    request: Request,
    page: int = 1,
    pageSize: int = Query(10, ge=1, le=500),
    userId: Optional[str] = None,
    keyword: str = "",
    db: Session = Depends(get_db),
    user: BxUserContext = Depends(get_bx_current_user),
):
    tenant_no = InvoiceService._tenant_no(db, user)
    params = _merge_query(request, userId=userId, keyword=keyword or None)
    extra_filters = _invoice_filters_from_params(params, tenant_no=tenant_no)
    sf, so = parse_sort_params(params)
    rows, total = InvoiceService.list_invoices(
        db,
        user=user,
        owner_user_id=userId,
        page=page,
        page_size=pageSize,
        extra_filters=extra_filters or None,
        order_by=order_invoice(sf, so),
    )
    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(
    request: Request,
    userId: Optional[str] = None,
    keyword: str = "",
    db: Session = Depends(get_db),
    user: BxUserContext = Depends(get_bx_current_user),
):
    params = _merge_query(request, userId=userId, keyword=keyword or None)
    extra_filters = _ledger_filters_from_params(params)
    sf, so = parse_sort_params(params)
    rows = InvoiceService.list_ledgers(
        db,
        user=user,
        owner_user_id=userId,
        extra_filters=extra_filters or None,
        order_by=order_ledger(sf, so),
    )
    return ApiResponse(data=[LedgerOut.model_validate(r) for r in rows])