duxingchen
71e660b428
业务调整：车间驳回不必须拍照（编号错误、选错工序等场景无需照片举证）。 - schemas/task.py: TaskRejectRequest.images 由必填(min_length=1)改为 default_factory=list，保留 max_length=9 与 URL 总长保护；reason 增加 mode="before" 的 strip 校验器，堵住纯空格凑长度绕过 min_length 的口子 （顺带保证落库的 reject_reason 不带首尾空白） - task_service.py: 驳回时把原因+图片写入 TaskRecord 挂到被驳回任务下， 修复前端时间线看不到驳回详情的问题；转交时把交接备注也挂一条记录到 下家任务，补齐接手人的上下文视角 - task_service.py: create_task / receive_task 写入 overall_status 后调用 sync_product_status 对齐 status 字段
1259 lines
49 KiB
Python
1259 lines
49 KiB
Python
"""任务服务 — 核心业务逻辑：接收、驳回返工、裂变转交、无限嵌套子任务"""
|
||
from __future__ import annotations
|
||
import json
|
||
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, TASK_STATUS_CANCELED, TASK_STATUS_ARCHIVED
|
||
from app.models.notification import Notification, NOTIFY_TRANSFER, NOTIFY_REJECT
|
||
from app.core.time_utils import get_beijing_time
|
||
from app.core.lifecycle import (
|
||
LIFECYCLE_AFTER_SALES,
|
||
PRODUCTION_ONLY_STEPS,
|
||
allowed_steps,
|
||
is_step_allowed,
|
||
phase_label,
|
||
resolve_phase_for_step,
|
||
sync_product_status,
|
||
)
|
||
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"
|
||
|
||
# 管理员/主管角色白名单 — 拥有上帝视角操作权限
|
||
ADMIN_ROLES = {"SUPER_ADMIN", "SUPERVISOR"}
|
||
|
||
|
||
async def _recalc_product_location(
|
||
db: AsyncSession, product_id: uuid.UUID, completed_task_id: uuid.UUID | None = None,
|
||
) -> None:
|
||
"""
|
||
任务完工/结束时触发：只跟随主干任务（主分支），无视协助分支。
|
||
|
||
主干任务定义：
|
||
parent_task_id IS NULL OR task_type IN ('TRANSFER', 'RECOVERY')
|
||
|
||
优先级：
|
||
WIP > PENDING > COMPLETED/ARCHIVED > None
|
||
"""
|
||
from sqlalchemy import select as sa_select, case as sa_case, or_ as sa_or_
|
||
|
||
product_result = await db.execute(
|
||
sa_select(Product).where(Product.id == product_id)
|
||
)
|
||
product = product_result.scalar_one_or_none()
|
||
if not product:
|
||
return
|
||
|
||
# ── 只查主干任务: parent_task_id IS NULL 或 task_type IN (TRANSFER, RECOVERY) ──
|
||
stmt = (
|
||
sa_select(Task)
|
||
.where(
|
||
Task.product_id == product_id,
|
||
sa_or_(
|
||
Task.parent_task_id.is_(None),
|
||
Task.task_type.in_(["TRANSFER", "RECOVERY"]),
|
||
),
|
||
)
|
||
.order_by(
|
||
# 优先级排序: WIP=3, PENDING=2, COMPLETED=1, ARCHIVED=1, else=0
|
||
sa_case(
|
||
(Task.status == TASK_STATUS_WIP, 3),
|
||
(Task.status == TASK_STATUS_PENDING, 2),
|
||
(Task.status == TASK_STATUS_COMPLETED, 1),
|
||
(Task.status == TASK_STATUS_ARCHIVED, 1),
|
||
else_=0,
|
||
).desc(),
|
||
Task.created_at.desc(),
|
||
)
|
||
.limit(1)
|
||
)
|
||
result = await db.execute(stmt)
|
||
main_task = result.scalar_one_or_none()
|
||
|
||
new_location = main_task.assignee_id if main_task else None
|
||
|
||
if product.current_location_id != new_location:
|
||
product.current_location_id = new_location
|
||
await db.flush() # 唯一的落盘点
|
||
|
||
|
||
async def _mark_after_sales_if_reactivated(
|
||
db: AsyncSession, product_id: uuid.UUID,
|
||
) -> bool:
|
||
"""出库后又被派发新任务 = 设备回流返厂 → 生命周期切到 AFTER_SALES。
|
||
|
||
判定依据：产品当前处于「已出库」终态（overall_status == '已出库' 或
|
||
status == 'OUTBOUND'），却又产生了新的在制任务。
|
||
|
||
该标志单向：一旦进入 AFTER_SALES 不再回退，这样前端就能把
|
||
生产阶段的「发货测试」与回流后的「售后维修」区分开。
|
||
|
||
调用时机必须早于调用方改写 overall_status，否则会漏判。
|
||
|
||
返回是否发生了翻转。product 已在本 session 加载时 db.get 直接命中
|
||
identity map，不产生额外查询。
|
||
"""
|
||
product = await db.get(Product, product_id)
|
||
if product is None or product.lifecycle_phase == LIFECYCLE_AFTER_SALES:
|
||
return False
|
||
|
||
is_outbound = (
|
||
product.overall_status == "已出库"
|
||
or (product.status or "").upper() == "OUTBOUND"
|
||
)
|
||
if not is_outbound:
|
||
return False
|
||
|
||
product.lifecycle_phase = LIFECYCLE_AFTER_SALES
|
||
await db.flush()
|
||
return True
|
||
|
||
|
||
def _enforce_step_isolation(
|
||
product: Product, step: str | None, *, action: str,
|
||
) -> None:
|
||
"""选项隔离守卫 — 售后回流设备禁止被重新排产回「备货 / 生产」等前期工序。
|
||
|
||
行为两步：
|
||
1. 选定售后专属工序（发货测试 / 售后维修）→ 设备随即进入售后生命周期。
|
||
这是「无历史记录的老设备」进入售后阶段的入口。
|
||
2. 再校验该工序在当前阶段是否合法，非法直接 400 —— 前端下拉被绕过、
|
||
请求被伪造时，仍在此拦下。
|
||
|
||
必须在 _mark_after_sales_if_reactivated 之后调用，这样出库回流的设备
|
||
已处于 AFTER_SALES，自然排不回前期工序。
|
||
|
||
注意：「待确认」与仓库虚拟节点属建单占位符，由 is_step_allowed 直接放行，
|
||
否则建单 / 转交入库整条链路会被卡死。
|
||
"""
|
||
phase = resolve_phase_for_step(product.lifecycle_phase, step)
|
||
if not is_step_allowed(phase, step):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=(
|
||
f"产品当前处于{phase_label(phase)}，{action}不允许使用工序「{step}」。"
|
||
f"该阶段可选：{'、'.join(allowed_steps(phase))}"
|
||
),
|
||
)
|
||
if phase != product.lifecycle_phase:
|
||
product.lifecycle_phase = phase
|
||
|
||
|
||
def _reject_cross_phase_steps(
|
||
phase: str | None, steps: list[str | None], *, action: str,
|
||
) -> None:
|
||
"""跨阶段工序拦截 — 用于「转交」这类自由文本工序名路径。
|
||
|
||
转交对话框允许用户手打工序名（喷漆 / 老化 / 待确认…），所以不能用
|
||
白名单，否则会误伤合法命名；但"售后回流设备被转交到【生产】"必须挡住。
|
||
因此这里只拒绝**生产阶段专属**词（备货 / 生产 / 测试 / 维修）。
|
||
|
||
「在库 / 已入库 / 已出库」两阶段通用，占位符同样放行。
|
||
"""
|
||
if (phase or "") != LIFECYCLE_AFTER_SALES:
|
||
return
|
||
bad = sorted({
|
||
s.strip() for s in steps
|
||
if s and s.strip() in PRODUCTION_ONLY_STEPS
|
||
})
|
||
if bad:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=(
|
||
f"产品当前处于{phase_label(LIFECYCLE_AFTER_SALES)}，{action}不允许使用前期工序："
|
||
f"{'、'.join(bad)}。该阶段可选：{'、'.join(allowed_steps(LIFECYCLE_AFTER_SALES))}"
|
||
),
|
||
)
|
||
|
||
|
||
def _check_permission(task_assignee_id: str | None, operator_id: str | None, operator_role: str | None = None) -> None:
|
||
"""权限校验：本人 或 管理员/主管 可操作"""
|
||
if operator_role and operator_role in ADMIN_ROLES:
|
||
return # 上帝视角，直接放行
|
||
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}",
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# 内部辅助函数
|
||
# ============================================================
|
||
|
||
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:
|
||
"""使用 PostgreSQL Recursive CTE 一次性加载任务及其所有子孙任务（消除 N+1）"""
|
||
from app.services.task_tree_loader import load_task_tree_by_root
|
||
return await load_task_tree_by_root(db, task_id)
|
||
|
||
|
||
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,
|
||
task_type=task.task_type,
|
||
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,
|
||
task_type=task.task_type,
|
||
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
|
||
|
||
|
||
def _admin_proxy_note(operator_id: str | None, task_assignee_id: str | None) -> str:
|
||
"""检测管理员代办 → 返回审计标记后缀"""
|
||
if operator_id and task_assignee_id and operator_id != task_assignee_id:
|
||
return f" [管理员 {operator_id} 代办]"
|
||
return ""
|
||
|
||
|
||
async def _create_task_log(
|
||
db: AsyncSession,
|
||
task_id: uuid.UUID,
|
||
action_type: str,
|
||
operator_id: str | None = None,
|
||
remark: str | None = None,
|
||
task_assignee_id: str | None = None,
|
||
) -> TaskLog:
|
||
"""创建任务操作日志。自动检测管理员代办并拼接审计标记。"""
|
||
final_remark = (remark or "") + _admin_proxy_note(operator_id, task_assignee_id)
|
||
log = TaskLog(
|
||
task_id=task_id,
|
||
operator_id=operator_id,
|
||
action_type=action_type,
|
||
remark=final_remark.strip() or None,
|
||
)
|
||
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:
|
||
# 🔧 出库后再次派发任务 = 设备回流返厂 → 切到售后生命周期（须早于下方改写 overall_status）
|
||
await _mark_after_sales_if_reactivated(db, data.product_id)
|
||
# 🔧 选项隔离：售后回流设备禁止被排回「备货 / 生产」等前期工序（防伪造传参）
|
||
_enforce_step_isolation(product, data.task_name, action="创建任务")
|
||
if data.task_name and (not data.parent_task_id or data.task_type in ("TRANSFER", "RECOVERY")):
|
||
product.overall_status = "已入库" if "virtual_warehouse" in data.task_name else data.task_name
|
||
# 🔧 双字段同步：overall_status 改动后必须对齐 status，
|
||
# 否则出库回流设备的 status 会永远停在 OUTBOUND，污染统计口径
|
||
sync_product_status(product)
|
||
# 派发给人 → 产品离开仓库
|
||
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.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,
|
||
operator_role: str | None = None,
|
||
) -> TaskResponse:
|
||
"""
|
||
结束当前分支：标记任务为 COMPLETED，不创建下游任务。
|
||
用于工序已完结、无需转交下一人的场景。
|
||
"""
|
||
task = await _get_task_or_404(db, task_id)
|
||
|
||
# 权限校验：本人 或 管理员/主管 可结束
|
||
_check_permission(task.assignee_id, operator_id, operator_role)
|
||
|
||
# 校验：仅 SPAWN 协助分支可以结束，主分支（TRANSFER/RECOVERY）不能通过此接口终止
|
||
if not task.parent_task_id:
|
||
raise HTTPException(status_code=409, detail="根任务无法结束，请使用完工转交")
|
||
if task.task_type != "SPAWN":
|
||
raise HTTPException(status_code=409, detail="仅协助分支可以结束，主分支请使用完工转交")
|
||
|
||
# 校验：必须等待所有协助分支完成
|
||
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}」已终止（无下游）",
|
||
task_assignee_id=task.assignee_id)
|
||
|
||
# 🔧 位置回溯：分支结束后优先回溯到父任务负责人
|
||
await _recalc_product_location(db, task.product_id, task.id)
|
||
|
||
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,
|
||
operator_role: str | None = None,
|
||
) -> TaskResponse:
|
||
"""撤回 PENDING 转交：标记为 CANCELED，以被撤回节点为父生成接力新任务给操作人。
|
||
|
||
权限校验：
|
||
- 管理员（SUPER_ADMIN / SUPERVISOR）：直接放行
|
||
- 操作人 必须等于 上游任务的负责人（谁发出的谁才能撤回），否则 403
|
||
"""
|
||
task = await _get_task_or_404(db, task_id)
|
||
|
||
if task.status != TASK_STATUS_PENDING:
|
||
raise HTTPException(status_code=409, detail="只有待接收(PENDING)的任务可以撤回")
|
||
|
||
# ── 权限校验：谁发出的谁才能撤回 ──
|
||
if not (operator_role and operator_role in ADMIN_ROLES):
|
||
# 查找上游任务的负责人（发出者）
|
||
upstream_assignee: str | None = None
|
||
if task.parent_task_id:
|
||
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:
|
||
upstream_assignee = parent_task.assignee_id
|
||
else:
|
||
# 无父任务：从任务日志追溯创建人
|
||
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:
|
||
upstream_assignee = create_log.operator_id
|
||
|
||
# 如果找不到上游负责人，或者操作人不等于上游负责人 → 403
|
||
if not upstream_assignee or operator_id != upstream_assignee:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="您不是该任务的发起人，无法撤回",
|
||
)
|
||
|
||
now = get_beijing_time()
|
||
|
||
# 1. 废掉当前待接收任务
|
||
task.status = TASK_STATUS_CANCELED
|
||
task.completed_at = now
|
||
await _create_task_log(db, task_id, action_type="recall", operator_id=operator_id,
|
||
remark=f"撤回转交「{task.task_name}」→ {task.assignee_id}",
|
||
task_assignee_id=task.assignee_id)
|
||
db.add(TaskRecord(task_id=task.id, remark=f"[撤回] 转交至 {task.assignee_id} 已撤回", images="[]"))
|
||
|
||
# 2. 生成接力新任务（以撤回节点为父，还给操作人）
|
||
recovery = Task(
|
||
product_id=task.product_id,
|
||
parent_task_id=task.id,
|
||
task_name=task.task_name,
|
||
assignee_id=operator_id,
|
||
status=TASK_STATUS_WIP,
|
||
task_type="RECOVERY",
|
||
notify_parent_on_complete=False,
|
||
is_rework=False,
|
||
remark=f"撤回「{task.task_name}」后重新接手",
|
||
)
|
||
db.add(recovery)
|
||
await db.flush()
|
||
await _create_task_log(db, recovery.id, action_type="create", operator_id=operator_id,
|
||
remark=f"撤回接力：撤回「{task.task_name}」→ {task.assignee_id} 后重新指派给 {operator_id}",
|
||
task_assignee_id=recovery.assignee_id)
|
||
db.add(TaskRecord(task_id=recovery.id, remark=f"[重新接手] 撤回转交后系统自动生成接力节点", images="[]"))
|
||
|
||
# 3. 更新产品位置
|
||
product_result = await db.execute(select(Product).where(Product.id == task.product_id))
|
||
product = product_result.scalar_one_or_none()
|
||
if product and operator_id:
|
||
product.current_location_id = operator_id
|
||
|
||
await db.commit()
|
||
await db.refresh(recovery)
|
||
return _to_response(recovery)
|
||
|
||
|
||
# ============================================================
|
||
# 核心业务 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,
|
||
task_type="SPAWN",
|
||
notify_parent_on_complete=False,
|
||
is_rework=False,
|
||
remark=data.remark or None,
|
||
)
|
||
db.add(child)
|
||
await db.flush()
|
||
|
||
# 协助分支不改变产品宏观状态（只有主分支影响 overall_status）
|
||
db.add(TaskRecord(task_id=task.id, remark=data.remark or f"[派发协助] 分配给 {data.assignee_id}", images="[]"))
|
||
|
||
await _create_task_log(db, child.id, action_type="create", operator_id=operator_id,
|
||
remark=f"协助分支（由「{task.task_name}」派发，分配给 {data.assignee_id}）",
|
||
task_assignee_id=child.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, TASK_STATUS_CANCELED, 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,
|
||
operator_role: str | None = None,
|
||
) -> TaskResponse:
|
||
"""
|
||
操作员确认接收任务。工人选定工序名称后接收。
|
||
|
||
校验：只有状态为 PENDING 的任务可接收。
|
||
动作：状态改为 WIP，记录 received_at，更新 task_name，同步产品宏观状态。
|
||
"""
|
||
task = await _get_task_or_404(db, task_id)
|
||
|
||
# 权限校验：本人 或 管理员/主管 可操作
|
||
_check_permission(task.assignee_id, operator_id, operator_role)
|
||
|
||
# 校验：只有 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}」",
|
||
task_assignee_id=task.assignee_id,
|
||
)
|
||
db.add(TaskRecord(task_id=task.id,
|
||
remark=(remark or f"[接收] 操作员已确认接收") + _admin_proxy_note(operator_id, task.assignee_id),
|
||
images="[]"))
|
||
|
||
# 接收时同步产品位置到接收人 + 宏观状态同步
|
||
product_result = await db.execute(select(Product).where(Product.id == task.product_id))
|
||
product = product_result.scalar_one_or_none()
|
||
if product:
|
||
# 🔧 出库后任务被重新接收 = 设备回流返厂 → 切到售后生命周期（须早于下方改写 overall_status）
|
||
await _mark_after_sales_if_reactivated(db, task.product_id)
|
||
# 🔧 选项隔离：接收时选定的工序必须落在该产品当前生命周期阶段的合法集合内
|
||
#（task_name 为 None 时表示本次未指定工序，跳过校验，不阻断 PC 端「直接接收」）
|
||
_enforce_step_isolation(product, task_name, action="接收任务")
|
||
if task.assignee_id:
|
||
product.current_location_id = task.assignee_id
|
||
if task_name and (
|
||
not task.parent_task_id or task.task_type in ("TRANSFER", "RECOVERY")
|
||
):
|
||
product.overall_status = task_name # 只主线任务同步宏观状态
|
||
# 🔧 双字段同步：无条件对齐一次，顺带治愈历史残留
|
||
# （如整体已是「发货测试」而 status 还停在 OUTBOUND）
|
||
sync_product_status(product)
|
||
|
||
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,
|
||
operator_role: str | None = None,
|
||
) -> TaskResponse:
|
||
"""
|
||
品质驳回：将当前任务标记为 REJECTED，并自动创建返工任务给上一道工序负责人。
|
||
|
||
防呆闭环逻辑：
|
||
1. 将当前任务状态改为 REJECTED，记录 reject_reason 和 completed_at。
|
||
2. 查找当前任务的父任务 (parent_task) 的负责人 (assignee_id)。
|
||
- 若有父任务：返工任务分配给父任务的 assignee_id。
|
||
- 若无父任务（顶层任务）：返工任务分配给当前任务自己的 assignee_id。
|
||
3. 把驳回原因 + 异常图片写入 TaskRecord，挂到被驳回的任务下（前端时间线可见）。
|
||
4. 为该负责人新建一个完全一样的任务，但 is_rework=True，status=PENDING。
|
||
5. 新返工任务挂在同一个 parent_task_id 下（与原任务同级）。
|
||
"""
|
||
task = await _get_task_or_404(db, task_id)
|
||
|
||
# 权限校验：本人 或 管理员/主管 可驳回
|
||
_check_permission(task.assignee_id, operator_id, operator_role)
|
||
|
||
# 校验：不能重复驳回已完成/已驳回的任务
|
||
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}",
|
||
task_assignee_id=task.assignee_id,
|
||
)
|
||
|
||
# --- 2. 确定返工任务的负责人（追溯上一道工序的转交人） ---
|
||
rework_assignee_id: str | None = None
|
||
|
||
# 🚀 优先方式：从任务日志追溯创建人（准确记录是谁发起的转交）
|
||
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
|
||
|
||
# 兜底方式1：有父任务 → 返工给父任务的负责人
|
||
if not rework_assignee_id and task.parent_task_id:
|
||
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
|
||
|
||
# 兜底方式2：用当前任务的负责人
|
||
if not rework_assignee_id:
|
||
rework_assignee_id = task.assignee_id
|
||
|
||
# --- 3. 写入驳回记录（原因 + 异常图片）→ 修复前端时间线缺失 ---
|
||
# 图片为选填（编号错误等场景可不传），空数组落库为 "[]"，前端时间线照常渲染
|
||
db.add(TaskRecord(
|
||
task_id=task.id,
|
||
remark=f"[品质驳回] {request.reason}（返工任务已派发给 {rework_assignee_id or '未指派'}）",
|
||
images=json.dumps(request.images, ensure_ascii=False),
|
||
created_at=now,
|
||
))
|
||
|
||
# --- 4. 创建返工任务 ---
|
||
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,
|
||
task_type=task.task_type, # 🚀 继承被驳回任务的基因：主线→主线，协助→协助
|
||
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}",
|
||
task_assignee_id=rework_task.assignee_id,
|
||
)
|
||
|
||
# 🔔 通知：品质驳回
|
||
product_sn = ""
|
||
try:
|
||
product_result = await db.execute(select(Product).where(Product.id == task.product_id))
|
||
p = product_result.scalar_one_or_none()
|
||
if p:
|
||
product_sn = p.serial_number or ""
|
||
except Exception:
|
||
pass
|
||
if rework_assignee_id:
|
||
db.add(Notification(
|
||
user_id=rework_assignee_id,
|
||
title="🔴 品质驳回提醒",
|
||
content=f"产品 [{product_sn}] 的「{task.task_name}」被驳回，原因: {request.reason}",
|
||
type=NOTIFY_REJECT,
|
||
task_id=rework_task.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,
|
||
operator_role: 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)
|
||
|
||
# 权限校验：本人 或 管理员/主管 可转交
|
||
_check_permission(task.assignee_id, operator_id, operator_role)
|
||
|
||
# 校验：不能重复完成
|
||
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}」，转交至下一道工序",
|
||
task_assignee_id=task.assignee_id,
|
||
)
|
||
db.add(TaskRecord(task_id=task.id,
|
||
remark=(request.note or f"[完工转交] 移交下一工序") + _admin_proxy_note(operator_id, task.assignee_id),
|
||
images="[]"))
|
||
|
||
# --- 动作 2：解析下家 & 裂变 ---
|
||
# 🔧 提前查产品：下面的阶段隔离校验依赖它的生命周期阶段。
|
||
# （通知也需要 product_sn，原本在创建任务之后才查，提前不增加查询次数）
|
||
product_result = await db.execute(
|
||
select(Product).where(Product.id == task.product_id)
|
||
)
|
||
product = product_result.scalar_one_or_none()
|
||
|
||
# 🔧 出库后又被转交出新任务 = 设备回流返厂 → 切到售后生命周期。
|
||
# 必须在阶段校验之前执行，否则"已出库设备被转交到生产工序"会被误放行。
|
||
if product:
|
||
await _mark_after_sales_if_reactivated(db, task.product_id)
|
||
|
||
# 兼容新旧格式
|
||
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]
|
||
|
||
# 🔧 选项隔离：售后回流设备不允许被转交到「备货 / 生产 / 测试 / 维修」。
|
||
# 转交允许自由填写工序名（喷漆/老化…），故只精准拦跨阶段词，不用白名单。
|
||
if product:
|
||
_reject_cross_phase_steps(
|
||
product.lifecycle_phase, [tn for tn, _ in real_branches], action="转交",
|
||
)
|
||
|
||
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:
|
||
# 🔧 动态推导 task_type：禁止分支任务转交被硬编码为 TRANSFER（否则 is_main 判定误升为主线）
|
||
if is_fission:
|
||
new_parent_task_id = task.id
|
||
new_task_type = "SPAWN" # 裂变产生新分支
|
||
elif is_child_task:
|
||
new_parent_task_id = task.parent_task_id
|
||
new_task_type = task.task_type or "SPAWN" # 继承父任务的分支血统
|
||
else:
|
||
new_parent_task_id = task.id
|
||
new_task_type = "TRANSFER" # 主干常规流转
|
||
|
||
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,
|
||
task_type=new_task_type, # ← 使用动态推导的类型
|
||
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()
|
||
|
||
product_sn = product.serial_number if product else ""
|
||
|
||
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}",
|
||
task_assignee_id=nt.assignee_id,
|
||
)
|
||
# 🔧 转交留言留痕：把上一工序的嘱咐也挂一条记录到**下家任务**上。
|
||
# 原先留言只落在「已完工的上游任务记录」+「新任务的 Task.remark 字段」里，
|
||
# 接手人打开自己任务的「操作日志」只能看到 [接收]，看不到交接上下文，
|
||
# 转交备注的协作意义就丢了。这里补齐下游视角。
|
||
if request.note and request.note.strip():
|
||
db.add(TaskRecord(
|
||
task_id=nt.id,
|
||
remark=f"[上一工序转交留言]：{request.note.strip()}",
|
||
images="[]",
|
||
))
|
||
# 🔔 通知：新任务派发
|
||
if nt.assignee_id:
|
||
db.add(Notification(
|
||
user_id=nt.assignee_id,
|
||
title=f"🟢 新任务派发",
|
||
content=f"产品 [{product_sn}] 的「{nt.task_name}」任务已分配给你",
|
||
type=NOTIFY_TRANSFER,
|
||
task_id=nt.id,
|
||
))
|
||
|
||
# --- 更新 Product 的 current_location_id ---
|
||
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]
|
||
if not task.parent_task_id or task.task_type in ("TRANSFER", "RECOVERY"):
|
||
product.overall_status = real_branches[0][0]
|
||
# 🔧 双字段同步：两个分支统一在这里对齐一次（原先只有入库分支硬编码
|
||
# product.status="COMPLETED"，普通转交分支完全没同步 → 残留 OUTBOUND）
|
||
sync_product_status(product)
|
||
|
||
# 🔧 位置回溯：如果有新任务创建，优先新任务负责人；否则回溯到父任务
|
||
if not real_branches and not has_warehouse:
|
||
await _recalc_product_location(db, task.product_id, task_id)
|
||
|
||
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,
|
||
operator_role: str | None = None,
|
||
) -> 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)
|
||
|
||
# 权限校验：本人 或 管理员/主管 可操作
|
||
_check_permission(task.assignee_id, request.operator_id, operator_role)
|
||
|
||
# --- 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}",
|
||
task_assignee_id=task.assignee_id,
|
||
)
|
||
# 🔧 完工留痕（与 transfer_task 同口径）：备注要写成 TaskRecord，
|
||
# 否则「操作日志」时间线只读 TaskRecord、看不到 TaskLog，交接内容就丢了。
|
||
_will_transfer = bool(request.next_task_name and request.next_assignee_id)
|
||
db.add(TaskRecord(
|
||
task_id=task.id,
|
||
remark=(
|
||
request.remark
|
||
or ("[完工转交] 移交下一工序" if _will_transfer else "[完工] 任务已完成")
|
||
) + _admin_proxy_note(request.operator_id, task.assignee_id),
|
||
images="[]",
|
||
))
|
||
|
||
# --- 4. 可选：创建下一步任务（转交） ---
|
||
next_task = None
|
||
if request.next_task_name and request.next_assignee_id:
|
||
# 🔧 出库后又被转交出新任务 = 设备回流返厂 → 切到售后生命周期
|
||
await _mark_after_sales_if_reactivated(db, task.product_id)
|
||
# 🚀 智能父节点继承算法
|
||
# 主线任务转交 → 保持平级继承（主分支永远在一维主干上）
|
||
# 协助分支转交 → 认当前任务为父（形成向外无限延伸的孙子节点树枝）
|
||
# 注：TASK_TYPE 实际值为 TRANSFER/RECOVERY/SPAWN，不存在 "MAIN"
|
||
is_main_line = (
|
||
not task.parent_task_id
|
||
or task.task_type in ("TRANSFER", "RECOVERY")
|
||
)
|
||
new_parent_id = task.parent_task_id if is_main_line else task.id
|
||
|
||
next_task = Task(
|
||
product_id=task.product_id,
|
||
parent_task_id=new_parent_id, # 👈 智能计算
|
||
task_name=request.next_task_name,
|
||
assignee_id=request.next_assignee_id,
|
||
status=TASK_STATUS_PENDING,
|
||
task_type=task.task_type, # 👈 基因严格继承（绝不篡位成 MAIN）
|
||
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}」完成后转交创建",
|
||
task_assignee_id=next_task.assignee_id,
|
||
)
|
||
# 🔧 转交留言留痕：把上一工序的嘱咐也挂一条记录到**下家任务**上，
|
||
# 让接手人打开自己的「操作日志」就能看到交接上下文
|
||
# （与 transfer_task 同口径，保证两条转交路径行为一致）
|
||
if request.remark and request.remark.strip():
|
||
db.add(TaskRecord(
|
||
task_id=next_task.id,
|
||
remark=f"[上一工序转交留言]：{request.remark.strip()}",
|
||
images="[]",
|
||
))
|
||
|
||
# 🔧 位置回溯：老接口也触发（父任务优先）
|
||
await _recalc_product_location(db, task.product_id, task_id)
|
||
|
||
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}」",
|
||
task_assignee_id=subtask.assignee_id,
|
||
)
|
||
await db.commit()
|
||
|
||
return _to_response(subtask)
|
||
|
||
|
||
# ============================================================
|
||
# 任务进度记录 — 随时备注/传图
|
||
# ============================================================
|
||
|
||
async def add_task_record(
|
||
db: AsyncSession, task_id: uuid.UUID, data: TaskRecordCreate,
|
||
current_user: dict | None = None,
|
||
) -> 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.flush()
|
||
|
||
# 🚀 留言通知：给任务当前负责人发送提醒（不给自己发）
|
||
if (
|
||
current_user
|
||
and task.assignee_id
|
||
and task.assignee_id != current_user.get("username", "")
|
||
and data.remark
|
||
):
|
||
# 截取留言内容前 30 字作为摘要
|
||
remark_text = data.remark.strip()
|
||
short_content = remark_text[:30] + ("..." if len(remark_text) > 30 else "")
|
||
# 查询产品条码
|
||
product_result = await db.execute(
|
||
select(Product).where(Product.id == task.product_id)
|
||
)
|
||
product = product_result.scalar_one_or_none()
|
||
product_sn = product.serial_number if product else "未知"
|
||
db.add(Notification(
|
||
user_id=task.assignee_id,
|
||
title="💬 收到新留言",
|
||
content=f"产品 [{product_sn}] 的「{task.task_name}」有新留言：{short_content}",
|
||
type="COMMENT",
|
||
task_id=task.id,
|
||
))
|
||
|
||
await db.commit()
|
||
await db.refresh(record)
|
||
|
||
# 手动追加新 record，避免二次加载整棵树（task 已在 L1051 由 CTE 完整加载）
|
||
task.records.append(record)
|
||
return _to_response(task)
|