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

"""运输计划 · 入库单 / 出库单管理端 API"""
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import 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.core.tenant_context import get_tenant_no
from app.models.pl_models import (
    PlInboundOrder,
    PlInboundOrderItem,
    PlOutboundOrder,
    PlOutboundOrderItem,
)
from app.schemas.pl.io_order import IoOrderListResponse, IoOrderPatch, IoOrderRead
from app.services.pl_io_order_service import (
    dump_feedback_images,
    order_header_to_dict,
    update_io_order_items_qty_and_brand,
)

inbound_router = APIRouter(prefix="/inbound", tags=["运输计划-入库单"])
outbound_router = APIRouter(prefix="/outbound", tags=["运输计划-出库单"])

_STATUS_OK = {"in_progress", "completed", "voided"}


def _models(kind: str):
    if kind == "inbound":
        return PlInboundOrder, PlInboundOrderItem
    return PlOutboundOrder, PlOutboundOrderItem


def _get_order(db: Session, tenant_no: str, kind: str, order_id: str):
    header_m, item_m = _models(kind)
    header = (
        db.query(header_m)
        .filter(header_m.id == order_id, header_m.tenant_no == tenant_no, header_m.del_flag == 0)
        .first()
    )
    if not header:
        raise HTTPException(status_code=404, detail="单据不存在")
    items = (
        db.query(item_m)
        .filter(item_m.order_id == header.id, item_m.del_flag == 0)
        .order_by(item_m.order_num.asc(), item_m.create_time.asc())
        .all()
    )
    return header, items


def _to_read(kind: str, header, items, db: Session) -> IoOrderRead:
    return IoOrderRead.model_validate(
        order_header_to_dict(kind, header, items, db=db)
    )


def _list_orders(
    *,
    kind: str,
    order_no: Optional[str],
    plan_no: Optional[str],
    status: Optional[str],
    partner_company: Optional[str],
    vehicle_no: Optional[str],
    q: Optional[str],
    page: int,
    page_size: int,
    db: Session,
    user: DbUserContext,
) -> IoOrderListResponse:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header_m, item_m = _models(kind)
    query = db.query(header_m).filter(header_m.tenant_no == tenant_no, header_m.del_flag == 0)
    if order_no:
        query = query.filter(header_m.order_no.contains(order_no))
    if plan_no:
        query = query.filter(header_m.plan_no.contains(plan_no))
    if status:
        query = query.filter(header_m.status == status)
    if partner_company:
        query = query.filter(header_m.partner_company.contains(partner_company))
    if vehicle_no:
        query = query.filter(header_m.vehicle_no.contains(vehicle_no))
    if q:
        query = query.filter(
            or_(
                header_m.order_no.contains(q),
                header_m.plan_no.contains(q),
                header_m.partner_company.contains(q),
                header_m.vehicle_no.contains(q),
            )
        )
    total = query.count()
    rows = (
        query.order_by(header_m.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    result: list[IoOrderRead] = []
    for row in rows:
        items = (
            db.query(item_m)
            .filter(item_m.order_id == row.id, item_m.del_flag == 0)
            .order_by(item_m.order_num.asc())
            .all()
        )
        result.append(_to_read(kind, row, items, db))
    return IoOrderListResponse(total=total, list=result)


def _patch_order(
    *,
    kind: str,
    order_id: str,
    payload: IoOrderPatch,
    db: Session,
    user: DbUserContext,
) -> IoOrderRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, items = _get_order(db, tenant_no, kind, order_id)
    if payload.status is not None:
        if payload.status not in _STATUS_OK:
            raise HTTPException(status_code=400, detail="无效状态")
        header.status = payload.status
    if payload.site_feedback_remark is not None:
        header.site_feedback_remark = (payload.site_feedback_remark or "").strip() or None
    if payload.site_feedback_images is not None:
        header.site_feedback_images = dump_feedback_images(payload.site_feedback_images)
        header.site_feedback_time = datetime.now()
    if payload.remark is not None:
        header.remark = payload.remark
    if payload.items:
        by_id = {it.id: it for it in items}
        qty_map: dict[str, Decimal] = {}
        brand_qty_map: dict[str, list[dict]] | None = None
        for patch in payload.items:
            row = by_id.get(patch.id)
            if not row:
                continue
            qty_map[patch.id] = Decimal(str(patch.actual_qty))
            if patch.brand_qtys is not None:
                if brand_qty_map is None:
                    brand_qty_map = {}
                brand_qty_map[patch.id] = [
                    b.model_dump(by_alias=True) for b in patch.brand_qtys
                ]
        effective_status = payload.status if payload.status is not None else header.status
        try:
            update_io_order_items_qty_and_brand(
                db,
                kind=kind,
                tenant_no=tenant_no,
                order_id=header.id,
                items=items,
                item_qty=qty_map,
                item_brand_qtys=brand_qty_map,
                enforce_brand=(effective_status == "completed"),
                user_id=user.id,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
    header.update_by = user.id
    db.commit()
    header, items = _get_order(db, tenant_no, kind, order_id)
    return _to_read(kind, header, items, db)


def _get_by_plan(db: Session, tenant_no: str, kind: str, plan_id: str):
    """按计划查找入库/出库单；未落单时返回 (None, [])，与运费 by-transport 一致。"""
    header_m, item_m = _models(kind)
    header = (
        db.query(header_m)
        .filter(
            header_m.tenant_no == tenant_no,
            header_m.plan_id == plan_id,
            header_m.del_flag == 0,
        )
        .order_by(header_m.create_time.desc())
        .first()
    )
    if not header:
        return None, []
    items = (
        db.query(item_m)
        .filter(item_m.order_id == header.id, item_m.del_flag == 0)
        .order_by(item_m.order_num.asc(), item_m.create_time.asc())
        .all()
    )
    return header, items


@inbound_router.get("/by-plan/{plan_id}", response_model=Optional[IoOrderRead])
def get_inbound_by_plan(
    plan_id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> Optional[IoOrderRead]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, items = _get_by_plan(db, tenant_no, "inbound", plan_id)
    if not header:
        return None
    return _to_read("inbound", header, items, db)


@inbound_router.get("/", response_model=IoOrderListResponse)
def list_inbound(
    order_no: Optional[str] = Query(default=None, alias="orderNo"),
    plan_no: Optional[str] = Query(default=None, alias="planNo"),
    status: Optional[str] = Query(default=None),
    partner_company: Optional[str] = Query(default=None, alias="partnerCompany"),
    vehicle_no: Optional[str] = Query(default=None, alias="vehicleNo"),
    q: Optional[str] = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> IoOrderListResponse:
    return _list_orders(
        kind="inbound",
        order_no=order_no,
        plan_no=plan_no,
        status=status,
        partner_company=partner_company,
        vehicle_no=vehicle_no,
        q=q,
        page=page,
        page_size=page_size,
        db=db,
        user=user,
    )


@inbound_router.get("/{id}", response_model=IoOrderRead)
def get_inbound(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> IoOrderRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, items = _get_order(db, tenant_no, "inbound", id)
    return _to_read("inbound", header, items, db)


@inbound_router.patch("/{id}", response_model=IoOrderRead)
def patch_inbound(
    id: str,
    payload: IoOrderPatch,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> IoOrderRead:
    return _patch_order(kind="inbound", order_id=id, payload=payload, db=db, user=user)


@inbound_router.get("/{id}/site-url")
def inbound_site_url(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, _ = _get_order(db, tenant_no, "inbound", id)
    data = order_header_to_dict("inbound", header, None)
    return {"siteUrl": data.get("siteUrl"), "siteAccessToken": data.get("siteAccessToken")}


@outbound_router.get("/by-plan/{plan_id}", response_model=Optional[IoOrderRead])
def get_outbound_by_plan(
    plan_id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> Optional[IoOrderRead]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, items = _get_by_plan(db, tenant_no, "outbound", plan_id)
    if not header:
        return None
    return _to_read("outbound", header, items, db)


@outbound_router.get("/", response_model=IoOrderListResponse)
def list_outbound(
    order_no: Optional[str] = Query(default=None, alias="orderNo"),
    plan_no: Optional[str] = Query(default=None, alias="planNo"),
    status: Optional[str] = Query(default=None),
    partner_company: Optional[str] = Query(default=None, alias="partnerCompany"),
    vehicle_no: Optional[str] = Query(default=None, alias="vehicleNo"),
    q: Optional[str] = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100, alias="pageSize"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> IoOrderListResponse:
    return _list_orders(
        kind="outbound",
        order_no=order_no,
        plan_no=plan_no,
        status=status,
        partner_company=partner_company,
        vehicle_no=vehicle_no,
        q=q,
        page=page,
        page_size=page_size,
        db=db,
        user=user,
    )


@outbound_router.get("/{id}", response_model=IoOrderRead)
def get_outbound(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> IoOrderRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, items = _get_order(db, tenant_no, "outbound", id)
    return _to_read("outbound", header, items, db)


@outbound_router.patch("/{id}", response_model=IoOrderRead)
def patch_outbound(
    id: str,
    payload: IoOrderPatch,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> IoOrderRead:
    return _patch_order(kind="outbound", order_id=id, payload=payload, db=db, user=user)


@outbound_router.get("/{id}/site-url")
def outbound_site_url(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> dict:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    header, _ = _get_order(db, tenant_no, "outbound", id)
    data = order_header_to_dict("outbound", header, None)
    return {"siteUrl": data.get("siteUrl"), "siteAccessToken": data.get("siteAccessToken")}