474 lines
17 KiB
Python
474 lines
17 KiB
Python
"""工艺库 CRUD（biz_process + biz_process_reagent/consumable/output）"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple, Union
|
||
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.models.models import (
|
||
BizProcess,
|
||
BizProcessCategory,
|
||
BizProcessConsumable,
|
||
BizProcessDataCollection,
|
||
BizProcessOutput,
|
||
BizProcessReagent,
|
||
)
|
||
|
||
_UNSET = object()
|
||
|
||
|
||
def _reagent_row_to_dict(r: BizProcessReagent) -> Dict[str, Any]:
|
||
return {
|
||
"id": r.id,
|
||
"processId": r.process_id,
|
||
"reagentCode": r.reagent_code,
|
||
"reagentName": r.reagent_name,
|
||
"spec": r.spec or "",
|
||
"standardUsage": r.standard_usage or "",
|
||
"unit": r.unit or "",
|
||
"isKeyMaterial": r.is_key_material if r.is_key_material is not None else 0,
|
||
"remark": r.remark or "",
|
||
"orderNum": r.order_num if r.order_num is not None else 0,
|
||
}
|
||
|
||
|
||
def _consumable_row_to_dict(c: BizProcessConsumable) -> Dict[str, Any]:
|
||
return {
|
||
"id": c.id,
|
||
"processId": c.process_id,
|
||
"consumableCode": c.consumable_code,
|
||
"consumableName": c.consumable_name,
|
||
"specModel": c.spec_model or "",
|
||
"standardUsage": c.standard_usage or "",
|
||
"unit": c.unit or "",
|
||
"isRequired": c.is_required if c.is_required is not None else 0,
|
||
"remark": c.remark or "",
|
||
}
|
||
|
||
|
||
def _output_row_to_dict(o: BizProcessOutput) -> Dict[str, Any]:
|
||
return {
|
||
"id": o.id,
|
||
"processId": o.process_id,
|
||
"outputTypeId": o.output_type_id or "",
|
||
"outputName": o.output_name or "",
|
||
"outputBarcode": o.output_barcode or "",
|
||
"remark": o.remark or "",
|
||
"orderNum": o.order_num if o.order_num is not None else 0,
|
||
}
|
||
|
||
|
||
def _data_collection_row_to_dict(d: BizProcessDataCollection) -> Dict[str, Any]:
|
||
return {
|
||
"id": d.id,
|
||
"processId": d.process_id,
|
||
"collectionName": d.collection_name or "",
|
||
"unit": d.unit or "",
|
||
"orderNum": d.order_num if d.order_num is not None else 0,
|
||
"remark": d.remark or "",
|
||
}
|
||
|
||
|
||
def _row_to_dict(p: BizProcess) -> Dict[str, Any]:
|
||
return {
|
||
"id": p.id,
|
||
"processCode": p.process_code,
|
||
"categoryId": p.category_id or "",
|
||
"processName": p.process_name,
|
||
"description": p.description or "",
|
||
"estimatedDuration": p.estimated_duration or "",
|
||
"sopIds": p.sop_ids or "",
|
||
"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_processes(
|
||
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(BizProcess).filter(BizProcess.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(BizProcess.tenant_no == tenant_no)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q0 = q0.filter(
|
||
or_(
|
||
BizProcess.process_code.like(like),
|
||
BizProcess.process_name.like(like),
|
||
)
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizProcess, user_info, db, tenant_no)
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(BizProcess.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
_filtered()
|
||
.outerjoin(
|
||
BizProcessCategory,
|
||
and_(
|
||
BizProcess.category_id == BizProcessCategory.id,
|
||
BizProcessCategory.del_flag == 0,
|
||
),
|
||
)
|
||
.order_by(
|
||
BizProcessCategory.category_name.is_(None),
|
||
BizProcessCategory.category_name.asc(),
|
||
BizProcess.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_processes(db: Session, tenant_no: str, user_info=None) -> List[Dict[str, Any]]:
|
||
q0 = db.query(BizProcess).filter(and_(BizProcess.tenant_no == tenant_no, BizProcess.del_flag == 0))
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizProcess, user_info, db, tenant_no)
|
||
rows = q0.order_by(BizProcess.create_time.desc()).all()
|
||
return [_row_to_dict(x) for x in rows]
|
||
|
||
|
||
def get_process(db: Session, process_id: str) -> Optional[BizProcess]:
|
||
return (
|
||
db.query(BizProcess)
|
||
.filter(and_(BizProcess.id == process_id, BizProcess.del_flag == 0))
|
||
.first()
|
||
)
|
||
|
||
|
||
def get_process_detail(db: Session, process_id: str) -> Optional[Dict[str, Any]]:
|
||
proc = get_process(db, process_id)
|
||
if not proc:
|
||
return None
|
||
result = _row_to_dict(proc)
|
||
|
||
reagents = (
|
||
db.query(BizProcessReagent)
|
||
.filter(and_(BizProcessReagent.process_id == process_id, BizProcessReagent.del_flag == 0))
|
||
.order_by(BizProcessReagent.order_num.asc(), BizProcessReagent.create_time.asc(), BizProcessReagent.id.asc())
|
||
.all()
|
||
)
|
||
consumables = (
|
||
db.query(BizProcessConsumable)
|
||
.filter(and_(BizProcessConsumable.process_id == process_id, BizProcessConsumable.del_flag == 0))
|
||
.order_by(BizProcessConsumable.order_num.asc(), BizProcessConsumable.create_time.asc(), BizProcessConsumable.id.asc())
|
||
.all()
|
||
)
|
||
outputs = (
|
||
db.query(BizProcessOutput)
|
||
.filter(and_(BizProcessOutput.process_id == process_id, BizProcessOutput.del_flag == 0))
|
||
.order_by(BizProcessOutput.order_num.asc(), BizProcessOutput.create_time.asc(), BizProcessOutput.id.asc())
|
||
.all()
|
||
)
|
||
|
||
result["reagents"] = [_reagent_row_to_dict(r) for r in reagents]
|
||
result["consumables"] = [_consumable_row_to_dict(c) for c in consumables]
|
||
result["outputs"] = [_output_row_to_dict(o) for o in outputs]
|
||
|
||
data_collections = (
|
||
db.query(BizProcessDataCollection)
|
||
.filter(and_(BizProcessDataCollection.process_id == process_id, BizProcessDataCollection.del_flag == 0))
|
||
.order_by(BizProcessDataCollection.order_num.asc(), BizProcessDataCollection.create_time.asc())
|
||
.all()
|
||
)
|
||
result["dataCollections"] = [_data_collection_row_to_dict(d) for d in data_collections]
|
||
|
||
return result
|
||
|
||
|
||
def process_code_exists(db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
|
||
q = db.query(BizProcess).filter(
|
||
and_(BizProcess.process_code == code.strip(), BizProcess.tenant_no == tenant_no, BizProcess.del_flag == 0)
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(BizProcess.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
def _create_sub_reagent(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
|
||
obj = BizProcessReagent(
|
||
id=generate_uuid(),
|
||
process_id=process_id,
|
||
reagent_code=(data.get("reagent_code") or "").strip(),
|
||
reagent_name=(data.get("reagent_name") or "").strip(),
|
||
spec=(data.get("spec") or "").strip() or None,
|
||
standard_usage=(data.get("standard_usage") or "").strip() or None,
|
||
unit=(data.get("unit") or "").strip() or None,
|
||
is_key_material=data.get("is_key_material", 0),
|
||
remark=(data.get("remark") or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=data.get("order_num", 0) if data.get("order_num") is not None else 0,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
|
||
|
||
def _create_sub_consumable(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
|
||
obj = BizProcessConsumable(
|
||
id=generate_uuid(),
|
||
process_id=process_id,
|
||
consumable_code=(data.get("consumable_code") or "").strip(),
|
||
consumable_name=(data.get("consumable_name") or "").strip(),
|
||
spec_model=(data.get("spec_model") or "").strip() or None,
|
||
standard_usage=(data.get("standard_usage") or "").strip() or None,
|
||
unit=(data.get("unit") or "").strip() or None,
|
||
is_required=data.get("is_required", 0),
|
||
remark=(data.get("remark") or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=data.get("order_num", 0) if data.get("order_num") is not None else 0,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
|
||
|
||
def _create_sub_output(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
|
||
obj = BizProcessOutput(
|
||
id=generate_uuid(),
|
||
process_id=process_id,
|
||
output_type_id=(data.get("output_type_id") or "").strip() or None,
|
||
output_name=(data.get("output_name") or "").strip() or None,
|
||
output_barcode=(data.get("output_barcode") or "").strip() or None,
|
||
remark=(data.get("remark") or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=data.get("order_num", 0) if data.get("order_num") is not None else 0,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
|
||
|
||
def _sub_item_with_order(data: Dict[str, Any], index: int) -> Dict[str, Any]:
|
||
"""保存子表时按列表下标写入 order_num，避免同批 create_time 导致查询顺序漂移。"""
|
||
item = dict(data)
|
||
item["order_num"] = index
|
||
return item
|
||
|
||
|
||
def _create_sub_data_collection(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
|
||
obj = BizProcessDataCollection(
|
||
id=generate_uuid(),
|
||
process_id=process_id,
|
||
collection_name=(data.get("collection_name") or "").strip(),
|
||
unit=(data.get("unit") or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=data.get("order_num", 0) or 0,
|
||
remark=(data.get("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)
|
||
|
||
|
||
def create_process(
|
||
db: Session,
|
||
*,
|
||
process_code: str,
|
||
process_name: str,
|
||
category_id: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
equipment_ids: Optional[Union[str, List[str]]] = None,
|
||
estimated_duration: Optional[str] = None,
|
||
sop_ids: Optional[str] = None,
|
||
reagents: Optional[List[Dict[str, Any]]] = None,
|
||
consumables: Optional[List[Dict[str, Any]]] = None,
|
||
outputs: Optional[List[Dict[str, Any]]] = None,
|
||
data_collections: Optional[List[Dict[str, Any]]] = None,
|
||
tenant_no: str = "zkxw",
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizProcess:
|
||
pc = (process_code or "").strip()
|
||
if not pc:
|
||
raise ValueError("工艺编码不能为空")
|
||
if process_code_exists(db, pc, tenant_no):
|
||
raise ValueError("工艺编码已存在")
|
||
now = datetime.now()
|
||
obj = BizProcess(
|
||
id=generate_uuid(),
|
||
process_code=pc,
|
||
process_name=(process_name or "").strip() or pc,
|
||
category_id=(category_id or "").strip() or None,
|
||
description=(description or "").strip() or None,
|
||
equipment_ids=None,
|
||
equipment_name=None,
|
||
estimated_duration=(estimated_duration or "").strip() or None,
|
||
sop_ids=(sop_ids 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=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.flush()
|
||
|
||
if reagents:
|
||
for idx, r in enumerate(reagents):
|
||
_create_sub_reagent(db, obj.id, tenant_no, user_id, now, _sub_item_with_order(r, idx))
|
||
if consumables:
|
||
for idx, c in enumerate(consumables):
|
||
_create_sub_consumable(db, obj.id, tenant_no, user_id, now, _sub_item_with_order(c, idx))
|
||
if outputs:
|
||
for idx, o in enumerate(outputs):
|
||
_create_sub_output(db, obj.id, tenant_no, user_id, now, _sub_item_with_order(o, idx))
|
||
if data_collections:
|
||
for d in data_collections:
|
||
_create_sub_data_collection(db, obj.id, tenant_no, user_id, now, d)
|
||
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def _soft_delete_sub_records(db: Session, model_class, process_id: str, now: datetime) -> None:
|
||
rows = (
|
||
db.query(model_class)
|
||
.filter(and_(model_class.process_id == process_id, model_class.del_flag == 0))
|
||
.all()
|
||
)
|
||
for r in rows:
|
||
r.del_flag = 1
|
||
r.update_time = now
|
||
db.add(r)
|
||
|
||
|
||
def update_process(
|
||
db: Session,
|
||
process_id: str,
|
||
*,
|
||
process_code: Optional[str] = None,
|
||
process_name: Optional[str] = None,
|
||
category_id: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
equipment_ids: Any = _UNSET,
|
||
estimated_duration: Optional[str] = None,
|
||
sop_ids: Optional[str] = None,
|
||
reagents: Optional[List[Dict[str, Any]]] = None,
|
||
consumables: Optional[List[Dict[str, Any]]] = None,
|
||
outputs: Optional[List[Dict[str, Any]]] = None,
|
||
data_collections: Optional[List[Dict[str, Any]]] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> BizProcess:
|
||
obj = get_process(db, process_id)
|
||
if not obj:
|
||
raise ValueError("工艺不存在")
|
||
if process_code is not None:
|
||
new_code = process_code.strip()
|
||
if not new_code:
|
||
raise ValueError("工艺编码不能为空")
|
||
old_code = (obj.process_code or "").strip()
|
||
if new_code != old_code and process_code_exists(db, new_code, obj.tenant_no, exclude_id=process_id):
|
||
raise ValueError("工艺编码已存在")
|
||
obj.process_code = new_code
|
||
if process_name is not None:
|
||
obj.process_name = (process_name or "").strip() or obj.process_name
|
||
if category_id is not None:
|
||
obj.category_id = (category_id or "").strip() or None
|
||
if description is not None:
|
||
obj.description = (description or "").strip() or None
|
||
if estimated_duration is not None:
|
||
obj.estimated_duration = (estimated_duration or "").strip() or None
|
||
if sop_ids is not None:
|
||
obj.sop_ids = (sop_ids 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)
|
||
|
||
sub_updated = reagents is not None or consumables is not None or outputs is not None or data_collections is not None
|
||
if sub_updated:
|
||
now = datetime.now()
|
||
if reagents is not None:
|
||
_soft_delete_sub_records(db, BizProcessReagent, process_id, now)
|
||
for idx, r in enumerate(reagents):
|
||
_create_sub_reagent(db, process_id, obj.tenant_no, user_id, now, _sub_item_with_order(r, idx))
|
||
if consumables is not None:
|
||
_soft_delete_sub_records(db, BizProcessConsumable, process_id, now)
|
||
for idx, c in enumerate(consumables):
|
||
_create_sub_consumable(db, process_id, obj.tenant_no, user_id, now, _sub_item_with_order(c, idx))
|
||
if outputs is not None:
|
||
_soft_delete_sub_records(db, BizProcessOutput, process_id, now)
|
||
for idx, o in enumerate(outputs):
|
||
_create_sub_output(db, process_id, obj.tenant_no, user_id, now, _sub_item_with_order(o, idx))
|
||
if data_collections is not None:
|
||
_soft_delete_sub_records(db, BizProcessDataCollection, process_id, now)
|
||
for d in data_collections:
|
||
_create_sub_data_collection(db, process_id, obj.tenant_no, user_id, now, d)
|
||
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_process(db: Session, process_id: str) -> None:
|
||
obj = get_process(db, process_id)
|
||
if not obj:
|
||
raise ValueError("工艺不存在")
|
||
now = datetime.now()
|
||
obj.del_flag = 1
|
||
obj.update_time = now
|
||
db.add(obj)
|
||
|
||
_soft_delete_sub_records(db, BizProcessReagent, process_id, now)
|
||
_soft_delete_sub_records(db, BizProcessConsumable, process_id, now)
|
||
_soft_delete_sub_records(db, BizProcessOutput, process_id, now)
|
||
_soft_delete_sub_records(db, BizProcessDataCollection, process_id, now)
|
||
|
||
db.commit()
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|