diff --git a/backend/app/crud/production_task.py b/backend/app/crud/production_task.py new file mode 100644 index 000000000..08fdc3d3f --- /dev/null +++ b/backend/app/crud/production_task.py @@ -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)))