207 lines
8.3 KiB
Python
207 lines
8.3 KiB
Python
"""CRUD API for **task-scoped** roundtable-planning drafts (聊天记录).
|
||
|
||
This is the dedicated store for roundtable sessions opened via a taskId deep-link
|
||
(无界嵌入抽屉). It is physically and semantically separate from the per-user
|
||
``/api/roundtable-drafts`` store:
|
||
|
||
* records are keyed **only by ``task_id``** (one task → many records);
|
||
* there is **no user 分权** — any authenticated caller may read/write/delete a
|
||
task's records (the deep-link audience is shared); ``created_by`` is recorded
|
||
for audit only.
|
||
|
||
Routes (prefix ``/api/roundtable-task-drafts``):
|
||
GET "/by-task/{task_id}" list all drafts for a task (lightweight)
|
||
GET "/by-task/{task_id}/latest" newest draft for a task (full)
|
||
POST "" create (``task_id`` required)
|
||
GET "/{draft_id}" full draft (incl. step1/step2/step3)
|
||
PUT "/{draft_id}" partial update
|
||
DELETE "/{draft_id}" delete
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from fastapi import APIRouter, HTTPException, Request
|
||
from pydantic import BaseModel, Field
|
||
|
||
from deerflow.persistence.roundtable_drafts.sql import DraftConcurrentWriteError
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/roundtable-task-drafts", tags=["roundtable-task-drafts"])
|
||
|
||
|
||
# ── schemas ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TaskDraftMetaResponse(BaseModel):
|
||
id: str
|
||
task_id: str
|
||
title: str = ""
|
||
furthest_step: int = 1
|
||
# 乐观锁版本号：PUT 时经 expected_version 回传，冲突 → 409 + current_version。
|
||
version: int = 0
|
||
# 本次会商所用业务链名（历史卡片展示）；无链条（自由/recommend 模式）时为 None。
|
||
chain_title: str | None = None
|
||
created_at: datetime | str | None = None
|
||
updated_at: datetime | str | None = None
|
||
# Latest background-job status for this draft (history-dropdown loading badge).
|
||
status: str | None = None
|
||
job_id: str | None = None
|
||
|
||
|
||
class TaskDraftListResponse(BaseModel):
|
||
drafts: list[TaskDraftMetaResponse]
|
||
|
||
|
||
class TaskDraftResponse(TaskDraftMetaResponse):
|
||
step1: dict[str, Any] | None = None
|
||
step2: dict[str, Any] | None = None
|
||
step3: dict[str, Any] | None = None
|
||
|
||
|
||
class TaskDraftCreateRequest(BaseModel):
|
||
id: str | None = Field(default=None, max_length=64)
|
||
task_id: str = Field(max_length=128)
|
||
title: str = Field(default="", max_length=512)
|
||
furthest_step: int = Field(default=1, ge=1, le=3)
|
||
step1: dict[str, Any] | None = None
|
||
step2: dict[str, Any] | None = None
|
||
step3: dict[str, Any] | None = None
|
||
|
||
|
||
class TaskDraftUpdateRequest(BaseModel):
|
||
title: str | None = Field(default=None, max_length=512)
|
||
furthest_step: int | None = Field(default=None, ge=1, le=3)
|
||
step1: dict[str, Any] | None = None
|
||
step2: dict[str, Any] | None = None
|
||
step3: dict[str, Any] | None = None
|
||
# 乐观锁：客户端基于的草稿版本号。缺省（旧客户端）= 盲写兼容；
|
||
# 提供时版本不匹配 → 409，body 带 current_version 供重拉合并。
|
||
expected_version: int | None = Field(default=None, ge=0)
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _current_user_id(request: Request) -> str:
|
||
user = getattr(request.state, "user", None)
|
||
if user is not None:
|
||
return str(user.id)
|
||
return get_effective_user_id()
|
||
|
||
|
||
def _get_store(request: Request):
|
||
store = getattr(request.app.state, "roundtable_task_draft_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Roundtable task draft store not available")
|
||
return store
|
||
|
||
|
||
async def _enrich_status(request: Request, rows: list[dict[str, Any]]) -> None:
|
||
"""Attach the latest background-job status/jobId per draft (best-effort).
|
||
|
||
Uses the job store's draft-id keyed lookup (no user filter) — a task's drafts
|
||
span users. Failure or a missing job store silently degrades to no status.
|
||
"""
|
||
job_store = getattr(request.app.state, "roundtable_job_store", None)
|
||
if job_store is None or not rows:
|
||
return
|
||
try:
|
||
status_map = await job_store.status_map_by_drafts([r.get("id") for r in rows])
|
||
except AttributeError:
|
||
status_map = {}
|
||
except Exception: # noqa: BLE001 — 富化失败不能拖垮列表
|
||
logger.warning("roundtable task draft list: job status enrichment failed", exc_info=True)
|
||
status_map = {}
|
||
for r in rows:
|
||
info = status_map.get(r.get("id"))
|
||
if info:
|
||
r["status"] = info.get("status")
|
||
r["job_id"] = info.get("jobId")
|
||
|
||
|
||
# ── routes ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.get("/by-task/{task_id}", response_model=TaskDraftListResponse)
|
||
async def list_drafts_by_task(request: Request, task_id: str) -> TaskDraftListResponse:
|
||
"""All drafts for a task, newest-first (一个 taskId 多条记录，不分权)。"""
|
||
store = _get_store(request)
|
||
rows = await store.list_drafts_by_task(task_id)
|
||
await _enrich_status(request, rows)
|
||
return TaskDraftListResponse(drafts=[TaskDraftMetaResponse(**r) for r in rows])
|
||
|
||
|
||
@router.get("/by-task/{task_id}/latest", response_model=TaskDraftResponse)
|
||
async def get_latest_draft_by_task(request: Request, task_id: str) -> TaskDraftResponse:
|
||
"""Newest draft for a task; 404 if none. 回显最新一条聊天记录用。"""
|
||
store = _get_store(request)
|
||
row = await store.get_latest_draft_by_task(task_id)
|
||
if row is None:
|
||
raise HTTPException(status_code=404, detail="No draft for this task")
|
||
return TaskDraftResponse(**row)
|
||
|
||
|
||
@router.post("", response_model=TaskDraftResponse, status_code=201)
|
||
async def create_draft(request: Request, body: TaskDraftCreateRequest) -> TaskDraftResponse:
|
||
store = _get_store(request)
|
||
task_id = (body.task_id or "").strip()
|
||
if not task_id:
|
||
raise HTTPException(status_code=400, detail="task_id is required")
|
||
row = await store.create_draft(
|
||
{
|
||
"id": (body.id or uuid4().hex),
|
||
"task_id": task_id,
|
||
"title": body.title.strip(),
|
||
"furthest_step": body.furthest_step,
|
||
"step1": body.step1,
|
||
"step2": body.step2,
|
||
"step3": body.step3,
|
||
},
|
||
created_by=_current_user_id(request),
|
||
)
|
||
return TaskDraftResponse(**row)
|
||
|
||
|
||
@router.get("/{draft_id}", response_model=TaskDraftResponse)
|
||
async def get_draft(request: Request, draft_id: str) -> TaskDraftResponse:
|
||
store = _get_store(request)
|
||
row = await store.get_draft(draft_id)
|
||
if row is None:
|
||
raise HTTPException(status_code=404, detail="Draft not found")
|
||
return TaskDraftResponse(**row)
|
||
|
||
|
||
@router.put("/{draft_id}", response_model=TaskDraftResponse)
|
||
async def update_draft(request: Request, draft_id: str, body: TaskDraftUpdateRequest) -> TaskDraftResponse:
|
||
store = _get_store(request)
|
||
fields = body.model_dump(exclude_unset=True)
|
||
try:
|
||
row = await store.update_draft(draft_id, **fields)
|
||
except DraftConcurrentWriteError as err:
|
||
# 版本冲突：detail 里带 current_version，前端据此重拉最新草稿、本地合并后重试。
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"error": "draft_version_conflict",
|
||
"message": f"草稿已被并发修改（expected={err.expected}, actual={err.actual}）",
|
||
"current_version": err.actual,
|
||
},
|
||
) from err
|
||
if row is None:
|
||
raise HTTPException(status_code=404, detail="Draft not found")
|
||
return TaskDraftResponse(**row)
|
||
|
||
|
||
@router.delete("/{draft_id}", status_code=204)
|
||
async def delete_draft(request: Request, draft_id: str) -> None:
|
||
store = _get_store(request)
|
||
deleted = await store.delete_draft(draft_id)
|
||
if not deleted:
|
||
raise HTTPException(status_code=404, detail="Draft not found")
|