534 lines
17 KiB
Python
534 lines
17 KiB
Python
"""耗材 CRUD（biz_consumable）"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.models.models import BizConsumable, BizConsumableStockLog
|
||
|
||
|
||
def _dec(v: Any) -> float:
|
||
if v is None:
|
||
return 0.0
|
||
return float(v)
|
||
|
||
|
||
def _norm_status(v: Any) -> str:
|
||
if v is None:
|
||
return "1"
|
||
if isinstance(v, bool):
|
||
return "1" if v else "0"
|
||
if isinstance(v, (int, float)):
|
||
if int(v) == 0:
|
||
return "0"
|
||
if int(v) == 1:
|
||
return "1"
|
||
s = str(v).strip()
|
||
if s in ("0", "1"):
|
||
return s
|
||
return "1"
|
||
|
||
|
||
def _is_enabled_status(v: Any) -> bool:
|
||
return _norm_status(v) == "1"
|
||
|
||
|
||
def _row_to_dict(c: BizConsumable) -> Dict[str, Any]:
|
||
return {
|
||
"id": c.id,
|
||
"consumableCode": c.consumable_code,
|
||
"consumableName": c.consumable_name,
|
||
"specModel": c.spec_model or "",
|
||
"standardUsage": c.standard_usage or "",
|
||
"unit": c.unit or "",
|
||
"description": c.description or "",
|
||
"stockQty": _dec(c.stock_qty),
|
||
"warningQty": _dec(c.warning_qty),
|
||
"status": _norm_status(c.status),
|
||
"tenantNo": c.tenant_no,
|
||
"orderNum": c.order_num if c.order_num is not None else 0,
|
||
"remark": c.remark or "",
|
||
"createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "",
|
||
"updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "",
|
||
}
|
||
|
||
|
||
def list_consumables(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
def _filtered():
|
||
q0 = db.query(BizConsumable).filter(BizConsumable.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(BizConsumable.tenant_no == tenant_no)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q0 = q0.filter(
|
||
or_(
|
||
BizConsumable.consumable_code.like(like),
|
||
BizConsumable.consumable_name.like(like),
|
||
)
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizConsumable, user_info, db, tenant_no)
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(BizConsumable.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
_filtered()
|
||
.order_by(BizConsumable.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return [_row_to_dict(x) for x in rows], int(total)
|
||
|
||
|
||
def get_consumable(db: Session, consumable_id: str) -> Optional[BizConsumable]:
|
||
return (
|
||
db.query(BizConsumable)
|
||
.filter(and_(BizConsumable.id == consumable_id, BizConsumable.del_flag == 0))
|
||
.first()
|
||
)
|
||
|
||
|
||
def consumable_code_exists(db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
|
||
q = db.query(BizConsumable).filter(
|
||
and_(BizConsumable.consumable_code == code.strip(), BizConsumable.tenant_no == tenant_no, BizConsumable.del_flag == 0)
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(BizConsumable.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
def create_consumable(
|
||
db: Session,
|
||
*,
|
||
consumable_code: str,
|
||
consumable_name: str,
|
||
spec_model: Optional[str] = None,
|
||
standard_usage: Optional[str] = None,
|
||
unit: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
tenant_no: str = "zkxw",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizConsumable:
|
||
cc = (consumable_code or "").strip()
|
||
if not cc:
|
||
raise ValueError("耗材编码不能为空")
|
||
if consumable_code_exists(db, cc, tenant_no):
|
||
raise ValueError("耗材编码已存在")
|
||
obj = BizConsumable(
|
||
id=generate_uuid(),
|
||
consumable_code=cc,
|
||
consumable_name=(consumable_name or "").strip() or cc,
|
||
spec_model=(spec_model or "").strip() or None,
|
||
standard_usage=(standard_usage or "").strip() or None,
|
||
unit=(unit or "").strip() or None,
|
||
description=(description or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=order_num if order_num is not None else 0,
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_consumable(
|
||
db: Session,
|
||
consumable_id: str,
|
||
*,
|
||
consumable_code: Optional[str] = None,
|
||
consumable_name: Optional[str] = None,
|
||
spec_model: Optional[str] = None,
|
||
standard_usage: Optional[str] = None,
|
||
unit: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizConsumable:
|
||
obj = get_consumable(db, consumable_id)
|
||
if not obj:
|
||
raise ValueError("耗材不存在")
|
||
if consumable_code is not None:
|
||
new_code = consumable_code.strip()
|
||
if not new_code:
|
||
raise ValueError("耗材编码不能为空")
|
||
if new_code != obj.consumable_code and consumable_code_exists(db, new_code, obj.tenant_no, exclude_id=consumable_id):
|
||
raise ValueError("耗材编码已存在")
|
||
obj.consumable_code = new_code
|
||
if consumable_name is not None:
|
||
obj.consumable_name = (consumable_name or "").strip() or obj.consumable_name
|
||
if spec_model is not None:
|
||
obj.spec_model = (spec_model or "").strip() or None
|
||
if standard_usage is not None:
|
||
obj.standard_usage = (standard_usage or "").strip() or None
|
||
if unit is not None:
|
||
obj.unit = (unit or "").strip() or None
|
||
if description is not None:
|
||
obj.description = (description or "").strip() or None
|
||
if order_num is not None:
|
||
obj.order_num = order_num
|
||
if remark is not None:
|
||
obj.remark = (remark or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_consumable(db: Session, consumable_id: str) -> None:
|
||
obj = get_consumable(db, consumable_id)
|
||
if not obj:
|
||
raise ValueError("耗材不存在")
|
||
if obj.stock_qty is not None and _dec(obj.stock_qty) > 0:
|
||
raise ValueError("该耗材仍有库存，无法删除")
|
||
obj.del_flag = 1
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
|
||
|
||
def _build_manual_out_remark(
|
||
*,
|
||
receiver: Optional[str] = None,
|
||
purpose: Optional[str] = None,
|
||
remark: Optional[str] = None,
|
||
line_remark: Optional[str] = None,
|
||
) -> Optional[str]:
|
||
parts: List[str] = []
|
||
r = (receiver or "").strip()
|
||
if r:
|
||
parts.append(f"领用人：{r}")
|
||
p = (purpose or "").strip()
|
||
if p:
|
||
parts.append(f"用途：{p}")
|
||
extra = (line_remark or remark or "").strip()
|
||
if extra:
|
||
parts.append(extra)
|
||
return "；".join(parts) if parts else None
|
||
|
||
|
||
def stock_in(
|
||
db: Session,
|
||
row_id: str,
|
||
*,
|
||
qty: Decimal,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
operate_time: Optional[datetime] = None,
|
||
) -> BizConsumable:
|
||
if qty is None or float(qty) <= 0:
|
||
raise ValueError("入库数量必须大于0")
|
||
obj = (
|
||
db.query(BizConsumable)
|
||
.filter(BizConsumable.id == row_id, BizConsumable.del_flag == 0)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not obj:
|
||
raise ValueError("耗材不存在")
|
||
if tenant_no and obj.tenant_no != tenant_no:
|
||
raise ValueError("耗材不存在")
|
||
if not _is_enabled_status(obj.status):
|
||
raise ValueError(f"耗材「{obj.consumable_name}」已停用，无法入库")
|
||
op_time = operate_time or datetime.now()
|
||
old_stock = obj.stock_qty or 0
|
||
new_stock = old_stock + qty
|
||
obj.stock_qty = new_stock
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
log = BizConsumableStockLog(
|
||
id=generate_uuid(),
|
||
material_id=obj.id,
|
||
material_code=obj.consumable_code,
|
||
material_name=obj.consumable_name,
|
||
change_type="in",
|
||
change_qty=qty,
|
||
stock_before=old_stock,
|
||
stock_after=new_stock,
|
||
source_type="manual_in",
|
||
tenant_no=obj.tenant_no,
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id=obj.create_dept_id or "",
|
||
create_time=op_time,
|
||
update_by=user_id,
|
||
update_time=op_time,
|
||
del_flag=0,
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
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
|
||
|
||
try:
|
||
for it in items:
|
||
material_id = (it.get("material_id") or it.get("materialId") or "").strip()
|
||
qty = it.get("qty")
|
||
if not material_id:
|
||
raise ValueError("耗材ID不能为空")
|
||
if qty is None or float(qty) <= 0:
|
||
raise ValueError("入库数量必须大于0")
|
||
qty_dec = Decimal(str(qty))
|
||
line_remark = (it.get("remark") or "").strip()
|
||
log_remark = line_remark or shared_remark or None
|
||
|
||
obj = (
|
||
db.query(BizConsumable)
|
||
.filter(
|
||
BizConsumable.id == material_id,
|
||
BizConsumable.tenant_no == tenant_no,
|
||
BizConsumable.del_flag == 0,
|
||
)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not obj:
|
||
raise ValueError("耗材不存在")
|
||
if not _is_enabled_status(obj.status):
|
||
raise ValueError(f"耗材「{obj.consumable_name}」已停用，无法入库")
|
||
|
||
old_stock = obj.stock_qty or 0
|
||
new_stock = old_stock + qty_dec
|
||
obj.stock_qty = new_stock
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
|
||
log = BizConsumableStockLog(
|
||
id=generate_uuid(),
|
||
material_id=obj.id,
|
||
material_code=obj.consumable_code,
|
||
material_name=obj.consumable_name,
|
||
change_type="in",
|
||
change_qty=qty_dec,
|
||
stock_before=old_stock,
|
||
stock_after=new_stock,
|
||
source_type="manual_in",
|
||
tenant_no=obj.tenant_no,
|
||
remark=log_remark,
|
||
create_by=user_id,
|
||
create_dept_id=obj.create_dept_id or "",
|
||
create_time=op_time,
|
||
update_by=user_id,
|
||
update_time=op_time,
|
||
del_flag=0,
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
return len(items)
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
|
||
|
||
def stock_out(
|
||
db: Session,
|
||
row_id: str,
|
||
*,
|
||
qty: Decimal,
|
||
receiver: Optional[str] = None,
|
||
purpose: Optional[str] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
operate_time: Optional[datetime] = None,
|
||
) -> BizConsumable:
|
||
if qty is None or float(qty) <= 0:
|
||
raise ValueError("出库数量必须大于0")
|
||
qty_dec = Decimal(str(qty))
|
||
if not (receiver or "").strip():
|
||
raise ValueError("请选择领用人")
|
||
|
||
obj = (
|
||
db.query(BizConsumable)
|
||
.filter(BizConsumable.id == row_id, BizConsumable.del_flag == 0)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not obj:
|
||
raise ValueError("耗材不存在")
|
||
if tenant_no and obj.tenant_no != tenant_no:
|
||
raise ValueError("耗材不存在")
|
||
if not _is_enabled_status(obj.status):
|
||
raise ValueError(f"耗材「{obj.consumable_name}」已停用，无法出库")
|
||
|
||
stock_before = obj.stock_qty if obj.stock_qty is not None else Decimal("0")
|
||
if stock_before < qty_dec:
|
||
raise ValueError(
|
||
f"耗材「{obj.consumable_name}」库存不足，当前库存 {stock_before}，需出库 {qty_dec}"
|
||
)
|
||
|
||
op_time = operate_time or datetime.now()
|
||
new_stock = stock_before - qty_dec
|
||
obj.stock_qty = new_stock
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
|
||
log = BizConsumableStockLog(
|
||
id=generate_uuid(),
|
||
material_id=obj.id,
|
||
material_code=obj.consumable_code,
|
||
material_name=obj.consumable_name,
|
||
change_type="use",
|
||
change_qty=-qty_dec,
|
||
stock_before=stock_before,
|
||
stock_after=new_stock,
|
||
source_type="manual_out",
|
||
tenant_no=obj.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=obj.create_dept_id or "",
|
||
create_time=op_time,
|
||
update_by=user_id,
|
||
update_time=op_time,
|
||
del_flag=0,
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
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("请至少选择一条耗材")
|
||
if not (receiver or "").strip():
|
||
raise ValueError("请选择领用人")
|
||
|
||
op_time = operate_time or datetime.now()
|
||
recv = (receiver or "").strip() or None
|
||
purp = (purpose or "").strip() or None
|
||
shared_remark = (remark or "").strip() or None
|
||
|
||
try:
|
||
for it in items:
|
||
material_id = (it.get("material_id") or it.get("materialId") or "").strip()
|
||
qty = it.get("qty")
|
||
if not material_id:
|
||
raise ValueError("耗材ID不能为空")
|
||
if qty is None or float(qty) <= 0:
|
||
raise ValueError("出库数量必须大于0")
|
||
qty_dec = Decimal(str(qty))
|
||
line_remark = (it.get("remark") or "").strip() or shared_remark
|
||
|
||
obj = (
|
||
db.query(BizConsumable)
|
||
.filter(
|
||
BizConsumable.id == material_id,
|
||
BizConsumable.tenant_no == tenant_no,
|
||
BizConsumable.del_flag == 0,
|
||
)
|
||
.with_for_update()
|
||
.first()
|
||
)
|
||
if not obj:
|
||
raise ValueError("耗材不存在")
|
||
if not _is_enabled_status(obj.status):
|
||
raise ValueError(f"耗材「{obj.consumable_name}」已停用，无法出库")
|
||
|
||
stock_before = obj.stock_qty if obj.stock_qty is not None else Decimal("0")
|
||
if stock_before < qty_dec:
|
||
raise ValueError(
|
||
f"耗材「{obj.consumable_name}」库存不足，当前库存 {stock_before}，需出库 {qty_dec}"
|
||
)
|
||
|
||
new_stock = stock_before - qty_dec
|
||
obj.stock_qty = new_stock
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
|
||
log = BizConsumableStockLog(
|
||
id=generate_uuid(),
|
||
material_id=obj.id,
|
||
material_code=obj.consumable_code,
|
||
material_name=obj.consumable_name,
|
||
change_type="use",
|
||
change_qty=-qty_dec,
|
||
stock_before=stock_before,
|
||
stock_after=new_stock,
|
||
source_type="manual_out",
|
||
tenant_no=obj.tenant_no,
|
||
receiver=recv,
|
||
purpose=purp,
|
||
remark=line_remark,
|
||
create_by=user_id,
|
||
create_dept_id=obj.create_dept_id or "",
|
||
create_time=op_time,
|
||
update_by=user_id,
|
||
update_time=op_time,
|
||
del_flag=0,
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
return len(items)
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|