"""输注计划出库交接 — 细胞产品状态与市场物料库存联动。""" from __future__ import annotations from datetime import datetime from decimal import Decimal from typing import Any, Dict, Iterable, List, Optional, Set from fastapi import HTTPException from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.crud import storage_occupancy as occupancy_svc from app.crud.material import _is_enabled_status from app.crud.material_stock_batch import _get_material_for_update, _usage_batch_snapshots from app.models.material import BizMaterialStockBatch, BizMaterialUsage from app.models.models import BizCellProduct, BizStorageSlotOccupancy from app.models.qc_material import BizQcMaterial, BizQcMaterialUsage # 出库交接后细胞产品目标状态（已售卖/已出库） HANDOVER_CELL_OUT_STATUS = "sold" HANDOVER_MARKET_SOURCE_TYPE = "outbound_handover" # 历史数据：曾扣减质检物料库存 HANDOVER_QC_SOURCE_TYPE = "outbound_handover" def _handover_usage_remark(schedule_id: str) -> str: return f"输注计划出库交接；scheduleId={schedule_id}" def _handover_usage_remark_prefix(schedule_id: str) -> str: return f"scheduleId={schedule_id}" def _remove_cell_occupancy_no_commit( db: Session, item_id: str, *, user_id: Optional[str] = None, ) -> None: """按细胞产品 ID 释放库位占用（不 commit）。""" obj = ( db.query(BizStorageSlotOccupancy) .filter( BizStorageSlotOccupancy.item_type == "cell_product", BizStorageSlotOccupancy.item_id == item_id, BizStorageSlotOccupancy.occupancy_status == "stored", BizStorageSlotOccupancy.del_flag == 0, ) .first() ) if not obj: return now = datetime.now() obj.occupancy_status = "removed" obj.removed_at = now obj.update_by = user_id obj.update_time = now db.add(obj) db.flush() occupancy_svc._refresh_slot_cache(db, obj.slot_id) def _restore_cell_occupancy_no_commit( db: Session, product: BizCellProduct, *, tenant_no: str, user_id: Optional[str] = None, ) -> None: """出库交接回滚时恢复库位占用（不 commit）。""" slot_id = (product.storage_slot_id or "").strip() if not slot_id: return existing_stored = ( db.query(BizStorageSlotOccupancy) .filter( BizStorageSlotOccupancy.item_type == "cell_product", BizStorageSlotOccupancy.item_id == product.id, BizStorageSlotOccupancy.occupancy_status == "stored", BizStorageSlotOccupancy.del_flag == 0, ) .first() ) if existing_stored: return now = datetime.now() latest_removed = ( db.query(BizStorageSlotOccupancy) .filter( BizStorageSlotOccupancy.item_type == "cell_product", BizStorageSlotOccupancy.item_id == product.id, BizStorageSlotOccupancy.occupancy_status == "removed", BizStorageSlotOccupancy.del_flag == 0, ) .order_by(BizStorageSlotOccupancy.removed_at.desc()) .first() ) if latest_removed: occupancy_svc.assert_slot_vacant( db, latest_removed.slot_id, tenant_no=tenant_no, exclude_product_id=product.id, ) latest_removed.occupancy_status = "stored" latest_removed.removed_at = None latest_removed.update_by = user_id latest_removed.update_time = now db.add(latest_removed) db.flush() occupancy_svc._refresh_slot_cache(db, latest_removed.slot_id) return occupancy_svc.assert_slot_vacant( db, slot_id, tenant_no=tenant_no, exclude_product_id=product.id, ) db.add( BizStorageSlotOccupancy( id=generate_uuid(), slot_id=slot_id, item_type="cell_product", item_id=product.id, item_code=(product.product_code or "").strip() or None, item_name=(product.product_name or "").strip() or None, item_source="outbound_handover_revert", stored_at=now, removed_at=None, occupancy_status="stored", tenant_no=tenant_no, order_num=0, remark="出库交接回滚恢复占用", create_by=user_id, create_dept_id=product.create_dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) ) db.flush() occupancy_svc._refresh_slot_cache(db, slot_id) def _get_cell_product_locked( db: Session, cell_product_id: str, tenant_no: str ) -> BizCellProduct: product = ( db.query(BizCellProduct) .with_for_update() .filter( BizCellProduct.id == cell_product_id, BizCellProduct.tenant_no == tenant_no, BizCellProduct.del_flag == 0, ) .first() ) if not product: raise HTTPException( status_code=400, detail=f"细胞产品不存在或已删除（ID: {cell_product_id}）", ) return product def _revert_legacy_qc_handover_usages( db: Session, *, schedule_id: str, tenant_no: str, user_id: Optional[str], remark_prefix: str, now: datetime, ) -> None: """回滚历史版本曾写入的质检物料出库交接记录。""" usage_rows = ( db.query(BizQcMaterialUsage) .filter( BizQcMaterialUsage.tenant_no == tenant_no, BizQcMaterialUsage.del_flag == 0, BizQcMaterialUsage.source_type == HANDOVER_QC_SOURCE_TYPE, BizQcMaterialUsage.change_type == "use", BizQcMaterialUsage.remark.like(f"%{remark_prefix}%"), ) .all() ) for usage in usage_rows: restore_qty = abs(usage.change_qty) if usage.change_qty is not None else Decimal("0") if restore_qty > 0: material = ( db.query(BizQcMaterial) .with_for_update() .filter( BizQcMaterial.id == usage.material_id, BizQcMaterial.tenant_no == tenant_no, BizQcMaterial.del_flag == 0, ) .first() ) if material: material.stock_qty = (material.stock_qty or Decimal("0")) + restore_qty material.update_by = user_id material.update_time = now db.add(material) usage.del_flag = 1 usage.update_by = user_id usage.update_time = now db.add(usage) def _revert_market_handover_usages( db: Session, *, schedule_id: str, tenant_no: str, user_id: Optional[str], remark_prefix: str, now: datetime, ) -> None: """回滚市场物料出库交接记录（含批次剩余量）。""" usage_rows = ( db.query(BizMaterialUsage) .filter( BizMaterialUsage.tenant_no == tenant_no, BizMaterialUsage.del_flag == 0, BizMaterialUsage.source_type == HANDOVER_MARKET_SOURCE_TYPE, BizMaterialUsage.change_type == "use", BizMaterialUsage.remark.like(f"%{remark_prefix}%"), ) .all() ) for usage in usage_rows: restore_qty = abs(usage.change_qty) if usage.change_qty is not None else Decimal("0") if restore_qty > 0: material = _get_material_for_update(db, usage.material_id, tenant_no) material.stock_qty = (material.stock_qty or Decimal("0")) + restore_qty material.update_by = user_id material.update_time = now db.add(material) batch_id = (usage.batch_id or "").strip() if batch_id: batch = ( db.query(BizMaterialStockBatch) .with_for_update() .filter( BizMaterialStockBatch.id == batch_id, BizMaterialStockBatch.tenant_no == tenant_no, BizMaterialStockBatch.del_flag == 0, ) .first() ) if batch: batch.remaining_qty = (batch.remaining_qty or Decimal("0")) + restore_qty batch.update_by = user_id batch.update_time = now db.add(batch) usage.del_flag = 1 usage.update_by = user_id usage.update_time = now db.add(usage) def revert_handover_inventory( db: Session, schedule_id: str, *, cell_product_ids: Iterable[str], tenant_no: str, user_id: Optional[str] = None, ) -> None: """回滚本输注计划上次出库交接产生的库存/状态变更（不 commit）。""" sid = (schedule_id or "").strip() if not sid: return remark_prefix = _handover_usage_remark_prefix(sid) now = datetime.now() _revert_legacy_qc_handover_usages( db, schedule_id=sid, tenant_no=tenant_no, user_id=user_id, remark_prefix=remark_prefix, now=now, ) _revert_market_handover_usages( db, schedule_id=sid, tenant_no=tenant_no, user_id=user_id, remark_prefix=remark_prefix, now=now, ) seen: Set[str] = set() for raw_id in cell_product_ids: cp_id = (raw_id or "").strip() if not cp_id or cp_id in seen: continue seen.add(cp_id) product = _get_cell_product_locked(db, cp_id, tenant_no) current = (product.product_status or "frozen").strip() if current != HANDOVER_CELL_OUT_STATUS: continue product.product_status = "frozen" product.update_by = user_id product.update_time = now db.add(product) _restore_cell_occupancy_no_commit( db, product, tenant_no=tenant_no, user_id=user_id ) db.flush() def _deduct_market_material_for_handover( db: Session, material_id: str, qty: Decimal, *, schedule_id: str, tenant_no: str, user_id: Optional[str], create_dept_id: str, receiver_person: str, line_remark: str, now: datetime, ) -> None: """按 FIFO 从市场物料入库批次扣减，并写入变动记录。""" material = _get_material_for_update(db, material_id, tenant_no) if not _is_enabled_status(material.status): raise HTTPException( status_code=400, detail=f"市场物料「{material.material_name}」已停用，无法出库", ) stock_before = material.stock_qty if material.stock_qty is not None else Decimal("0") if stock_before < qty: raise HTTPException( status_code=400, detail=( f"物料「{material.material_name}」库存不足，" f"当前库存 {stock_before}，需出库 {qty}" ), ) batches = ( db.query(BizMaterialStockBatch) .with_for_update() .filter( BizMaterialStockBatch.material_id == material_id, BizMaterialStockBatch.tenant_no == tenant_no, BizMaterialStockBatch.del_flag == 0, BizMaterialStockBatch.remaining_qty > 0, ) .order_by( BizMaterialStockBatch.stock_in_date.asc(), BizMaterialStockBatch.create_time.asc(), ) .all() ) usage_remark = _handover_usage_remark(schedule_id) receiver = (receiver_person or "").strip() or None remaining = qty running_stock = stock_before for batch in batches: if remaining <= 0: break rem = batch.remaining_qty if batch.remaining_qty is not None else Decimal("0") if rem <= 0: continue take = min(rem, remaining) batch.remaining_qty = rem - take batch.update_by = user_id batch.update_time = now db.add(batch) stock_after = running_stock - take snaps = _usage_batch_snapshots(batch, out_qty=take) db.add( BizMaterialUsage( id=generate_uuid(), material_id=material.id, material_code=material.material_code, material_name=material.material_name, change_type="use", change_qty=-take, usage_unit=(material.unit or "").strip() or None, stock_before=running_stock, stock_after=stock_after, source_type=HANDOVER_MARKET_SOURCE_TYPE, requisition_status="closed", tenant_no=tenant_no, receiver=receiver, purpose="输注计划出库交接", remark=usage_remark if not line_remark else f"{usage_remark}；{line_remark}", create_by=user_id, create_dept_id=create_dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, **snaps, ) ) running_stock = stock_after remaining -= take if remaining > 0: raise HTTPException( status_code=400, detail=( f"物料「{material.material_name}」批次库存不足，" f"汇总库存 {stock_before}，需出库 {qty}，批次可分配 {qty - remaining}" ), ) material.stock_qty = stock_before - qty material.update_by = user_id material.update_time = now db.add(material) def apply_handover_inventory( db: Session, schedule_id: str, *, cells: List[Dict[str, Any]], qc_materials: List[Dict[str, Any]], tenant_no: str, user_id: Optional[str] = None, create_dept_id: str = "", receiver_person: str = "", ) -> None: """应用出库交接：细胞产品出库 + 市场物料扣减（不 commit）。""" sid = (schedule_id or "").strip() now = datetime.now() for item in cells or []: cp_id = (item.get("cellProductId") or item.get("cell_product_id") or "").strip() if not cp_id: continue product = _get_cell_product_locked(db, cp_id, tenant_no) current = (product.product_status or "frozen").strip() if current != "frozen": label = product.product_code or product.product_name or cp_id raise HTTPException( status_code=400, detail=( f"细胞产品「{label}」当前状态不可出库交接" f"（仅「已冻存」可出库，当前为 {current or '未知'}）" ), ) product.product_status = HANDOVER_CELL_OUT_STATUS product.update_by = user_id product.update_time = now db.add(product) _remove_cell_occupancy_no_commit(db, cp_id, user_id=user_id) for item in qc_materials or []: mid = (item.get("materialId") or item.get("material_id") or "").strip() qty_raw = item.get("qty") if not mid: raise HTTPException(status_code=400, detail="市场物料ID不能为空") try: qty_dec = Decimal(str(qty_raw)) except Exception: raise HTTPException(status_code=400, detail="物料用量格式无效") if qty_dec <= 0: raise HTTPException(status_code=400, detail="物料用量必须大于0") line_remark = (item.get("remark") or "").strip() _deduct_market_material_for_handover( db, mid, qty_dec, schedule_id=sid, tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id, receiver_person=receiver_person, line_remark=line_remark, now=now, ) db.flush()