feat(crud): add production task CRUD with route-cloning logic
This commit is contained in:
zlwu committed
1 parent
9d9f43579d
commit
fb7846f2ed
1 file changed
+692
@@ -0,0 +1,692 @@
|
||||
"""生产任务 CRUD（biz_production_task + 步骤 + 子表 + 投入/产出）"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from sqlalchemy import and_, func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.utils import generate_uuid
|
||||
from app.models.models import (
|
||||
BizProcessRouteStep,
|
||||
BizProcessRouteStepConsumable,
|
||||
BizProcessRouteStepDataCollection,
|
||||
BizProcessRouteStepOutput,
|
||||
BizProcessRouteStepReagent,
|
||||
BizProductionTask,
|
||||
BizProductionTaskInput,
|
||||
BizProductionTaskOutput,
|
||||
BizProductionTaskStep,
|
||||
BizProductionTaskStepConsumable,
|
||||
BizProductionTaskStepDataCollection,
|
||||
BizProductionTaskStepOutput,
|
||||
BizProductionTaskStepReagent,
|
||||
)
|
||||
|
||||
|
||||
def _fmt_dt(t: Optional[datetime]) -> str:
|
||||
return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""
|
||||
|
||||
|
||||
def _generate_task_no(db: Session, tenant_no: str) -> str:
|
||||
day = datetime.now().strftime("%Y%m%d")
|
||||
like = f"RW{day}%"
|
||||
cnt = (
|
||||
db.query(func.count(BizProductionTask.id))
|
||||
.filter(
|
||||
BizProductionTask.tenant_no == tenant_no,
|
||||
BizProductionTask.task_no.like(like),
|
||||
)
|
||||
.scalar()
|
||||
)
|
||||
return f"RW{day}{(cnt or 0) + 1:04d}"
|
||||
|
||||
|
||||
def _task_to_dict(t: BizProductionTask) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": t.id,
|
||||
"taskNo": t.task_no,
|
||||
"workOrderId": t.work_order_id,
|
||||
"taskName": t.task_name or "",
|
||||
"description": t.description or "",
|
||||
"routeId": t.route_id or "",
|
||||
"taskStatus": t.task_status,
|
||||
"startedTime": _fmt_dt(t.started_time),
|
||||
"completedTime": _fmt_dt(t.completed_time),
|
||||
"remark": t.remark or "",
|
||||
"attachments": t.attachments or "",
|
||||
"tenantNo": t.tenant_no,
|
||||
"orderNum": t.order_num if t.order_num is not None else 0,
|
||||
"createTime": _fmt_dt(t.create_time),
|
||||
"updateTime": _fmt_dt(t.update_time),
|
||||
}
|
||||
|
||||
|
||||
def _step_to_dict(s: BizProductionTaskStep) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": s.id,
|
||||
"taskId": s.task_id,
|
||||
"stepOrder": s.step_order if s.step_order is not None else 0,
|
||||
"processCode": s.process_code or "",
|
||||
"categoryId": s.category_id or "",
|
||||
"processName": s.process_name or "",
|
||||
"description": s.description or "",
|
||||
"equipmentName": s.equipment_name or "",
|
||||
"estimatedDuration": s.estimated_duration or "",
|
||||
"actualDuration": s.actual_duration or "",
|
||||
"sopIds": s.sop_ids or "",
|
||||
"stepStatus": s.step_status or "dzx",
|
||||
"startedTime": _fmt_dt(s.started_time),
|
||||
"completedTime": _fmt_dt(s.completed_time),
|
||||
"remark": s.remark or "",
|
||||
}
|
||||
|
||||
|
||||
def _step_reagent_to_dict(r: BizProductionTaskStepReagent) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": r.id,
|
||||
"stepId": r.step_id,
|
||||
"reagentCode": r.reagent_code or "",
|
||||
"reagentName": r.reagent_name or "",
|
||||
"spec": r.spec or "",
|
||||
"standardUsage": r.standard_usage or "",
|
||||
"actualUsage": r.actual_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 "",
|
||||
}
|
||||
|
||||
|
||||
def _step_consumable_to_dict(c: BizProductionTaskStepConsumable) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": c.id,
|
||||
"stepId": c.step_id,
|
||||
"consumableCode": c.consumable_code or "",
|
||||
"consumableName": c.consumable_name or "",
|
||||
"specModel": c.spec_model or "",
|
||||
"standardUsage": c.standard_usage or "",
|
||||
"actualUsage": c.actual_usage or "",
|
||||
"unit": c.unit or "",
|
||||
"isRequired": c.is_required if c.is_required is not None else "1",
|
||||
"remark": c.remark or "",
|
||||
}
|
||||
|
||||
|
||||
def _step_output_to_dict(o: BizProductionTaskStepOutput) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": o.id,
|
||||
"stepId": o.step_id,
|
||||
"outputTypeId": o.output_type_id or "",
|
||||
"outputName": o.output_name or "",
|
||||
"outputBarcode": o.output_barcode or "",
|
||||
"remark": o.remark or "",
|
||||
}
|
||||
|
||||
|
||||
def _step_data_collection_to_dict(d: BizProductionTaskStepDataCollection) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": d.id,
|
||||
"stepId": d.step_id,
|
||||
"collectionName": d.collection_name or "",
|
||||
"unit": d.unit or "",
|
||||
"collectionValue": d.collection_value or "",
|
||||
"remark": d.remark or "",
|
||||
}
|
||||
|
||||
|
||||
def _task_input_to_dict(i: BizProductionTaskInput) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": i.id,
|
||||
"taskId": i.task_id,
|
||||
"inputType": i.input_type or "",
|
||||
"inputId": i.input_id or "",
|
||||
"inputCode": i.input_code or "",
|
||||
"inputName": i.input_name or "",
|
||||
"remark": i.remark or "",
|
||||
}
|
||||
|
||||
|
||||
def _task_output_to_dict(o: BizProductionTaskOutput) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": o.id,
|
||||
"taskId": o.task_id,
|
||||
"cellProductId": o.cell_product_id or "",
|
||||
"storageSlotId": o.storage_slot_id or "",
|
||||
"outputName": o.output_name or "",
|
||||
"outputBarcode": o.output_barcode or "",
|
||||
"quantity": float(o.quantity) if o.quantity is not None else None,
|
||||
"unit": o.unit or "",
|
||||
"remark": o.remark or "",
|
||||
"attachments": o.attachments or "",
|
||||
}
|
||||
|
||||
|
||||
def _soft_delete_sub(db: Session, model_class, parent_field: str, parent_id: str, now: datetime) -> None:
|
||||
rows = db.query(model_class).filter(
|
||||
and_(
|
||||
getattr(model_class, parent_field) == parent_id,
|
||||
model_class.del_flag == 0,
|
||||
)
|
||||
).all()
|
||||
for r in rows:
|
||||
r.del_flag = 1
|
||||
r.update_time = now
|
||||
db.add(r)
|
||||
|
||||
|
||||
def _clone_step_reagents(
|
||||
db: Session, src_step_id: str, new_step_id: str,
|
||||
tenant_no: str, user_id: Optional[str], now: datetime,
|
||||
) -> None:
|
||||
for r in db.query(BizProcessRouteStepReagent).filter(
|
||||
and_(BizProcessRouteStepReagent.step_id == src_step_id, BizProcessRouteStepReagent.del_flag == 0)
|
||||
).all():
|
||||
db.add(BizProductionTaskStepReagent(
|
||||
id=generate_uuid(), step_id=new_step_id, tenant_no=tenant_no,
|
||||
reagent_code=r.reagent_code, reagent_name=r.reagent_name,
|
||||
spec=r.spec, standard_usage=r.standard_usage,
|
||||
unit=r.unit, is_key_material=r.is_key_material,
|
||||
remark=r.remark,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
|
||||
|
||||
def _clone_step_consumables(
|
||||
db: Session, src_step_id: str, new_step_id: str,
|
||||
tenant_no: str, user_id: Optional[str], now: datetime,
|
||||
) -> None:
|
||||
for c in db.query(BizProcessRouteStepConsumable).filter(
|
||||
and_(BizProcessRouteStepConsumable.step_id == src_step_id, BizProcessRouteStepConsumable.del_flag == 0)
|
||||
).all():
|
||||
db.add(BizProductionTaskStepConsumable(
|
||||
id=generate_uuid(), step_id=new_step_id, tenant_no=tenant_no,
|
||||
consumable_code=c.consumable_code, consumable_name=c.consumable_name,
|
||||
spec_model=c.spec_model, standard_usage=c.standard_usage,
|
||||
unit=c.unit, is_required=c.is_required,
|
||||
remark=c.remark,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
|
||||
|
||||
def _clone_step_outputs(
|
||||
db: Session, src_step_id: str, new_step_id: str,
|
||||
tenant_no: str, user_id: Optional[str], now: datetime,
|
||||
) -> None:
|
||||
for o in db.query(BizProcessRouteStepOutput).filter(
|
||||
and_(BizProcessRouteStepOutput.step_id == src_step_id, BizProcessRouteStepOutput.del_flag == 0)
|
||||
).all():
|
||||
db.add(BizProductionTaskStepOutput(
|
||||
id=generate_uuid(), step_id=new_step_id, tenant_no=tenant_no,
|
||||
output_type_id=o.output_type_id, output_name=o.output_name,
|
||||
output_barcode=o.output_barcode,
|
||||
remark=o.remark,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
|
||||
|
||||
def _clone_step_data_collections(
|
||||
db: Session, src_step_id: str, new_step_id: str,
|
||||
tenant_no: str, user_id: Optional[str], now: datetime,
|
||||
) -> None:
|
||||
for d in db.query(BizProcessRouteStepDataCollection).filter(
|
||||
and_(BizProcessRouteStepDataCollection.step_id == src_step_id, BizProcessRouteStepDataCollection.del_flag == 0)
|
||||
).all():
|
||||
db.add(BizProductionTaskStepDataCollection(
|
||||
id=generate_uuid(), step_id=new_step_id, tenant_no=tenant_no,
|
||||
collection_name=d.collection_name, unit=d.unit,
|
||||
remark=d.remark,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
|
||||
|
||||
def _clone_route_to_task(
|
||||
db: Session, task_id: str, route_id: str,
|
||||
tenant_no: str, user_id: Optional[str], create_dept_id: str,
|
||||
) -> None:
|
||||
route_steps = (
|
||||
db.query(BizProcessRouteStep)
|
||||
.filter(and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0))
|
||||
.order_by(BizProcessRouteStep.step_order.asc(), BizProcessRouteStep.create_time.asc())
|
||||
.all()
|
||||
)
|
||||
now = datetime.now()
|
||||
for rs in route_steps:
|
||||
new_step = BizProductionTaskStep(
|
||||
id=generate_uuid(), task_id=task_id,
|
||||
step_order=rs.step_order,
|
||||
process_code=rs.process_code, category_id=rs.category_id,
|
||||
process_name=rs.process_name, description=rs.description,
|
||||
equipment_name=rs.equipment_name, estimated_duration=rs.estimated_duration,
|
||||
sop_ids=rs.sop_ids,
|
||||
step_status="dzx",
|
||||
tenant_no=tenant_no, order_num=0,
|
||||
remark=None, create_by=user_id, create_dept_id=create_dept_id or "",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
)
|
||||
db.add(new_step)
|
||||
db.flush()
|
||||
|
||||
_clone_step_reagents(db, rs.id, new_step.id, tenant_no, user_id, now)
|
||||
_clone_step_consumables(db, rs.id, new_step.id, tenant_no, user_id, now)
|
||||
_clone_step_outputs(db, rs.id, new_step.id, tenant_no, user_id, now)
|
||||
_clone_step_data_collections(db, rs.id, new_step.id, tenant_no, user_id, now)
|
||||
|
||||
|
||||
def create_production_task(
|
||||
db: Session, *, tenant_no: str, user_id: Optional[str] = None,
|
||||
create_dept_id: str = "", data: Dict[str, Any],
|
||||
) -> BizProductionTask:
|
||||
task_no = _generate_task_no(db, tenant_no)
|
||||
now = datetime.now()
|
||||
task = BizProductionTask(
|
||||
id=generate_uuid(),
|
||||
task_no=task_no,
|
||||
work_order_id=data.get("work_order_id", ""),
|
||||
task_name=(data.get("task_name") or "").strip() or None,
|
||||
description=(data.get("description") or "").strip() or None,
|
||||
route_id=data.get("route_id") or None,
|
||||
task_status="dpz",
|
||||
remark=(data.get("remark") or "").strip() or None,
|
||||
attachments=(data.get("attachments") or "").strip() or None,
|
||||
tenant_no=tenant_no,
|
||||
order_num=0,
|
||||
create_by=user_id,
|
||||
create_dept_id=create_dept_id or "",
|
||||
create_time=now,
|
||||
update_by=user_id,
|
||||
update_time=now,
|
||||
del_flag=0,
|
||||
)
|
||||
db.add(task)
|
||||
db.flush()
|
||||
|
||||
route_id = data.get("route_id")
|
||||
if route_id:
|
||||
_clone_route_to_task(db, task.id, route_id, tenant_no, user_id, create_dept_id)
|
||||
|
||||
db.flush()
|
||||
return task
|
||||
|
||||
|
||||
def list_production_tasks(
|
||||
db: Session, *, tenant_no: str,
|
||||
page: int = 1, page_size: int = 10,
|
||||
work_order_id: Optional[str] = None,
|
||||
task_no: Optional[str] = None,
|
||||
task_status: Optional[str] = None,
|
||||
) -> Tuple[List[Dict[str, Any]], int]:
|
||||
def _filtered():
|
||||
q = db.query(BizProductionTask).filter(
|
||||
BizProductionTask.tenant_no == tenant_no,
|
||||
BizProductionTask.del_flag == 0,
|
||||
)
|
||||
if work_order_id and work_order_id.strip():
|
||||
q = q.filter(BizProductionTask.work_order_id == work_order_id.strip())
|
||||
if task_no and task_no.strip():
|
||||
q = q.filter(BizProductionTask.task_no.like(f"%{task_no.strip()}%"))
|
||||
if task_status and task_status.strip():
|
||||
q = q.filter(BizProductionTask.task_status == task_status.strip())
|
||||
return q
|
||||
|
||||
total = _filtered().order_by(None).with_entities(func.count(BizProductionTask.id)).scalar() or 0
|
||||
page = max(1, page)
|
||||
page_size = min(max(1, page_size), 100)
|
||||
rows = (
|
||||
_filtered()
|
||||
.order_by(BizProductionTask.create_time.desc())
|
||||
.offset((page - 1) * page_size)
|
||||
.limit(page_size)
|
||||
.all()
|
||||
)
|
||||
return [_task_to_dict(x) for x in rows], int(total)
|
||||
|
||||
|
||||
def get_production_task_detail(db: Session, task_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
|
||||
task = db.query(BizProductionTask).filter(
|
||||
and_(
|
||||
BizProductionTask.id == task_id,
|
||||
BizProductionTask.tenant_no == tenant_no,
|
||||
BizProductionTask.del_flag == 0,
|
||||
)
|
||||
).first()
|
||||
if not task:
|
||||
return None
|
||||
|
||||
result = _task_to_dict(task)
|
||||
|
||||
steps = (
|
||||
db.query(BizProductionTaskStep)
|
||||
.filter(and_(BizProductionTaskStep.task_id == task_id, BizProductionTaskStep.del_flag == 0))
|
||||
.order_by(BizProductionTaskStep.step_order.asc(), BizProductionTaskStep.create_time.asc())
|
||||
.all()
|
||||
)
|
||||
step_list = []
|
||||
for s in steps:
|
||||
sd = _step_to_dict(s)
|
||||
reagents = db.query(BizProductionTaskStepReagent).filter(
|
||||
and_(BizProductionTaskStepReagent.step_id == s.id, BizProductionTaskStepReagent.del_flag == 0)
|
||||
).order_by(BizProductionTaskStepReagent.create_time.asc()).all()
|
||||
consumables = db.query(BizProductionTaskStepConsumable).filter(
|
||||
and_(BizProductionTaskStepConsumable.step_id == s.id, BizProductionTaskStepConsumable.del_flag == 0)
|
||||
).order_by(BizProductionTaskStepConsumable.create_time.asc()).all()
|
||||
outputs = db.query(BizProductionTaskStepOutput).filter(
|
||||
and_(BizProductionTaskStepOutput.step_id == s.id, BizProductionTaskStepOutput.del_flag == 0)
|
||||
).order_by(BizProductionTaskStepOutput.create_time.asc()).all()
|
||||
data_collections = db.query(BizProductionTaskStepDataCollection).filter(
|
||||
and_(BizProductionTaskStepDataCollection.step_id == s.id, BizProductionTaskStepDataCollection.del_flag == 0)
|
||||
).order_by(BizProductionTaskStepDataCollection.order_num.asc()).all()
|
||||
|
||||
sd["reagents"] = [_step_reagent_to_dict(r) for r in reagents]
|
||||
sd["consumables"] = [_step_consumable_to_dict(c) for c in consumables]
|
||||
sd["outputs"] = [_step_output_to_dict(o) for o in outputs]
|
||||
sd["dataCollections"] = [_step_data_collection_to_dict(d) for d in data_collections]
|
||||
step_list.append(sd)
|
||||
|
||||
inputs = db.query(BizProductionTaskInput).filter(
|
||||
and_(BizProductionTaskInput.task_id == task_id, BizProductionTaskInput.del_flag == 0)
|
||||
).order_by(BizProductionTaskInput.create_time.asc()).all()
|
||||
|
||||
outputs = db.query(BizProductionTaskOutput).filter(
|
||||
and_(BizProductionTaskOutput.task_id == task_id, BizProductionTaskOutput.del_flag == 0)
|
||||
).order_by(BizProductionTaskOutput.create_time.asc()).all()
|
||||
|
||||
result["steps"] = step_list
|
||||
result["inputs"] = [_task_input_to_dict(i) for i in inputs]
|
||||
result["outputs"] = [_task_output_to_dict(o) for o in outputs]
|
||||
return result
|
||||
|
||||
|
||||
def update_production_task(
|
||||
db: Session, task_id: str, tenant_no: str,
|
||||
user_id: Optional[str], data: Dict[str, Any],
|
||||
) -> BizProductionTask:
|
||||
task = db.query(BizProductionTask).filter(
|
||||
and_(
|
||||
BizProductionTask.id == task_id,
|
||||
BizProductionTask.tenant_no == tenant_no,
|
||||
BizProductionTask.del_flag == 0,
|
||||
)
|
||||
).first()
|
||||
if not task:
|
||||
raise ValueError("任务不存在")
|
||||
for field in ["task_name", "description", "remark", "attachments"]:
|
||||
val = data.get(field)
|
||||
if val is not None:
|
||||
setattr(task, field, (val or "").strip() or None)
|
||||
task.update_by = user_id
|
||||
task.update_time = datetime.now()
|
||||
db.add(task)
|
||||
db.commit()
|
||||
db.refresh(task)
|
||||
return task
|
||||
|
||||
|
||||
def update_task_status(
|
||||
db: Session, task_id: str, tenant_no: str,
|
||||
user_id: Optional[str], status: str,
|
||||
) -> BizProductionTask:
|
||||
task = db.query(BizProductionTask).filter(
|
||||
and_(
|
||||
BizProductionTask.id == task_id,
|
||||
BizProductionTask.tenant_no == tenant_no,
|
||||
BizProductionTask.del_flag == 0,
|
||||
)
|
||||
).first()
|
||||
if not task:
|
||||
raise ValueError("任务不存在")
|
||||
now = datetime.now()
|
||||
task.task_status = status
|
||||
if status == "jxz" and not task.started_time:
|
||||
task.started_time = now
|
||||
if status == "ywc" and not task.completed_time:
|
||||
task.completed_time = now
|
||||
task.update_by = user_id
|
||||
task.update_time = now
|
||||
db.add(task)
|
||||
db.commit()
|
||||
db.refresh(task)
|
||||
return task
|
||||
|
||||
|
||||
def add_task_step(
|
||||
db: Session, task_id: str, tenant_no: str,
|
||||
user_id: Optional[str], data: Dict[str, Any],
|
||||
) -> Dict[str, Any]:
|
||||
task = db.query(BizProductionTask).filter(
|
||||
and_(
|
||||
BizProductionTask.id == task_id,
|
||||
BizProductionTask.tenant_no == tenant_no,
|
||||
BizProductionTask.del_flag == 0,
|
||||
)
|
||||
).first()
|
||||
if not task:
|
||||
raise ValueError("任务不存在")
|
||||
|
||||
now = datetime.now()
|
||||
step_order = data.get("step_order")
|
||||
if step_order is None:
|
||||
max_order = db.query(func.max(BizProductionTaskStep.step_order)).filter(
|
||||
and_(BizProductionTaskStep.task_id == task_id, BizProductionTaskStep.del_flag == 0)
|
||||
).scalar()
|
||||
step_order = (max_order or 0) + 1
|
||||
|
||||
step = BizProductionTaskStep(
|
||||
id=generate_uuid(), task_id=task_id,
|
||||
step_order=step_order,
|
||||
process_code=(data.get("process_code") or "").strip() or None,
|
||||
category_id=(data.get("category_id") or "").strip() or None,
|
||||
process_name=(data.get("process_name") or "").strip() or None,
|
||||
description=(data.get("description") or "").strip() or None,
|
||||
equipment_name=(data.get("equipment_name") or "").strip() or None,
|
||||
estimated_duration=(data.get("estimated_duration") or "").strip() or None,
|
||||
actual_duration=(data.get("actual_duration") or "").strip() or None,
|
||||
sop_ids=(data.get("sop_ids") or "").strip() or None,
|
||||
step_status=(data.get("step_status") or "dzx"),
|
||||
tenant_no=tenant_no, order_num=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(step)
|
||||
db.commit()
|
||||
return _step_to_dict(step)
|
||||
|
||||
|
||||
def update_task_step(
|
||||
db: Session, step_id: str, tenant_no: str,
|
||||
user_id: Optional[str], data: Dict[str, Any],
|
||||
) -> BizProductionTaskStep:
|
||||
step = db.query(BizProductionTaskStep).filter(
|
||||
and_(
|
||||
BizProductionTaskStep.id == step_id,
|
||||
BizProductionTaskStep.tenant_no == tenant_no,
|
||||
BizProductionTaskStep.del_flag == 0,
|
||||
)
|
||||
).first()
|
||||
if not step:
|
||||
raise ValueError("步骤不存在")
|
||||
for field in [
|
||||
"step_order", "process_code", "category_id", "process_name",
|
||||
"description", "equipment_name", "estimated_duration", "actual_duration",
|
||||
"sop_ids", "step_status", "remark",
|
||||
]:
|
||||
val = data.get(field)
|
||||
if val is not None:
|
||||
if field == "step_order":
|
||||
setattr(step, field, int(val))
|
||||
else:
|
||||
setattr(step, field, (val or "").strip() or None)
|
||||
if data.get("started_time"):
|
||||
from dateutil.parser import parse as _parse_dt
|
||||
step.started_time = _parse_dt(data["started_time"])
|
||||
if data.get("completed_time"):
|
||||
from dateutil.parser import parse as _parse_dt
|
||||
step.completed_time = _parse_dt(data["completed_time"])
|
||||
step.update_by = user_id
|
||||
step.update_time = datetime.now()
|
||||
db.add(step)
|
||||
db.commit()
|
||||
db.refresh(step)
|
||||
return step
|
||||
|
||||
|
||||
def delete_task_step(db: Session, step_id: str, tenant_no: str, user_id: Optional[str]) -> None:
|
||||
step = db.query(BizProductionTaskStep).filter(
|
||||
and_(
|
||||
BizProductionTaskStep.id == step_id,
|
||||
BizProductionTaskStep.tenant_no == tenant_no,
|
||||
BizProductionTaskStep.del_flag == 0,
|
||||
)
|
||||
).first()
|
||||
if not step:
|
||||
raise ValueError("步骤不存在")
|
||||
now = datetime.now()
|
||||
step.del_flag = 1
|
||||
step.update_by = user_id
|
||||
step.update_time = now
|
||||
db.add(step)
|
||||
_soft_delete_sub(db, BizProductionTaskStepReagent, "step_id", step_id, now)
|
||||
_soft_delete_sub(db, BizProductionTaskStepConsumable, "step_id", step_id, now)
|
||||
_soft_delete_sub(db, BizProductionTaskStepOutput, "step_id", step_id, now)
|
||||
_soft_delete_sub(db, BizProductionTaskStepDataCollection, "step_id", step_id, now)
|
||||
db.commit()
|
||||
|
||||
|
||||
def replace_step_reagents(
|
||||
db: Session, step_id: str, tenant_no: str,
|
||||
user_id: Optional[str], items: List[Dict[str, Any]],
|
||||
) -> None:
|
||||
now = datetime.now()
|
||||
_soft_delete_sub(db, BizProductionTaskStepReagent, "step_id", step_id, now)
|
||||
for it in items:
|
||||
db.add(BizProductionTaskStepReagent(
|
||||
id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
|
||||
reagent_code=(it.get("reagent_code") or "").strip() or None,
|
||||
reagent_name=(it.get("reagent_name") or "").strip() or None,
|
||||
spec=(it.get("spec") or "").strip() or None,
|
||||
standard_usage=(it.get("standard_usage") or "").strip() or None,
|
||||
actual_usage=(it.get("actual_usage") or "").strip() or None,
|
||||
unit=(it.get("unit") or "").strip() or None,
|
||||
is_key_material=it.get("is_key_material", "0"),
|
||||
remark=(it.get("remark") or "").strip() or None,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def replace_step_consumables(
|
||||
db: Session, step_id: str, tenant_no: str,
|
||||
user_id: Optional[str], items: List[Dict[str, Any]],
|
||||
) -> None:
|
||||
now = datetime.now()
|
||||
_soft_delete_sub(db, BizProductionTaskStepConsumable, "step_id", step_id, now)
|
||||
for it in items:
|
||||
db.add(BizProductionTaskStepConsumable(
|
||||
id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
|
||||
consumable_code=(it.get("consumable_code") or "").strip() or None,
|
||||
consumable_name=(it.get("consumable_name") or "").strip() or None,
|
||||
spec_model=(it.get("spec_model") or "").strip() or None,
|
||||
standard_usage=(it.get("standard_usage") or "").strip() or None,
|
||||
actual_usage=(it.get("actual_usage") or "").strip() or None,
|
||||
unit=(it.get("unit") or "").strip() or None,
|
||||
is_required=it.get("is_required", "1"),
|
||||
remark=(it.get("remark") or "").strip() or None,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def replace_step_outputs(
|
||||
db: Session, step_id: str, tenant_no: str,
|
||||
user_id: Optional[str], items: List[Dict[str, Any]],
|
||||
) -> None:
|
||||
now = datetime.now()
|
||||
_soft_delete_sub(db, BizProductionTaskStepOutput, "step_id", step_id, now)
|
||||
for it in items:
|
||||
db.add(BizProductionTaskStepOutput(
|
||||
id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
|
||||
output_type_id=(it.get("output_type_id") or "").strip() or None,
|
||||
output_name=(it.get("output_name") or "").strip() or None,
|
||||
output_barcode=(it.get("output_barcode") or "").strip() or None,
|
||||
remark=(it.get("remark") or "").strip() or None,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def replace_step_data_collections(
|
||||
db: Session, step_id: str, tenant_no: str,
|
||||
user_id: Optional[str], items: List[Dict[str, Any]],
|
||||
) -> None:
|
||||
now = datetime.now()
|
||||
_soft_delete_sub(db, BizProductionTaskStepDataCollection, "step_id", step_id, now)
|
||||
for it in items:
|
||||
db.add(BizProductionTaskStepDataCollection(
|
||||
id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
|
||||
collection_name=(it.get("collection_name") or "").strip() or None,
|
||||
unit=(it.get("unit") or "").strip() or None,
|
||||
collection_value=(it.get("collection_value") or "").strip() or None,
|
||||
remark=(it.get("remark") or "").strip() or None,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def set_task_inputs(
|
||||
db: Session, task_id: str, tenant_no: str,
|
||||
user_id: Optional[str], items: List[Dict[str, Any]],
|
||||
) -> None:
|
||||
now = datetime.now()
|
||||
_soft_delete_sub(db, BizProductionTaskInput, "task_id", task_id, now)
|
||||
for it in items:
|
||||
db.add(BizProductionTaskInput(
|
||||
id=generate_uuid(), task_id=task_id, tenant_no=tenant_no,
|
||||
input_type=(it.get("input_type") or "").strip() or None,
|
||||
input_id=(it.get("input_id") or "").strip() or None,
|
||||
input_code=(it.get("input_code") or "").strip() or None,
|
||||
input_name=(it.get("input_name") or "").strip() or None,
|
||||
remark=(it.get("remark") or "").strip() or None,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def set_task_outputs(
|
||||
db: Session, task_id: str, tenant_no: str,
|
||||
user_id: Optional[str], items: List[Dict[str, Any]],
|
||||
) -> None:
|
||||
now = datetime.now()
|
||||
_soft_delete_sub(db, BizProductionTaskOutput, "task_id", task_id, now)
|
||||
for it in items:
|
||||
quantity = it.get("quantity")
|
||||
db.add(BizProductionTaskOutput(
|
||||
id=generate_uuid(), task_id=task_id, tenant_no=tenant_no,
|
||||
cell_product_id=(it.get("cell_product_id") or "").strip() or None,
|
||||
storage_slot_id=(it.get("storage_slot_id") or "").strip() or None,
|
||||
output_name=(it.get("output_name") or "").strip() or None,
|
||||
output_barcode=(it.get("output_barcode") or "").strip() or None,
|
||||
quantity=quantity if quantity is not None else None,
|
||||
unit=(it.get("unit") or "").strip() or None,
|
||||
remark=(it.get("remark") or "").strip() or None,
|
||||
attachments=(it.get("attachments") or "").strip() or None,
|
||||
order_num=0, create_by=user_id, create_dept_id="",
|
||||
create_time=now, update_by=user_id, update_time=now, del_flag=0,
|
||||
))
|
||||
db.commit()
|
||||
|
||||
|
||||
def total_pages(total: int, page_size: int) -> int:
|
||||
if total <= 0:
|
||||
return 0
|
||||
return int(math.ceil(total / float(page_size)))
|
||||
Reference in new issue
Block a user