shanxia2.0/backend/app/api/pl/inventory.py

"""运输计划管理 - 盘存统计 API"""
from datetime import date, datetime
from decimal import Decimal
from typing import Any, List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field
from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.db_auth import DbUserContext, get_db_current_user
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.core.tenant_context import get_tenant_no
from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)
from app.core.utils import generate_uuid
from app.models.pl_models import (
    PlDocumentReminder,
    PlInventory,
    PlInventoryTransaction,
    PlInventoryWarning,
)
from app.schemas.pl.inventory import (
    DocumentReminderRead,
    DocumentReminderStatsRead,
    DocumentReminderUpdate,
    InventoryLedgerQuery,
    InventoryLedgerResponse,
    InventoryRead,
    InventorySummaryItem,
    InventorySummaryResponse,
    InventoryTransactionQuery,
    InventoryTransactionRead,
    InventoryTransactionResponse,
    InventoryWarningRead,
    InventoryWarningThreshold,
    InventoryWarningUpdate,
    TransportStatsItem,
    TransportStatsResponse,
)
from app.services.pl_cert_service import scan_cert_expiry

router = APIRouter(prefix="/inventory", tags=["运输计划-盘存统计"])


# =============================================================================
# 库存台账
# =============================================================================


@router.get("/ledger", response_model=InventoryLedgerResponse)
def get_inventory_ledger(
    product_name: Optional[str] = Query(default=None, alias="productName", description="产品名称"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> InventoryLedgerResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlInventory).filter(
        PlInventory.tenant_no == tenant_no,
        PlInventory.del_flag == 0,
    )

    if product_name:
        query = query.filter(PlInventory.product_name.contains(product_name))

    query = apply_pl_scope(query, PlInventory, user, db, tenant_no)

    total = query.count()
    offset = (page - 1) * page_size
    items = query.order_by(PlInventory.product_name).offset(offset).limit(page_size).all()

    return InventoryLedgerResponse(total=total, list=items)


# =============================================================================
# 库存预警阈值设置
# =============================================================================



LEDGER_EXPORT_COLUMNS = [
    ExportColumn(title="产品名称", field="productName"),
    ExportColumn(title="品牌", field="brandName"),
    ExportColumn(title="产品编号", field="productNo"),
    ExportColumn(title="规格型号", field="model"),
    ExportColumn(title="单位", field="unit"),
    ExportColumn(title="库存数量", field="stockQty"),
    ExportColumn(title="预警阈值", field="warnThreshold"),
    ExportColumn(title="状态", field="status"),
]


class InventoryLedgerExportBody(BaseModel):
    product_name: Optional[str] = Field(default=None, alias="productName")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None
    model_config = {"populate_by_name": True}


@router.post("/ledger/export")
def export_inventory_ledger(
    body: InventoryLedgerExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        query = db.query(PlInventory).filter(
            PlInventory.tenant_no == tenant_no,
            PlInventory.del_flag == 0,
        )
        if body.product_name:
            query = query.filter(PlInventory.product_name.contains(body.product_name))
        if body.ids:
            query = query.filter(PlInventory.id.in_(body.ids))
        query = apply_pl_scope(query, PlInventory, user, db, tenant_no)
        items = query.order_by(PlInventory.product_name).limit(EXPORT_MAX_ROWS + 1).all()
        rows = []
        for x in items:
            data = InventoryRead.model_validate(x).model_dump(by_alias=True)
            low = data.get("lowStockWarning")
            high = data.get("highStockWarning")
            data["warnThreshold"] = f"{low or '-'} ~ {high or '-'}"
            stock = data.get("stockQty") or 0
            try:
                stock_f = float(stock)
                low_f = float(low or 0)
            except (TypeError, ValueError):
                stock_f, low_f = 0.0, 0.0
            data["status"] = "库存不足" if stock_f < low_f else "正常"
            rows.append(data)
        cols = filter_export_columns(LEDGER_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db, tenant_no=tenant_no, sheet_title="库存台账", columns=cols, rows=rows,
            filename_stem=f"库存台账_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


@router.post("/warning-threshold")
def set_warning_threshold(
    payload: InventoryWarningThreshold,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    brand_id = payload.brand_id or ""
    inventory = (
        db.query(PlInventory)
        .filter(
            PlInventory.product_id == payload.product_id,
            PlInventory.brand_id == brand_id,
            PlInventory.tenant_no == tenant_no,
            PlInventory.del_flag == 0,
        )
        .first()
    )

    if inventory:
        assert_pl_row_in_scope(db, inventory, PlInventory, user, tenant_no)
        inventory.low_stock_warning = payload.low_stock_warning
        inventory.high_stock_warning = payload.high_stock_warning
        if payload.brand_name is not None:
            inventory.brand_name = payload.brand_name
    else:
        # 创建库存记录
        inventory = PlInventory(
            id=generate_uuid(),
            tenant_no=tenant_no,
            product_id=payload.product_id,
            brand_id=brand_id,
            brand_name=payload.brand_name,
            product_no="",
            product_name="",
            unit="吨",
            stock_qty=0,
            stock_amount=0,
            low_stock_warning=payload.low_stock_warning,
            high_stock_warning=payload.high_stock_warning,
            create_by=user.id,
            create_dept_id=user.dept_id,
        )
        db.add(inventory)

    db.commit()
    return {"success": True}


# =============================================================================
# 库存预警列表
# =============================================================================


@router.get("/warnings", response_model=list[InventoryWarningRead])
def list_warnings(
    status: Optional[str] = Query(default=None, description="状态"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[InventoryWarningRead]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlInventoryWarning).filter(
        PlInventoryWarning.tenant_no == tenant_no,
        PlInventoryWarning.del_flag == 0,
    )

    if status:
        query = query.filter(PlInventoryWarning.status == status)

    query = apply_pl_scope(query, PlInventoryWarning, user, db, tenant_no)
    return query.order_by(PlInventoryWarning.create_time.desc()).all()


@router.put("/warnings/{id}", response_model=InventoryWarningRead)
def update_warning(
    id: str,
    payload: InventoryWarningUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> InventoryWarningRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    warning = (
        db.query(PlInventoryWarning)
        .filter(
            PlInventoryWarning.id == id,
            PlInventoryWarning.tenant_no == tenant_no,
            PlInventoryWarning.del_flag == 0,
        )
        .first()
    )

    if not warning:
        raise HTTPException(status_code=404, detail="预警不存在")
    assert_pl_row_in_scope(db, warning, PlInventoryWarning, user, tenant_no)

    if payload.status == "resolved":
        warning.status = "resolved"
        from datetime import datetime

        warning.resolve_time = datetime.now()
        warning.resolve_remark = payload.resolve_remark

    db.commit()
    db.refresh(warning)
    return warning


# =============================================================================
# 出入库明细
# =============================================================================


@router.get("/transactions", response_model=InventoryTransactionResponse)
def list_transactions(
    product_id: Optional[str] = Query(default=None, alias="productId", description="产品ID"),
    transaction_type: Optional[str] = Query(default=None, alias="transactionType", description="事务类型"),
    date_from: Optional[date] = Query(default=None, alias="dateFrom", description="开始日期"),
    date_to: Optional[date] = Query(default=None, alias="dateTo", description="结束日期"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> InventoryTransactionResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = db.query(PlInventoryTransaction).filter(
        PlInventoryTransaction.tenant_no == tenant_no,
        PlInventoryTransaction.del_flag == 0,
    )

    if product_id:
        query = query.filter(PlInventoryTransaction.product_id == product_id)

    if transaction_type:
        query = query.filter(PlInventoryTransaction.transaction_type == transaction_type)

    if date_from:
        query = query.filter(PlInventoryTransaction.create_time >= date_from)

    if date_to:
        query = query.filter(PlInventoryTransaction.create_time <= date_to)

    query = apply_pl_scope(query, PlInventoryTransaction, user, db, tenant_no)

    total = query.count()
    offset = (page - 1) * page_size
    items = query.order_by(PlInventoryTransaction.create_time.desc()).offset(offset).limit(page_size).all()

    return InventoryTransactionResponse(total=total, list=items)


# =============================================================================
# 出入库汇总
# =============================================================================


@router.get("/summary", response_model=InventorySummaryResponse)
def get_summary(
    group_by: str = Query(default="month", description="汇总维度 month/week"),
    date_from: date = Query(..., alias="dateFrom", description="开始日期"),
    date_to: date = Query(..., alias="dateTo", description="结束日期"),
    product_id: Optional[str] = Query(default=None, alias="productId", description="产品ID"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> InventorySummaryResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    query = db.query(PlInventoryTransaction).filter(
        PlInventoryTransaction.tenant_no == tenant_no,
        PlInventoryTransaction.del_flag == 0,
        PlInventoryTransaction.create_time >= date_from,
        PlInventoryTransaction.create_time <= date_to,
    )
    query = apply_pl_scope(query, PlInventoryTransaction, user, db, tenant_no)

    if product_id:
        query = query.filter(PlInventoryTransaction.product_id == product_id)

    transactions = query.all()

    # 按产品分组汇总
    summary_map = {}
    for t in transactions:
        key = t.product_id

        if key not in summary_map:
            summary_map[key] = {
                "product_id": t.product_id,
                "product_name": t.product_name,
                "inbound_qty": Decimal("0"),
                "outbound_qty": Decimal("0"),
            }

        if t.transaction_type == "inbound":
            summary_map[key]["inbound_qty"] += t.qty
        else:
            summary_map[key]["outbound_qty"] += abs(t.qty)

    items = []
    for key, data in summary_map.items():
        net = data["inbound_qty"] - data["outbound_qty"]
        items.append(InventorySummaryItem(
            period="汇总",
            product_id=data["product_id"],
            product_name=data["product_name"],
            inbound_qty=data["inbound_qty"],
            outbound_qty=data["outbound_qty"],
            net_change=net,
        ))

    return InventorySummaryResponse(list=items)


# =============================================================================
# 证件到期提醒
# =============================================================================


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 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 _body_to_params(body: BaseModel) -> dict:
    data = body.model_dump(by_alias=True, exclude_none=True)
    for key in ("ids", "columns"):
        data.pop(key, None)
    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


def _document_reminder_list_kwargs(params: dict) -> dict:
    return dict(
        q=_param_get(params, "q"),
        owner_label=_param_get(params, "ownerLabel", "owner_label"),
        owner_type=_param_get(params, "ownerType", "owner_type"),
        owner_type_in=_param_get(params, "ownerTypeIn", "owner_type_in"),
        document_type=_param_get(params, "documentType", "document_type"),
        document_no=_param_get(params, "documentNo", "document_no"),
        status=_param_get(params, "status"),
        status_in=_param_get(params, "statusIn", "status_in"),
        expire_date_from=_param_get(params, "expireDateFrom", "expire_date_from"),
        expire_date_to=_param_get(params, "expireDateTo", "expire_date_to"),
        sort_field=_param_get(params, "sortField", "sort_field"),
        sort_order=_param_get(params, "sortOrder", "sort_order"),
    )


def _query_document_reminders(
    db: Session,
    *,
    tenant_no: str,
    user: DbUserContext,
    q: Optional[str] = None,
    owner_label: Optional[str] = None,
    owner_type: Optional[str] = None,
    owner_type_in: Optional[str] = None,
    document_type: Optional[str] = None,
    document_no: Optional[str] = None,
    status: Optional[str] = None,
    status_in: Optional[str] = None,
    expire_date_from=None,
    expire_date_to=None,
    ids: Optional[List[str]] = None,
    sort_field: Optional[str] = None,
    sort_order: Optional[str] = None,
):
    from app.services.pl.document_reminder_list_filters import (
        build_document_reminder_column_filters,
    )
    from app.services.pl.list_sort import order_reminder

    query = db.query(PlDocumentReminder).filter(
        PlDocumentReminder.tenant_no == tenant_no,
        PlDocumentReminder.del_flag == 0,
    )
    if q:
        keyword = str(q).strip()
        if keyword:
            query = query.filter(
                or_(
                    PlDocumentReminder.owner_label.contains(keyword),
                    PlDocumentReminder.driver_name.contains(keyword),
                    PlDocumentReminder.document_no.contains(keyword),
                    PlDocumentReminder.document_type.contains(keyword),
                )
            )
    column_filters = build_document_reminder_column_filters(
        owner_label=owner_label,
        owner_type=owner_type,
        owner_type_in=owner_type_in,
        document_type=document_type,
        document_no=document_no,
        status=status,
        status_in=status_in,
        expire_date_from=expire_date_from,
        expire_date_to=expire_date_to,
    )
    if column_filters:
        query = query.filter(*column_filters)
    if ids:
        query = query.filter(PlDocumentReminder.id.in_(ids))
    query = apply_pl_scope(query, PlDocumentReminder, user, db, tenant_no)
    return query.order_by(order_reminder(sort_field, sort_order))


@router.get("/document-reminders/stats", response_model=DocumentReminderStatsRead)
def document_reminder_stats(
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DocumentReminderStatsRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    scan_cert_expiry(db, tenant_no)
    db.commit()

    query = db.query(PlDocumentReminder).filter(
        PlDocumentReminder.tenant_no == tenant_no,
        PlDocumentReminder.del_flag == 0,
        PlDocumentReminder.status == "pending",
        PlDocumentReminder.days_remaining <= 0,
    )
    query = apply_pl_scope(query, PlDocumentReminder, user, db, tenant_no)
    expired_pending_count = query.count()
    return DocumentReminderStatsRead(expired_pending_count=expired_pending_count)


@router.get("/document-reminders", response_model=list[DocumentReminderRead])
def list_document_reminders(
    request: Request,
    status: Optional[str] = Query(default=None, description="状态"),
    q: Optional[str] = Query(default=None),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[DocumentReminderRead]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    scan_cert_expiry(db, tenant_no)
    db.commit()

    params = _merge_query(request, status=status, q=q)
    query = _query_document_reminders(
        db,
        tenant_no=tenant_no,
        user=user,
        **_document_reminder_list_kwargs(params),
    )
    return query.all()



DOCUMENT_REMINDER_EXPORT_COLUMNS = [
    ExportColumn(title="主体名称", field="ownerLabel"),
    ExportColumn(title="主体类型", field="ownerType"),
    ExportColumn(title="证件类型", field="documentType"),
    ExportColumn(title="证件号", field="documentNo"),
    ExportColumn(title="到期日期", field="expireDate"),
    ExportColumn(title="剩余天数", field="daysRemaining"),
    ExportColumn(title="状态", field="status"),
    ExportColumn(title="处理备注", field="remark"),
]


class DocumentReminderExportBody(BaseModel):
    q: Optional[str] = None
    status: Optional[str] = None
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None
    model_config = {"populate_by_name": True, "extra": "allow"}


@router.post("/document-reminders/export")
def export_document_reminders(
    body: DocumentReminderExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    try:
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        scan_cert_expiry(db, tenant_no)
        db.commit()
        params = _body_to_params(body)
        query = _query_document_reminders(
            db,
            tenant_no=tenant_no,
            user=user,
            **_document_reminder_list_kwargs(params),
            ids=body.ids,
        )
        items = query.limit(EXPORT_MAX_ROWS + 1).all()
        status_labels = {"pending": "待处理", "resolved": "已处理"}
        owner_labels = {"vehicle": "车辆", "personnel": "人员", "driver": "司机"}
        rows = []
        for x in items:
            data = DocumentReminderRead.model_validate(x).model_dump(by_alias=True)
            data["remark"] = getattr(x, "remark", None) or ""
            data["status"] = status_labels.get(str(data.get("status") or ""), data.get("status"))
            data["ownerType"] = owner_labels.get(str(data.get("ownerType") or ""), data.get("ownerType"))
            for k, v in list(data.items()):
                if hasattr(v, "isoformat"):
                    data[k] = str(v)[:19]
            rows.append(data)
        cols = filter_export_columns(DOCUMENT_REMINDER_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        return export_rows_to_xlsx_response(
            db, tenant_no=tenant_no, sheet_title="证件提醒", columns=cols, rows=rows,
            filename_stem=f"证件到期提醒_{date_str}",
        )
    except ExportValidationError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e


@router.put("/document-reminders/{id}", response_model=DocumentReminderRead)
def update_document_reminder(
    id: str,
    payload: DocumentReminderUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> DocumentReminderRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())

    reminder = (
        db.query(PlDocumentReminder)
        .filter(
            PlDocumentReminder.id == id,
            PlDocumentReminder.tenant_no == tenant_no,
            PlDocumentReminder.del_flag == 0,
        )
        .first()
    )

    if not reminder:
        raise HTTPException(status_code=404, detail="提醒不存在")
    assert_pl_row_in_scope(db, reminder, PlDocumentReminder, user, tenant_no)

    reminder.status = payload.status
    if payload.remark:
        reminder.remark = payload.remark

    db.commit()
    db.refresh(reminder)
    return reminder


# =============================================================================
# 生成证件到期提醒（定时任务调用）
# =============================================================================


@router.post("/check-document-expiry")
def check_document_expiry(db: Session = Depends(get_db)) -> dict:
    """检查证件到期，生成提醒（供定时任务调用）"""
    created_count = scan_cert_expiry(db)
    db.commit()
    return {"created_count": created_count}