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

from typing import List, Optional

from datetime import datetime

from fastapi import APIRouter, Depends, HTTPException, Query, status
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.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, WeighingSlipRead, WeighingSlipUpdate
from app.services.db.mappers import (
    datetime_end_of_day,
    inventory_date_to_api,
    inventory_to_read,
    load_user_display_name_map,
    parse_date_str,
    parse_datetime_str,
    plan_to_read,
    weighing_to_read,
)
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 _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("/", response_model=list[WeighingSlipRead])
def list_weighing(
    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="业务类型"),
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
) -> list[WeighingSlipRead]:
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    query = _base_query(db, tenant_no).order_by(DbWeighingSlip.gross_weight_time.desc())

    if q:
        keyword = 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),
            )
        )

    if date_from:
        start_date = parse_date_str(date_from)
        if start_date:
            query = query.filter(
                DbWeighingSlip.gross_weight_time >= datetime.combine(start_date, datetime.min.time())
            )

    if date_to:
        end_date = parse_date_str(date_to)
        if end_date:
            query = query.filter(DbWeighingSlip.gross_weight_time <= datetime_end_of_day(end_date))

    if plan_no:
        query = query.filter(DbWeighingSlip.plan_no.contains(plan_no))

    if batch_no:
        query = query.filter(DbWeighingSlip.batch_no.contains(batch_no))

    if vehicle_no:
        query = query.filter(DbWeighingSlip.vehicle_no.contains(vehicle_no))

    if cargo_name:
        query = query.filter(DbWeighingSlip.cargo_name.contains(cargo_name))

    if supplier:
        query = query.filter(func.lower(DbWeighingSlip.supplier).contains(supplier.lower()))

    if consignee:
        query = query.filter(func.lower(DbWeighingSlip.consignee).contains(consignee.lower()))

    if business_type:
        query = query.filter(DbWeighingSlip.business_type == business_type)

    query = apply_db_scope(query, DbWeighingSlip, user, db, tenant_no)
    weighings = query.all()
    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
    ]


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")
    ids: Optional[List[str]] = None
    columns: Optional[List[str]] = None

    model_config = {"populate_by_name": True}


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 = list_weighing(
            q=body.q,
            date_from=body.date_from,
            date_to=body.date_to,
            plan_no=body.plan_no,
            batch_no=body.batch_no,
            vehicle_no=body.vehicle_no,
            cargo_name=body.cargo_name,
            supplier=body.supplier,
            consignee=body.consignee,
            business_type=body.business_type,
            db=db,
            user=user,
        )
        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,
    )