feat: implement production task set CRUD and batch dispatch

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/crud/production_task_set.py

@@ -1,8 +1,21 @@
 """生产任务集合 CRUD"""
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any, Dict, List, Optional, Set, Tuple
 
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.models import (
+    BizProcessRoute,
+    BizProductionTask,
+    BizProductionTaskSet,
+    BizProductionTaskSetItem,
+    BizProductionWorkOrder,
+)
+
 
 def _strip_or_none(raw: Any) -> Optional[str]:
     text = str(raw or "").strip()
@@ -56,3 +69,539 @@ def collect_missing_route_ids(items: List[Dict[str, Any]], existing_ids: Set[str
         seen.add(rid)
         missing.append(rid)
     return missing
+
+
+def total_pages(total: int, page_size: int) -> int:
+    if page_size <= 0:
+        return 0
+    return (int(total) + int(page_size) - 1) // int(page_size)
+
+
+def _item_to_dict(r: BizProductionTaskSetItem) -> Dict[str, Any]:
+    return {
+        "id": r.id,
+        "setId": r.set_id,
+        "taskName": r.task_name or "",
+        "routeId": r.route_id or "",
+        "routeName": r.route_name or "",
+        "description": r.description or "",
+        "remark": r.remark or "",
+        "orderNum": r.order_num if r.order_num is not None else 0,
+    }
+
+
+def _set_to_dict(
+    r: BizProductionTaskSet,
+    item_count: int = 0,
+    items: Optional[List[Dict[str, Any]]] = None,
+) -> Dict[str, Any]:
+    data: Dict[str, Any] = {
+        "id": r.id,
+        "setName": r.set_name or "",
+        "status": r.status or "1",
+        "itemCount": item_count,
+        "orderNum": r.order_num if r.order_num is not None else 0,
+        "remark": r.remark or "",
+        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
+        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
+    }
+    if items is not None:
+        data["items"] = items
+    return data
+
+
+def set_name_exists(
+    db: Session, name: str, tenant_no: str, exclude_id: Optional[str] = None
+) -> bool:
+    n = (name or "").strip()
+    if not n:
+        return False
+    q = db.query(BizProductionTaskSet).filter(
+        BizProductionTaskSet.set_name == n,
+        BizProductionTaskSet.tenant_no == tenant_no,
+        BizProductionTaskSet.del_flag == 0,
+    )
+    if exclude_id:
+        q = q.filter(BizProductionTaskSet.id != exclude_id)
+    return q.first() is not None
+
+
+def _require_status(status: Optional[str]) -> str:
+    st = (status or "1").strip()
+    if st not in ("0", "1"):
+        raise ValueError("状态无效")
+    return st
+
+
+def _load_route_name_map(db: Session, tenant_no: str, route_ids: List[str]) -> Dict[str, str]:
+    ids = [rid for rid in route_ids if rid]
+    if not ids:
+        return {}
+    rows = (
+        db.query(BizProcessRoute)
+        .filter(
+            BizProcessRoute.tenant_no == tenant_no,
+            BizProcessRoute.del_flag == 0,
+            BizProcessRoute.id.in_(ids),
+        )
+        .all()
+    )
+    return {r.id: (r.route_name or "") for r in rows}
+
+
+def _normalize_items_or_raise(raw_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    out: List[Dict[str, Any]] = []
+    for raw in raw_items or []:
+        item = normalize_item_payload(raw)
+        if item:
+            out.append(item)
+    if not out:
+        raise ValueError("请至少添加一条任务名称非空的条目")
+    return out
+
+
+def _assert_routes_exist(
+    db: Session, tenant_no: str, items: List[Dict[str, Any]]
+) -> Dict[str, str]:
+    wanted = [i["route_id"] for i in items if i.get("route_id")]
+    name_map = _load_route_name_map(db, tenant_no, wanted)
+    missing = collect_missing_route_ids(items, set(name_map.keys()))
+    if missing:
+        raise ValueError("存在无效或已删除的工艺路线")
+    return name_map
+
+
+def _replace_items(
+    db: Session,
+    *,
+    set_id: str,
+    tenant_no: str,
+    items: List[Dict[str, Any]],
+    route_names: Dict[str, str],
+    user_id: Optional[str],
+    create_dept_id: str,
+) -> None:
+    now = datetime.now()
+    existing = (
+        db.query(BizProductionTaskSetItem)
+        .filter(
+            BizProductionTaskSetItem.set_id == set_id,
+            BizProductionTaskSetItem.tenant_no == tenant_no,
+            BizProductionTaskSetItem.del_flag == 0,
+        )
+        .all()
+    )
+    for row in existing:
+        row.del_flag = 1
+        row.update_by = user_id
+        row.update_time = now
+        db.add(row)
+    for idx, item in enumerate(items, start=1):
+        rid = item.get("route_id")
+        db.add(
+            BizProductionTaskSetItem(
+                id=generate_uuid(),
+                set_id=set_id,
+                task_name=item["task_name"],
+                route_id=rid,
+                route_name=route_names.get(rid or "") if rid else None,
+                description=item.get("description"),
+                remark=item.get("remark"),
+                tenant_no=tenant_no,
+                order_num=idx,
+                create_by=user_id,
+                create_dept_id=create_dept_id or "",
+                create_time=now,
+                update_by=user_id,
+                update_time=now,
+                del_flag=0,
+            )
+        )
+
+
+def _item_count_map(db: Session, tenant_no: str, set_ids: List[str]) -> Dict[str, int]:
+    if not set_ids:
+        return {}
+    rows = (
+        db.query(BizProductionTaskSetItem.set_id, func.count(BizProductionTaskSetItem.id))
+        .filter(
+            BizProductionTaskSetItem.tenant_no == tenant_no,
+            BizProductionTaskSetItem.del_flag == 0,
+            BizProductionTaskSetItem.set_id.in_(set_ids),
+        )
+        .group_by(BizProductionTaskSetItem.set_id)
+        .all()
+    )
+    return {sid: int(cnt) for sid, cnt in rows}
+
+
+def list_sets(
+    db: Session,
+    *,
+    tenant_no: str,
+    page: int = 1,
+    page_size: int = 10,
+    keyword: Optional[str] = None,
+    status: Optional[str] = None,
+) -> Tuple[List[Dict[str, Any]], int]:
+    q = db.query(BizProductionTaskSet).filter(
+        BizProductionTaskSet.tenant_no == tenant_no,
+        BizProductionTaskSet.del_flag == 0,
+    )
+    kw = (keyword or "").strip()
+    if kw:
+        q = q.filter(BizProductionTaskSet.set_name.like(f"%{kw}%"))
+    st = (status or "").strip()
+    if st in ("0", "1"):
+        q = q.filter(BizProductionTaskSet.status == st)
+    total = q.count()
+    rows = (
+        q.order_by(BizProductionTaskSet.create_time.desc())
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+    counts = _item_count_map(db, tenant_no, [r.id for r in rows])
+    return [_set_to_dict(r, item_count=counts.get(r.id, 0)) for r in rows], int(total)
+
+
+def list_active_sets(db: Session, *, tenant_no: str) -> List[Dict[str, Any]]:
+    rows = (
+        db.query(BizProductionTaskSet)
+        .filter(
+            BizProductionTaskSet.tenant_no == tenant_no,
+            BizProductionTaskSet.del_flag == 0,
+            BizProductionTaskSet.status == "1",
+        )
+        .order_by(BizProductionTaskSet.create_time.desc())
+        .all()
+    )
+    counts = _item_count_map(db, tenant_no, [r.id for r in rows])
+    return [_set_to_dict(r, item_count=counts.get(r.id, 0)) for r in rows]
+
+
+def get_set(db: Session, set_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
+    row = (
+        db.query(BizProductionTaskSet)
+        .filter(
+            BizProductionTaskSet.id == set_id,
+            BizProductionTaskSet.tenant_no == tenant_no,
+            BizProductionTaskSet.del_flag == 0,
+        )
+        .first()
+    )
+    if not row:
+        return None
+    items = (
+        db.query(BizProductionTaskSetItem)
+        .filter(
+            BizProductionTaskSetItem.set_id == set_id,
+            BizProductionTaskSetItem.tenant_no == tenant_no,
+            BizProductionTaskSetItem.del_flag == 0,
+        )
+        .order_by(
+            BizProductionTaskSetItem.order_num.asc(),
+            BizProductionTaskSetItem.create_time.asc(),
+        )
+        .all()
+    )
+    return _set_to_dict(row, item_count=len(items), items=[_item_to_dict(i) for i in items])
+
+
+def create_set(
+    db: Session,
+    *,
+    tenant_no: str,
+    user_id: Optional[str] = None,
+    create_dept_id: str = "",
+    set_name: str,
+    status: str = "1",
+    remark: Optional[str] = None,
+    items: Optional[List[Dict[str, Any]]] = None,
+) -> Dict[str, Any]:
+    name = (set_name or "").strip()
+    if not name:
+        raise ValueError("集合名称不能为空")
+    if set_name_exists(db, name, tenant_no):
+        raise ValueError("集合名称已存在")
+    normalized = _normalize_items_or_raise(items or [])
+    route_names = _assert_routes_exist(db, tenant_no, normalized)
+    now = datetime.now()
+    obj = BizProductionTaskSet(
+        id=generate_uuid(),
+        set_name=name,
+        status=_require_status(status),
+        tenant_no=tenant_no,
+        order_num=0,
+        remark=(remark or "").strip() or 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(obj)
+    db.flush()
+    _replace_items(
+        db,
+        set_id=obj.id,
+        tenant_no=tenant_no,
+        items=normalized,
+        route_names=route_names,
+        user_id=user_id,
+        create_dept_id=create_dept_id,
+    )
+    db.flush()
+    detail = get_set(db, obj.id, tenant_no)
+    if not detail:
+        raise ValueError("集合创建失败")
+    return detail
+
+
+def update_set(
+    db: Session,
+    set_id: str,
+    *,
+    tenant_no: str,
+    user_id: Optional[str] = None,
+    set_name: Optional[str] = None,
+    status: Optional[str] = None,
+    remark: Optional[str] = None,
+    items: Optional[List[Dict[str, Any]]] = None,
+) -> Dict[str, Any]:
+    obj = (
+        db.query(BizProductionTaskSet)
+        .filter(
+            BizProductionTaskSet.id == set_id,
+            BizProductionTaskSet.tenant_no == tenant_no,
+            BizProductionTaskSet.del_flag == 0,
+        )
+        .first()
+    )
+    if not obj:
+        raise ValueError("集合不存在")
+    if set_name is not None:
+        name = set_name.strip()
+        if not name:
+            raise ValueError("集合名称不能为空")
+        if set_name_exists(db, name, tenant_no, exclude_id=set_id):
+            raise ValueError("集合名称已存在")
+        obj.set_name = name
+    if status is not None:
+        obj.status = _require_status(status)
+    if remark is not None:
+        obj.remark = remark.strip() or None
+    if items is not None:
+        normalized = _normalize_items_or_raise(items)
+        route_names = _assert_routes_exist(db, tenant_no, normalized)
+        _replace_items(
+            db,
+            set_id=obj.id,
+            tenant_no=tenant_no,
+            items=normalized,
+            route_names=route_names,
+            user_id=user_id,
+            create_dept_id=obj.create_dept_id or "",
+        )
+    obj.update_by = user_id
+    obj.update_time = datetime.now()
+    db.add(obj)
+    db.flush()
+    detail = get_set(db, obj.id, tenant_no)
+    if not detail:
+        raise ValueError("集合不存在")
+    return detail
+
+
+def delete_set(db: Session, set_id: str, *, tenant_no: str, user_id: Optional[str] = None) -> None:
+    obj = (
+        db.query(BizProductionTaskSet)
+        .filter(
+            BizProductionTaskSet.id == set_id,
+            BizProductionTaskSet.tenant_no == tenant_no,
+            BizProductionTaskSet.del_flag == 0,
+        )
+        .first()
+    )
+    if not obj:
+        raise ValueError("集合不存在")
+    now = datetime.now()
+    obj.del_flag = 1
+    obj.update_by = user_id
+    obj.update_time = now
+    db.add(obj)
+    items = (
+        db.query(BizProductionTaskSetItem)
+        .filter(
+            BizProductionTaskSetItem.set_id == set_id,
+            BizProductionTaskSetItem.del_flag == 0,
+        )
+        .all()
+    )
+    for row in items:
+        row.del_flag = 1
+        row.update_by = user_id
+        row.update_time = now
+        db.add(row)
+    db.flush()
+
+
+def create_set_from_work_order(
+    db: Session,
+    *,
+    tenant_no: str,
+    user_id: Optional[str] = None,
+    create_dept_id: str = "",
+    work_order_id: str,
+    set_name: str,
+    remark: Optional[str] = None,
+) -> Dict[str, Any]:
+    wo_id = (work_order_id or "").strip()
+    wo = (
+        db.query(BizProductionWorkOrder)
+        .filter(
+            BizProductionWorkOrder.id == wo_id,
+            BizProductionWorkOrder.tenant_no == tenant_no,
+            BizProductionWorkOrder.del_flag == 0,
+        )
+        .first()
+    )
+    if not wo:
+        raise ValueError("工单不存在")
+    tasks = (
+        db.query(BizProductionTask)
+        .filter(
+            BizProductionTask.work_order_id == wo_id,
+            BizProductionTask.tenant_no == tenant_no,
+            BizProductionTask.del_flag == 0,
+        )
+        .order_by(BizProductionTask.create_time.asc(), BizProductionTask.id.asc())
+        .all()
+    )
+    raw = [
+        {
+            "task_name": t.task_name,
+            "route_id": t.route_id,
+            "description": t.description,
+            "remark": t.remark,
+        }
+        for t in tasks
+    ]
+    kept, skipped = skip_empty_named_tasks(raw)
+    if not kept:
+        raise ValueError("无可保存的生产任务（任务名称均为空）")
+    route_ids = [i["route_id"] for i in kept if i.get("route_id")]
+    name_map = _load_route_name_map(db, tenant_no, route_ids)
+    payload_items = []
+    for i in kept:
+        rid = i.get("route_id")
+        if rid and rid not in name_map:
+            rid = None
+        payload_items.append(
+            {
+                "taskName": i["task_name"],
+                "routeId": rid,
+                "description": i.get("description"),
+                "remark": i.get("remark"),
+            }
+        )
+    detail = create_set(
+        db,
+        tenant_no=tenant_no,
+        user_id=user_id,
+        create_dept_id=create_dept_id,
+        set_name=set_name,
+        status="1",
+        remark=remark,
+        items=payload_items,
+    )
+    return {**detail, "skippedEmptyNameCount": skipped}
+
+
+def create_tasks_from_set(
+    db: Session,
+    *,
+    tenant_no: str,
+    user_id: Optional[str] = None,
+    create_dept_id: str = "",
+    work_order_id: str,
+    set_id: str,
+) -> Dict[str, Any]:
+    from app.crud import production_task as production_task_svc
+
+    wo_id = (work_order_id or "").strip()
+    wo = (
+        db.query(BizProductionWorkOrder)
+        .filter(
+            BizProductionWorkOrder.id == wo_id,
+            BizProductionWorkOrder.tenant_no == tenant_no,
+            BizProductionWorkOrder.del_flag == 0,
+        )
+        .first()
+    )
+    if not wo:
+        raise ValueError("工单不存在")
+    status = getattr(wo, "work_order_status", None) or ""
+    if status not in ("dpz", "jxz"):
+        raise ValueError("仅待配置或进行中的工单可派发任务")
+    header = (
+        db.query(BizProductionTaskSet)
+        .filter(
+            BizProductionTaskSet.id == (set_id or "").strip(),
+            BizProductionTaskSet.tenant_no == tenant_no,
+            BizProductionTaskSet.del_flag == 0,
+        )
+        .first()
+    )
+    if not header:
+        raise ValueError("集合不存在")
+    if (header.status or "1") != "1":
+        raise ValueError("集合已停用")
+    items = (
+        db.query(BizProductionTaskSetItem)
+        .filter(
+            BizProductionTaskSetItem.set_id == header.id,
+            BizProductionTaskSetItem.tenant_no == tenant_no,
+            BizProductionTaskSetItem.del_flag == 0,
+        )
+        .order_by(
+            BizProductionTaskSetItem.order_num.asc(),
+            BizProductionTaskSetItem.create_time.asc(),
+        )
+        .all()
+    )
+    if not items:
+        raise ValueError("集合没有可派发的任务")
+    payloads = [
+        {
+            "route_id": row.route_id,
+            "task_name": row.task_name,
+            "description": row.description,
+            "remark": row.remark,
+        }
+        for row in items
+    ]
+    name_map = _load_route_name_map(
+        db, tenant_no, [p["route_id"] for p in payloads if p.get("route_id")]
+    )
+    missing = collect_missing_route_ids(payloads, set(name_map.keys()))
+    if missing:
+        raise ValueError("工艺路线已删除，无法派发：" + "、".join(missing))
+    created = []
+    for row in items:
+        task = production_task_svc.create_production_task(
+            db,
+            tenant_no=tenant_no,
+            user_id=user_id,
+            create_dept_id=create_dept_id,
+            data={
+                "work_order_id": wo_id,
+                "task_name": row.task_name,
+                "route_id": row.route_id,
+                "description": row.description,
+                "remark": row.remark,
+            },
+        )
+        created.append(production_task_svc._task_to_dict(task))
+    return {"createdCount": len(created), "list": created}