代码提交

offline-backend-20260512/backend/app/gateway/routers/taskcop_tasks.py

@@ -0,0 +1,336 @@
+"""Temporary compatibility API for the TaskCOP situation-overview frontend.
+
+The Vue page at ``/situation-overview/action/sentiment`` currently talks to a
+legacy Consumer service. These routes retain that service's paths and response
+envelopes while persisting data in DeerFlow's own database:
+
+* ``POST /cop/saveSuperiorTask``
+* ``GET /taskAnalyseSearch/cop-task-three-list``
+* ``GET /taskAnalyseSearch/cop-my-tasks``
+* ``GET /taskAnalyseSearch/cop-task-detail``
+
+The separately deployed legacy UI signs in through DeerFlow before calling
+these routes. The Gateway authentication middleware therefore supplies the
+creator identity, which keeps the "my tasks" endpoint correctly scoped.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Body, HTTPException, Query, Request
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, ValidationError, field_validator
+
+from deerflow.runtime.user_context import DEFAULT_USER_ID, get_effective_user_id
+
+router = APIRouter(tags=["taskcop-compat"])
+
+
+class SaveSuperiorTaskRequest(BaseModel):
+    """Payload posted by the legacy ``AddTaskDialog`` component."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    issuing_dept: str = Field(default="", validation_alias=AliasChoices("issuingDept", "issuing_dept"))
+    overview: str = Field(min_length=1, max_length=50_000)
+    task_direction: str = Field(
+        default="W方向", validation_alias=AliasChoices("taskDirection", "task_direction"), max_length=64
+    )
+    task_end_date: datetime | None = Field(
+        default=None, validation_alias=AliasChoices("taskEndDate", "task_end_date")
+    )
+    task_level: str = Field(
+        default="1", validation_alias=AliasChoices("taskLevel", "task_level"), max_length=32
+    )
+    task_name: str = Field(
+        min_length=1, max_length=512, validation_alias=AliasChoices("taskName", "task_name")
+    )
+    task_no: str = Field(default="", validation_alias=AliasChoices("taskNo", "task_no"), max_length=255)
+    task_start_date: datetime | None = Field(
+        default=None, validation_alias=AliasChoices("taskStartDate", "task_start_date")
+    )
+    task_status: int = Field(default=0, validation_alias=AliasChoices("taskStatus", "task_status"))
+    task_superior_flag: int = Field(
+        default=0, validation_alias=AliasChoices("taskSuperiorFlag", "task_superior_flag")
+    )
+    task_type: int = Field(default=1, validation_alias=AliasChoices("taskType", "task_type"))
+    parent_task_id: str | None = Field(
+        default=None, validation_alias=AliasChoices("pTaskId", "parent_task_id"), max_length=64
+    )
+
+
+class TaskImportRecord(BaseModel):
+    """One task accepted by the JSON import page.
+
+    Both the legacy ``taskName/taskContent`` keys and the internal snake_case
+    keys are accepted, so exported task-list files can be uploaded unchanged.
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: str = Field(..., validation_alias=AliasChoices("id", "taskId"), max_length=64)
+    task_name: str = Field(
+        ..., validation_alias=AliasChoices("taskName", "task_name"), min_length=1, max_length=512
+    )
+    overview: str = Field(
+        default="", validation_alias=AliasChoices("taskContent", "overview"), max_length=50_000
+    )
+    issuing_dept: str = Field(default="", validation_alias=AliasChoices("issuingDept", "issuing_dept"))
+    task_direction: str = Field(
+        default="W方向", validation_alias=AliasChoices("taskDirection", "task_direction"), max_length=64
+    )
+    task_end_date: datetime | None = Field(
+        default=None, validation_alias=AliasChoices("taskEndDate", "task_end_date")
+    )
+    task_level: str = Field(
+        default="1", validation_alias=AliasChoices("taskLevel", "task_level"), max_length=32
+    )
+    task_no: str = Field(default="", validation_alias=AliasChoices("taskNo", "task_no"), max_length=255)
+    task_start_date: datetime | None = Field(
+        default=None, validation_alias=AliasChoices("taskStartDate", "task_start_date")
+    )
+    task_status: int = Field(default=0, validation_alias=AliasChoices("taskStatus", "task_status"))
+    task_superior_flag: int = Field(
+        default=0, validation_alias=AliasChoices("taskSuperiorFlag", "task_superior_flag")
+    )
+    task_type: int = Field(default=1, validation_alias=AliasChoices("taskType", "task_type"))
+    parent_task_id: str | None = Field(
+        default=None, validation_alias=AliasChoices("pTaskId", "parent_task_id"), max_length=64
+    )
+
+    @field_validator("id", "task_name")
+    @classmethod
+    def strip_required_text(cls, value: str) -> str:
+        value = value.strip()
+        if not value:
+            raise ValueError("must not be blank")
+        return value
+
+
+def _store(request: Request):
+    store = getattr(request.app.state, "taskcop_task_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="TaskCOP task store not available")
+    return store
+
+
+def _request_user_id(request: Request) -> str:
+    user = getattr(request.state, "user", None)
+    if user is not None:
+        return str(user.id)
+    try:
+        return get_effective_user_id()
+    except RuntimeError:
+        return DEFAULT_USER_ID
+
+
+def _parse_statuses(task_status: str | None) -> set[int] | None:
+    if not task_status or not task_status.strip():
+        return None
+    try:
+        return {int(value.strip()) for value in task_status.split(",") if value.strip()}
+    except ValueError:
+        # The legacy service treats malformed filters as no filter.
+        return None
+
+
+def _bounded_page(page_num: int, page_size: int) -> tuple[int, int]:
+    return max(page_num, 1), min(max(page_size, 1), 100)
+
+
+async def _list_tasks(
+    request: Request,
+    *,
+    mine: bool,
+    content: str,
+    page_num: int,
+    page_size: int,
+    task_status: str | None,
+    task_direction: str | None,
+    start_time: datetime | None,
+    end_time: datetime | None,
+) -> dict:
+    page_num, page_size = _bounded_page(page_num, page_size)
+    total, records = await _store(request).list_tasks(
+        created_by=_request_user_id(request) if mine else None,
+        keyword=content,
+        task_direction=task_direction,
+        task_statuses=_parse_statuses(task_status),
+        start_time=start_time,
+        end_time=end_time,
+        page_num=page_num,
+        page_size=page_size,
+    )
+    return {"state": 200, "msg": "查询成功", "data": {"total": total, "records": records}}
+
+
+TaskListQuery = Annotated[str, Query(max_length=512)]
+
+
+@router.post("/cop/saveSuperiorTask", status_code=201)
+async def save_superior_task(request: Request, body: SaveSuperiorTaskRequest) -> dict:
+    """Create a task using the legacy ``{state, msg, data}`` response shape."""
+    row = await _store(request).create_task(
+        {
+            **body.model_dump(),
+            "task_name": body.task_name.strip(),
+            "overview": body.overview.strip(),
+            "created_by": _request_user_id(request),
+        }
+    )
+    return {"state": 201, "msg": "创建任务成功", "data": row}
+
+
+def _import_task_rows(payload: Any) -> list[Any]:
+    """Accept array, ``{tasks: [...]}``, or legacy list-response envelopes."""
+    if isinstance(payload, list):
+        rows = payload
+    elif isinstance(payload, dict):
+        data = payload.get("tasks", payload.get("records", payload.get("data", payload)))
+        if isinstance(data, dict):
+            data = data.get("records", data.get("tasks", data))
+        rows = data if isinstance(data, list) else [data]
+    else:
+        raise HTTPException(status_code=422, detail="任务导入文件必须是 JSON 数组或包含 tasks/records 的对象")
+
+    if not rows or not all(isinstance(row, dict) for row in rows):
+        raise HTTPException(status_code=422, detail="任务导入文件中至少需要一条任务对象")
+    return rows
+
+
+@router.post("/api/taskcop/import/tasks")
+async def import_tasks(request: Request, payload: Any = Body(...)) -> dict:
+    """Import tasks by id; a duplicate id atomically replaces that task."""
+    rows = _import_task_rows(payload)
+    created = 0
+    updated = 0
+    records: list[dict[str, Any]] = []
+    for index, raw in enumerate(rows, start=1):
+        try:
+            item = TaskImportRecord.model_validate(raw)
+        except ValidationError as error:
+            raise HTTPException(
+                status_code=422,
+                detail={"message": f"第 {index} 条任务数据格式不正确", "errors": error.errors()},
+            ) from error
+        record, inserted = await _store(request).upsert_task(
+            {
+                **item.model_dump(),
+                "task_name": item.task_name,
+                "overview": item.overview,
+                "created_by": _request_user_id(request),
+            }
+        )
+        records.append(record)
+        created += int(inserted)
+        updated += int(not inserted)
+
+    return {
+        "state": 200,
+        "msg": "任务导入成功",
+        "data": {"created": created, "updated": updated, "records": records},
+    }
+
+
+@router.delete("/cop/tasks/{task_id}")
+async def delete_task(request: Request, task_id: str) -> dict:
+    """Delete a TaskCOP task and its shared report document."""
+    task_id = task_id.strip()
+    if not task_id:
+        raise HTTPException(status_code=422, detail="task id is required")
+
+    task_store = _store(request)
+    if await task_store.get_task(task_id) is None:
+        raise HTTPException(status_code=404, detail="task not found")
+    if not await task_store.delete_task(task_id):
+        raise HTTPException(status_code=404, detail="task not found")
+
+    report_store = getattr(request.app.state, "task_report_store", None)
+    if report_store is not None:
+        await report_store.delete_by_task(task_id)
+    return {"state": 200, "msg": "task deleted", "data": {"id": task_id}}
+
+
+@router.get("/taskAnalyseSearch/cop-task-three-list")
+async def list_task_three(
+    request: Request,
+    content: TaskListQuery = "",
+    pageNum: int = Query(default=1),
+    pageSize: int = Query(default=10),
+    taskStatus: str | None = Query(default=None),
+    taskDirection: str | None = Query(default=None),
+    startTime: datetime | None = Query(default=None),
+    endTime: datetime | None = Query(default=None),
+) -> dict:
+    """List all tasks for the sentiment screen's default "全部" tab."""
+    return await _list_tasks(
+        request,
+        mine=False,
+        content=content,
+        page_num=pageNum,
+        page_size=pageSize,
+        task_status=taskStatus,
+        task_direction=taskDirection,
+        start_time=startTime,
+        end_time=endTime,
+    )
+
+
+@router.get("/taskAnalyseSearch/cop-my-tasks")
+async def list_my_tasks(
+    request: Request,
+    content: TaskListQuery = "",
+    pageNum: int = Query(default=1),
+    pageSize: int = Query(default=10),
+    taskStatus: str | None = Query(default=None),
+    taskDirection: str | None = Query(default=None),
+    startTime: datetime | None = Query(default=None),
+    endTime: datetime | None = Query(default=None),
+) -> dict:
+    """List the caller's tasks for the sentiment screen's "我的" tab."""
+    return await _list_tasks(
+        request,
+        mine=True,
+        content=content,
+        page_num=pageNum,
+        page_size=pageSize,
+        task_status=taskStatus,
+        task_direction=taskDirection,
+        start_time=startTime,
+        end_time=endTime,
+    )
+
+
+@router.get("/taskAnalyseSearch/cop-task-list")
+async def list_task_analysis(
+    request: Request,
+    content: TaskListQuery = "",
+    pageNum: int = Query(default=1),
+    pageSize: int = Query(default=10),
+    taskStatus: str | None = Query(default=None),
+    taskDirection: str | None = Query(default=None),
+    startTime: datetime | None = Query(default=None),
+    endTime: datetime | None = Query(default=None),
+) -> dict:
+    """Compatibility alias used by the neighbouring task-analysis screen."""
+    return await _list_tasks(
+        request,
+        mine=False,
+        content=content,
+        page_num=pageNum,
+        page_size=pageSize,
+        task_status=taskStatus,
+        task_direction=taskDirection,
+        start_time=startTime,
+        end_time=endTime,
+    )
+
+
+@router.get("/taskAnalyseSearch/cop-task-detail")
+async def get_task_detail(request: Request, taskId: str = Query(min_length=1, max_length=64)) -> dict:
+    """Return a task selected in the list or opened via a task link."""
+    row = await _store(request).get_task(taskId)
+    if row is None:
+        raise HTTPException(status_code=404, detail="任务不存在")
+    return {"state": 200, "msg": "查询成功", "data": row}