482 lines
17 KiB
Python
482 lines
17 KiB
Python
"""Scheduled task management API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import zipfile
|
|
from io import BytesIO
|
|
import mimetypes
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import FileResponse, Response, StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import get_current_user, get_local_provider
|
|
from deerflow.runtime.scheduler import get_scheduled_task_service
|
|
|
|
router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
|
|
|
|
|
|
class ScheduledTaskCreateRequest(BaseModel):
|
|
name: str = Field(default="Scheduled task")
|
|
prompt: str
|
|
schedule: str
|
|
timezone: str | None = None
|
|
execution_agent_name: str | None = None
|
|
execution_context: dict[str, Any] | None = None
|
|
|
|
|
|
class ScheduledTaskUpdateRequest(BaseModel):
|
|
name: str | None = None
|
|
prompt: str | None = None
|
|
schedule: str | None = None
|
|
timezone: str | None = None
|
|
enabled: bool | None = None
|
|
execution_agent_name: str | None = None
|
|
execution_context: dict[str, Any] | None = None
|
|
|
|
|
|
class ScheduledTaskDeliveryProfileUpdateRequest(BaseModel):
|
|
element_user_id: str | None = None
|
|
|
|
|
|
class ScheduledTaskSubscribeRequest(BaseModel):
|
|
notify_element: bool = False
|
|
|
|
|
|
class ScheduledTaskRunFileUpdateRequest(BaseModel):
|
|
content: str = Field(..., description="New text content to write into the run file")
|
|
|
|
|
|
_MAX_RUN_FILE_EDIT_BYTES = 5 * 1024 * 1024 # 5 MB
|
|
|
|
|
|
async def _user_id(request: Request) -> str:
|
|
user_id = await get_current_user(request)
|
|
if not user_id:
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
return user_id
|
|
|
|
|
|
async def _enrich_with_creator(tasks: list[dict[str, Any]] | dict[str, Any] | None) -> Any:
|
|
"""Attach ``creator_email`` / ``creator_user_id`` to one or many task dicts.
|
|
|
|
Looks up the auth user once per distinct ``user_id`` in the batch so we do
|
|
not pay N+1 round trips when rendering a long task list. Tolerates missing
|
|
user records (deleted accounts) by leaving the email blank.
|
|
"""
|
|
if tasks is None:
|
|
return tasks
|
|
items = tasks if isinstance(tasks, list) else [tasks]
|
|
|
|
user_ids = {str(t.get("user_id")) for t in items if isinstance(t, dict) and t.get("user_id")}
|
|
if not user_ids:
|
|
return tasks
|
|
|
|
try:
|
|
provider = get_local_provider()
|
|
except Exception:
|
|
# Auth provider not initialised (e.g. memory-mode test harness): keep
|
|
# the bare user_id and let the frontend render the raw id.
|
|
provider = None
|
|
|
|
email_by_id: dict[str, str] = {}
|
|
if provider is not None:
|
|
for uid in user_ids:
|
|
try:
|
|
user = await provider.get_user(uid)
|
|
except Exception:
|
|
user = None
|
|
if user is not None:
|
|
email_by_id[uid] = user.email
|
|
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
uid = item.get("user_id")
|
|
if uid is None:
|
|
continue
|
|
uid_str = str(uid)
|
|
item["creator_user_id"] = uid_str
|
|
item["creator_email"] = email_by_id.get(uid_str, "")
|
|
return tasks
|
|
|
|
|
|
def _safe_filename(value: str, default: str = "scheduled-task-result") -> str:
|
|
name = re.sub(r'[\\/:*?"<>|\r\n]+', "-", value).strip(" .")
|
|
return name or default
|
|
|
|
|
|
def _result_download_name(result: dict[str, Any], *, extension: str = "md") -> str:
|
|
title = result.get("title") if isinstance(result, dict) else None
|
|
return f"{_safe_filename(str(title or 'scheduled-task-result'))}.{extension}"
|
|
|
|
|
|
def _result_markdown(result: dict[str, Any]) -> str:
|
|
title = str(result.get("title") or "Scheduled Task Result")
|
|
content = str(result.get("content") or "")
|
|
return f"# {title}\n\n{content}".strip() + "\n"
|
|
|
|
|
|
def _utf8_sig_bytes(content: str) -> bytes:
|
|
return content.encode("utf-8-sig")
|
|
|
|
|
|
def _content_disposition(filename: str) -> str:
|
|
quoted = quote(filename)
|
|
return f"attachment; filename*=UTF-8''{quoted}"
|
|
|
|
|
|
def _run_files(run: dict[str, Any]) -> list:
|
|
result = run.get("result") or {}
|
|
return result.get("files") if isinstance(result, dict) and isinstance(result.get("files"), list) else []
|
|
|
|
|
|
def _run_file_path(run: dict[str, Any], file_index: int) -> tuple[dict[str, Any], Path]:
|
|
files = _run_files(run)
|
|
if file_index < 0 or file_index >= len(files):
|
|
raise HTTPException(status_code=404, detail="Result file not found")
|
|
item = files[file_index]
|
|
if not isinstance(item, dict) or not isinstance(item.get("path"), str):
|
|
raise HTTPException(status_code=404, detail="Result file not found")
|
|
path = Path(item["path"])
|
|
if not path.exists() or not path.is_file():
|
|
raise HTTPException(status_code=404, detail="Result file is no longer available")
|
|
return item, path
|
|
|
|
|
|
def _is_text_previewable(path: Path, mime_type: str | None) -> bool:
|
|
if mime_type and (mime_type.startswith("text/") or mime_type in {"application/json", "application/xml"}):
|
|
return True
|
|
return path.suffix.lower() in {".md", ".markdown", ".txt", ".json", ".csv", ".xml", ".yaml", ".yml", ".log"}
|
|
|
|
|
|
def _is_text_download(path: Path, mime_type: str | None) -> bool:
|
|
return _is_text_previewable(path, mime_type)
|
|
|
|
|
|
def _read_text_download_bytes(path: Path) -> bytes:
|
|
try:
|
|
content = path.read_text(encoding="utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
content = path.read_text(encoding="utf-8", errors="replace")
|
|
return _utf8_sig_bytes(content)
|
|
|
|
|
|
@router.get("/thread")
|
|
async def get_scheduler_thread(request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
return await service.ensure_scheduler_thread(user_id)
|
|
|
|
|
|
@router.get("")
|
|
async def list_scheduled_tasks(request: Request) -> list[dict]:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
tasks = await service.list_tasks(user_id)
|
|
await _enrich_with_creator(tasks)
|
|
return tasks
|
|
|
|
|
|
@router.get("/{task_id}/runs")
|
|
async def list_scheduled_task_runs(task_id: str, request: Request, limit: int = 50) -> list[dict]:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
runs = await service.list_runs(user_id, task_id, limit=max(1, min(limit, 200)))
|
|
return runs
|
|
|
|
|
|
@router.get("/runs/{run_id}")
|
|
async def get_scheduled_task_run(run_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
run = await service.get_run(user_id, run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
return run
|
|
|
|
|
|
@router.delete("/runs/{run_id}")
|
|
async def delete_scheduled_task_run(run_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
try:
|
|
deleted = await service.delete_run(user_id, run_id)
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc))
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/runs/{run_id}/files/{file_index}")
|
|
async def get_scheduled_task_run_file(run_id: str, file_index: int, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
run = await service.get_run(user_id, run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
item, path = _run_file_path(run, file_index)
|
|
mime_type = item.get("mime_type") or mimetypes.guess_type(path.name)[0]
|
|
previewable = _is_text_previewable(path, mime_type)
|
|
content = None
|
|
if previewable:
|
|
try:
|
|
content = path.read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
content = path.read_text(encoding="utf-8", errors="replace")
|
|
return {
|
|
"index": file_index,
|
|
"name": item.get("name") or path.name,
|
|
"mime_type": mime_type,
|
|
"kind": item.get("kind"),
|
|
"size": path.stat().st_size,
|
|
"previewable": previewable,
|
|
"content": content,
|
|
}
|
|
|
|
|
|
@router.put("/runs/{run_id}/files/{file_index}")
|
|
async def update_scheduled_task_run_file(
|
|
run_id: str,
|
|
file_index: int,
|
|
body: ScheduledTaskRunFileUpdateRequest,
|
|
request: Request,
|
|
) -> dict:
|
|
if len(body.content.encode("utf-8")) > _MAX_RUN_FILE_EDIT_BYTES:
|
|
raise HTTPException(status_code=413, detail="Edited content exceeds the 5 MB limit")
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
try:
|
|
return await service.update_run_file(user_id, run_id, file_index, body.content)
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc))
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
|
|
|
|
@router.get("/runs/{run_id}/download")
|
|
async def download_scheduled_task_run(run_id: str, request: Request):
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
run = await service.get_run(user_id, run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
|
|
result = run.get("result") or {}
|
|
if not isinstance(result, dict):
|
|
result = {}
|
|
files = result.get("files") if isinstance(result.get("files"), list) else []
|
|
accessible_files: list[tuple[str, Path]] = []
|
|
for item in files:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
path_value = item.get("path")
|
|
if not isinstance(path_value, str) or not path_value:
|
|
continue
|
|
path = Path(path_value)
|
|
if path.exists() and path.is_file():
|
|
accessible_files.append((str(item.get("name") or path.name), path))
|
|
|
|
if not accessible_files:
|
|
body = _utf8_sig_bytes(_result_markdown(result))
|
|
return Response(
|
|
content=body,
|
|
media_type="text/markdown; charset=utf-8",
|
|
headers={"Content-Disposition": _content_disposition(_result_download_name(result))},
|
|
)
|
|
|
|
archive = BytesIO()
|
|
with zipfile.ZipFile(archive, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr("result.md", _utf8_sig_bytes(_result_markdown(result)))
|
|
for name, path in accessible_files:
|
|
arcname = _safe_filename(name, path.name)
|
|
mime_type = mimetypes.guess_type(path.name)[0]
|
|
if _is_text_download(path, mime_type):
|
|
zf.writestr(arcname, _read_text_download_bytes(path))
|
|
else:
|
|
zf.write(path, arcname=arcname)
|
|
archive.seek(0)
|
|
return StreamingResponse(
|
|
archive,
|
|
media_type="application/zip",
|
|
headers={"Content-Disposition": _content_disposition(_result_download_name(result, extension="zip"))},
|
|
)
|
|
|
|
|
|
@router.get("/runs/{run_id}/files/{file_index}/download")
|
|
async def download_scheduled_task_run_file(run_id: str, file_index: int, request: Request):
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
run = await service.get_run(user_id, run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
item, path = _run_file_path(run, file_index)
|
|
filename = _safe_filename(str(item.get("name") or path.name), path.name)
|
|
mime_type = item.get("mime_type") or mimetypes.guess_type(path.name)[0]
|
|
if _is_text_download(path, mime_type):
|
|
return Response(
|
|
content=_read_text_download_bytes(path),
|
|
media_type=f"{mime_type or 'text/plain'}; charset=utf-8",
|
|
headers={"Content-Disposition": _content_disposition(filename)},
|
|
)
|
|
return FileResponse(path, filename=filename, media_type=item.get("mime_type") or "application/octet-stream")
|
|
|
|
|
|
@router.get("/delivery-profile")
|
|
async def get_delivery_profile(request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
return await service.get_delivery_profile(user_id)
|
|
|
|
|
|
@router.put("/delivery-profile")
|
|
async def update_delivery_profile(body: ScheduledTaskDeliveryProfileUpdateRequest, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
return await service.update_delivery_profile(user_id, body.model_dump(exclude_unset=True))
|
|
|
|
|
|
@router.post("")
|
|
async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.create_task(
|
|
user_id=user_id,
|
|
name=body.name,
|
|
prompt=body.prompt,
|
|
schedule=body.schedule,
|
|
timezone=body.timezone,
|
|
execution_agent_name=body.execution_agent_name,
|
|
execution_context=body.execution_context,
|
|
)
|
|
await _enrich_with_creator(task)
|
|
return task
|
|
|
|
|
|
@router.patch("/{task_id}")
|
|
async def update_scheduled_task(task_id: str, body: ScheduledTaskUpdateRequest, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.update_task(user_id, task_id, body.model_dump(exclude_unset=True))
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
await _enrich_with_creator(task)
|
|
return task
|
|
|
|
|
|
@router.delete("/{task_id}")
|
|
async def delete_scheduled_task(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
deleted = await service.delete_task(user_id, task_id)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return {"success": True}
|
|
|
|
|
|
@router.post("/{task_id}/pause")
|
|
async def pause_scheduled_task(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.pause_task(user_id, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return task
|
|
|
|
|
|
@router.post("/{task_id}/resume")
|
|
async def resume_scheduled_task(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.resume_task(user_id, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return task
|
|
|
|
|
|
@router.post("/{task_id}/trigger")
|
|
async def trigger_scheduled_task(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.trigger_task(user_id, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return {"success": True, "task": task}
|
|
|
|
|
|
# --- Publish / subscribe ---
|
|
|
|
|
|
@router.get("/published")
|
|
async def list_published_scheduled_tasks(request: Request, limit: int = 50, offset: int = 0) -> list[dict]:
|
|
service = get_scheduled_task_service()
|
|
await _user_id(request)
|
|
tasks = await service.list_published(limit=max(1, min(limit, 200)), offset=max(0, offset))
|
|
# published list uses owner_user_id; resolve it into a creator_email here too
|
|
# so the discovery view can show the author's account.
|
|
for task in tasks:
|
|
if isinstance(task, dict) and task.get("owner_user_id"):
|
|
task["user_id"] = task["owner_user_id"]
|
|
await _enrich_with_creator(tasks)
|
|
for task in tasks:
|
|
if isinstance(task, dict):
|
|
task.pop("user_id", None)
|
|
return tasks
|
|
|
|
|
|
@router.post("/{task_id}/publish")
|
|
async def publish_scheduled_task(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.publish_task(user_id, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return task
|
|
|
|
|
|
@router.post("/{task_id}/unpublish")
|
|
async def unpublish_scheduled_task(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
task = await service.unpublish_task(user_id, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return task
|
|
|
|
|
|
@router.post("/{task_id}/subscribe")
|
|
async def subscribe_scheduled_task(task_id: str, body: ScheduledTaskSubscribeRequest, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
sub = await service.subscribe_task(user_id, task_id, notify_element=body.notify_element)
|
|
if sub is None:
|
|
raise HTTPException(status_code=404, detail="Task is not available for subscription")
|
|
return sub
|
|
|
|
|
|
@router.patch("/{task_id}/subscribe")
|
|
async def update_scheduled_task_subscription(task_id: str, body: ScheduledTaskSubscribeRequest, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
sub = await service.subscribe_task(user_id, task_id, notify_element=body.notify_element)
|
|
if sub is None:
|
|
raise HTTPException(status_code=404, detail="Subscription not found")
|
|
return sub
|
|
|
|
|
|
@router.delete("/{task_id}/subscribe")
|
|
async def delete_scheduled_task_subscription(task_id: str, request: Request) -> dict:
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
deleted = await service.unsubscribe_task(user_id, task_id)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail="Subscription not found")
|
|
return {"success": True}
|