deerflow-code/offline-backend-20260512/backend/app/gateway/routers/scheduled_tasks.py

"""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_current_user, get_local_provider, get_tag_store
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,
    search: str | None = Query(default=None, description="Case-insensitive fuzzy match on task name"),
    tag_id: str | None = Query(default=None, description="Keep only tasks carrying this tag"),
) -> 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_id:
        tasks = [t for t in tasks if any(tg.get("id") == tag_id 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 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")
    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}