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

import logging
import mimetypes
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import quote

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import FileResponse, PlainTextResponse, Response
from pydantic import BaseModel

from app.gateway.authz import require_permission
from app.gateway.path_utils import aresolve_thread_virtual_path
from deerflow.config.paths import get_paths
from deerflow.runtime.thread_paths import aresolve_path_user_id

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api", tags=["artifacts"])

ACTIVE_CONTENT_MIME_TYPES = {
    "text/html",
    "application/xhtml+xml",
    "image/svg+xml",
}

EDITABLE_TEXT_SUFFIXES = {
    ".md",
    ".markdown",
    ".txt",
    ".json",
    ".yaml",
    ".yml",
    ".csv",
    ".xml",
    ".log",
}


class ArtifactUpdateRequest(BaseModel):
    content: str


def _build_content_disposition(disposition_type: str, filename: str) -> str:
    """Build an RFC 5987 encoded Content-Disposition header value."""
    return f"{disposition_type}; filename*=UTF-8''{quote(filename)}"


def _build_attachment_headers(filename: str, extra_headers: dict[str, str] | None = None) -> dict[str, str]:
    headers = {"Content-Disposition": _build_content_disposition("attachment", filename)}
    if extra_headers:
        headers.update(extra_headers)
    return headers


def is_text_file_by_content(path: Path, sample_size: int = 8192) -> bool:
    """Check if file is text by examining content for null bytes."""
    try:
        with open(path, "rb") as f:
            chunk = f.read(sample_size)
            # Text files shouldn't contain null bytes
            return b"\x00" not in chunk
    except Exception:
        return False


def _extract_file_from_skill_archive(zip_path: Path, internal_path: str) -> bytes | None:
    """Extract a file from a .skill ZIP archive.

    Args:
        zip_path: Path to the .skill file (ZIP archive).
        internal_path: Path to the file inside the archive (e.g., "SKILL.md").

    Returns:
        The file content as bytes, or None if not found.
    """
    if not zipfile.is_zipfile(zip_path):
        return None

    try:
        with zipfile.ZipFile(zip_path, "r") as zip_ref:
            # List all files in the archive
            namelist = zip_ref.namelist()

            # Try direct path first
            if internal_path in namelist:
                return zip_ref.read(internal_path)

            # Try with any top-level directory prefix (e.g., "skill-name/SKILL.md")
            for name in namelist:
                if name.endswith("/" + internal_path) or name == internal_path:
                    return zip_ref.read(name)

            # Not found
            return None
    except (zipfile.BadZipFile, KeyError):
        return None


def collect_outputs_listing(outputs_dir: Path) -> list[dict]:
    """Walk an outputs directory and return artifact listing entries.

    Each entry has ``path`` (virtual ``mnt/user-data/outputs/...``), ``name``,
    ``size_bytes``, ``mime_type``, and ``modified_at`` (ISO 8601 UTC). Symlinks
    that escape the outputs root are dropped to keep the listing scoped to the
    thread. The result is sorted by ``modified_at`` descending (newest first).
    """
    if not outputs_dir.exists() or not outputs_dir.is_dir():
        return []

    outputs_root = outputs_dir.resolve()
    files: list[dict] = []
    for entry in outputs_root.rglob("*"):
        if not entry.is_file():
            continue
        try:
            resolved = entry.resolve()
            resolved.relative_to(outputs_root)
        except (OSError, ValueError):
            continue
        relative = resolved.relative_to(outputs_root).as_posix()
        mime_type, _ = mimetypes.guess_type(resolved.name)
        stat = resolved.stat()
        files.append(
            {
                "path": f"mnt/user-data/outputs/{relative}",
                "name": resolved.name,
                "size_bytes": stat.st_size,
                "mime_type": mime_type,
                "modified_at": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
            }
        )

    files.sort(key=lambda f: f["modified_at"], reverse=True)
    return files


@router.get(
    "/threads/{thread_id}/artifacts",
    summary="List Artifact Files",
    description="List all files under the thread's outputs directory (/mnt/user-data/outputs/). Returns an empty list if the thread has no outputs.",
)
@require_permission("threads", "read", owner_check=True)
async def list_artifacts(thread_id: str, request: Request) -> dict:
    """列出 thread 在 ``outputs/`` 下生成的产物文件。

    圆桌 Step 3（``HighFidelityReport``）对每个**子席位** thread_id 调用本接口，
    展示各智能体写入沙箱 ``mnt/user-data/outputs/`` 的文件；协调智能体 thread
    通常无独立交付文件。

    遍历路径::

        .deer-flow/users/{user_id}/threads/{thread_id}/user-data/outputs/

    返回 ``{ "thread_id", "files": [{ path, name, size_bytes, mime_type, modified_at }] }``，
    目录不存在时 ``files`` 为空数组（非 404）。
    """
    outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=await aresolve_path_user_id(thread_id))
    return {"thread_id": thread_id, "files": collect_outputs_listing(outputs_dir)}


@router.get(
    "/threads/{thread_id}/artifacts/{path:path}",
    summary="Get Artifact File",
    description="Retrieve an artifact file generated by the AI agent. Text and binary files can be viewed inline, while active web content is always downloaded.",
)
@require_permission("threads", "read", owner_check=True)
async def get_artifact(thread_id: str, path: str, request: Request, download: bool = False) -> Response:
    """Get an artifact file by its path.

    The endpoint automatically detects file types and returns appropriate content types.
    Use the `download` query parameter to force file download for non-active content.

    Args:
        thread_id: The thread ID.
        path: The artifact path with virtual prefix (e.g., mnt/user-data/outputs/file.txt).
        request: FastAPI request object (automatically injected).

    Returns:
        The file content as a FileResponse with appropriate content type:
        - Active content (HTML/XHTML/SVG): Served as download attachment
        - Text files: Plain text with proper MIME type
        - Binary files: Inline display with download option

    Raises:
        HTTPException:
            - 400 if path is invalid or not a file
            - 403 if access denied (path traversal detected)
            - 404 if file not found

    Query Parameters:
        download (bool): If true, forces attachment download for file types that are
            otherwise returned inline or as plain text. Active HTML/XHTML/SVG content
            is always downloaded regardless of this flag.

    Example:
        - Get text file inline: `/api/threads/abc123/artifacts/mnt/user-data/outputs/notes.txt`
        - Download file: `/api/threads/abc123/artifacts/mnt/user-data/outputs/data.csv?download=true`
        - Active web content such as `.html`, `.xhtml`, and `.svg` artifacts is always downloaded
    """
    # Check if this is a request for a file inside a .skill archive (e.g., xxx.skill/SKILL.md)
    if ".skill/" in path:
        # Split the path at ".skill/" to get the ZIP file path and internal path
        skill_marker = ".skill/"
        marker_pos = path.find(skill_marker)
        skill_file_path = path[: marker_pos + len(".skill")]  # e.g., "mnt/user-data/outputs/my-skill.skill"
        internal_path = path[marker_pos + len(skill_marker) :]  # e.g., "SKILL.md"

        actual_skill_path = await aresolve_thread_virtual_path(thread_id, skill_file_path)

        if not actual_skill_path.exists():
            raise HTTPException(status_code=404, detail=f"Skill file not found: {skill_file_path}")

        if not actual_skill_path.is_file():
            raise HTTPException(status_code=400, detail=f"Path is not a file: {skill_file_path}")

        # Extract the file from the .skill archive
        content = _extract_file_from_skill_archive(actual_skill_path, internal_path)
        if content is None:
            raise HTTPException(status_code=404, detail=f"File '{internal_path}' not found in skill archive")

        # Determine MIME type based on the internal file
        mime_type, _ = mimetypes.guess_type(internal_path)
        # Add cache headers to avoid repeated ZIP extraction (cache for 5 minutes)
        cache_headers = {"Cache-Control": "private, max-age=300"}
        download_name = Path(internal_path).name or actual_skill_path.stem
        if download or mime_type in ACTIVE_CONTENT_MIME_TYPES:
            return Response(content=content, media_type=mime_type or "application/octet-stream", headers=_build_attachment_headers(download_name, cache_headers))

        if mime_type and mime_type.startswith("text/"):
            return PlainTextResponse(content=content.decode("utf-8"), media_type=mime_type, headers=cache_headers)

        # Default to plain text for unknown types that look like text
        try:
            return PlainTextResponse(content=content.decode("utf-8"), media_type="text/plain", headers=cache_headers)
        except UnicodeDecodeError:
            return Response(content=content, media_type=mime_type or "application/octet-stream", headers=cache_headers)

    actual_path = await aresolve_thread_virtual_path(thread_id, path)

    logger.info(f"Resolving artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")

    if not actual_path.exists():
        raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")

    if not actual_path.is_file():
        raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")

    mime_type, _ = mimetypes.guess_type(actual_path)

    if download:
        return FileResponse(path=actual_path, filename=actual_path.name, media_type=mime_type, headers=_build_attachment_headers(actual_path.name))

    # Always force download for active content types to prevent script execution
    # in the application origin when users open generated artifacts.
    if mime_type in ACTIVE_CONTENT_MIME_TYPES:
        return FileResponse(path=actual_path, filename=actual_path.name, media_type=mime_type, headers=_build_attachment_headers(actual_path.name))

    if mime_type and mime_type.startswith("text/"):
        return PlainTextResponse(content=actual_path.read_text(encoding="utf-8"), media_type=mime_type)

    if is_text_file_by_content(actual_path):
        return PlainTextResponse(content=actual_path.read_text(encoding="utf-8"), media_type=mime_type)

    return Response(content=actual_path.read_bytes(), media_type=mime_type, headers={"Content-Disposition": _build_content_disposition("inline", actual_path.name)})


@router.patch(
    "/threads/{thread_id}/artifacts/{path:path}",
    summary="Update Artifact File",
    description="Update a text artifact file in the current thread workspace.",
)
@require_permission("threads", "write", owner_check=True)
async def update_artifact(
    thread_id: str,
    path: str,
    request: Request,
    body: ArtifactUpdateRequest,
) -> Response:
    """Update a text artifact.

    This endpoint is intentionally limited to normal text artifacts. Files inside
    .skill archives and active web content remain read-only from the artifact UI.
    """
    if ".skill/" in path:
        raise HTTPException(status_code=400, detail="Skill archive contents are read-only.")

    actual_path = await aresolve_thread_virtual_path(thread_id, path)
    logger.info(f"Updating artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")

    if not actual_path.exists():
        raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")

    if not actual_path.is_file():
        raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")

    mime_type, _ = mimetypes.guess_type(actual_path)
    suffix = actual_path.suffix.lower()
    if mime_type in ACTIVE_CONTENT_MIME_TYPES:
        raise HTTPException(status_code=400, detail="Active web artifacts are read-only.")

    if suffix not in EDITABLE_TEXT_SUFFIXES and not (mime_type and mime_type.startswith("text/")):
        raise HTTPException(status_code=400, detail="Only text artifacts can be edited.")

    actual_path.write_text(body.content, encoding="utf-8")
    return PlainTextResponse(content=body.content, media_type=mime_type or "text/plain")