"""质检物料入库批次 CRUD""" from __future__ import annotations from datetime import date, datetime from decimal import Decimal from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.crud.qc_material import _dec, _fmt_dt, _is_enabled_status, _norm_status from app.crud.qc_material_batch_stock_status import normalize_batch_stock_status from app.models.qc_material import BizQcMaterial, BizQcMaterialStockBatch, BizQcMaterialUsage def _fmt_date(d: Optional[date]) -> str: return d.strftime("%Y-%m-%d") if d else "" def _parse_date(v: Any, *, label: str = "入库日期") -> date: if isinstance(v, date) and not isinstance(v, datetime): return v s = str(v or "").strip() if not s: raise ValueError(f"{label}不能为空") if len(s) >= 10: s = s[:10] try: return datetime.strptime(s, "%Y-%m-%d").date() except ValueError as e: raise ValueError("入库日期格式无效，应为 YYYY-MM-DD") from e def _parse_expire_date(v: Any) -> date: return _parse_date(v, label="过期日期") def _require_str(v: Any, label: str) -> str: s = str(v or "").strip() if not s: raise ValueError(f"{label}不能为空") return s def _require_decimal(v: Any, label: str) -> Decimal: if v is None or str(v).strip() == "": raise ValueError(f"{label}不能为空") try: d = Decimal(str(v)) except Exception as e: raise ValueError(f"{label}格式无效") from e if d < 0: raise ValueError(f"{label}不能为负数") return d def stock_in_no_exists( db: Session, stock_in_no: str, tenant_no: str, exclude_id: Optional[str] = None ) -> bool: s = (stock_in_no or "").strip() if not s: return False q = db.query(BizQcMaterialStockBatch).filter( BizQcMaterialStockBatch.stock_in_no == s, BizQcMaterialStockBatch.tenant_no == tenant_no, BizQcMaterialStockBatch.del_flag == 0, ) if exclude_id: q = q.filter(BizQcMaterialStockBatch.id != exclude_id) return q.first() is not None def _batch_row_to_dict(b: BizQcMaterialStockBatch, material: Optional[BizQcMaterial] = None) -> Dict[str, Any]: mat = material warning_qty = _dec(mat.warning_qty) if mat else 0 stock_qty = _dec(mat.stock_qty) if mat else _dec(b.remaining_qty) status = _norm_status(mat.status) if mat else "1" return { "id": b.id, "batchId": b.id, "materialId": b.material_id, "materialCode": b.material_code or "", "materialName": b.material_name or "", "specModel": b.spec_model or "", "unit": b.unit or "", "stockInNo": b.stock_in_no or "", "stockInDate": _fmt_date(b.stock_in_date), "expireDate": _fmt_date(b.expire_date), "catalogNo": b.catalog_no or "", "batchNo": b.batch_no or "", "manufacturer": b.manufacturer or "", "unitPrice": float(b.unit_price) if b.unit_price is not None else 0, "amount": float(b.amount) if b.amount is not None else 0, "inQty": _dec(b.in_qty), "remainingQty": _dec(b.remaining_qty), "qcAttachmentUrls": b.qc_attachment_urls or "", "stockQty": stock_qty, "warningQty": warning_qty, "status": status, "remark": b.remark or "", "createTime": _fmt_dt(b.create_time), "updateTime": _fmt_dt(b.update_time), } def _usage_batch_snapshots( batch: BizQcMaterialStockBatch, *, out_qty: Optional[Decimal] = None, ) -> Dict[str, Any]: unit_price = batch.unit_price if out_qty is not None: amount = (Decimal(str(unit_price or 0)) * abs(out_qty)).quantize(Decimal("0.01")) else: amount = batch.amount return { "batch_id": batch.id, "stock_in_no": batch.stock_in_no, "catalog_no": batch.catalog_no, "batch_no": batch.batch_no, "manufacturer": batch.manufacturer, "unit_price": unit_price, "amount": amount, } def _get_material_for_update( db: Session, material_id: str, tenant_no: str ) -> BizQcMaterial: obj = ( db.query(BizQcMaterial) .filter( BizQcMaterial.id == material_id, BizQcMaterial.tenant_no == tenant_no, BizQcMaterial.del_flag == 0, ) .with_for_update() .first() ) if not obj: raise ValueError("物料不存在") return obj def list_batches( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, material_id: Optional[str] = None, status: Optional[str] = None, stock_status: Optional[str] = None, tenant_no: Optional[str] = None, user_info=None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = ( db.query(BizQcMaterialStockBatch, BizQcMaterial) .join( BizQcMaterial, BizQcMaterial.id == BizQcMaterialStockBatch.material_id, ) .filter( BizQcMaterialStockBatch.del_flag == 0, BizQcMaterial.del_flag == 0, ) ) if tenant_no: q0 = q0.filter(BizQcMaterialStockBatch.tenant_no == tenant_no) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter( or_( BizQcMaterialStockBatch.material_code.like(like), BizQcMaterialStockBatch.material_name.like(like), BizQcMaterialStockBatch.stock_in_no.like(like), BizQcMaterialStockBatch.batch_no.like(like), BizQcMaterialStockBatch.catalog_no.like(like), ) ) if material_id: q0 = q0.filter(BizQcMaterialStockBatch.material_id == material_id) st = (status or "").strip() if st in ("0", "1"): q0 = q0.filter(BizQcMaterial.status == st) mode = normalize_batch_stock_status(stock_status) if mode == "available": q0 = q0.filter(BizQcMaterialStockBatch.remaining_qty > 0) elif mode == "exhausted": q0 = q0.filter(BizQcMaterialStockBatch.remaining_qty == 0) if user_info: from app.core.data_scope import apply_data_scope q0 = apply_data_scope(q0, BizQcMaterialStockBatch, user_info, db, tenant_no) return q0 total = _filtered().order_by(None).with_entities(func.count(BizQcMaterialStockBatch.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by( BizQcMaterialStockBatch.stock_in_date.desc(), BizQcMaterialStockBatch.create_time.desc(), ) .offset((page - 1) * page_size) .limit(page_size) .all() ) return [_batch_row_to_dict(b, m) for b, m in rows], int(total) def batch_options( db: Session, tenant_no: str, *, keyword: Optional[str] = None, material_id: Optional[str] = None, only_available: bool = True, ) -> List[Dict[str, Any]]: q = ( db.query(BizQcMaterialStockBatch, BizQcMaterial) .join(BizQcMaterial, BizQcMaterial.id == BizQcMaterialStockBatch.material_id) .filter( BizQcMaterialStockBatch.tenant_no == tenant_no, BizQcMaterialStockBatch.del_flag == 0, BizQcMaterial.del_flag == 0, BizQcMaterial.status == "1", ) ) if only_available: q = q.filter(BizQcMaterialStockBatch.remaining_qty > 0) if material_id: q = q.filter(BizQcMaterialStockBatch.material_id == material_id) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q = q.filter( or_( BizQcMaterialStockBatch.stock_in_no.like(like), BizQcMaterialStockBatch.material_name.like(like), BizQcMaterialStockBatch.material_code.like(like), BizQcMaterialStockBatch.batch_no.like(like), BizQcMaterialStockBatch.catalog_no.like(like), ) ) rows = q.order_by(BizQcMaterialStockBatch.stock_in_date.asc()).all() out: List[Dict[str, Any]] = [] for b, m in rows: unit = b.unit or "" rem = _dec(b.remaining_qty) mfr = (b.manufacturer or "").strip() cat = (b.catalog_no or "").strip() label = ( f"{b.stock_in_no} / {b.material_name}" f" / 厂家:{mfr or '—'}" f" / 批号:{b.batch_no or '—'}" f" / 货号:{cat or '—'}" f" / 剩:{rem}{unit}" ) unit_price = float(b.unit_price) if b.unit_price is not None else 0 out.append( { "id": b.id, "batchId": b.id, "stockInNo": b.stock_in_no or "", "stockInDate": _fmt_date(b.stock_in_date), "expireDate": _fmt_date(b.expire_date), "materialId": b.material_id, "materialCode": b.material_code or "", "materialName": b.material_name or "", "specModel": b.spec_model or "", "unit": unit, "batchNo": b.batch_no or "", "catalogNo": b.catalog_no or "", "manufacturer": b.manufacturer or "", "unitPrice": unit_price, "amount": float(b.amount) if b.amount is not None else 0, "remainingQty": rem, "label": label, } ) return out def _create_batch_and_usage( db: Session, *, material: BizQcMaterial, stock_in_no: str, stock_in_date: date, expire_date: date, catalog_no: str, batch_no: str, manufacturer: str, unit_price: Decimal, amount: Decimal, qty: Decimal, remark: Optional[str], tenant_no: str, user_id: Optional[str], operate_time: datetime, commit: bool = False, ) -> BizQcMaterialStockBatch: if qty <= 0: raise ValueError("入库数量必须大于0") sno = _require_str(stock_in_no, "入库编号") if stock_in_no_exists(db, sno, tenant_no): raise ValueError(f"入库编号「{sno}」已存在") if not _is_enabled_status(material.status): raise ValueError(f"物料「{material.material_name}」已停用，无法入库") old_stock = material.stock_qty or Decimal("0") new_stock = old_stock + qty material.stock_qty = new_stock material.update_by = user_id material.update_time = datetime.now() db.add(material) batch = BizQcMaterialStockBatch( id=generate_uuid(), material_id=material.id, material_code=material.material_code, material_name=material.material_name, spec_model=material.spec_model, unit=material.unit, stock_in_no=sno, stock_in_date=stock_in_date, expire_date=expire_date, catalog_no=catalog_no, batch_no=batch_no, manufacturer=manufacturer, unit_price=unit_price, amount=amount, in_qty=qty, remaining_qty=qty, tenant_no=tenant_no, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id=material.create_dept_id or "", create_time=operate_time, update_by=user_id, update_time=operate_time, del_flag=0, ) db.add(batch) snaps = _usage_batch_snapshots(batch) usage = BizQcMaterialUsage( id=generate_uuid(), material_id=material.id, material_code=material.material_code, material_name=material.material_name, change_type="in", change_qty=qty, stock_before=old_stock, stock_after=new_stock, source_type="manual_in", tenant_no=tenant_no, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id=material.create_dept_id or "", create_time=operate_time, update_by=user_id, update_time=operate_time, del_flag=0, **snaps, ) db.add(usage) if commit: db.commit() db.refresh(batch) return batch def stock_in_one( db: Session, material_id: str, *, stock_in_no: str, stock_in_date: Any, expire_date: Any, catalog_no: str, batch_no: str, manufacturer: str, unit_price: Any, amount: Any, qty: Decimal, remark: Optional[str] = None, tenant_no: str, user_id: Optional[str] = None, operate_time: Optional[datetime] = None, ) -> BizQcMaterialStockBatch: op_time = operate_time or datetime.now() material = _get_material_for_update(db, material_id, tenant_no) batch = _create_batch_and_usage( db, material=material, stock_in_no=stock_in_no, stock_in_date=_parse_date(stock_in_date), expire_date=_parse_expire_date(expire_date), catalog_no=_require_str(catalog_no, "货号"), batch_no=_require_str(batch_no, "批号"), manufacturer=_require_str(manufacturer, "厂家"), unit_price=_require_decimal(unit_price, "单价"), amount=_require_decimal(amount, "金额"), qty=qty, remark=remark, tenant_no=tenant_no, user_id=user_id, operate_time=op_time, commit=True, ) return batch def batch_stock_in( db: Session, *, items: List[Dict[str, Any]], remark: Optional[str] = None, tenant_no: str, user_id: Optional[str] = None, operate_time: Optional[datetime] = None, ) -> int: if not items: raise ValueError("请至少添加一条入库明细") op_time = operate_time or datetime.now() shared_remark = (remark or "").strip() or None seen_nos: set = set() try: for it in items: material_id = (it.get("material_id") or it.get("materialId") or "").strip() if not material_id: raise ValueError("请选择物资") sno = _require_str(it.get("stock_in_no") or it.get("stockInNo"), "入库编号") if sno in seen_nos: raise ValueError(f"入库编号「{sno}」在本单中重复") seen_nos.add(sno) qty = it.get("qty") if qty is None or Decimal(str(qty)) <= 0: raise ValueError("入库数量必须大于0") qty_dec = Decimal(str(qty)) line_remark = (it.get("remark") or "").strip() or shared_remark material = _get_material_for_update(db, material_id, tenant_no) _create_batch_and_usage( db, material=material, stock_in_no=sno, stock_in_date=_parse_date(it.get("stock_in_date") or it.get("stockInDate")), expire_date=_parse_expire_date(it.get("expire_date") or it.get("expireDate")), catalog_no=_require_str(it.get("catalog_no") or it.get("catalogNo"), "货号"), batch_no=_require_str(it.get("batch_no") or it.get("batchNo"), "批号"), manufacturer=_require_str(it.get("manufacturer"), "厂家"), unit_price=_require_decimal(it.get("unit_price") or it.get("unitPrice"), "单价"), amount=_require_decimal(it.get("amount"), "金额"), qty=qty_dec, remark=line_remark, tenant_no=tenant_no, user_id=user_id, operate_time=op_time, commit=False, ) db.commit() return len(items) except Exception: db.rollback() raise def _stock_out_batch( db: Session, batch_id: str, *, qty: Decimal, receiver: Optional[str], purpose: Optional[str], remark: Optional[str], usage_unit: Optional[str] = None, tenant_no: str, user_id: Optional[str], operate_time: datetime, ) -> None: if qty <= 0: raise ValueError("出库数量必须大于0") if not (receiver or "").strip(): raise ValueError("请选择领用人") batch = ( db.query(BizQcMaterialStockBatch) .filter( BizQcMaterialStockBatch.id == batch_id, BizQcMaterialStockBatch.tenant_no == tenant_no, BizQcMaterialStockBatch.del_flag == 0, ) .with_for_update() .first() ) if not batch: raise ValueError("入库批次不存在") material = _get_material_for_update(db, batch.material_id, tenant_no) if not _is_enabled_status(material.status): raise ValueError(f"物料「{material.material_name}」已停用，无法出库") rem = batch.remaining_qty or Decimal("0") if rem < qty: raise ValueError( f"批次「{batch.stock_in_no}」库存不足，当前剩余 {rem}，需出库 {qty}" ) batch.remaining_qty = rem - qty batch.update_by = user_id batch.update_time = datetime.now() db.add(batch) stock_before = material.stock_qty or Decimal("0") new_stock = stock_before - qty material.stock_qty = new_stock material.update_by = user_id material.update_time = datetime.now() db.add(material) snaps = _usage_batch_snapshots(batch, out_qty=qty) unit_label = (usage_unit or "").strip() or (batch.unit or "").strip() or None usage = BizQcMaterialUsage( id=generate_uuid(), material_id=material.id, material_code=material.material_code, material_name=material.material_name, change_type="use", change_qty=-qty, usage_unit=unit_label, stock_before=stock_before, stock_after=new_stock, source_type="manual_out", tenant_no=tenant_no, receiver=(receiver or "").strip() or None, purpose=(purpose or "").strip() or None, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id=material.create_dept_id or "", create_time=operate_time, update_by=user_id, update_time=operate_time, del_flag=0, **snaps, ) db.add(usage) def batch_stock_out( db: Session, *, items: List[Dict[str, Any]], receiver: Optional[str] = None, purpose: Optional[str] = None, remark: Optional[str] = None, tenant_no: str, user_id: Optional[str] = None, operate_time: Optional[datetime] = None, ) -> int: if not items: raise ValueError("请至少添加一条出库明细") op_time = operate_time or datetime.now() shared_remark = (remark or "").strip() or None seen_batches: set = set() try: for it in items: batch_id = (it.get("batch_id") or it.get("batchId") or "").strip() if not batch_id: raise ValueError("请选择入库批次") if batch_id in seen_batches: raise ValueError("同一入库批次不可重复添加") seen_batches.add(batch_id) qty = it.get("qty") if qty is None or Decimal(str(qty)) <= 0: raise ValueError("出库数量必须大于0") qty_dec = Decimal(str(qty)) line_remark = (it.get("remark") or "").strip() or shared_remark usage_unit = (it.get("usageUnit") or it.get("usage_unit") or "").strip() or None _stock_out_batch( db, batch_id, qty=qty_dec, receiver=receiver, purpose=purpose, remark=line_remark, usage_unit=usage_unit, tenant_no=tenant_no, user_id=user_id, operate_time=op_time, ) db.commit() return len(items) except Exception: db.rollback() raise DISCARD_REASONS = frozenset({"过期", "质量问题", "污染/破损", "其他"}) def _validate_discard_reason(discard_reason: Optional[str], remark: Optional[str]) -> str: reason = (discard_reason or "").strip() if reason not in DISCARD_REASONS: raise ValueError("请选择有效的废弃原因") if reason == "其他" and not (remark or "").strip(): raise ValueError("选择「其他」时请填写备注说明") return reason def _discard_batch( db: Session, batch_id: str, *, qty: Decimal, discard_reason: str, remark: Optional[str], usage_unit: Optional[str] = None, tenant_no: str, user_id: Optional[str], operate_time: datetime, ) -> None: if qty <= 0: raise ValueError("废弃数量必须大于0") batch = ( db.query(BizQcMaterialStockBatch) .filter( BizQcMaterialStockBatch.id == batch_id, BizQcMaterialStockBatch.tenant_no == tenant_no, BizQcMaterialStockBatch.del_flag == 0, ) .with_for_update() .first() ) if not batch: raise ValueError("入库批次不存在") material = _get_material_for_update(db, batch.material_id, tenant_no) if not _is_enabled_status(material.status): raise ValueError(f"物料「{material.material_name}」已停用，无法废弃") rem = batch.remaining_qty or Decimal("0") if rem < qty: raise ValueError( f"批次「{batch.stock_in_no}」库存不足，当前剩余 {rem}，需废弃 {qty}" ) batch.remaining_qty = rem - qty batch.update_by = user_id batch.update_time = datetime.now() db.add(batch) stock_before = material.stock_qty or Decimal("0") new_stock = stock_before - qty material.stock_qty = new_stock material.update_by = user_id material.update_time = datetime.now() db.add(material) snaps = _usage_batch_snapshots(batch, out_qty=qty) unit_label = (usage_unit or "").strip() or (batch.unit or "").strip() or None usage = BizQcMaterialUsage( id=generate_uuid(), material_id=material.id, material_code=material.material_code, material_name=material.material_name, change_type="discard", change_qty=-qty, usage_unit=unit_label, stock_before=stock_before, stock_after=new_stock, source_type="manual_discard", tenant_no=tenant_no, purpose=discard_reason, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id=material.create_dept_id or "", create_time=operate_time, update_by=user_id, update_time=operate_time, del_flag=0, **snaps, ) db.add(usage) def batch_discard( db: Session, *, items: List[Dict[str, Any]], discard_reason: Optional[str] = None, remark: Optional[str] = None, tenant_no: str, user_id: Optional[str] = None, operate_time: Optional[datetime] = None, ) -> int: if not items: raise ValueError("请至少添加一条废弃明细") op_time = operate_time or datetime.now() shared_remark = (remark or "").strip() or None reason = _validate_discard_reason(discard_reason, shared_remark) seen_batches: set = set() try: for it in items: batch_id = (it.get("batch_id") or it.get("batchId") or "").strip() if not batch_id: raise ValueError("请选择入库批次") if batch_id in seen_batches: raise ValueError("同一入库批次不可重复添加") seen_batches.add(batch_id) qty = it.get("qty") if qty is None or Decimal(str(qty)) <= 0: raise ValueError("废弃数量必须大于0") qty_dec = Decimal(str(qty)) line_remark = (it.get("remark") or "").strip() or shared_remark if reason == "其他" and not line_remark: raise ValueError("选择「其他」时请填写备注说明") usage_unit = (it.get("usageUnit") or it.get("usage_unit") or "").strip() or None _discard_batch( db, batch_id, qty=qty_dec, discard_reason=reason, remark=line_remark, usage_unit=usage_unit, tenant_no=tenant_no, user_id=user_id, operate_time=op_time, ) db.commit() return len(items) except Exception: db.rollback() raise def get_batch(db: Session, batch_id: str, tenant_no: str) -> Optional[BizQcMaterialStockBatch]: return ( db.query(BizQcMaterialStockBatch) .filter( BizQcMaterialStockBatch.id == batch_id, BizQcMaterialStockBatch.tenant_no == tenant_no, BizQcMaterialStockBatch.del_flag == 0, ) .first() ) def _sum_batch_out_total( db: Session, batch_id: str, tenant_no: str ) -> Tuple[Decimal, int]: row = ( db.query( func.coalesce(func.sum(func.abs(BizQcMaterialUsage.change_qty)), 0), func.count(BizQcMaterialUsage.id), ) .filter( BizQcMaterialUsage.batch_id == batch_id, BizQcMaterialUsage.tenant_no == tenant_no, BizQcMaterialUsage.del_flag == 0, BizQcMaterialUsage.change_type.in_(["use", "discard"]), ) .one() ) total = Decimal(str(row[0] or 0)) count = int(row[1] or 0) return total, count def get_batch_detail( db: Session, batch_id: str, tenant_no: str ) -> Dict[str, Any]: batch = get_batch(db, batch_id, tenant_no) if not batch: raise ValueError("入库批次不存在") material = ( db.query(BizQcMaterial) .filter( BizQcMaterial.id == batch.material_id, BizQcMaterial.tenant_no == tenant_no, BizQcMaterial.del_flag == 0, ) .first() ) out_total, out_count = _sum_batch_out_total(db, batch_id, tenant_no) payload = _batch_row_to_dict(batch, material) payload["outTotal"] = _dec(out_total) payload["outCount"] = out_count return payload def _append_batch_qty_adjust( db: Session, *, batch: BizQcMaterialStockBatch, material: BizQcMaterial, change_qty: Decimal, stock_before: Decimal, stock_after: Decimal, remark: str, tenant_no: str, user_id: Optional[str], ) -> None: if change_qty == 0: return now = datetime.now() snaps = _usage_batch_snapshots(batch) db.add( BizQcMaterialUsage( id=generate_uuid(), material_id=material.id, material_code=material.material_code, material_name=material.material_name, change_type="adjust", change_qty=change_qty, stock_before=stock_before, stock_after=stock_after, source_type="batch_qty_edit", tenant_no=tenant_no, remark=remark, create_by=user_id, create_dept_id=material.create_dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, **snaps, ) ) def update_batch( db: Session, batch_id: str, *, tenant_no: str, user_id: Optional[str] = None, stock_in_no: Optional[str] = None, stock_in_date: Optional[Any] = None, expire_date: Optional[Any] = None, catalog_no: Optional[str] = None, batch_no: Optional[str] = None, manufacturer: Optional[str] = None, unit_price: Optional[Any] = None, amount: Optional[Any] = None, in_qty: Optional[Any] = None, remaining_qty: Optional[Any] = None, remark: Optional[str] = None, qc_attachment_urls: Optional[str] = None, material_code: Optional[str] = None, material_name: Optional[str] = None, spec_model: Optional[str] = None, unit: Optional[str] = None, warning_qty: Optional[Any] = None, status: Optional[str] = None, ) -> BizQcMaterialStockBatch: from app.crud import qc_material as mat_svc batch = ( db.query(BizQcMaterialStockBatch) .filter( BizQcMaterialStockBatch.id == batch_id, BizQcMaterialStockBatch.tenant_no == tenant_no, BizQcMaterialStockBatch.del_flag == 0, ) .with_for_update() .first() ) if not batch: raise ValueError("入库批次不存在") material = _get_material_for_update(db, batch.material_id, tenant_no) mat_data: Dict[str, Any] = {} if material_code is not None: mat_data["material_code"] = material_code if material_name is not None: mat_data["material_name"] = material_name if spec_model is not None: mat_data["spec_model"] = spec_model if unit is not None: mat_data["unit"] = unit if warning_qty is not None: mat_data["warning_qty"] = Decimal(str(warning_qty)) if status is not None: mat_data["status"] = _norm_status(status) if mat_data: material = mat_svc.update_material( db, batch.material_id, tenant_no=tenant_no, user_id=user_id, **mat_data, ) if stock_in_no is not None: sno = _require_str(stock_in_no, "入库编号") if stock_in_no_exists(db, sno, tenant_no, exclude_id=batch_id): raise ValueError(f"入库编号「{sno}」已存在") batch.stock_in_no = sno if stock_in_date is not None: batch.stock_in_date = _parse_date(stock_in_date) if expire_date is not None: batch.expire_date = _parse_expire_date(expire_date) if catalog_no is not None: batch.catalog_no = _require_str(catalog_no, "货号") if batch_no is not None: batch.batch_no = _require_str(batch_no, "批号") if manufacturer is not None: batch.manufacturer = _require_str(manufacturer, "厂家") if unit_price is not None: batch.unit_price = _require_decimal(unit_price, "单价") if amount is not None: batch.amount = _require_decimal(amount, "金额") if remark is not None: batch.remark = (remark or "").strip() or None if qc_attachment_urls is not None: batch.qc_attachment_urls = (qc_attachment_urls or "").strip() or None old_in_qty = batch.in_qty or Decimal("0") old_remaining = batch.remaining_qty or Decimal("0") qty_updated = in_qty is not None or remaining_qty is not None qty_values_changed = False if qty_updated: new_in_qty = _require_decimal(in_qty, "入库数量") if in_qty is not None else old_in_qty new_remaining = ( _require_decimal(remaining_qty, "剩余数量") if remaining_qty is not None else old_remaining ) qty_values_changed = new_in_qty != old_in_qty or new_remaining != old_remaining if qty_values_changed: if new_in_qty <= 0: raise ValueError("入库数量必须大于0") if new_remaining < 0: raise ValueError("剩余数量不能为负数") if new_remaining > new_in_qty: raise ValueError("剩余数量不能大于入库数量") out_total, _ = _sum_batch_out_total(db, batch_id, tenant_no) if new_in_qty < out_total: raise ValueError(f"入库数量不能小于已出库量 {_dec(out_total)}") delta_remaining = new_remaining - old_remaining if in_qty is not None and amount is None: unit_p = batch.unit_price or Decimal("0") batch.amount = (unit_p * new_in_qty).quantize(Decimal("0.01")) stock_before = material.stock_qty or Decimal("0") material.stock_qty = stock_before + delta_remaining material.update_by = user_id material.update_time = datetime.now() db.add(material) batch.in_qty = new_in_qty batch.remaining_qty = new_remaining in_qty_delta = new_in_qty - old_in_qty remaining_extra = new_remaining - (old_remaining + in_qty_delta) running_stock = stock_before user_remark = (remark or "").strip() if remark is not None else "" if in_qty_delta != 0: next_stock = running_stock + in_qty_delta adjust_remark = ( f"编辑修正入库数量（{_dec(old_in_qty)} → {_dec(new_in_qty)}）" ) if user_remark: adjust_remark = f"{adjust_remark}；{user_remark}" _append_batch_qty_adjust( db, batch=batch, material=material, change_qty=in_qty_delta, stock_before=running_stock, stock_after=next_stock, remark=adjust_remark, tenant_no=tenant_no, user_id=user_id, ) running_stock = next_stock if remaining_extra != 0: expected_remaining = old_remaining + in_qty_delta next_stock = running_stock + remaining_extra adjust_remark = ( f"编辑修正剩余数量（{_dec(expected_remaining)} → {_dec(new_remaining)}）" ) if user_remark and in_qty_delta == 0: adjust_remark = f"{adjust_remark}；{user_remark}" _append_batch_qty_adjust( db, batch=batch, material=material, change_qty=remaining_extra, stock_before=running_stock, stock_after=next_stock, remark=adjust_remark, tenant_no=tenant_no, user_id=user_id, ) batch.material_code = material.material_code batch.material_name = material.material_name batch.spec_model = material.spec_model batch.unit = material.unit batch.update_by = user_id batch.update_time = datetime.now() db.add(batch) in_usage = ( db.query(BizQcMaterialUsage) .filter( BizQcMaterialUsage.batch_id == batch_id, BizQcMaterialUsage.tenant_no == tenant_no, BizQcMaterialUsage.del_flag == 0, BizQcMaterialUsage.change_type == "in", BizQcMaterialUsage.source_type == "manual_in", ) .first() ) if in_usage: in_usage.material_code = material.material_code in_usage.material_name = material.material_name in_usage.stock_in_no = batch.stock_in_no in_usage.catalog_no = batch.catalog_no in_usage.batch_no = batch.batch_no in_usage.manufacturer = batch.manufacturer in_usage.unit_price = batch.unit_price if not qty_values_changed: in_usage.amount = batch.amount in_usage.remark = batch.remark in_usage.update_by = user_id in_usage.update_time = datetime.now() db.add(in_usage) db.commit() db.refresh(batch) return batch def delete_batch( db: Session, batch_id: str, *, tenant_no: str, user_id: Optional[str] = None, ) -> None: batch = get_batch(db, batch_id, tenant_no) if not batch: raise ValueError("入库批次不存在") used = ( db.query(BizQcMaterialUsage) .filter( BizQcMaterialUsage.batch_id == batch_id, BizQcMaterialUsage.tenant_no == tenant_no, BizQcMaterialUsage.del_flag == 0, BizQcMaterialUsage.change_type == "use", ) .count() ) if used > 0: raise ValueError("该批次已有领用或出库记录，无法删除") material = _get_material_for_update(db, batch.material_id, tenant_no) rem = batch.remaining_qty or Decimal("0") material.stock_qty = (material.stock_qty or Decimal("0")) - rem material.update_by = user_id material.update_time = datetime.now() db.add(material) now = datetime.now() batch.del_flag = 1 batch.update_by = user_id batch.update_time = now db.add(batch) usages = ( db.query(BizQcMaterialUsage) .filter( BizQcMaterialUsage.batch_id == batch_id, BizQcMaterialUsage.tenant_no == tenant_no, BizQcMaterialUsage.del_flag == 0, ) .all() ) for u in usages: u.del_flag = 1 u.update_by = user_id u.update_time = now db.add(u) db.commit()