shanxia2.0/backend/app/api/db/weighing.py

from typing import Any, List, Optional, Tuple

from datetime import datetime

from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from sqlalchemy import and_, 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.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
from app.core.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
from app.services.db_data_scope import apply_db_scope, assert_db_row_in_scope
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
from app.schemas.db.inventory import InboundOutboundRead
from app.schemas.db.plans import PurchasePlanRead
from app.schemas.db.weighing import (
    WeighingSlipCreate,
    WeighingSlipListResponse,
    WeighingSlipRead,
    WeighingSlipUpdate,
)
from app.services.db.mappers import (
    inventory_date_to_api,
    inventory_to_read,
    load_user_display_name_map,
    parse_datetime_str,
    plan_to_read,
    weighing_to_read,
)
from app.services.db.list_sort import (
    order_weighing,
    parse_sort_params,
    weighing_needs_inventory_join,
)
from app.services.db.weighing_list_filters import build_weighing_column_filters
from app.services.db.weighing_logic import process_weighing_business_logic
from app.services.db.product_helpers import (
    apply_product_snapshot_to_weighing,
    resolve_crm_product,
)
from app.utils.excel_export import (
    EXPORT_MAX_ROWS,
    ExportColumn,
    ExportValidationError,
    export_rows_to_xlsx_response,
    filter_export_columns,
)

router = APIRouter(prefix="/weighing", tags=["weighing"])

_WEIGHING_REMARK_FIELD = "remarks"


def _base_query(db: Session, tenant_no: str):
    return db.query(DbWeighingSlip).filter(
        DbWeighingSlip.tenant_no == tenant_no,
        DbWeighingSlip.del_flag == 0,
        DbWeighingSlip.is_voided == 0,
    )


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 _filters_from_params(params: dict) -> list:
    return build_weighing_column_filters(
        plan_no=_param_get(params, "planNo", "plan_no"),
        batch_no=_param_get(params, "batchNo", "batch_no"),
        vehicle_no=_param_get(params, "vehicleNo", "vehicle_no"),
        cargo_name=_param_get(params, "cargoName", "cargo_name"),
        supplier=_param_get(params, "supplier"),
        consignee=_param_get(params, "consignee"),
        business_type=_param_get(params, "businessType", "business_type"),
        business_type_in=_param_get(params, "businessTypeIn", "business_type_in"),
        date_from=_param_get(params, "dateFrom", "date_from"),
        date_to=_param_get(params, "dateTo", "date_to"),
        net_weight_op=_param_get(params, "netWeightOp", "net_weight_op"),
        net_weight=_param_get(params, "netWeight", "net_weight"),
        net_weight_min=_param_get(params, "netWeightMin", "net_weight_min"),
        net_weight_max=_param_get(params, "netWeightMax", "net_weight_max"),
        gross_weight_op=_param_get(params, "grossWeightOp", "gross_weight_op"),
        gross_weight=_param_get(params, "grossWeight", "gross_weight"),
        gross_weight_min=_param_get(params, "grossWeightMin", "gross_weight_min"),
        gross_weight_max=_param_get(params, "grossWeightMax", "gross_weight_max"),
    )


def _body_to_params(body: "WeighingExportBody") -> dict:
    data = body.model_dump(by_alias=True, exclude_none=True)
    data.pop("ids", None)
    data.pop("columns", 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 _weighing_query(db: Session, user: DbUserContext, params: dict):
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = _base_query(db, tenant_no)

    q = _param_get(params, "q")
    if q:
        keyword = str(q).lower()
        query = query.filter(
            or_(
                func.lower(DbWeighingSlip.vehicle_no).contains(keyword),
                func.lower(DbWeighingSlip.plan_no).contains(keyword),
                func.lower(DbWeighingSlip.batch_no).contains(keyword),
                func.lower(DbWeighingSlip.cargo_name).contains(keyword),
                func.lower(DbWeighingSlip.supplier).contains(keyword),
                func.lower(DbWeighingSlip.consignee).contains(keyword),
            )
        )

    column_filters = _filters_from_params(params)
    if column_filters:
        query = query.filter(*column_filters)

    query = apply_db_scope(query, DbWeighingSlip, user, db, tenant_no)
    sf, so = parse_sort_params(params)
    if weighing_needs_inventory_join(sf, so):
        query = query.outerjoin(
            DbInboundOutbound,
            and_(
                DbWeighingSlip.inbound_outbound_id == DbInboundOutbound.id,
                DbInboundOutbound.del_flag == 0,
            ),
        )
    query = query.order_by(
        order_weighing(sf, so, default=DbWeighingSlip.gross_weight_time.desc()),
        DbWeighingSlip.id.desc(),
    )
    return query


def _map_weighings(db: Session, weighings: list) -> list[WeighingSlipRead]:
    inventory_ids = [w.inbound_outbound_id for w in weighings if w.inbound_outbound_id]
    inventory_date_map: dict[str, str] = {}
    if inventory_ids:
        inventories = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.id.in_(inventory_ids),
                DbInboundOutbound.del_flag == 0,
            )
            .all()
        )
        for inventory in inventories:
            inventory_date_map[inventory.id] = inventory_date_to_api(inventory.date)

    user_name_map = load_user_display_name_map(db, [w.create_by for w in weighings])
    return [
        weighing_to_read(
            weighing,
            inventory_date=inventory_date_map.get(weighing.inbound_outbound_id or ""),
            user_name_map=user_name_map,
        )
        for weighing in weighings
    ]


def _fetch_weighing(
    db: Session,
    user: DbUserContext,
    params: dict,
    *,
    page: Optional[int] = None,
    page_size: Optional[int] = None,
) -> Tuple[list[WeighingSlipRead], int]:
    query = _weighing_query(db, user, params)
    if page is not None and page_size is not None:
        total = query.order_by(None).count()
        weighings = query.offset((page - 1) * page_size).limit(page_size).all()
        return _map_weighings(db, weighings), total
    weighings = query.all()
    items = _map_weighings(db, weighings)
    return items, len(items)


def _get_weighing(
    db: Session, tenant_no: str, weighing_id: str, user: DbUserContext
) -> DbWeighingSlip:
    row = (
        db.query(DbWeighingSlip)
        .filter(
            DbWeighingSlip.id == weighing_id,
            DbWeighingSlip.tenant_no == tenant_no,
            DbWeighingSlip.del_flag == 0,
        )
        .one_or_none()
    )
    if row is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Weighing {weighing_id} not found",
        )
    assert_db_row_in_scope(db, row, DbWeighingSlip, user, tenant_no)
    return row


def _apply_weighing_product_update(
    db: Session, tenant_no: str, row: DbWeighingSlip, payload_dict: dict
) -> None:
    if "cargo_name" in payload_dict and "product_id" not in payload_dict:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="请从 CRM 重新选择产品",
        )
    product_id = payload_dict.pop("product_id", None)
    payload_dict.pop("cargo_name", None)
    payload_dict.pop("specification", None)
    if product_id is not None:
        snapshot = resolve_crm_product(db, tenant_no, product_id)
        apply_product_snapshot_to_weighing(row, snapshot)


def _apply_weighing_update(row: DbWeighingSlip, payload_dict: dict) -> None:
    for key, value in payload_dict.items():
        if key == _WEIGHING_REMARK_FIELD:
            if isinstance(value, str):
                value = value.strip()
            row.remark = value
            continue
        if key in {"gross_weight_time", "tare_weight_time"}:
            if value is None or (isinstance(value, str) and not str(value).strip()):
                if key == "tare_weight_time":
                    row.tare_weight_time = None
            else:
                parsed = parse_datetime_str(str(value))
                if parsed:
                    setattr(row, key, parsed)
            continue
        if isinstance(value, str):
            value = value.strip()
        setattr(row, key, value)


@router.get("/")
def list_weighing(
    request: Request,
    q: Optional[str] = Query(default=None, description="模糊搜索关键词"),
    date_from: Optional[str] = Query(default=None, alias="dateFrom", description="开始日期"),
    date_to: Optional[str] = Query(default=None, alias="dateTo", description="结束日期"),
    plan_no: Optional[str] = Query(default=None, alias="planNo", description="计划号"),
    batch_no: Optional[str] = Query(default=None, alias="batchNo", description="出入库单号"),
    vehicle_no: Optional[str] = Query(default=None, alias="vehicleNo", description="车牌号"),
    cargo_name: Optional[str] = Query(default=None, alias="cargoName", description="货物名称"),
    supplier: Optional[str] = Query(default=None, description="供货单位"),
    consignee: Optional[str] = Query(default=None, description="收货单位"),
    business_type: Optional[str] = Query(default=None, alias="businessType", description="业务类型"),
    page: Optional[int] = Query(default=None, ge=1, description="页码；传入则返回分页结构"),
    page_size: Optional[int] = Query(
        default=None, ge=1, le=500, alias="pageSize", description="每页条数"
    ),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    params = _merge_query(
        request,
        q=q,
        dateFrom=date_from,
        dateTo=date_to,
        planNo=plan_no,
        batchNo=batch_no,
        vehicleNo=vehicle_no,
        cargoName=cargo_name,
        supplier=supplier,
        consignee=consignee,
        businessType=business_type,
    )
    # 管理端未传 page：仍返回全量数组（前端本地分页）；小程序传 page/pageSize 走服务端分页
    paged = page is not None
    size = page_size if page_size is not None else 20
    page_num = page if page is not None else 1
    items, total = _fetch_weighing(
        db,
        user,
        params,
        page=page_num if paged else None,
        page_size=size if paged else None,
    )
    if paged:
        body = WeighingSlipListResponse(
            items=items, total=total, page=page_num, pageSize=size
        )
        return JSONResponse(content=body.model_dump(by_alias=True))
    return JSONResponse(
        content=[item.model_dump(by_alias=True) for item in items]
    )


WEIGHING_EXPORT_COLUMNS = [
    ExportColumn(title="日期", field="inventoryDate"),
    ExportColumn(title="计划单号", field="planNo"),
    ExportColumn(title="出入库单号", field="batchNo"),
    ExportColumn(title="车牌号", field="vehicleNo"),
    ExportColumn(title="货物名称", field="cargoName"),
    ExportColumn(title="规格", field="specification"),
    ExportColumn(title="业务类型", field="businessType"),
    ExportColumn(title="单据状态", field="docStatus"),
    ExportColumn(title="供货单位", field="supplier"),
    ExportColumn(title="收货单位", field="consignee"),
    ExportColumn(title="本次量(吨)", field="currentQuantity"),
    ExportColumn(title="净重(吨)", field="netWeight"),
    ExportColumn(title="毛重(kg)", field="grossWeight"),
    ExportColumn(title="毛重时间", field="grossWeightTime"),
    ExportColumn(title="皮重(kg)", field="tareWeight"),
    ExportColumn(title="皮重时间", field="tareWeightTime"),
    ExportColumn(title="扣杂(kg)", field="deduction"),
    ExportColumn(title="创建人", field="createByName"),
]


class WeighingExportBody(BaseModel):
    q: Optional[str] = None
    date_from: Optional[str] = Field(default=None, alias="dateFrom")
    date_to: Optional[str] = Field(default=None, alias="dateTo")
    plan_no: Optional[str] = Field(default=None, alias="planNo")
    batch_no: Optional[str] = Field(default=None, alias="batchNo")
    vehicle_no: Optional[str] = Field(default=None, alias="vehicleNo")
    cargo_name: Optional[str] = Field(default=None, alias="cargoName")
    supplier: Optional[str] = None
    consignee: Optional[str] = None
    business_type: Optional[str] = Field(default=None, alias="businessType")
    sort_field: Optional[str] = Field(default=None, alias="sortField")
    sort_order: Optional[str] = Field(default=None, alias="sortOrder")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True, "extra": "allow"}


def _kg_to_ton(value) -> float:
    try:
        return round(float(value or 0) / 1000, 2)
    except (TypeError, ValueError):
        return 0.0


@router.post("/export")
def export_weighing(
    body: WeighingExportBody,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    """导出地磅单列表（当前筛选全量，或 ids 已选）。"""
    try:
        items, _total = _fetch_weighing(db, user, _body_to_params(body))
        if body.ids:
            id_set = set(body.ids)
            items = [x for x in items if x.id in id_set]
        if len(items) > EXPORT_MAX_ROWS:
            raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条，请缩小筛选范围")
        rows = []
        for item in items:
            data = item.model_dump(by_alias=True)
            data["docStatus"] = "已作废" if data.get("isVoided") == 1 else "正常"
            data["currentQuantity"] = _kg_to_ton(data.get("currentQuantity"))
            data["netWeight"] = _kg_to_ton(data.get("netWeight"))
            if not data.get("inventoryDate"):
                gwt = data.get("grossWeightTime") or ""
                data["inventoryDate"] = str(gwt)[:10] if gwt else ""
            if not data.get("createByName"):
                data["createByName"] = data.get("createBy") or ""
            rows.append(data)
        cols = filter_export_columns(WEIGHING_EXPORT_COLUMNS, body.columns)
        date_str = datetime.now().strftime("%Y%m%d")
        tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
        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("/", response_model=WeighingSlipRead, status_code=status.HTTP_201_CREATED)
def create_weighing(
    payload: WeighingSlipCreate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> WeighingSlipRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    inbound_outbound_id = payload.inbound_outbound_id

    if inbound_outbound_id:
        existing = (
            db.query(DbWeighingSlip)
            .filter(
                DbWeighingSlip.inbound_outbound_id == inbound_outbound_id,
                DbWeighingSlip.is_voided == 0,
                DbWeighingSlip.del_flag == 0,
            )
            .first()
        )
        if existing:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="该出入库单已经存在关联的地磅单，无法重复创建",
            )

    planned_quantity = payload.planned_quantity
    current_quantity = payload.current_quantity
    purchase_plan_id = payload.purchase_plan_id
    batch_no = payload.batch_no.strip() if payload.batch_no else None

    if inbound_outbound_id:
        inventory = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.id == inbound_outbound_id,
                DbInboundOutbound.tenant_no == tenant_no,
                DbInboundOutbound.del_flag == 0,
            )
            .one_or_none()
        )
        if inventory:
            if current_quantity is None:
                current_quantity = inventory.quantity * 1000
            if purchase_plan_id is None and inventory.purchase_plan_id:
                purchase_plan_id = inventory.purchase_plan_id
            if purchase_plan_id:
                plan = (
                    db.query(DbPurchasePlan)
                    .filter(DbPurchasePlan.id == purchase_plan_id)
                    .one_or_none()
                )
                if plan and planned_quantity is None:
                    planned_quantity = plan.planned_quantity * 1000
            if not batch_no and inventory.batch_no:
                batch_no = inventory.batch_no

    gross_time = parse_datetime_str(payload.gross_weight_time) if payload.gross_weight_time else None
    tare_time = (
        parse_datetime_str(payload.tare_weight_time)
        if payload.tare_weight_time and payload.tare_weight_time.strip()
        else None
    )

    snapshot = resolve_crm_product(db, tenant_no, payload.product_id)

    row = DbWeighingSlip(
        id=generate_uuid(),
        tenant_no=tenant_no,
        vehicle_no=payload.vehicle_no.strip(),
        plan_no=payload.plan_no.strip() if payload.plan_no else None,
        business_type=payload.business_type,
        batch_no=batch_no,
        supplier=payload.supplier.strip(),
        consignee=payload.consignee.strip(),
        gross_weight=payload.gross_weight or 0,
        gross_weight_time=gross_time or datetime.now(),
        tare_weight=payload.tare_weight or 0,
        tare_weight_time=tare_time,
        deduction=payload.deduction or 0,
        net_weight=payload.net_weight or 0,
        planned_quantity=planned_quantity,
        executed_quantity=payload.executed_quantity,
        purchase_plan_id=purchase_plan_id,
        inbound_outbound_id=inbound_outbound_id,
        weighing_point=payload.weighing_point or "山峡化工",
        remark=payload.remarks,
        current_quantity=current_quantity,
        is_voided=0,
        create_by=user.id,
        create_dept_id=user.dept_id,
    )
    apply_product_snapshot_to_weighing(row, snapshot)
    db.add(row)
    db.commit()
    db.refresh(row)

    process_weighing_business_logic(db, row)
    db.refresh(row)
    return weighing_to_read(row, db=db)


@router.get("/{id}", response_model=WeighingSlipRead)
def get_weighing(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> WeighingSlipRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    row = _get_weighing(db, tenant_no, id, user)
    return weighing_to_read(row, db=db)


@router.patch("/{id}", response_model=WeighingSlipRead)
def update_weighing(
    id: str,
    payload: WeighingSlipUpdate,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> WeighingSlipRead:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    row = _get_weighing(db, tenant_no, id, user)

    if row.is_voided == 1:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Cannot update voided weighing record",
        )

    payload_dict = payload.model_dump(exclude_unset=True)
    _apply_weighing_product_update(db, tenant_no, row, payload_dict)
    _apply_weighing_update(row, payload_dict)
    row.update_by = user.id
    db.commit()
    db.refresh(row)

    process_weighing_business_logic(db, row)
    db.refresh(row)
    return weighing_to_read(row, db=db)


@router.post("/{id}/void", status_code=status.HTTP_204_NO_CONTENT)
def void_weighing(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    row = _get_weighing(db, tenant_no, id, user)

    if row.is_voided == 1:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Weighing record is already voided",
        )

    row.is_voided = 1
    row.update_by = user.id
    db.commit()
    process_weighing_business_logic(db, row)


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_weighing(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> None:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    row = _get_weighing(db, tenant_no, id, user)
    assert_row_deletable_by_creator(row, user_info_from_context(user))
    row.is_voided = 1
    row.update_by = user.id
    db.commit()
    process_weighing_business_logic(db, row)


class WeighingRelatedData(BaseModel):
    plan: Optional[PurchasePlanRead]
    inventory: Optional[InboundOutboundRead]


@router.get("/{id}/related", response_model=WeighingRelatedData)
def get_weighing_related(
    id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> WeighingRelatedData:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    row = _get_weighing(db, tenant_no, id, user)

    plan = None
    if row.purchase_plan_id:
        plan = (
            db.query(DbPurchasePlan)
            .filter(
                DbPurchasePlan.id == row.purchase_plan_id,
                DbPurchasePlan.del_flag == 0,
            )
            .one_or_none()
        )
    elif row.plan_no:
        plan = (
            db.query(DbPurchasePlan)
            .filter(
                DbPurchasePlan.plan_no == row.plan_no,
                DbPurchasePlan.tenant_no == tenant_no,
                DbPurchasePlan.del_flag == 0,
            )
            .limit(1)
            .first()
        )

    inventory = None
    if row.inbound_outbound_id:
        inventory = (
            db.query(DbInboundOutbound)
            .filter(
                DbInboundOutbound.id == row.inbound_outbound_id,
                DbInboundOutbound.del_flag == 0,
            )
            .one_or_none()
        )

    return WeighingRelatedData(
        plan=plan_to_read(plan, db=db) if plan else None,
        inventory=inventory_to_read(inventory, db=db) if inventory else None,
    )