"""任务服务 — 核心业务逻辑：接收、驳回返工、裂变转交、无限嵌套子任务""" from __future__ import annotations import uuid from fastapi import HTTPException, status from sqlalchemy import select, delete from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.models.task import Task, TaskRecord, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED from app.core.time_utils import get_beijing_time from app.models.product import Product from app.models.task_log import TaskLog from app.schemas.task import ( TaskCreate, TaskUpdate, TaskCompleteRequest, TaskRejectRequest, TaskTransferRequest, SubtaskCreate, TaskRecordCreate, TaskRecordResponse, TaskResponse, TaskCompleteResponse, TaskTransferResponse, TaskSummaryResponse, TaskListResponse, ) # 特殊位置常量 VIRTUAL_WAREHOUSE = "virtual_warehouse" # ============================================================ # 内部辅助函数 # ============================================================ async def _get_task_or_404(db: AsyncSession, task_id: uuid.UUID) -> Task: """获取任务，不存在则 404""" result = await db.execute( select(Task) .options( selectinload(Task.child_tasks), selectinload(Task.parent_task), selectinload(Task.product), selectinload(Task.records), ) .where(Task.id == task_id) ) task = result.scalar_one_or_none() if not task: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"任务不存在: {task_id}", ) return task async def _get_task_with_children_recursive(db: AsyncSession, task_id: uuid.UUID) -> Task: """递归加载任务及其所有子孙任务""" result = await db.execute( select(Task) .options( selectinload(Task.child_tasks), selectinload(Task.records), ) .where(Task.id == task_id) ) task = result.scalar_one_or_none() if not task: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"任务不存在: {task_id}", ) # 递归加载每一层子任务 async def _load_children(t: Task): for child in t.child_tasks: child_result = await db.execute( select(Task) .options( selectinload(Task.child_tasks), selectinload(Task.records), ) .where(Task.id == child.id) ) refreshed_child = child_result.scalar_one() t.child_tasks[t.child_tasks.index(child)] = refreshed_child await _load_children(refreshed_child) await _load_children(task) return task def _to_flat_response(task: Task) -> TaskResponse: """扁平序列化，不递归 children（避免 MissingGreenlet）""" product_sn = "" product_material = "" try: if task.product: product_sn = task.product.serial_number or "" product_material = (task.product.material_name or task.product.material_id or "") except Exception: pass return TaskResponse( id=task.id, product_id=task.product_id, product_sn=product_sn, product_material=product_material, parent_task_id=task.parent_task_id, task_name=task.task_name, assignee_id=task.assignee_id, status=task.status, notify_parent_on_complete=task.notify_parent_on_complete, is_rework=task.is_rework, remark=task.remark, reject_reason=task.reject_reason, received_at=task.received_at, completed_at=task.completed_at, created_at=task.created_at, child_tasks=[], records=[TaskRecordResponse.model_validate(r) for r in (task.records or [])], ) def _to_response(task: Task) -> TaskResponse: """将 Task ORM 对象转为递归 TaskResponse""" product_sn = "" product_material = "" try: if task.product: product_sn = task.product.serial_number or "" product_material = (task.product.material_name or task.product.material_id or "") except Exception: pass return TaskResponse( id=task.id, product_id=task.product_id, product_sn=product_sn, product_material=product_material, parent_task_id=task.parent_task_id, task_name=task.task_name, assignee_id=task.assignee_id, status=task.status, notify_parent_on_complete=task.notify_parent_on_complete, is_rework=task.is_rework, remark=task.remark, reject_reason=task.reject_reason, received_at=task.received_at, completed_at=task.completed_at, created_at=task.created_at, child_tasks=[_to_response(c) for c in task.child_tasks], records=[TaskRecordResponse.model_validate(r) for r in (task.records or [])], ) async def _check_all_critical_children_completed( db: AsyncSession, task_id: uuid.UUID ) -> tuple[bool, list[str]]: """ 检查指定任务下所有 notify_parent_on_complete=True 的子任务是否都已完成。 返回 (是否全部完成, 未完成的子任务名称列表)。 """ result = await db.execute( select(Task).where( Task.parent_task_id == task_id, Task.notify_parent_on_complete.is_(True), ) ) critical_children = result.scalars().all() incomplete = [ child.task_name for child in critical_children if child.status != TASK_STATUS_COMPLETED ] return len(incomplete) == 0, incomplete async def _create_task_log( db: AsyncSession, task_id: uuid.UUID, action_type: str, operator_id: str | None = None, remark: str | None = None, ) -> TaskLog: """创建任务操作日志""" log = TaskLog( task_id=task_id, operator_id=operator_id, action_type=action_type, remark=remark, ) db.add(log) return log # ============================================================ # 公开 API — 查询 # ============================================================ async def get_task(db: AsyncSession, task_id: uuid.UUID) -> TaskResponse: """获取任务详情 — 递归包含所有子任务""" task = await _get_task_with_children_recursive(db, task_id) return _to_response(task) async def get_top_level_tasks(db: AsyncSession, product_id: uuid.UUID) -> list[TaskSummaryResponse]: """获取产品的顶层任务列表""" result = await db.execute( select(Task) .where( Task.product_id == product_id, Task.parent_task_id.is_(None), ) .order_by(Task.created_at) ) tasks = result.scalars().all() return [TaskSummaryResponse.model_validate(t) for t in tasks] async def create_task(db: AsyncSession, data: TaskCreate) -> TaskResponse: """创建任务，并同步产品宏观状态""" task = Task(**data.model_dump()) db.add(task) # 同步产品宏观状态 + 当前位置 product_result = await db.execute(select(Product).where(Product.id == data.product_id)) product = product_result.scalar_one_or_none() if product: if data.task_name: product.overall_status = "在库" if "virtual_warehouse" in data.task_name else data.task_name # 派发给人 → 产品离开仓库 if data.assignee_id and data.assignee_id != VIRTUAL_WAREHOUSE: product.current_location_id = data.assignee_id await db.commit() await db.refresh(task) return _to_response(task) async def update_task(db: AsyncSession, task_id: uuid.UUID, data: TaskUpdate) -> TaskResponse: """更新任务""" task = await _get_task_or_404(db, task_id) update_data = data.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(task, field, value) await db.commit() await db.refresh(task) return _to_response(task) async def get_all_tasks( db: AsyncSession, product_id: uuid.UUID | None = None, assignee_id: str | None = None, skip: int = 0, limit: int = 50 ) -> TaskListResponse: """获取任务列表，可按产品/负责人筛选""" stmt = select(Task).options( selectinload(Task.child_tasks), selectinload(Task.records), selectinload(Task.product), ) if product_id: stmt = stmt.where(Task.product_id == product_id) if assignee_id: stmt = stmt.where(Task.assignee_id == assignee_id) stmt = stmt.offset(skip).limit(limit).order_by(Task.created_at.desc()) result = await db.execute(stmt) tasks = result.scalars().all() # 返回扁平列表（不递归 children，避免 MissingGreenlet） flat_tasks = [_to_flat_response(t) for t in tasks] return TaskListResponse(tasks=flat_tasks, total=len(flat_tasks)) # ============================================================ # 核心业务 0：结束分支（终止当前节点） # ============================================================ async def end_task( db: AsyncSession, task_id: uuid.UUID, operator_id: str | None = None ) -> TaskResponse: """ 结束当前分支：标记任务为 COMPLETED，不创建下游任务。 用于工序已完结、无需转交下一人的场景。 """ task = await _get_task_or_404(db, task_id) # 校验：必须等待所有协助分支完成 await _check_children_done(db, task_id) if task.status == TASK_STATUS_COMPLETED: raise HTTPException(status_code=409, detail="此分支已经结束") if task.status == TASK_STATUS_PENDING: raise HTTPException(status_code=409, detail="请先接收任务再结束分支") now = get_beijing_time() task.status = TASK_STATUS_COMPLETED task.completed_at = now await _create_task_log(db, task_id, action_type="end", operator_id=operator_id, remark=f"分支「{task.task_name}」已终止（无下游）") await db.commit() await db.refresh(task) return _to_response(task) # ============================================================ # 核心业务 0.3：撤回转交 # ============================================================ async def recall_task( db: AsyncSession, task_id: uuid.UUID, operator_id: str | None = None ) -> TaskResponse: """撤回 PENDING 转交任务，恢复父任务为 WIP。""" task = await _get_task_or_404(db, task_id) if task.status != TASK_STATUS_PENDING: raise HTTPException(status_code=409, detail="只有待接收(PENDING)的任务可以撤回") if not task.parent_task_id: raise HTTPException(status_code=400, detail="顶层任务无法撤回") parent = await _get_task_or_404(db, task.parent_task_id) if parent.status != TASK_STATUS_COMPLETED: raise HTTPException(status_code=409, detail="父任务状态异常，无法撤回") # 恢复父任务 parent.status = TASK_STATUS_WIP parent.completed_at = None # 更新产品位置到父任务负责人 product_result = await db.execute(select(Product).where(Product.id == parent.product_id)) product = product_result.scalar_one_or_none() if product and parent.assignee_id: product.current_location_id = parent.assignee_id product.overall_status = parent.task_name # 先删除关联的 task_logs 和 task_records await db.execute(delete(TaskLog).where(TaskLog.task_id == task_id)) await db.execute(delete(TaskRecord).where(TaskRecord.task_id == task_id)) await db.flush() # 删除 PENDING 子任务 await db.delete(task) await _create_task_log(db, parent.id, action_type="recall", operator_id=operator_id, remark=f"撤回转交「{task.task_name}」→ {task.assignee_id}，恢复父任务为 WIP") await db.commit() await db.refresh(parent) return _to_response(parent) # ============================================================ # 核心业务 0.5：派发协助分支（不改变父任务状态） # ============================================================ async def spawn_subtask( db: AsyncSession, task_id: uuid.UUID, data, operator_id: str | None = None ) -> TaskResponse: """在当前任务下创建并行子任务，父任务状态保持不变。""" task = await _get_task_or_404(db, task_id) if task.status == TASK_STATUS_COMPLETED: raise HTTPException(status_code=409, detail="任务已完成，无法派发协助分支") if task.status == TASK_STATUS_REJECTED: raise HTTPException(status_code=409, detail="任务已驳回，无法派发协助分支") child = Task( product_id=task.product_id, parent_task_id=task.id, task_name=data.task_name, assignee_id=data.assignee_id, status=TASK_STATUS_PENDING, notify_parent_on_complete=False, is_rework=False, remark=data.remark or None, ) db.add(child) await db.flush() # 协助分支不改变产品宏观状态（只有主分支影响 overall_status） await _create_task_log(db, child.id, action_type="create", operator_id=operator_id, remark=f"协助分支（由「{task.task_name}」派发，分配给 {data.assignee_id}）") await db.commit() await db.refresh(child) return _to_response(child) async def _check_children_done(db: AsyncSession, task_id: uuid.UUID): """检查当前任务的所有子任务是否都已完结。未完结则抛出 409。""" result = await db.execute( select(Task).where(Task.parent_task_id == task_id) ) children = result.scalars().all() incomplete = [c for c in children if c.status not in ( TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED, ARCHIVED_STATUS )] if incomplete: names = "、".join(c.task_name for c in incomplete) raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"当前工序还有未完成的协助分支（{names}），必须等待分支结束才能转交或完工！", ) return True ARCHIVED_STATUS = "ARCHIVED" # ============================================================ # 核心业务 1：确认接收 (PENDING → WIP) # ============================================================ async def receive_task( db: AsyncSession, task_id: uuid.UUID, operator_id: str | None = None, remark: str | None = None, task_name: str | None = None, ) -> TaskResponse: """ 操作员确认接收任务。工人选定工序名称后接收。 校验：只有状态为 PENDING 的任务可接收。 动作：状态改为 WIP，记录 received_at，更新 task_name，同步产品宏观状态。 """ task = await _get_task_or_404(db, task_id) # 权限校验：只有负责人本人可接收 if operator_id and task.assignee_id and operator_id != task.assignee_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"您无权操作此任务，当前任务负责人为 {task.assignee_id}", ) # 校验：只有 PENDING 状态可接收 if task.status != TASK_STATUS_PENDING: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"只有待接收(PENDING)状态的任务可接收，当前状态: {task.status}", ) now = get_beijing_time() task.status = TASK_STATUS_WIP task.received_at = now if remark: task.remark = remark if task_name: task.task_name = task_name await _create_task_log( db, task_id, action_type="receive", operator_id=operator_id, remark=remark or f"操作员确认接收任务「{task.task_name}」", ) # 接收时同步产品位置到接收人 + 宏观状态同步 product_result = await db.execute(select(Product).where(Product.id == task.product_id)) product = product_result.scalar_one_or_none() if product: if task.assignee_id: product.current_location_id = task.assignee_id if task_name: product.overall_status = task_name await db.commit() await db.refresh(task) return _to_response(task) # ============================================================ # 核心业务 2：品质驳回 (→ REJECTED + 返工闭环) # ============================================================ async def reject_task( db: AsyncSession, task_id: uuid.UUID, request: TaskRejectRequest, operator_id: str | None = None ) -> TaskResponse: """ 品质驳回：将当前任务标记为 REJECTED，并自动创建返工任务给上一道工序负责人。 防呆闭环逻辑： 1. 将当前任务状态改为 REJECTED，记录 reject_reason 和 completed_at。 2. 查找当前任务的父任务 (parent_task) 的负责人 (assignee_id)。 - 若有父任务：返工任务分配给父任务的 assignee_id。 - 若无父任务（顶层任务）：返工任务分配给当前任务自己的 assignee_id。 3. 为该负责人新建一个完全一样的任务，但 is_rework=True，status=PENDING。 4. 新返工任务挂在同一个 parent_task_id 下（与原任务同级）。 """ task = await _get_task_or_404(db, task_id) # 权限校验：只有负责人本人可驳回 if operator_id and task.assignee_id and operator_id != task.assignee_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"您无权操作此任务，当前任务负责人为 {task.assignee_id}", ) # 校验：不能重复驳回已完成/已驳回的任务 if task.status in (TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"任务状态为 {task.status}，无法驳回", ) now = get_beijing_time() # --- 1. 标记当前任务为已驳回 --- task.status = TASK_STATUS_REJECTED task.reject_reason = request.reason task.completed_at = now await _create_task_log( db, task_id, action_type="reject", operator_id=operator_id, remark=f"品质驳回: {request.reason}", ) # --- 2. 确定返工任务的负责人（追溯上一道工序的转交人） --- rework_assignee_id: str | None = None if task.parent_task_id: # 有父任务：返工给父任务的负责人（即上一环的转交人 A） parent_result = await db.execute( select(Task).where(Task.id == task.parent_task_id) ) parent_task = parent_result.scalar_one_or_none() if parent_task: rework_assignee_id = parent_task.assignee_id if not rework_assignee_id: # 无父任务（顶层转交）：从任务日志追溯创建人（转交发起者 A） log_result = await db.execute( select(TaskLog).where( TaskLog.task_id == task_id, TaskLog.action_type == "create", ).order_by(TaskLog.created_at.asc()).limit(1) ) create_log = log_result.scalar_one_or_none() if create_log and create_log.operator_id: rework_assignee_id = create_log.operator_id if not rework_assignee_id: # 最后兜底：用当前任务的负责人（通常不应该走到这里） rework_assignee_id = task.assignee_id # --- 3. 创建返工任务 --- rework_task = Task( product_id=task.product_id, parent_task_id=task.parent_task_id, # 与原任务同级 task_name=task.task_name, assignee_id=rework_assignee_id, status=TASK_STATUS_PENDING, notify_parent_on_complete=task.notify_parent_on_complete, is_rework=True, ) db.add(rework_task) await db.flush() await _create_task_log( db, rework_task.id, action_type="create", operator_id=operator_id, remark=f"返工任务（驳回自「{task.task_name}」，原因: {request.reason}），分配给 {rework_assignee_id}", ) await db.commit() await db.refresh(task) return _to_response(task) # ============================================================ # 核心业务 3：完工并裂变转交 (→ COMPLETED + 裂变创建下家任务) # ============================================================ async def transfer_task( db: AsyncSession, task_id: uuid.UUID, request: TaskTransferRequest, operator_id: str | None = None ) -> TaskTransferResponse: """ 完工并裂变转交： 动作 1（闭环当前节点）： - 将当前任务状态改为 COMPLETED，记录 completed_at。 动作 2（解析下家）： - 遍历 next_assignees 列表。 - 如果包含 'virtual_warehouse'，则将 Product 的 current_location_id 设为 'virtual_warehouse'。 - 为每一个 assignee_id（非 virtual_warehouse）新建一条 Task 记录（状态 PENDING）。 裂变逻辑： - 如果 len(next_assignees) > 1：多路裂变 → 所有新任务挂到当前任务下（parent_task_id = 当前任务ID）。 - 如果当前任务本身就是子任务（有 parent_task_id）：单路转交也挂到同一父任务下。 - 否则（顶层单路转交）：新任务与当前任务同级（parent_task_id = None）。 - 更新 Product 的 current_location_id 为对应的人员（非仓库）。 """ task = await _get_task_or_404(db, task_id) # 校验：不能重复完成 if task.status == TASK_STATUS_COMPLETED: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"任务「{task.task_name}」已经完成（当前状态: {task.status}），请刷新页面", ) # 校验：PENDING 状态不允许转交（必须先接收） if task.status == TASK_STATUS_PENDING: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"任务「{task.task_name}」尚未接收（当前状态: PENDING），请先接收再转交", ) # 校验：卡点逻辑 — 检查关键子任务 all_done, incomplete_names = await _check_all_critical_children_completed(db, task_id) if not all_done: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"请等待相关子任务完成：{', '.join(incomplete_names)}", ) now = get_beijing_time() # --- 动作 1：闭环当前节点 --- task.status = TASK_STATUS_COMPLETED task.completed_at = now await _create_task_log( db, task_id, action_type="complete", operator_id=operator_id, remark=request.note or f"完成任务「{task.task_name}」，转交至下一道工序", ) # --- 动作 2：解析下家 & 裂变 --- # 兼容新旧格式 if request.next_tasks: branches = [ (b.task_name, a) for b in request.next_tasks for a in (b.assignees or []) ] else: branches = [ (request.next_task_name, a) for a in (request.next_assignees or []) ] has_warehouse = any(a == VIRTUAL_WAREHOUSE for _, a in branches) real_branches = [(tn, a) for tn, a in branches if a != VIRTUAL_WAREHOUSE] is_fission = len(real_branches) > 1 or (request.next_tasks and len(request.next_tasks) > 1) is_child_task = task.parent_task_id is not None created_tasks: list[Task] = [] for task_name, assignee_id in real_branches: if is_fission: new_parent_task_id = task.id elif is_child_task: new_parent_task_id = task.parent_task_id else: new_parent_task_id = None new_task = Task( product_id=task.product_id, parent_task_id=new_parent_task_id, task_name=task_name, assignee_id=assignee_id, status=TASK_STATUS_PENDING, notify_parent_on_complete=False, is_rework=False, remark=request.note or None, ) db.add(new_task) created_tasks.append(new_task) # 批量 flush 以生成 ID await db.flush() for nt in created_tasks: await _create_task_log( db, nt.id, action_type="create", operator_id=operator_id, remark=request.note or f"由任务「{task.task_name}」裂变转交创建，分配给 {nt.assignee_id}", ) # --- 更新 Product 的 current_location_id --- product_result = await db.execute( select(Product).where(Product.id == task.product_id) ) product = product_result.scalar_one_or_none() if product: if has_warehouse and not real_branches: product.current_location_id = VIRTUAL_WAREHOUSE product.overall_status = "在库" elif real_branches: product.current_location_id = real_branches[0][1] product.overall_status = real_branches[0][0] await db.commit() # --- 构建响应 --- refreshed_task = await _get_task_with_children_recursive(db, task_id) created_task_responses = [] for nt in created_tasks: await db.refresh(nt) created_task_responses.append(_to_response(nt)) assignee_list = ", ".join(a for _, a in real_branches) if real_branches else "仓库" location_info = "" if has_warehouse: location_info = "，产品已入库(virtual_warehouse)" return TaskTransferResponse( completed_task=_to_response(refreshed_task), created_tasks=created_task_responses, message=( f"任务「{task.task_name}」已完成，" f"已创建 {len(created_tasks)} 个下一道工序任务「{request.next_task_name}」" f"（接收人: {assignee_list}）{location_info}" ), ) # ============================================================ # 保留兼容：旧版 complete_task（单步完成/转交） # ============================================================ async def complete_task( db: AsyncSession, task_id: uuid.UUID, request: TaskCompleteRequest ) -> TaskCompleteResponse: """ 核心业务：完成任务 + 可选创建下一步任务。 逻辑： 1. 检查当前任务是否已完成（幂等） 2. 检查所有 notify_parent_on_complete=True 的子任务是否都已完成 → 如果存在未完成的关键子任务，返回 400 错误 3. 将当前任务状态改为 completed，记录日志 4. 如果提供了 next_task_name 和 next_assignee_id，创建下一步任务 5. 返回完成结果 """ task = await _get_task_or_404(db, task_id) # 注入子任务数据到当前对象以便后续检查 children_result = await db.execute( select(Task).where(Task.parent_task_id == task_id) ) task.child_tasks = children_result.scalars().all() # --- 1. 幂等检查 --- if task.status == TASK_STATUS_COMPLETED: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"任务「{task.task_name}」已经完成，无需重复操作", ) # --- 2. 卡点逻辑：检查关键子任务 --- all_done, incomplete_names = await _check_all_critical_children_completed(db, task_id) if not all_done: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"请等待相关子任务完成：{', '.join(incomplete_names)}", ) now = get_beijing_time() # --- 3. 标记当前任务为已完成 --- task.status = TASK_STATUS_COMPLETED task.completed_at = now await _create_task_log( db, task_id, action_type="complete", operator_id=request.operator_id, remark=request.remark or f"完成任务: {task.task_name}", ) # --- 4. 可选：创建下一步任务（转交） --- next_task = None if request.next_task_name and request.next_assignee_id: next_task = Task( product_id=task.product_id, parent_task_id=task.parent_task_id, # 与已完成任务同级 task_name=request.next_task_name, assignee_id=request.next_assignee_id, status=TASK_STATUS_PENDING, notify_parent_on_complete=False, ) db.add(next_task) await db.flush() await _create_task_log( db, next_task.id, action_type="create", operator_id=request.operator_id, remark=f"由任务「{task.task_name}」完成后转交创建", ) await db.commit() # --- 5. 构建响应 --- refreshed_task = await _get_task_with_children_recursive(db, task_id) next_task_response = None if next_task: await db.refresh(next_task) next_task_response = _to_response(next_task) return TaskCompleteResponse( completed_task=_to_response(refreshed_task), next_task=next_task_response, message=f"任务「{task.task_name}」已完成" + (f"，已创建下一步任务「{request.next_task_name}」" if next_task else ""), ) # ============================================================ # 子任务 # ============================================================ async def create_subtask( db: AsyncSession, parent_task_id: uuid.UUID, data: SubtaskCreate ) -> TaskResponse: """ 在现有任务下创建子任务 — 支持无限层级嵌套。 新子任务继承父任务的 product_id。 """ parent = await _get_task_or_404(db, parent_task_id) # 禁止在已完成/已驳回的任务下创建子任务 if parent.status in (TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"无法在状态为 {parent.status} 的任务「{parent.task_name}」下创建子任务", ) subtask = Task( product_id=parent.product_id, parent_task_id=parent_task_id, task_name=data.task_name, assignee_id=data.assignee_id, status=TASK_STATUS_PENDING, notify_parent_on_complete=data.notify_parent_on_complete, ) db.add(subtask) await db.commit() await db.refresh(subtask) await _create_task_log( db, subtask.id, action_type="create", operator_id=data.assignee_id, remark=f"创建子任务「{data.task_name}」，父任务: 「{parent.task_name}」", ) await db.commit() return _to_response(subtask) # ============================================================ # 任务进度记录 — 随时备注/传图 # ============================================================ async def add_task_record( db: AsyncSession, task_id: uuid.UUID, data: TaskRecordCreate ) -> TaskResponse: """追加进度记录（备注+图片），不改变任务状态""" import json task = await _get_task_with_children_recursive(db, task_id) record = TaskRecord( task_id=task_id, remark=data.remark or None, images=json.dumps(data.images) if data.images else None, ) db.add(record) await db.commit() await db.refresh(record) # 重新加载 task 带上新 record return await get_task(db, task_id)