from typing import Any, List, Optional from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, Request, status 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, 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 _fetch_weighing( db: Session, user: DbUserContext, params: dict ) -> list[WeighingSlipRead]: 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(), ) 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 ] 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( 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="业务类型"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> list[WeighingSlipRead]: 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, ) return _fetch_weighing(db, user, params) 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 = _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, )