Files
nas-tools/dashboard/backend/routers/opc_tasks.py
T
Gan, Jimmy 2c8f4fcd5c
Deploy Dashboard (Dev) / deploy-dev (push) Failing after 2m19s
Run Tests / Frontend Tests (push) Has been cancelled
Run Tests / Backend Tests (push) Has started running
Run Tests / Test Summary (push) Has been cancelled
fix: auto-assign tasks to PM agent when moved to in_progress, fix datetime serialization in WebSocket, add executor logging
2026-04-02 10:22:15 +08:00

300 lines
8.4 KiB
Python

"""
OPC Task Management Router
"""
from fastapi import APIRouter, Depends, Query, HTTPException, Request
from typing import Optional, List
from pydantic import BaseModel
from datetime import datetime
from db import opc_db
from rbac import require_page, User
from routers import opc_ws
router = APIRouter()
class TaskCreate(BaseModel):
title: str
description: Optional[str] = None
status: str = "parking_lot"
priority: str = "medium"
project_id: Optional[int] = None
assigned_to: Optional[str] = None
assigned_type: str = "human"
tags: Optional[List[str]] = None
due_date: Optional[datetime] = None
class TaskUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
status: Optional[str] = None
priority: Optional[str] = None
project_id: Optional[int] = None
assigned_to: Optional[str] = None
assigned_type: Optional[str] = None
tags: Optional[List[str]] = None
due_date: Optional[datetime] = None
completed_at: Optional[datetime] = None
class TaskMove(BaseModel):
status: str
class TaskAssign(BaseModel):
assigned_to: str
assigned_type: str = "human"
@router.get("/tasks")
async def list_tasks(
request: Request,
status: Optional[str] = Query(None),
assigned_to: Optional[str] = Query(None),
project_id: Optional[int] = Query(None),
priority: Optional[str] = Query(None),
tags: Optional[str] = Query(None),
limit: int = Query(50, le=100),
offset: int = Query(0, ge=0),
):
"""List tasks with filters"""
user: User = request.state.user
tag_list = tags.split(",") if tags else None
tasks = await opc_db.get_tasks(
status=status,
assigned_to=assigned_to,
project_id=project_id,
priority=priority,
tags=tag_list,
limit=limit,
offset=offset
)
return {"items": tasks, "total": len(tasks)}
@router.post("/tasks")
async def create_task(
request: Request,
task: TaskCreate,
):
"""Create a new task"""
user: User = request.state.user
# Strip timezone info from due_date if present (database uses TIMESTAMP not TIMESTAMPTZ)
due_date = task.due_date.replace(tzinfo=None) if task.due_date else None
created_task = await opc_db.create_task(
title=task.title,
description=task.description,
status=task.status,
priority=task.priority,
project_id=task.project_id,
assigned_to=task.assigned_to,
assigned_type=task.assigned_type,
tags=task.tags,
due_date=due_date,
actor=user.username
)
# Start automatic timer if status is in_progress
if task.status == "in_progress":
await opc_db.start_time_entry(
task_id=created_task["id"],
project_id=task.project_id,
actor=user.username
)
# Broadcast WebSocket update
await opc_ws.broadcast_task_update(created_task["id"], "created", created_task)
return created_task
@router.get("/tasks/{task_id}")
async def get_task(
task_id: int,
request: Request,
):
"""Get task details"""
user: User = request.state.user
tasks = await opc_db.get_tasks(limit=1, offset=0)
task = next((t for t in tasks if t["id"] == task_id), None)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.put("/tasks/{task_id}")
async def update_task(
task_id: int,
task: TaskUpdate,
request: Request,
):
"""Update a task"""
user: User = request.state.user
updates = task.dict(exclude_unset=True)
# Get current task status
tasks = await opc_db.get_tasks(limit=1000, offset=0)
current_task = next((t for t in tasks if t["id"] == task_id), None)
if not current_task:
raise HTTPException(status_code=404, detail="Task not found")
# Handle automatic time tracking
old_status = current_task["status"]
new_status = updates.get("status", old_status)
# Start timer when moving to in_progress
if old_status != "in_progress" and new_status == "in_progress":
await opc_db.start_time_entry(
task_id=task_id,
project_id=current_task.get("project_id"),
actor=user.username
)
# Stop timer when moving away from in_progress
if old_status == "in_progress" and new_status != "in_progress":
await opc_db.stop_time_entry(task_id=task_id)
# Set completed_at when moving to done
if new_status == "done" and old_status != "done":
updates["completed_at"] = datetime.utcnow()
# Strip timezone info from datetime fields (database uses TIMESTAMP not TIMESTAMPTZ)
if "due_date" in updates and updates["due_date"] is not None:
updates["due_date"] = updates["due_date"].replace(tzinfo=None)
if "completed_at" in updates and updates["completed_at"] is not None:
updates["completed_at"] = updates["completed_at"].replace(tzinfo=None)
updated_task = await opc_db.update_task(
task_id=task_id,
updates=updates,
actor=user.username
)
# Broadcast WebSocket update
await opc_ws.broadcast_task_update(task_id, "updated", updated_task)
return updated_task
@router.delete("/tasks/{task_id}")
async def delete_task(
task_id: int,
request: Request,
):
"""Delete a task"""
user: User = request.state.user
await opc_db.delete_task(task_id=task_id, actor=user.username)
# Broadcast WebSocket update
await opc_ws.broadcast_task_update(task_id, "deleted", {"id": task_id})
return {"ok": True}
@router.put("/tasks/{task_id}/move")
async def move_task(
task_id: int,
move: TaskMove,
request: Request,
):
"""Move task to a different column"""
user: User = request.state.user
updates = {"status": move.status}
# Get current task
tasks = await opc_db.get_tasks(limit=1000, offset=0)
current_task = next((t for t in tasks if t["id"] == task_id), None)
if not current_task:
raise HTTPException(status_code=404, detail="Task not found")
old_status = current_task["status"]
# Handle automatic time tracking
if old_status != "in_progress" and move.status == "in_progress":
await opc_db.start_time_entry(
task_id=task_id,
project_id=current_task.get("project_id"),
actor=user.username
)
# Auto-assign to default agent (PM) if not already assigned
if not current_task.get("assigned_to"):
updates["assigned_to"] = "pm"
updates["assigned_type"] = "agent"
# Create agent execution record
await opc_db.create_agent_execution(
task_id=task_id,
agent_id="pm",
actor=user.username
)
if old_status == "in_progress" and move.status != "in_progress":
await opc_db.stop_time_entry(task_id=task_id)
if move.status == "done":
updates["completed_at"] = datetime.utcnow().replace(tzinfo=None)
updated_task = await opc_db.update_task(
task_id=task_id,
updates=updates,
actor=user.username
)
return updated_task
@router.post("/tasks/{task_id}/assign")
async def assign_task(
task_id: int,
assign: TaskAssign,
request: Request,
):
"""Assign task to human or agent"""
user: User = request.state.user
updates = {
"assigned_to": assign.assigned_to,
"assigned_type": assign.assigned_type
}
updated_task = await opc_db.update_task(
task_id=task_id,
updates=updates,
actor=user.username
)
# If assigning to agent, create execution record
if assign.assigned_type == "agent":
await opc_db.create_agent_execution(
task_id=task_id,
agent_id=assign.assigned_to,
actor=user.username
)
return updated_task
@router.get("/tasks/{task_id}/history")
async def get_task_history(
task_id: int,
request: Request,
):
"""Get task history"""
user: User = request.state.user
history = await opc_db.get_task_history(task_id=task_id)
return {"items": history}
@router.get("/tasks/{task_id}/time")
async def get_task_time(
task_id: int,
request: Request,
):
"""Get time entries for a task"""
user: User = request.state.user
entries = await opc_db.get_time_entries(task_id=task_id)
total_minutes = sum(e["duration_minutes"] for e in entries)
return {
"entries": entries,
"total_minutes": total_minutes,
"total_hours": round(total_minutes / 60, 2)
}