from typing import List, Optional import re 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, get_db_current_user_optional 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.services.db_data_scope import apply_db_scope, assert_db_row_in_scope from app.services.db.inventory_id import next_inbound_outbound_id from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip from app.schemas.db.inventory import ( CreateLinkForPlanPayload, InboundOutboundCreate, InboundOutboundRead, InboundOutboundUpdate, ) from app.schemas.db.public_inventory import InventorySiteUrlRead from app.schemas.db.plans import PurchasePlanRead from app.schemas.db.weighing import WeighingSlipRead from app.services.db.mappers import ( EMPTY_INVENTORY_DATE, inventory_to_read, is_empty_inventory_date, load_user_display_name_map, parse_date_str, plan_to_read, weighing_to_read, ) from app.services.db.product_helpers import ( apply_product_snapshot_to_inventory, resolve_crm_product, ) from app.services.db_inventory_site_token import build_site_inventory_url, ensure_site_access_token from app.utils.excel_export import ( EXPORT_MAX_ROWS, ExportColumn, ExportValidationError, export_rows_to_xlsx_response, filter_export_columns, ) router = APIRouter(prefix="/inventory", tags=["inventory"]) STATUS_THIRD_PARTY_FILL = "待三方填写" _INVENTORY_REMARK_FIELD = "remarks" def _base_query(db: Session, tenant_no: str): return db.query(DbInboundOutbound).filter( DbInboundOutbound.tenant_no == tenant_no, DbInboundOutbound.del_flag == 0, ) def _generate_batch_no(db: Session, tenant_no: str) -> str: rows = ( db.query(DbInboundOutbound.batch_no) .filter( DbInboundOutbound.tenant_no == tenant_no, DbInboundOutbound.del_flag == 0, DbInboundOutbound.batch_no.like("CRKDH%"), ) .all() ) max_seq = 0 pattern = re.compile(r"^CRKDH(\d+)$") for (batch_no,) in rows: if batch_no: match = pattern.match(batch_no) if match: max_seq = max(max_seq, int(match.group(1))) return f"CRKDH{max_seq + 1:07d}" def _resolve_tenant_no( db: Session, user: Optional[DbUserContext], purchase_plan_id: Optional[str] = None, ) -> str: if user: return get_tenant_no(db, user.to_tenant_user_dict()) if purchase_plan_id: plan = ( db.query(DbPurchasePlan) .filter(DbPurchasePlan.id == purchase_plan_id, DbPurchasePlan.del_flag == 0) .one_or_none() ) if plan: return plan.tenant_no raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="无法确定租户") def _get_inventory_by_id(db: Session, inventory_id: str) -> DbInboundOutbound: row = ( db.query(DbInboundOutbound) .filter(DbInboundOutbound.id == inventory_id, DbInboundOutbound.del_flag == 0) .one_or_none() ) if row is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Inventory {inventory_id} not found", ) return row def _get_inventory( db: Session, tenant_no: str, inventory_id: str, user: DbUserContext ) -> DbInboundOutbound: row = ( _base_query(db, tenant_no) .filter(DbInboundOutbound.id == inventory_id) .one_or_none() ) if row is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Inventory {inventory_id} not found", ) assert_db_row_in_scope(db, row, DbInboundOutbound, user, tenant_no) return row def _apply_inventory_product_update( db: Session, tenant_no: str, row: DbInboundOutbound, payload_dict: dict ) -> None: if "product" 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("product", 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_inventory(row, snapshot) def _apply_inventory_update(row: DbInboundOutbound, payload_dict: dict) -> None: for key, value in payload_dict.items(): if key == _INVENTORY_REMARK_FIELD: if isinstance(value, str): value = value.strip() row.remark = value continue if key == "date": if value is None or (isinstance(value, str) and not value.strip()): row.date = EMPTY_INVENTORY_DATE else: row.date = parse_date_str(str(value)) or EMPTY_INVENTORY_DATE continue if isinstance(value, str): value = value.strip() setattr(row, key, value) def _maybe_promote_third_party_status(row: DbInboundOutbound) -> None: if row.status != STATUS_THIRD_PARTY_FILL: return has_date = not is_empty_inventory_date(row.date) has_qty = row.quantity is not None and row.quantity != 0 has_vehicle = bool(row.vehicle_no and row.vehicle_no.strip()) if has_date and has_qty and has_vehicle: row.status = "待入库" if row.business_type == "入库" else "待出库" @router.get("/", response_model=list[InboundOutboundRead]) def list_inventory( q: Optional[str] = Query(default=None, description="模糊搜索关键词"), plan_no: Optional[str] = Query(default=None, alias="planNo", description="计划号"), batch_no: Optional[str] = Query(default=None, alias="batchNo", description="出入库单号"), partner_company: Optional[str] = Query( default=None, alias="partnerCompany", description="往来单位" ), product: Optional[str] = Query(default=None, description="产品"), vehicle_no: Optional[str] = Query(default=None, alias="vehicleNo", description="车牌号"), date_from: Optional[str] = Query(default=None, alias="dateFrom", description="开始日期"), date_to: Optional[str] = Query(default=None, alias="dateTo", description="结束日期"), status_filter: Optional[str] = Query(default=None, alias="status", description="状态筛选"), business_type: Optional[str] = Query(default=None, alias="businessType", description="业务类型"), db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> list[InboundOutboundRead]: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) query = _base_query(db, tenant_no).order_by( DbInboundOutbound.date.desc(), DbInboundOutbound.id.desc(), ) if q: keyword = q.lower() query = query.filter( or_( func.lower(DbInboundOutbound.partner_company).contains(keyword), func.lower(DbInboundOutbound.product).contains(keyword), func.lower(DbInboundOutbound.plan_no).contains(keyword), func.lower(DbInboundOutbound.batch_no).contains(keyword), func.lower(DbInboundOutbound.vehicle_no).contains(keyword), ) ) if plan_no: query = query.filter(DbInboundOutbound.plan_no.contains(plan_no)) if batch_no: query = query.filter(DbInboundOutbound.batch_no.contains(batch_no)) if partner_company: query = query.filter( func.lower(DbInboundOutbound.partner_company).contains(partner_company.lower()) ) if product: query = query.filter(func.lower(DbInboundOutbound.product).contains(product.lower())) if vehicle_no: query = query.filter(DbInboundOutbound.vehicle_no.contains(vehicle_no)) if date_from: start = parse_date_str(date_from) if start: query = query.filter(DbInboundOutbound.date >= start) if date_to: end = parse_date_str(date_to) if end: query = query.filter(DbInboundOutbound.date <= end) if status_filter: query = query.filter(DbInboundOutbound.status == status_filter) if business_type: query = query.filter(DbInboundOutbound.business_type == business_type) query = apply_db_scope(query, DbInboundOutbound, user, db, tenant_no) inventories = query.all() inventory_ids = [inv.id for inv in inventories] weighing_map: dict[str, DbWeighingSlip] = {} if inventory_ids: weighings = ( db.query(DbWeighingSlip) .filter( DbWeighingSlip.tenant_no == tenant_no, DbWeighingSlip.del_flag == 0, DbWeighingSlip.inbound_outbound_id.in_(inventory_ids), DbWeighingSlip.is_voided == 0, ) .order_by(DbWeighingSlip.id.asc()) .all() ) for weighing in weighings: if ( weighing.inbound_outbound_id and weighing.inbound_outbound_id not in weighing_map ): weighing_map[weighing.inbound_outbound_id] = weighing result_list: list[InboundOutboundRead] = [] user_name_map = load_user_display_name_map(db, [inv.create_by for inv in inventories]) for inventory in inventories: try: result_list.append( inventory_to_read( inventory, weighing=weighing_map.get(inventory.id), user_name_map=user_name_map, ) ) except Exception: continue return result_list INVENTORY_EXPORT_COLUMNS = [ ExportColumn(title="ID", field="id"), ExportColumn(title="计划号", field="planNo"), ExportColumn(title="出入库单号", field="batchNo"), ExportColumn(title="日期", field="date"), ExportColumn(title="往来单位", field="partnerCompany"), ExportColumn(title="业务类型", field="businessType"), ExportColumn(title="状态", field="status"), ExportColumn(title="产品", field="product"), ExportColumn(title="规格", field="specification"), ExportColumn(title="车牌号", field="vehicleNo"), ExportColumn(title="数量(吨)", field="quantity"), ExportColumn(title="净重(吨)", field="netWeight"), ExportColumn(title="毛重(吨)", field="grossWeight"), ExportColumn(title="毛重时间", field="grossWeightTime"), ExportColumn(title="皮重(吨)", field="tareWeight"), ExportColumn(title="皮重时间", field="tareWeightTime"), ExportColumn(title="扣杂(吨)", field="deduction"), ExportColumn(title="备注", field="remarks"), ExportColumn(title="创建人", field="createByName"), ] class InventoryExportBody(BaseModel): q: Optional[str] = None plan_no: Optional[str] = Field(default=None, alias="planNo") batch_no: Optional[str] = Field(default=None, alias="batchNo") partner_company: Optional[str] = Field(default=None, alias="partnerCompany") product: Optional[str] = None vehicle_no: Optional[str] = Field(default=None, alias="vehicleNo") date_from: Optional[str] = Field(default=None, alias="dateFrom") date_to: Optional[str] = Field(default=None, alias="dateTo") status: 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} @router.post("/export") def export_inventory( body: InventoryExportBody, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ): """导出出入库列表（当前筛选全量，或 ids 已选）。""" try: items = list_inventory( q=body.q, plan_no=body.plan_no, batch_no=body.batch_no, partner_company=body.partner_company, product=body.product, vehicle_no=body.vehicle_no, date_from=body.date_from, date_to=body.date_to, status_filter=body.status, 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) if not data.get("createByName"): data["createByName"] = data.get("createBy") or "" rows.append(data) cols = filter_export_columns(INVENTORY_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=InboundOutboundRead, status_code=status.HTTP_201_CREATED) def create_inventory( payload: InboundOutboundCreate, db: Session = Depends(get_db), user: Optional[DbUserContext] = Depends(get_db_current_user_optional), ) -> InboundOutboundRead: tenant_no = _resolve_tenant_no(db, user, payload.purchase_plan_id) batch_no = payload.batch_no.strip() if payload.batch_no else None if not batch_no: batch_no = _generate_batch_no(db, tenant_no) status_value = ( payload.status.strip() if payload.status and str(payload.status).strip() else ("待入库" if payload.business_type == "入库" else "待出库") ) snapshot = resolve_crm_product(db, tenant_no, payload.product_id) row = DbInboundOutbound( id=next_inbound_outbound_id(db, tenant_no), tenant_no=tenant_no, date=parse_date_str(payload.date) or datetime.now().date(), partner_company=payload.partner_company.strip(), business_type=payload.business_type, quantity=payload.quantity, plan_no=payload.plan_no.strip() if payload.plan_no else None, batch_no=batch_no, vehicle_no=payload.vehicle_no.strip(), status=status_value, remark=payload.remarks, purchase_plan_id=payload.purchase_plan_id, create_by=user.id if user else None, create_dept_id=user.dept_id if user else "", ) apply_product_snapshot_to_inventory(row, snapshot) ensure_site_access_token(row) db.add(row) db.commit() db.refresh(row) return inventory_to_read(row, db=db) @router.post( "/create-link-for-plan", response_model=InboundOutboundRead, status_code=status.HTTP_201_CREATED, ) def create_link_for_plan( payload: CreateLinkForPlanPayload, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> InboundOutboundRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) plan = ( db.query(DbPurchasePlan) .filter( DbPurchasePlan.id == payload.purchase_plan_id, DbPurchasePlan.tenant_no == tenant_no, DbPurchasePlan.del_flag == 0, ) .one_or_none() ) if not plan: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Plan not found") if not plan.product_id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="计划未关联 CRM 产品，请先编辑计划并选择产品", ) row = DbInboundOutbound( id=next_inbound_outbound_id(db, tenant_no), tenant_no=tenant_no, date=EMPTY_INVENTORY_DATE, partner_company=(plan.customer_name or "").strip() or "待填写", business_type=plan.business_type, product_id=plan.product_id, product=(plan.product or "").strip(), specification=plan.specification, quantity=0.0, plan_no=(plan.plan_no or "").strip() or None, batch_no=_generate_batch_no(db, tenant_no), vehicle_no="", status=STATUS_THIRD_PARTY_FILL, remark=payload.remarks, purchase_plan_id=plan.id, create_by=user.id, create_dept_id=user.dept_id, ) ensure_site_access_token(row) db.add(row) db.commit() db.refresh(row) return inventory_to_read(row, db=db) @router.get("/{id}/site-url", response_model=InventorySiteUrlRead) def get_inventory_site_url( id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(get_db_current_user), ) -> InventorySiteUrlRead: tenant_no = get_tenant_no(db, user.to_tenant_user_dict()) row = _get_inventory(db, tenant_no, id, user) token = ensure_site_access_token(row) db.commit() return InventorySiteUrlRead(site_url=build_site_inventory_url(token)) @router.get("/{id}", response_model=InboundOutboundRead) def get_inventory( id: str, db: Session = Depends(get_db), ) -> InboundOutboundRead: """单条查询（三方 H5 填写页无需门户登录）。""" row = _get_inventory_by_id(db, id) weighing = ( db.query(DbWeighingSlip) .filter( DbWeighingSlip.inbound_outbound_id == row.id, DbWeighingSlip.is_voided == 0, DbWeighingSlip.del_flag == 0, ) .order_by(DbWeighingSlip.id.asc()) .first() ) return inventory_to_read(row, weighing=weighing, db=db) @router.patch("/{id}", response_model=InboundOutboundRead) def update_inventory( id: str, payload: InboundOutboundUpdate, db: Session = Depends(get_db), ) -> InboundOutboundRead: """更新（三方 H5 填写页无需门户登录）。""" row = _get_inventory_by_id(db, id) payload_dict = payload.model_dump(exclude_unset=True) _apply_inventory_product_update(db, row.tenant_no, row, payload_dict) _apply_inventory_update(row, payload_dict) _maybe_promote_third_party_status(row) db.commit() db.refresh(row) return inventory_to_read(row, db=db) @router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) def delete_inventory( 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_inventory(db, tenant_no, id, user) if row.status == "已完成": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="已完成的记录不能删除", ) assert_row_deletable_by_creator(row, user_info_from_context(user)) row.del_flag = 1 row.update_by = user.id db.commit() class InventoryRelatedData(BaseModel): plan: Optional[PurchasePlanRead] weighing: Optional[WeighingSlipRead] @router.get("/{id}/related", response_model=InventoryRelatedData) def get_inventory_related( id: str, db: Session = Depends(get_db), ) -> InventoryRelatedData: row = _get_inventory_by_id(db, id) 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() ) weighing = ( db.query(DbWeighingSlip) .filter( DbWeighingSlip.inbound_outbound_id == row.id, DbWeighingSlip.is_voided == 0, DbWeighingSlip.del_flag == 0, ) .limit(1) .first() ) return InventoryRelatedData( plan=plan_to_read(plan, db=db) if plan else None, weighing=weighing_to_read(weighing, db=db) if weighing else None, )