858 lines
33 KiB
Python
858 lines
33 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, Query, Request
|
|
from fastapi.responses import FileResponse, Response, StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import (
|
|
get_checkpointer,
|
|
get_current_user,
|
|
get_local_provider,
|
|
get_optional_user_from_request,
|
|
get_run_event_store,
|
|
get_run_store,
|
|
get_tag_store,
|
|
)
|
|
from deerflow.runtime import serialize_channel_values
|
|
from deerflow.runtime.scheduler import DuplicateTaskNameError, get_scheduled_task_service
|
|
|
|
router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
|
|
|
|
|
|
class ScheduledTaskCreateRequest(BaseModel):
|
|
name: str = Field(default="Scheduled task")
|
|
description: str | None = None
|
|
prompt: str
|
|
schedule: str
|
|
timezone: str | None = None
|
|
execution_agent_name: str | None = None
|
|
execution_context: dict[str, Any] | None = None
|
|
require_markdown: bool = Field(
|
|
default=False,
|
|
description="Require the result to be Markdown and exceed the minimum length, else mark the run failed.",
|
|
)
|
|
|
|
|
|
class ScheduledTaskUpdateRequest(BaseModel):
|
|
name: str | None = None
|
|
description: 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
|
|
require_markdown: bool | 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 _is_admin(request: Request) -> bool:
|
|
"""True when the caller is an admin (or auth is disabled entirely)."""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return True
|
|
return getattr(user, "system_role", None) == "admin"
|
|
|
|
|
|
async def _load_run(service, request: Request, run_id: str, user_id: str) -> dict[str, Any] | None:
|
|
"""Owner/subscriber run lookup, with an admin fallback to any run.
|
|
|
|
Lets admins open run results for tasks they do not own (the admin
|
|
management view links into the shared run-detail page).
|
|
"""
|
|
run = await service.get_run(user_id, run_id)
|
|
if run is None and await _is_admin(request):
|
|
run = await service.admin_get_run(run_id)
|
|
return run
|
|
|
|
|
|
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("/html-templates")
|
|
async def list_html_templates() -> dict:
|
|
"""可选的 HTML 模板清单(供「模板数据转换器」内置 agent 填充用)。
|
|
|
|
返回 ``{"templates": [{id, name, description}]}``,前端定时任务表单在选中
|
|
「模板数据转换器」后据此渲染「HTML 模板」下拉。开放给任意登录用户。
|
|
"""
|
|
from deerflow.runtime.scheduler import template_fill
|
|
|
|
return {"templates": template_fill.list_templates()}
|
|
|
|
|
|
class TemplateRenderRequest(BaseModel):
|
|
"""「模板数据转换器」对话页的调试渲染入参。"""
|
|
|
|
template_id: str
|
|
# 智能体回复原文(含转换后的 JSON);后端用 extract_embedded_json 抽取后填充。
|
|
text: str | None = None
|
|
# 或直接给已解析好的对象(优先于 text)。
|
|
data: Any | None = None
|
|
|
|
|
|
@router.post("/template-fill/render")
|
|
async def render_template_fill(body: TemplateRenderRequest) -> dict:
|
|
"""调试用:把一段(智能体回复的)内容里的 JSON 填进所选 HTML 模板并返回完整 HTML。
|
|
|
|
供「模板数据转换器」对话页的「渲染预览」按钮调试智能体效果——不调模型,只做
|
|
抽取 → 校验 → 填充,把校验错误/缺失字段一并返回方便定位。「只展示」模板原样返回。
|
|
开放给任意登录用户。
|
|
"""
|
|
from deerflow.runtime.scheduler import template_fill
|
|
|
|
spec = template_fill.get_template(body.template_id)
|
|
if spec is None:
|
|
raise HTTPException(status_code=404, detail=f"未知模板：{body.template_id}")
|
|
|
|
# 「只展示」模板(no.2 等):原样返回模板 HTML。
|
|
if not spec.convertible:
|
|
html = template_fill.read_template_text(body.template_id)
|
|
return {"ok": True, "display_only": True, "errors": [], "gaps": [], "html": html}
|
|
|
|
obj = body.data
|
|
if obj is None and body.text:
|
|
obj = template_fill.extract_embedded_json(body.text)
|
|
if obj is None:
|
|
return {"ok": False, "errors": ["未从内容中识别到可用的 JSON"], "gaps": [], "html": None}
|
|
|
|
errors = template_fill.validate_data(body.template_id, obj)
|
|
gaps = template_fill.field_coverage_gaps(body.template_id, obj)
|
|
try:
|
|
html = template_fill.fill_template(body.template_id, obj)
|
|
except template_fill.TemplateError as exc:
|
|
return {"ok": False, "errors": [str(exc)], "gaps": gaps, "html": None}
|
|
return {"ok": not errors, "errors": errors, "gaps": gaps, "html": html}
|
|
|
|
|
|
@router.get("")
|
|
async def list_scheduled_tasks(
|
|
request: Request,
|
|
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on task name"),
|
|
tag_ids: list[str] | None = Query(default=None, description="Keep tasks carrying any of these tags (OR)"),
|
|
) -> 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)
|
|
|
|
tag_store = get_tag_store(request)
|
|
assignments = await tag_store.list_assignments("scheduled_task", [str(t.get("task_id")) for t in tasks])
|
|
for task in tasks:
|
|
task["tags"] = assignments.get(str(task.get("task_id")), [])
|
|
if search:
|
|
needle = search.strip().lower()
|
|
tasks = [t for t in tasks if needle in str(t.get("name", "")).lower()]
|
|
if tag_ids:
|
|
wanted = set(tag_ids)
|
|
tasks = [t for t in tasks if any(tg.get("id") in wanted for tg in t.get("tags", []))]
|
|
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 _load_run(service, request, run_id, user_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
return run
|
|
|
|
|
|
@router.get("/runs/{run_id}/html")
|
|
async def get_scheduled_task_run_html(run_id: str, request: Request) -> dict:
|
|
"""Return the generated HTML document for an ``html_page`` run."""
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
run = await _load_run(service, request, run_id, user_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
page = await service.get_html_page_for_run(run_id)
|
|
if page is None:
|
|
raise HTTPException(status_code=404, detail="该执行记录没有 HTML 页面")
|
|
return {
|
|
"run_id": run_id,
|
|
"task_id": page.get("task_id") or run.get("task_id"),
|
|
"title": page.get("title"),
|
|
"html": page.get("html_content"),
|
|
"style_summary": page.get("style_summary"),
|
|
"layout_summary": page.get("layout_summary"),
|
|
"page_id": page.get("id"),
|
|
"created_at": page.get("created_at"),
|
|
}
|
|
|
|
|
|
def _event_type_for_checkpoint_message(message: dict) -> str:
|
|
"""Map a serialized LangChain message to the same ``event_type`` strings
|
|
the run-event store uses, so the frontend's existing role labels keep
|
|
working for checkpoint-derived synthetic rows."""
|
|
raw_type = message.get("type") or message.get("role") or ""
|
|
if raw_type in {"human", "user"}:
|
|
return "llm.human.input"
|
|
if raw_type in {"ai", "assistant"}:
|
|
return "llm.ai.response"
|
|
if raw_type == "tool":
|
|
return "llm.tool.result"
|
|
if raw_type == "system":
|
|
return "llm.system"
|
|
return f"llm.{raw_type or 'message'}"
|
|
|
|
|
|
async def _checkpoint_messages_for_run(
|
|
request: Request, agent_run: dict, *, limit: int
|
|
) -> list[dict]:
|
|
"""Return a synthetic list of message rows pulled from the LangGraph
|
|
checkpoint for an in-flight run.
|
|
|
|
The event store is updated by the streaming pipeline and may stay empty
|
|
until the first message tuple is flushed — meanwhile the checkpoint
|
|
already carries the current state (input + accumulating AI deltas). We
|
|
project that into the same row shape the frontend uses for the event
|
|
store, so the run-detail page can show progress immediately."""
|
|
try:
|
|
checkpointer = get_checkpointer(request)
|
|
except Exception:
|
|
return []
|
|
thread_id = agent_run.get("thread_id")
|
|
if not thread_id:
|
|
return []
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
try:
|
|
tup = await checkpointer.aget_tuple(config)
|
|
except Exception:
|
|
return []
|
|
if tup is None:
|
|
return []
|
|
checkpoint = getattr(tup, "checkpoint", {}) or {}
|
|
channel_values = checkpoint.get("channel_values") or {}
|
|
try:
|
|
values = serialize_channel_values(channel_values)
|
|
except Exception:
|
|
return []
|
|
messages = values.get("messages") or []
|
|
|
|
rows: list[dict] = []
|
|
for index, message in enumerate(messages[-limit:]):
|
|
if not isinstance(message, dict):
|
|
continue
|
|
rows.append(
|
|
{
|
|
"seq": index + 1,
|
|
"event_type": _event_type_for_checkpoint_message(message),
|
|
"category": "message",
|
|
"content": message,
|
|
"metadata": {"source": "checkpoint"},
|
|
"created_at": None,
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
@router.get("/runs/{run_id}/messages")
|
|
async def get_scheduled_task_run_messages(run_id: str, request: Request, limit: int = 500) -> dict:
|
|
"""Agent messages produced by a scheduled run's underlying agent run.
|
|
|
|
Resolution order (most authoritative first):
|
|
|
|
1. Event-store rows for the underlying agent run — these are the per-tuple
|
|
streaming events the worker writes as the agent runs.
|
|
2. LangGraph checkpoint snapshot — the canonical conversation state. Used
|
|
as a fallback when the event store is still empty (early in a run) or
|
|
when only partial events have been flushed; we project the snapshot
|
|
into the same row shape the frontend already renders so the user can
|
|
see the prompt + current AI output while the task is mid-flight.
|
|
3. Synthetic human-input row built from ``task.prompt`` — shown even
|
|
before the agent run is registered, so a `pending` task at minimum
|
|
displays what was queued.
|
|
"""
|
|
service = get_scheduled_task_service()
|
|
user_id = await _user_id(request)
|
|
run = await _load_run(service, request, run_id, user_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
|
|
capped = max(1, min(limit, 1000))
|
|
|
|
# Always available: task prompt as the seeded human input. Lets the
|
|
# ``execution process`` tab show *something* even before the agent run is
|
|
# registered, which is the user-visible bug we're fixing.
|
|
fallback_rows: list[dict] = []
|
|
task_id = run.get("task_id")
|
|
if task_id:
|
|
task = await service.store.get_task_any(task_id)
|
|
prompt = (task or {}).get("prompt")
|
|
if prompt:
|
|
fallback_rows.append(
|
|
{
|
|
"seq": 0,
|
|
"event_type": "llm.human.input",
|
|
"category": "message",
|
|
# Frontend treats human-input content as a plain string.
|
|
"content": str(prompt),
|
|
"metadata": {"source": "task_prompt"},
|
|
"created_at": run.get("created_at") or run.get("scheduled_for"),
|
|
}
|
|
)
|
|
|
|
agent_run_id = run.get("agent_run_id")
|
|
if not agent_run_id:
|
|
return {"messages": fallback_rows, "has_more": False}
|
|
|
|
# Each scheduled run executes on its own throwaway runtime thread — NOT the
|
|
# run row's ``scheduler_thread_id`` — so resolve the real thread from the
|
|
# run store before querying its message events.
|
|
run_store = get_run_store(request)
|
|
agent_run = await run_store.get(agent_run_id, user_id=None)
|
|
if agent_run is None:
|
|
return {"messages": fallback_rows, "has_more": False}
|
|
|
|
event_store = get_run_event_store(request)
|
|
rows = await event_store.list_messages_by_run(
|
|
agent_run["thread_id"],
|
|
agent_run_id,
|
|
limit=capped + 1,
|
|
user_id=None,
|
|
)
|
|
|
|
# Event store has no rows yet (run just started, or events still buffering).
|
|
# Fall back to the live checkpoint so the user immediately sees the prompt
|
|
# + whatever AI output has been streamed so far.
|
|
if not rows:
|
|
checkpoint_rows = await _checkpoint_messages_for_run(request, agent_run, limit=capped)
|
|
if checkpoint_rows:
|
|
return {"messages": checkpoint_rows[:capped], "has_more": False}
|
|
return {"messages": fallback_rows, "has_more": False}
|
|
|
|
has_more = len(rows) > capped
|
|
return {"messages": rows[:capped] if has_more else rows, "has_more": has_more}
|
|
|
|
|
|
@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:
|
|
# Owner-only for regular users; admins may still delete any run.
|
|
if await _is_admin(request):
|
|
deleted = await service.admin_delete_run(run_id)
|
|
else:
|
|
raise HTTPException(status_code=403, detail=str(exc))
|
|
# A run the caller does not own returns deleted=False — fall back to an
|
|
# admin-scoped delete so admins can clean up other users' runs.
|
|
if not deleted and await _is_admin(request):
|
|
deleted = await service.admin_delete_run(run_id)
|
|
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 _load_run(service, request, run_id, user_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 _load_run(service, request, run_id, user_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 _load_run(service, request, run_id, user_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)
|
|
try:
|
|
task = await service.create_task(
|
|
user_id=user_id,
|
|
name=body.name,
|
|
description=body.description,
|
|
prompt=body.prompt,
|
|
schedule=body.schedule,
|
|
timezone=body.timezone,
|
|
execution_agent_name=body.execution_agent_name,
|
|
execution_context=body.execution_context,
|
|
require_markdown=body.require_markdown,
|
|
)
|
|
except DuplicateTaskNameError:
|
|
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
|
|
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)
|
|
try:
|
|
task = await service.update_task(user_id, task_id, body.model_dump(exclude_unset=True))
|
|
except DuplicateTaskNameError:
|
|
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
|
|
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")
|
|
try:
|
|
await get_tag_store(request).unassign_all("scheduled_task", task_id)
|
|
except Exception: # noqa: BLE001 — tag cleanup must not block deletion
|
|
pass
|
|
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}
|
|
|
|
|
|
# --- Admin (cross-user management) ---
|
|
|
|
|
|
async def _require_admin(request: Request) -> None:
|
|
"""Reject non-admins. Lets the call through when auth is disabled."""
|
|
user = await get_optional_user_from_request(request)
|
|
if user is None:
|
|
return
|
|
if getattr(user, "system_role", None) != "admin":
|
|
raise HTTPException(status_code=403, detail="定时任务管理仅限管理员")
|
|
|
|
|
|
@router.get("/admin/all")
|
|
async def admin_list_scheduled_tasks(
|
|
request: Request,
|
|
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on task name"),
|
|
) -> list[dict]:
|
|
"""List every scheduled task across all users. Admin-only."""
|
|
await _require_admin(request)
|
|
service = get_scheduled_task_service()
|
|
tasks = await service.admin_list_tasks()
|
|
await _enrich_with_creator(tasks)
|
|
|
|
tag_store = get_tag_store(request)
|
|
assignments = await tag_store.list_assignments("scheduled_task", [str(t.get("task_id")) for t in tasks])
|
|
for task in tasks:
|
|
task["tags"] = assignments.get(str(task.get("task_id")), [])
|
|
if search:
|
|
needle = search.strip().lower()
|
|
tasks = [t for t in tasks if needle in str(t.get("name", "")).lower()]
|
|
return tasks
|
|
|
|
|
|
@router.get("/admin/{task_id}/runs")
|
|
async def admin_list_scheduled_task_runs(task_id: str, request: Request, limit: int = 50) -> list[dict]:
|
|
"""List runs for any task regardless of owner. Admin-only."""
|
|
await _require_admin(request)
|
|
service = get_scheduled_task_service()
|
|
return await service.admin_list_runs(task_id, limit=max(1, min(limit, 200)))
|
|
|
|
|
|
@router.patch("/admin/{task_id}")
|
|
async def admin_update_scheduled_task(task_id: str, body: ScheduledTaskUpdateRequest, request: Request) -> dict:
|
|
"""Update any task regardless of owner. Admin-only."""
|
|
await _require_admin(request)
|
|
service = get_scheduled_task_service()
|
|
try:
|
|
task = await service.admin_update_task(task_id, body.model_dump(exclude_unset=True))
|
|
except DuplicateTaskNameError:
|
|
raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")
|
|
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("/admin/{task_id}")
|
|
async def admin_delete_scheduled_task(task_id: str, request: Request) -> dict:
|
|
"""Delete any task regardless of owner. Admin-only."""
|
|
await _require_admin(request)
|
|
service = get_scheduled_task_service()
|
|
deleted = await service.admin_delete_task(task_id)
|
|
if not deleted:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
try:
|
|
await get_tag_store(request).unassign_all("scheduled_task", task_id)
|
|
except Exception: # noqa: BLE001 — tag cleanup must not block deletion
|
|
pass
|
|
return {"success": True}
|
|
|
|
|
|
@router.post("/admin/{task_id}/trigger")
|
|
async def admin_trigger_scheduled_task(task_id: str, request: Request) -> dict:
|
|
"""Trigger an immediate run of any task regardless of owner. Admin-only."""
|
|
await _require_admin(request)
|
|
service = get_scheduled_task_service()
|
|
task = await service.admin_trigger_task(task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
|
|
return {"success": True, "task": task}
|