""" 任務 API。 階段 2a 提供基本 CRUD: - GET /api/tasks ─ 列任務(支援 filter) - POST /api/tasks ─ 建立任務 - GET /api/tasks/{id} ─ 取得單一任務 - PATCH /api/tasks/{id} ─ 更新任務(改狀態、優先順序、指派、due_date 都走這) - DELETE /api/tasks/{id} ─ 軟刪除任務 filter 參數: - status ─ 只看某狀態的(用 Kanban 看板時會用) - assignee_id ─ 只看指派給某人的 - project_id ─ 只看某專案的(可填 "null" 表示 inbox) - mine ─ 只看指派給「我」的(用「我的任務」分頁時用) 設計重點: 1. status 變 done 時自動填入 completed_at,變回其他狀態自動清空 2. 任何人(team member)都能建/改/刪任務(微型團隊不需要複雜權限) 3. 但只有 task creator / assignee / 公司 owner 能刪任務 """ from datetime import datetime, timezone from typing import Annotated from uuid import UUID from fastapi import APIRouter, HTTPException, Query, status from sqlalchemy import or_, select from app.core.deps import CurrentUser, DBSession from app.models.task import Task from app.schemas.task import TaskCreateRequest, TaskPublic, TaskUpdateRequest router = APIRouter(prefix="/api/tasks", tags=["tasks"]) @router.get("", response_model=list[TaskPublic], summary="列出任務(支援 filter)") async def list_tasks( current_user: CurrentUser, db: DBSession, status_filter: Annotated[str | None, Query(alias="status")] = None, assignee_id: UUID | None = None, project_id: Annotated[ str | None, Query(description="專案 ID,或填 'null' 看 inbox 任務"), ] = None, mine: bool = False, include_done: bool = True, ) -> list[Task]: """列出符合 filter 的任務。 預設回傳全部未軟刪的任務,按建立時間倒序。 """ stmt = select(Task).where(Task.deleted_at.is_(None)) if status_filter: stmt = stmt.where(Task.status == status_filter) elif not include_done: stmt = stmt.where(Task.status != "done", Task.status != "cancelled") if mine: # 階段 9-K(2026-05-08)修:「我的任務」= 只看指派給我的 # 原本含 created_by → owner 在群組大量派工後自己看到「全公司任務」 stmt = stmt.where(Task.assignee_id == current_user.id) if assignee_id: stmt = stmt.where(Task.assignee_id == assignee_id) if project_id is not None: if project_id == "null": stmt = stmt.where(Task.project_id.is_(None)) else: try: pid = UUID(project_id) except ValueError as e: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="project_id 不是合法 UUID", ) from e stmt = stmt.where(Task.project_id == pid) # 排序:先按 status 內的 order_index,再按建立時間 stmt = stmt.order_by(Task.order_index.asc(), Task.created_at.desc()) result = await db.execute(stmt) return list(result.scalars().all()) @router.post( "", response_model=TaskPublic, status_code=status.HTTP_201_CREATED, summary="建立任務", ) async def create_task( payload: TaskCreateRequest, current_user: CurrentUser, db: DBSession, ) -> Task: task = Task( title=payload.title, description=payload.description, project_id=payload.project_id, status=payload.status, priority=payload.priority, due_date=payload.due_date, assignee_id=payload.assignee_id, created_by=current_user.id, ) if payload.status == "done": task.completed_at = datetime.now(timezone.utc) db.add(task) await db.commit() await db.refresh(task) return task @router.get("/{task_id}/related", summary="找跟這個任務相關的過去討論 / 任務(階段 8 E)") async def get_related(task_id: UUID, current_user: CurrentUser, db: DBSession) -> dict: """RAG 找相關的過去訊息 + 字串相似度找相關的其他任務。""" from app.services.coba_memory import search_relevant task = (await db.execute( select(Task).where(Task.id == task_id, Task.deleted_at.is_(None)) )).scalar_one_or_none() if task is None: raise HTTPException(status_code=404, detail="找不到任務") # 1. RAG 找相關過去訊息 query = task.title if task.description: query = f"{query} {task.description[:200]}" msgs_raw = await search_relevant(query, limit=8) related_messages = [ { "message_id": m.get("message_id"), "content": (m.get("content") or "")[:200], "tags": m.get("tags") or [], "created_at": m.get("created_at"), "score": round(m.get("score", 0), 3), } for m in msgs_raw if m.get("score", 0) >= 0.4 ] # 2. 字串相似的其他任務(同一團隊內,排除自己) # 簡單做法:取 title 的關鍵詞模糊找 keywords = [] for w in (task.title or "").split(): # 中文分不了詞,用第一個字 + 前 2 字嘗試 if len(w) >= 2: keywords.append(w[:3]) if task.title and not keywords: keywords = [task.title[:4]] related_tasks = [] if keywords: from sqlalchemy import or_ like_clauses = [Task.title.contains(k) for k in keywords[:3]] stmt = ( select(Task) .where( Task.deleted_at.is_(None), Task.id != task_id, or_(*like_clauses), ) .order_by(Task.created_at.desc()) .limit(5) ) rows = (await db.execute(stmt)).scalars().all() for t in rows: related_tasks.append({ "id": str(t.id), "title": t.title, "status": t.status, "priority": t.priority, "due_date": t.due_date.isoformat() if t.due_date else None, }) return { "task_id": str(task_id), "related_messages": related_messages, "related_tasks": related_tasks, } async def _get_task_or_404(task_id: UUID, db: DBSession) -> Task: stmt = select(Task).where(Task.id == task_id, Task.deleted_at.is_(None)) task = (await db.execute(stmt)).scalar_one_or_none() if task is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="找不到此任務") return task @router.get("/{task_id}", response_model=TaskPublic, summary="取得單一任務") async def get_task( task_id: UUID, current_user: CurrentUser, db: DBSession, ) -> Task: return await _get_task_or_404(task_id, db) @router.patch("/{task_id}", response_model=TaskPublic, summary="更新任務") async def update_task( task_id: UUID, payload: TaskUpdateRequest, current_user: CurrentUser, db: DBSession, ) -> Task: task = await _get_task_or_404(task_id, db) # 紀錄原狀態,等下用來判斷要不要改 completed_at old_status = task.status # 用 model_dump(exclude_unset=True) 只取「前端有傳的欄位」, # 沒傳的欄位保持原值(這就是 PATCH semantics) update_data = payload.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(task, field, value) # 處理 completed_at 自動化: # - 從非 done 變成 done → 記下完成時間 # - 從 done 變回非 done → 清掉完成時間 if old_status != task.status: if task.status == "done": task.completed_at = datetime.now(timezone.utc) elif old_status == "done": task.completed_at = None await db.commit() await db.refresh(task) return task def _can_delete_task(user, task: Task) -> bool: """誰能刪這個任務?建立者、被指派的人、或公司 owner。""" return ( user.role == "owner" or task.created_by == user.id or task.assignee_id == user.id ) @router.delete( "/{task_id}", status_code=status.HTTP_204_NO_CONTENT, summary="軟刪除任務", ) async def delete_task( task_id: UUID, current_user: CurrentUser, db: DBSession, ) -> None: task = await _get_task_or_404(task_id, db) if not _can_delete_task(current_user, task): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="只有建立者、負責人或公司 owner 能刪除", ) task.deleted_at = datetime.now(timezone.utc) await db.commit()