296 lines
9.8 KiB
Python
296 lines
9.8 KiB
Python
"""耗材包 CRUD（biz_consumable_pack + biz_consumable_pack_item）"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
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 BizConsumablePack, BizConsumablePackItem
|
||
|
||
|
||
def _item_row_to_dict(item: BizConsumablePackItem) -> Dict[str, Any]:
|
||
return {
|
||
"id": item.id,
|
||
"packId": item.pack_id,
|
||
"consumableCode": item.consumable_code,
|
||
"consumableName": item.consumable_name,
|
||
"specModel": item.spec_model or "",
|
||
"standardUsage": item.standard_usage or "",
|
||
"unit": item.unit or "",
|
||
"remark": item.remark or "",
|
||
}
|
||
|
||
|
||
def _row_to_dict(p: BizConsumablePack) -> Dict[str, Any]:
|
||
return {
|
||
"id": p.id,
|
||
"packCode": p.pack_code,
|
||
"packName": p.pack_name,
|
||
"usageScene": p.usage_scene or "",
|
||
"packType": p.pack_type or "",
|
||
"status": int(p.status) if p.status not in (None, "") else 0,
|
||
"tenantNo": p.tenant_no,
|
||
"orderNum": p.order_num if p.order_num is not None else 0,
|
||
"remark": p.remark or "",
|
||
"createTime": p.create_time.strftime("%Y-%m-%d %H:%M:%S") if p.create_time else "",
|
||
"updateTime": p.update_time.strftime("%Y-%m-%d %H:%M:%S") if p.update_time else "",
|
||
}
|
||
|
||
|
||
def list_consumable_packs(
|
||
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(BizConsumablePack).filter(BizConsumablePack.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(BizConsumablePack.tenant_no == tenant_no)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q0 = q0.filter(
|
||
or_(
|
||
BizConsumablePack.pack_code.like(like),
|
||
BizConsumablePack.pack_name.like(like),
|
||
)
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizConsumablePack, user_info, db, tenant_no)
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(BizConsumablePack.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
_filtered()
|
||
.order_by(BizConsumablePack.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return [_row_to_dict(x) for x in rows], int(total)
|
||
|
||
|
||
def list_all_packs(db: Session, tenant_no: str, user_info=None) -> List[Dict[str, Any]]:
|
||
q0 = db.query(BizConsumablePack).filter(and_(BizConsumablePack.tenant_no == tenant_no, BizConsumablePack.del_flag == 0))
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizConsumablePack, user_info, db, tenant_no)
|
||
rows = q0.order_by(BizConsumablePack.create_time.desc()).all()
|
||
return [_row_to_dict(x) for x in rows]
|
||
|
||
|
||
def get_consumable_pack(db: Session, pack_id: str) -> Optional[BizConsumablePack]:
|
||
return (
|
||
db.query(BizConsumablePack)
|
||
.filter(and_(BizConsumablePack.id == pack_id, BizConsumablePack.del_flag == 0))
|
||
.first()
|
||
)
|
||
|
||
|
||
def get_pack_detail(db: Session, pack_id: str) -> Optional[Dict[str, Any]]:
|
||
pack = get_consumable_pack(db, pack_id)
|
||
if not pack:
|
||
return None
|
||
result = _row_to_dict(pack)
|
||
items = (
|
||
db.query(BizConsumablePackItem)
|
||
.filter(and_(BizConsumablePackItem.pack_id == pack_id, BizConsumablePackItem.del_flag == 0))
|
||
.order_by(BizConsumablePackItem.create_time.asc())
|
||
.all()
|
||
)
|
||
result["items"] = [_item_row_to_dict(i) for i in items]
|
||
return result
|
||
|
||
|
||
def pack_code_exists(db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
|
||
q = db.query(BizConsumablePack).filter(
|
||
and_(BizConsumablePack.pack_code == code.strip(), BizConsumablePack.tenant_no == tenant_no, BizConsumablePack.del_flag == 0)
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(BizConsumablePack.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
def create_consumable_pack(
|
||
db: Session,
|
||
*,
|
||
pack_code: str,
|
||
pack_name: str,
|
||
usage_scene: Optional[str] = None,
|
||
pack_type: Optional[str] = None,
|
||
status: Optional[int] = None,
|
||
items: Optional[List[Dict[str, Any]]] = None,
|
||
tenant_no: str = "zkxw",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizConsumablePack:
|
||
pc = (pack_code or "").strip()
|
||
if not pc:
|
||
raise ValueError("耗材包编码不能为空")
|
||
if pack_code_exists(db, pc, tenant_no):
|
||
raise ValueError("耗材包编码已存在")
|
||
now = datetime.now()
|
||
obj = BizConsumablePack(
|
||
id=generate_uuid(),
|
||
pack_code=pc,
|
||
pack_name=(pack_name or "").strip() or pc,
|
||
usage_scene=(usage_scene or "").strip() or None,
|
||
pack_type=(pack_type or "").strip() or None,
|
||
status=str(status) if status is not None else "1",
|
||
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=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.flush()
|
||
|
||
if items:
|
||
for it in items:
|
||
item_obj = BizConsumablePackItem(
|
||
id=generate_uuid(),
|
||
pack_id=obj.id,
|
||
consumable_code=(it.get("consumable_code") or "").strip(),
|
||
consumable_name=(it.get("consumable_name") or "").strip(),
|
||
spec_model=(it.get("spec_model") or "").strip() or None,
|
||
standard_usage=(it.get("standard_usage") or "").strip() or None,
|
||
unit=(it.get("unit") or "").strip() or None,
|
||
remark=(it.get("remark") or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(item_obj)
|
||
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_consumable_pack(
|
||
db: Session,
|
||
pack_id: str,
|
||
*,
|
||
pack_code: Optional[str] = None,
|
||
pack_name: Optional[str] = None,
|
||
usage_scene: Optional[str] = None,
|
||
pack_type: Optional[str] = None,
|
||
status: Optional[int] = None,
|
||
items: Optional[List[Dict[str, Any]]] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizConsumablePack:
|
||
obj = get_consumable_pack(db, pack_id)
|
||
if not obj:
|
||
raise ValueError("耗材包不存在")
|
||
if pack_code is not None:
|
||
new_code = pack_code.strip()
|
||
if not new_code:
|
||
raise ValueError("耗材包编码不能为空")
|
||
if new_code != obj.pack_code and pack_code_exists(db, new_code, obj.tenant_no, exclude_id=pack_id):
|
||
raise ValueError("耗材包编码已存在")
|
||
obj.pack_code = new_code
|
||
if pack_name is not None:
|
||
obj.pack_name = (pack_name or "").strip() or obj.pack_name
|
||
if usage_scene is not None:
|
||
obj.usage_scene = (usage_scene or "").strip() or None
|
||
if pack_type is not None:
|
||
obj.pack_type = (pack_type or "").strip() or None
|
||
if status is not None:
|
||
obj.status = str(status)
|
||
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)
|
||
|
||
if items is not None:
|
||
existing = (
|
||
db.query(BizConsumablePackItem)
|
||
.filter(and_(BizConsumablePackItem.pack_id == pack_id, BizConsumablePackItem.del_flag == 0))
|
||
.all()
|
||
)
|
||
now = datetime.now()
|
||
for ei in existing:
|
||
ei.del_flag = 1
|
||
ei.update_time = now
|
||
db.add(ei)
|
||
|
||
for it in items:
|
||
item_obj = BizConsumablePackItem(
|
||
id=generate_uuid(),
|
||
pack_id=pack_id,
|
||
consumable_code=(it.get("consumable_code") or "").strip(),
|
||
consumable_name=(it.get("consumable_name") or "").strip(),
|
||
spec_model=(it.get("spec_model") or "").strip() or None,
|
||
standard_usage=(it.get("standard_usage") or "").strip() or None,
|
||
unit=(it.get("unit") or "").strip() or None,
|
||
remark=(it.get("remark") or "").strip() or None,
|
||
tenant_no=obj.tenant_no,
|
||
order_num=0,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(item_obj)
|
||
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_consumable_pack(db: Session, pack_id: str) -> None:
|
||
obj = get_consumable_pack(db, pack_id)
|
||
if not obj:
|
||
raise ValueError("耗材包不存在")
|
||
now = datetime.now()
|
||
obj.del_flag = 1
|
||
obj.update_time = now
|
||
db.add(obj)
|
||
|
||
items = (
|
||
db.query(BizConsumablePackItem)
|
||
.filter(and_(BizConsumablePackItem.pack_id == pack_id, BizConsumablePackItem.del_flag == 0))
|
||
.all()
|
||
)
|
||
for it in items:
|
||
it.del_flag = 1
|
||
it.update_time = now
|
||
db.add(it)
|
||
|
||
db.commit()
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|