RollCalcPython/mcp_pdf_delivery.py

"""MCP-host-local storage and browser delivery for generated PDF reports."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import re
import secrets
import tempfile
import time
from typing import Any
from urllib.parse import urljoin, urlsplit

from roll_calculation import calculate_roll
from roll_report_service import (
    create_roll_calculation_pdf_from_result,
    load_report_build_info,
)


ARTIFACT_DIRECTORY_ENV = "ROLLCALC_MCP_ARTIFACT_DIRECTORY"
ARTIFACT_PUBLIC_BASE_URL_ENV = "ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL"
REPORT_TTL_SECONDS = 24 * 60 * 60
TOKEN_PATTERN = re.compile(r"[A-Za-z0-9_-]{43}")
SAFE_FILENAME_PATTERN = re.compile(r"[A-Za-z0-9._-]+\.pdf")


class McpReportNotFoundError(FileNotFoundError):
    """Raised for missing, malformed, or expired MCP-host report tokens."""


class McpArtifactConfigurationError(ValueError):
    """Raised when the MCP host lacks a browser-reachable artifact origin."""


@dataclass(frozen=True)
class StoredMcpReport:
    token: str
    filename: str
    expires_at: float

    @property
    def expires_at_iso(self) -> str:
        return datetime.fromtimestamp(self.expires_at, timezone.utc).isoformat()


class McpReportStore:
    """A local, opaque-token report store owned exclusively by the MCP host."""

    def __init__(self, directory: Path | None = None) -> None:
        configured = os.getenv(ARTIFACT_DIRECTORY_ENV, "").strip()
        self.directory = directory or (
            Path(configured).resolve()
            if configured
            else Path(tempfile.gettempdir(), "rollcalc-mcp-reports")
        )

    def _paths(self, token: str) -> tuple[Path, Path]:
        if not TOKEN_PATTERN.fullmatch(token):
            raise McpReportNotFoundError(token)
        return (
            self.directory / f"{token}.pdf",
            self.directory / f"{token}.json",
        )

    def _metadata(self, metadata_path: Path) -> dict[str, Any]:
        try:
            data = json.loads(metadata_path.read_text(encoding="utf-8"))
        except (FileNotFoundError, json.JSONDecodeError) as error:
            raise McpReportNotFoundError(metadata_path.name) from error
        if not isinstance(data, dict):
            raise McpReportNotFoundError(metadata_path.name)
        return data

    def cleanup_expired(self, now: float | None = None) -> None:
        if not self.directory.is_dir():
            return
        now = time.time() if now is None else now
        for metadata_path in self.directory.glob("*.json"):
            if not TOKEN_PATTERN.fullmatch(metadata_path.stem):
                continue
            try:
                expires_at = self._metadata(metadata_path).get("expires_at")
                if not isinstance(expires_at, (int, float)) or expires_at <= now:
                    pdf_path, _ = self._paths(metadata_path.stem)
                    pdf_path.unlink(missing_ok=True)
                    metadata_path.unlink(missing_ok=True)
            except McpReportNotFoundError:
                pdf_path, _ = self._paths(metadata_path.stem)
                pdf_path.unlink(missing_ok=True)
                metadata_path.unlink(missing_ok=True)

    def create(self, pdf: bytes, filename: str) -> StoredMcpReport:
        if not SAFE_FILENAME_PATTERN.fullmatch(filename):
            raise ValueError("report filename is unsafe")
        self.directory.mkdir(mode=0o700, parents=True, exist_ok=True)
        self.cleanup_expired()
        token = secrets.token_urlsafe(32)
        expires_at = time.time() + REPORT_TTL_SECONDS
        pdf_path, metadata_path = self._paths(token)
        pdf_path.write_bytes(pdf)
        metadata_path.write_text(
            json.dumps({"filename": filename, "expires_at": expires_at}),
            encoding="utf-8",
        )
        return StoredMcpReport(token, filename, expires_at)

    def read(self, token: str) -> tuple[bytes, StoredMcpReport]:
        pdf_path, metadata_path = self._paths(token)
        self.cleanup_expired()
        metadata = self._metadata(metadata_path)
        filename = metadata.get("filename")
        expires_at = metadata.get("expires_at")
        if (
            not isinstance(filename, str)
            or not SAFE_FILENAME_PATTERN.fullmatch(filename)
            or not isinstance(expires_at, (int, float))
            or expires_at <= time.time()
        ):
            raise McpReportNotFoundError(token)
        try:
            return pdf_path.read_bytes(), StoredMcpReport(token, filename, expires_at)
        except FileNotFoundError as error:
            raise McpReportNotFoundError(token) from error


def mcp_artifact_public_base_url() -> str:
    value = os.getenv(ARTIFACT_PUBLIC_BASE_URL_ENV, "").strip()
    parsed = urlsplit(value)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise McpArtifactConfigurationError(
            f"{ARTIFACT_PUBLIC_BASE_URL_ENV} must be an absolute http(s) URL"
        )
    return value.rstrip("/")


def mcp_report_download_url(token: str) -> str:
    if not TOKEN_PATTERN.fullmatch(token):
        raise McpReportNotFoundError(token)
    return urljoin(f"{mcp_artifact_public_base_url()}/", f"reports/{token}.pdf")


def generate_mcp_calculation_pdf(request: dict[str, Any]) -> dict[str, Any]:
    """Create an MCP-host artifact from authoritative direct-roll inputs."""
    build_info = load_report_build_info()
    result = calculate_roll(request, build_info=build_info)
    if result.get("status") != "success":
        return result
    try:
        mcp_artifact_public_base_url()
    except McpArtifactConfigurationError as error:
        return {
            "status": "artifact_delivery_unavailable",
            "error": str(error),
        }
    pdf, filename = create_roll_calculation_pdf_from_result(
        result,
        build_info=build_info,
    )
    stored = McpReportStore().create(pdf, filename)
    return {
        "status": "success",
        "filename": stored.filename,
        "expires_at": stored.expires_at_iso,
        "expires_in_seconds": REPORT_TTL_SECONDS,
        "download_url": mcp_report_download_url(stored.token),
    }