1022 lines
32 KiB
Python
1022 lines
32 KiB
Python
"""市场物料入库批次 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.material import _dec, _fmt_dt, _is_enabled_status, _norm_status
|
|
from app.models.material import BizMaterial, BizMaterialStockBatch, BizMaterialUsage
|
|
|
|
_SENTINEL_EXPIRE_DATE = date(9999, 12, 31)
|
|
_DEFAULT_STOCK_IN_QTY = Decimal("1")
|
|
|
|
|
|
def _fmt_date(d: Optional[date]) -> str:
|
|
if not d or d >= _SENTINEL_EXPIRE_DATE:
|
|
return ""
|
|
return d.strftime("%Y-%m-%d")
|
|
|
|
|
|
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:
|
|
if isinstance(v, date) and not isinstance(v, datetime):
|
|
if v >= _SENTINEL_EXPIRE_DATE:
|
|
return _SENTINEL_EXPIRE_DATE
|
|
return v
|
|
s = str(v or "").strip()
|
|
if not s:
|
|
return _SENTINEL_EXPIRE_DATE
|
|
return _parse_date(v, label="过期日期")
|
|
|
|
|
|
def _opt_str(v: Any, *, default: str = "") -> str:
|
|
return str(v or "").strip() or default
|
|
|
|
|
|
def _opt_decimal(v: Any, *, default: Decimal = Decimal("0")) -> Decimal:
|
|
if v is None or str(v).strip() == "":
|
|
return default
|
|
try:
|
|
d = Decimal(str(v))
|
|
except Exception as e:
|
|
raise ValueError("金额格式无效") from e
|
|
if d < 0:
|
|
raise ValueError("金额不能为负数")
|
|
return d
|
|
|
|
|
|
def _opt_stock_in_date(v: Any) -> date:
|
|
if isinstance(v, date) and not isinstance(v, datetime):
|
|
return v
|
|
s = str(v or "").strip()
|
|
if not s:
|
|
return date.today()
|
|
return _parse_date(v, label="入库日期")
|
|
|
|
|
|
def _opt_stock_in_qty(v: Any) -> Decimal:
|
|
if v is None or str(v).strip() == "":
|
|
return _DEFAULT_STOCK_IN_QTY
|
|
try:
|
|
qty = Decimal(str(v))
|
|
except Exception as e:
|
|
raise ValueError("入库数量格式无效") from e
|
|
if qty < 0:
|
|
raise ValueError("入库数量不能为负数")
|
|
return qty
|
|
|
|
|
|
def _gen_stock_in_no(db: Session, tenant_no: str) -> str:
|
|
for _ in range(10):
|
|
suffix = generate_uuid()[:8].upper()
|
|
sno = f"RK{datetime.now().strftime('%Y%m%d%H%M%S')}{suffix}"
|
|
if not stock_in_no_exists(db, sno, tenant_no):
|
|
return sno
|
|
raise ValueError("入库编号自动生成失败，请稍后重试")
|
|
|
|
|
|
def _resolve_stock_in_no(
|
|
db: Session,
|
|
tenant_no: str,
|
|
stock_in_no: Any,
|
|
*,
|
|
exclude_id: Optional[str] = None,
|
|
seen_nos: Optional[set] = None,
|
|
) -> str:
|
|
s = str(stock_in_no or "").strip()
|
|
if not s:
|
|
s = _gen_stock_in_no(db, tenant_no)
|
|
if seen_nos is not None:
|
|
if s in seen_nos:
|
|
raise ValueError(f"入库编号「{s}」在本单中重复")
|
|
seen_nos.add(s)
|
|
if stock_in_no_exists(db, s, tenant_no, exclude_id=exclude_id):
|
|
raise ValueError(f"入库编号「{s}」已存在")
|
|
return s
|
|
|
|
|
|
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(BizMaterialStockBatch).filter(
|
|
BizMaterialStockBatch.stock_in_no == s,
|
|
BizMaterialStockBatch.tenant_no == tenant_no,
|
|
BizMaterialStockBatch.del_flag == 0,
|
|
)
|
|
if exclude_id:
|
|
q = q.filter(BizMaterialStockBatch.id != exclude_id)
|
|
return q.first() is not None
|
|
|
|
|
|
def _batch_row_to_dict(b: BizMaterialStockBatch, material: Optional[BizMaterial] = 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 "",
|
|
"category": (mat.category or "") if mat else "",
|
|
"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),
|
|
"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: BizMaterialStockBatch,
|
|
*,
|
|
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
|
|
) -> BizMaterial:
|
|
obj = (
|
|
db.query(BizMaterial)
|
|
.filter(
|
|
BizMaterial.id == material_id,
|
|
BizMaterial.tenant_no == tenant_no,
|
|
BizMaterial.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,
|
|
category: Optional[str] = None,
|
|
status: Optional[str] = None,
|
|
tenant_no: Optional[str] = None,
|
|
user_info=None,
|
|
) -> Tuple[List[Dict[str, Any]], int]:
|
|
def _filtered():
|
|
q0 = (
|
|
db.query(BizMaterialStockBatch, BizMaterial)
|
|
.join(
|
|
BizMaterial,
|
|
BizMaterial.id == BizMaterialStockBatch.material_id,
|
|
)
|
|
.filter(
|
|
BizMaterialStockBatch.del_flag == 0,
|
|
BizMaterial.del_flag == 0,
|
|
)
|
|
)
|
|
if tenant_no:
|
|
q0 = q0.filter(BizMaterialStockBatch.tenant_no == tenant_no)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
q0 = q0.filter(
|
|
or_(
|
|
BizMaterialStockBatch.material_code.like(like),
|
|
BizMaterialStockBatch.material_name.like(like),
|
|
BizMaterialStockBatch.stock_in_no.like(like),
|
|
BizMaterialStockBatch.batch_no.like(like),
|
|
BizMaterialStockBatch.catalog_no.like(like),
|
|
)
|
|
)
|
|
if material_id:
|
|
q0 = q0.filter(BizMaterialStockBatch.material_id == material_id)
|
|
cat = (category or "").strip()
|
|
if cat:
|
|
q0 = q0.filter(BizMaterial.category == cat)
|
|
st = (status or "").strip()
|
|
if st in ("0", "1"):
|
|
q0 = q0.filter(BizMaterial.status == st)
|
|
if user_info:
|
|
from app.core.data_scope import apply_data_scope
|
|
|
|
q0 = apply_data_scope(q0, BizMaterialStockBatch, user_info, db, tenant_no)
|
|
return q0
|
|
|
|
total = _filtered().order_by(None).with_entities(func.count(BizMaterialStockBatch.id)).scalar() or 0
|
|
page = max(1, page)
|
|
page_size = min(max(1, page_size), 100)
|
|
rows = (
|
|
_filtered()
|
|
.order_by(
|
|
BizMaterialStockBatch.stock_in_date.desc(),
|
|
BizMaterialStockBatch.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(BizMaterialStockBatch, BizMaterial)
|
|
.join(BizMaterial, BizMaterial.id == BizMaterialStockBatch.material_id)
|
|
.filter(
|
|
BizMaterialStockBatch.tenant_no == tenant_no,
|
|
BizMaterialStockBatch.del_flag == 0,
|
|
BizMaterial.del_flag == 0,
|
|
BizMaterial.status == "1",
|
|
)
|
|
)
|
|
if only_available:
|
|
q = q.filter(BizMaterialStockBatch.remaining_qty > 0)
|
|
if material_id:
|
|
q = q.filter(BizMaterialStockBatch.material_id == material_id)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
q = q.filter(
|
|
or_(
|
|
BizMaterialStockBatch.stock_in_no.like(like),
|
|
BizMaterialStockBatch.material_name.like(like),
|
|
BizMaterialStockBatch.material_code.like(like),
|
|
BizMaterialStockBatch.batch_no.like(like),
|
|
BizMaterialStockBatch.catalog_no.like(like),
|
|
)
|
|
)
|
|
rows = q.order_by(BizMaterialStockBatch.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: BizMaterial,
|
|
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: Any,
|
|
remark: Optional[str],
|
|
operator_name: Optional[str],
|
|
tenant_no: str,
|
|
user_id: Optional[str],
|
|
operate_time: datetime,
|
|
commit: bool = False,
|
|
) -> BizMaterialStockBatch:
|
|
qty = _opt_stock_in_qty(qty)
|
|
if qty <= 0:
|
|
raise ValueError("入库数量必须大于0")
|
|
sno = _resolve_stock_in_no(db, tenant_no, stock_in_no)
|
|
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 = BizMaterialStockBatch(
|
|
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=_opt_stock_in_date(stock_in_date),
|
|
expire_date=_parse_expire_date(expire_date),
|
|
catalog_no=_opt_str(catalog_no),
|
|
batch_no=_opt_str(batch_no),
|
|
manufacturer=_opt_str(manufacturer),
|
|
unit_price=_opt_decimal(unit_price),
|
|
amount=_opt_decimal(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 = BizMaterialUsage(
|
|
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,
|
|
operator_name=(operator_name 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)
|
|
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: Any,
|
|
remark: Optional[str] = None,
|
|
operator_name: Optional[str] = None,
|
|
tenant_no: str,
|
|
user_id: Optional[str] = None,
|
|
operate_time: Optional[datetime] = None,
|
|
) -> BizMaterialStockBatch:
|
|
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=stock_in_date,
|
|
expire_date=expire_date,
|
|
catalog_no=catalog_no,
|
|
batch_no=batch_no,
|
|
manufacturer=manufacturer,
|
|
unit_price=unit_price,
|
|
amount=amount,
|
|
qty=qty,
|
|
remark=remark,
|
|
operator_name=operator_name,
|
|
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,
|
|
operator_name: 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 = _resolve_stock_in_no(
|
|
db,
|
|
tenant_no,
|
|
it.get("stock_in_no") or it.get("stockInNo"),
|
|
seen_nos=seen_nos,
|
|
)
|
|
qty_dec = _opt_stock_in_qty(it.get("qty"))
|
|
if qty_dec <= 0:
|
|
raise ValueError("入库数量必须大于0")
|
|
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=it.get("stock_in_date") or it.get("stockInDate"),
|
|
expire_date=it.get("expire_date") or it.get("expireDate"),
|
|
catalog_no=it.get("catalog_no") or it.get("catalogNo"),
|
|
batch_no=it.get("batch_no") or it.get("batchNo"),
|
|
manufacturer=it.get("manufacturer"),
|
|
unit_price=it.get("unit_price") or it.get("unitPrice"),
|
|
amount=it.get("amount"),
|
|
qty=qty_dec,
|
|
remark=line_remark,
|
|
operator_name=operator_name,
|
|
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],
|
|
operator_name: 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(BizMaterialStockBatch)
|
|
.filter(
|
|
BizMaterialStockBatch.id == batch_id,
|
|
BizMaterialStockBatch.tenant_no == tenant_no,
|
|
BizMaterialStockBatch.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 = BizMaterialUsage(
|
|
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,
|
|
operator_name=(operator_name 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,
|
|
operator_name: 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,
|
|
operator_name=operator_name,
|
|
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],
|
|
operator_name: 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(BizMaterialStockBatch)
|
|
.filter(
|
|
BizMaterialStockBatch.id == batch_id,
|
|
BizMaterialStockBatch.tenant_no == tenant_no,
|
|
BizMaterialStockBatch.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 = BizMaterialUsage(
|
|
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,
|
|
operator_name=(operator_name 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_discard(
|
|
db: Session,
|
|
*,
|
|
items: List[Dict[str, Any]],
|
|
discard_reason: Optional[str] = None,
|
|
remark: Optional[str] = None,
|
|
operator_name: 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,
|
|
operator_name=operator_name,
|
|
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[BizMaterialStockBatch]:
|
|
return (
|
|
db.query(BizMaterialStockBatch)
|
|
.filter(
|
|
BizMaterialStockBatch.id == batch_id,
|
|
BizMaterialStockBatch.tenant_no == tenant_no,
|
|
BizMaterialStockBatch.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
|
|
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,
|
|
remark: Optional[str] = None,
|
|
material_code: Optional[str] = None,
|
|
material_name: Optional[str] = None,
|
|
category: Optional[str] = None,
|
|
spec_model: Optional[str] = None,
|
|
unit: Optional[str] = None,
|
|
warning_qty: Optional[Any] = None,
|
|
status: Optional[str] = None,
|
|
) -> BizMaterialStockBatch:
|
|
from app.crud import material as mat_svc
|
|
|
|
batch = get_batch(db, batch_id, tenant_no)
|
|
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 category is not None:
|
|
mat_data["category"] = (category or "").strip() or None
|
|
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_fields(
|
|
db,
|
|
batch.material_id,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
**mat_data,
|
|
)
|
|
|
|
if stock_in_no is not None:
|
|
s = str(stock_in_no).strip()
|
|
if s:
|
|
if stock_in_no_exists(db, s, tenant_no, exclude_id=batch_id):
|
|
raise ValueError(f"入库编号「{s}」已存在")
|
|
batch.stock_in_no = s
|
|
if stock_in_date is not None:
|
|
s = str(stock_in_date).strip()
|
|
if s:
|
|
batch.stock_in_date = _parse_date(stock_in_date)
|
|
if expire_date is not None:
|
|
s = str(expire_date).strip()
|
|
if s:
|
|
batch.expire_date = _parse_expire_date(expire_date)
|
|
else:
|
|
batch.expire_date = _SENTINEL_EXPIRE_DATE
|
|
if catalog_no is not None:
|
|
batch.catalog_no = _opt_str(catalog_no)
|
|
if batch_no is not None:
|
|
batch.batch_no = _opt_str(batch_no)
|
|
if manufacturer is not None:
|
|
batch.manufacturer = _opt_str(manufacturer)
|
|
if unit_price is not None:
|
|
batch.unit_price = _opt_decimal(unit_price)
|
|
if amount is not None:
|
|
batch.amount = _opt_decimal(amount)
|
|
if remark is not None:
|
|
batch.remark = (remark or "").strip() or None
|
|
|
|
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(BizMaterialUsage)
|
|
.filter(
|
|
BizMaterialUsage.batch_id == batch_id,
|
|
BizMaterialUsage.tenant_no == tenant_no,
|
|
BizMaterialUsage.del_flag == 0,
|
|
BizMaterialUsage.change_type == "in",
|
|
BizMaterialUsage.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
|
|
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(BizMaterialUsage)
|
|
.filter(
|
|
BizMaterialUsage.batch_id == batch_id,
|
|
BizMaterialUsage.tenant_no == tenant_no,
|
|
BizMaterialUsage.del_flag == 0,
|
|
BizMaterialUsage.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(BizMaterialUsage)
|
|
.filter(
|
|
BizMaterialUsage.batch_id == batch_id,
|
|
BizMaterialUsage.tenant_no == tenant_no,
|
|
BizMaterialUsage.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()
|