bambuddy/backend/app/services/model_providers/manyfold/service.py

"""Manyfold API client (#1471).

Talks to a self-hosted Manyfold install through its v0 API: list and search
models, read a model's files, download one file and fetch a model's preview
image. Bambuddy signs in with an OAuth application's client ID and secret
(the client-credentials flow) and asks only for the ``public read`` scopes,
which cover everything here; it never writes to Manyfold.

Manyfold limits the token endpoint to 10 requests in 3 minutes, so the token
is cached until shortly before it expires (two hours by default) and shared
by every request.

Request URLs are always built from the configured base URL and validated ids.
The ``@id`` and ``contentUrl`` links in Manyfold's responses are read for the
ids and the file name they carry, never fetched as given.
"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import os
import re
import time
from typing import Any
from urllib.parse import unquote, urlsplit

import httpx

from backend.app.api.routes._url_safety import assert_safe_lan_service_url
from backend.app.services.model_providers.base import (
    ProviderAuthError,
    ProviderDownload,
    ProviderDownloadInfo,
    ProviderError,
    ProviderForbiddenError,
    ProviderNotFoundError,
    ProviderResolvedModel,
    ProviderResourceRef,
    ProviderService,
    ProviderStatus,
    ProviderUnavailableError,
)
from backend.app.services.model_providers.manyfold.config import ManyfoldConfig

logger = logging.getLogger(__name__)

API_MEDIA_TYPE = "application/vnd.manyfold.v0+json"
SCOPES = "public read"

# Manyfold ids are short public ids such as "x7q3k2pa".
_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
_MODEL_ID_IN_URL = re.compile(r"/models/([A-Za-z0-9_-]{1,64})(?:/|$)")
_FILE_ID_IN_URL = re.compile(r"/model_files/([A-Za-z0-9_-]{1,64})(?:[/.?]|$)")

# What Bambuddy can slice or print, by Manyfold's MIME type and by extension.
IMPORTABLE_MIME_TYPES = frozenset({"model/3mf", "model/stl", "model/step"})
IMPORTABLE_EXTENSIONS = (".3mf", ".stl", ".step", ".stp")

MAX_FILE_BYTES = 200 * 1024 * 1024
_MAX_REDIRECTS = 5
MAX_PREVIEW_BYTES = 10 * 1024 * 1024

# Renew this long before Manyfold's own expiry, so a request never starts
# with a token that dies on the way.
_TOKEN_MARGIN_SECONDS = 120

_IMAGE_SIGNATURES = (
    (b"\x89PNG\r\n\x1a\n", "image/png"),
    (b"\xff\xd8\xff", "image/jpeg"),
    (b"GIF87a", "image/gif"),
    (b"GIF89a", "image/gif"),
)

# (base url, client id, secret digest) -> (token, monotonic expiry)
_token_cache: dict[tuple[str, str, str], tuple[str, float]] = {}
# Credentials Manyfold just refused, and until when that answer is reused.
# Without it, every request of a page or a bulk import would try to sign in
# again with the same wrong secret and use up Manyfold's 10 sign-ins in 3
# minutes, so the corrected secret would then be refused as well.
_refused: dict[tuple[str, str, str], tuple[ProviderError, float]] = {}
_REFUSED_FOR_SECONDS = 30.0
# One sign-in at a time, so a page of previews opening at once asks for one
# token rather than a dozen. Bound to the running loop, recreated if it changes.
_token_lock: asyncio.Lock | None = None
_token_lock_loop: asyncio.AbstractEventLoop | None = None


def _sign_in_lock() -> asyncio.Lock:
    global _token_lock, _token_lock_loop
    loop = asyncio.get_running_loop()
    if _token_lock is None or _token_lock_loop is not loop:
        _token_lock = asyncio.Lock()
        _token_lock_loop = loop
    return _token_lock


class ManyfoldError(ProviderError):
    """Base exception for Manyfold API errors.

    ``code`` names the failure for the frontend, which shows its own
    translated text for it; the message is the English fallback.
    """

    default_code = "manyfold_failed"

    def __init__(self, message: str, code: str | None = None):
        super().__init__(message)
        self.code = code or self.default_code


class ManyfoldAuthError(ProviderAuthError, ManyfoldError):
    """Manyfold refused the configured credentials, or none are configured."""

    default_code = "manyfold_credentials"


class ManyfoldForbiddenError(ProviderForbiddenError, ManyfoldError):
    """Manyfold accepted the token but refused this request."""

    default_code = "manyfold_forbidden"


class ManyfoldNotFoundError(ProviderNotFoundError, ManyfoldError):
    """The model or file doesn't exist, or isn't visible to the application's owner."""

    default_code = "manyfold_not_found"


class ManyfoldUnavailableError(ProviderUnavailableError, ManyfoldError):
    """Manyfold is unreachable, failing, or sent something unexpected."""

    default_code = "manyfold_failed"


def clear_token_cache() -> None:
    """Forget cached tokens and refusals, e.g. after the credentials were changed."""
    _token_cache.clear()
    _refused.clear()


def valid_id(value: str) -> str:
    """Return ``value`` if it is a well-formed Manyfold id, else raise."""
    if not isinstance(value, str) or not _ID_RE.match(value):
        raise ManyfoldNotFoundError("Not a valid Manyfold id")
    return value


def _id_from(url: Any, pattern: re.Pattern[str]) -> str | None:
    if not isinstance(url, str):
        return None
    match = pattern.search(urlsplit(url).path)
    return match.group(1) if match else None


def _image_type(head: bytes) -> str | None:
    for signature, mime in _IMAGE_SIGNATURES:
        if head.startswith(signature):
            return mime
    if head[:4] == b"RIFF" and head[8:12] == b"WEBP":
        return "image/webp"
    return None


def is_importable_filename(filename: str) -> bool:
    return filename.lower().endswith(IMPORTABLE_EXTENSIONS)


class ManyfoldService(ProviderService):
    """Per-request Manyfold client."""

    def __init__(self, config: ManyfoldConfig, *, client: httpx.AsyncClient | None = None):
        self._config = config
        self._base = config.url.rstrip("/")
        self._host = (urlsplit(self._base).hostname or "").lower()
        if client is not None:
            self._client = client
            self._owns_client = False
        else:
            self._client = httpx.AsyncClient(timeout=30.0)
            self._owns_client = True

    async def close(self) -> None:
        if self._owns_client:
            await self._client.aclose()

    # ---- sign-in -------------------------------------------------------

    def _cache_key(self) -> tuple[str, str, str]:
        digest = hashlib.sha256(self._config.client_secret.encode()).hexdigest()
        return (self._base, self._config.client_id, digest)

    async def _token(self, *, renew: bool = False) -> str:
        if not self._config.configured:
            raise ManyfoldAuthError(
                "Manyfold is not set up. Enter its URL, client ID and secret first.", "manyfold_not_configured"
            )
        key = self._cache_key()
        async with _sign_in_lock():
            cached = _token_cache.get(key)
            if cached and not renew and cached[1] > time.monotonic():
                return cached[0]
            refused = _refused.get(key)
            if refused and refused[1] > time.monotonic():
                earlier = refused[0]
                raise type(earlier)(str(earlier), getattr(earlier, "code", None))
            try:
                return await self._sign_in(key)
            except ManyfoldAuthError as exc:
                _refused[key] = (exc, time.monotonic() + _REFUSED_FOR_SECONDS)
                raise

    async def _sign_in(self, key: tuple[str, str, str]) -> str:
        """Ask Manyfold for a token; the caller holds the sign-in lock."""
        try:
            response = await self._client.post(
                f"{self._base}/oauth/token",
                data={
                    "grant_type": "client_credentials",
                    "client_id": self._config.client_id,
                    "client_secret": self._config.client_secret,
                    "scope": SCOPES,
                },
                headers={"Accept": "application/json"},
            )
        except httpx.HTTPError as exc:
            raise ManyfoldUnavailableError(
                f"Could not reach Manyfold at {self._base}: {exc}", "manyfold_unreachable"
            ) from exc
        if response.status_code == 429:
            raise ManyfoldUnavailableError(
                "Manyfold is limiting sign-ins. Try again in a few minutes.", "manyfold_rate_limited"
            )
        body = self._json_or_none(response)
        error = body.get("error") if isinstance(body, dict) else None
        if error == "invalid_scope":
            raise ManyfoldAuthError(
                "The Manyfold application lacks the 'read' scope. Edit it in Manyfold under Settings -> API.",
                "manyfold_scope",
            )
        if response.status_code in (400, 401) or error in ("invalid_client", "unauthorized_client"):
            raise ManyfoldAuthError("Manyfold did not accept the client ID or secret.")
        if response.status_code != 200 or not isinstance(body, dict):
            raise ManyfoldUnavailableError(f"Manyfold sign-in failed with HTTP {response.status_code}")
        token = body.get("access_token")
        if not isinstance(token, str) or not token:
            raise ManyfoldUnavailableError("Manyfold sign-in returned no token")
        granted = str(body.get("scope") or "").split()
        if granted and "read" not in granted:
            raise ManyfoldAuthError(
                "The Manyfold application lacks the 'read' scope. Edit it in Manyfold under Settings -> API.",
                "manyfold_scope",
            )
        try:
            lifetime = float(body.get("expires_in") or 7200)
        except (TypeError, ValueError):
            lifetime = 7200.0
        _token_cache[key] = (token, time.monotonic() + max(0.0, lifetime - _TOKEN_MARGIN_SECONDS))
        return token

    @staticmethod
    def _json_or_none(response: httpx.Response) -> Any:
        try:
            return response.json()
        except ValueError:
            return None

    # ---- requests ------------------------------------------------------

    def _raise_for(self, response: httpx.Response, what: str) -> None:
        status = response.status_code
        if status == 401:
            raise ManyfoldAuthError("Manyfold did not accept Bambuddy's sign-in.")
        if status == 403:
            raise ManyfoldForbiddenError(
                f"Manyfold refused access to {what}. Check that the application has the 'read' scope "
                "and that its owner can see this model."
            )
        if status == 404:
            raise ManyfoldNotFoundError(f"Manyfold could not find {what}")
        if status >= 400:
            raise ManyfoldUnavailableError(f"Manyfold answered HTTP {status} for {what}")

    async def _send(
        self,
        url: str,
        *,
        accept: str | None = None,
        params: dict[str, Any] | None = None,
        follow_redirects: bool = False,
    ) -> httpx.Response:
        """GET with the token, renewing it once if Manyfold refuses it.

        The response is streamed; the caller reads and closes it.

        Redirects (Manyfold may hand files over to object storage) are
        followed here rather than by httpx, so every hop passes the same
        LAN-service check as the configured URL, and the token is only sent
        to Manyfold's own host.
        """
        headers = {"Accept": accept} if accept else {}
        for attempt in range(2):
            headers["Authorization"] = f"Bearer {await self._token(renew=attempt > 0)}"
            try:
                request = self._client.build_request("GET", url, headers=headers, params=params)
                response = await self._client.send(request, stream=True)
                hops = 0
                while follow_redirects and response.is_redirect and response.next_request is not None:
                    hops += 1
                    following = response.next_request
                    await response.aclose()
                    if hops > _MAX_REDIRECTS:
                        raise ManyfoldUnavailableError("Manyfold redirected too often")
                    try:
                        assert_safe_lan_service_url(str(following.url), label="Manyfold redirect")
                    except ValueError as exc:
                        raise ManyfoldUnavailableError(f"Refusing Manyfold's redirect: {exc}") from exc
                    if (following.url.host or "").lower() != self._host:
                        following.headers.pop("Authorization", None)
                    response = await self._client.send(following, stream=True)
            except httpx.HTTPError as exc:
                raise ManyfoldUnavailableError(f"Could not reach Manyfold: {exc}", "manyfold_unreachable") from exc
            # Only Manyfold's own 401 means the token was refused; object
            # storage behind a redirect never saw it.
            if response.status_code == 401 and attempt == 0 and (response.url.host or "").lower() == self._host:
                await response.aclose()
                continue
            return response
        return response  # pragma: no cover - the loop always returns

    async def _get_json(self, path: str, what: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        response = await self._send(f"{self._base}{path}", accept=API_MEDIA_TYPE, params=params)
        try:
            await response.aread()
        finally:
            await response.aclose()
        self._raise_for(response, what)
        body = self._json_or_none(response)
        if not isinstance(body, dict):
            raise ManyfoldUnavailableError(f"Manyfold sent an unexpected answer for {what}")
        return body

    async def _read_capped(self, response: httpx.Response, cap: int, what: str) -> bytes:
        chunks: list[bytes] = []
        size = 0
        async for chunk in response.aiter_bytes():
            size += len(chunk)
            if size > cap:
                raise ManyfoldUnavailableError(f"{what} is larger than {cap // (1024 * 1024)} MB", "manyfold_too_large")
            chunks.append(chunk)
        return b"".join(chunks)

    # ---- models --------------------------------------------------------

    async def list_models(self, *, query: str = "", page: int = 1) -> dict[str, Any]:
        params: dict[str, Any] = {"page": max(1, page)}
        if query.strip():
            params["q"] = query.strip()
        body = await self._get_json("/models", "the model list", params)
        members = body.get("member") if isinstance(body.get("member"), list) else []
        models = []
        for member in members:
            if not isinstance(member, dict):
                continue
            model_id = _id_from(member.get("@id"), _MODEL_ID_IN_URL)
            if model_id:
                models.append({"id": model_id, "name": str(member.get("name") or model_id)})
        view = body.get("view") if isinstance(body.get("view"), dict) else {}
        total = body.get("totalItems")
        return {
            "total": total if isinstance(total, int) else len(models),
            "page": params["page"],
            "has_next": bool(view.get("next")),
            "has_previous": bool(view.get("previous")),
            "models": models,
        }

    async def get_model(self, model_id: str) -> dict[str, Any]:
        model_id = valid_id(model_id)
        body = await self._get_json(f"/models/{model_id}", "this model")
        files = []
        for part in body.get("hasPart") or []:
            if not isinstance(part, dict):
                continue
            file_id = _id_from(part.get("@id"), _FILE_ID_IN_URL)
            if not file_id:
                continue
            mime = str(part.get("encodingFormat") or "")
            files.append(
                {
                    "id": file_id,
                    "name": str(part.get("name") or file_id),
                    "mime": mime,
                    "importable": mime in IMPORTABLE_MIME_TYPES,
                }
            )
        license_info = body.get("spdx:license")
        keywords = body.get("keywords")
        preview = body.get("preview_file")
        return {
            "id": model_id,
            "name": str(body.get("name") or model_id),
            "caption": body.get("caption") if isinstance(body.get("caption"), str) else None,
            "description": body.get("description") if isinstance(body.get("description"), str) else None,
            "license": license_info.get("licenseId") if isinstance(license_info, dict) else None,
            "tags": [str(tag) for tag in keywords] if isinstance(keywords, list) else [],
            "url": f"{self._base}/models/{model_id}",
            "preview_file_id": _id_from(preview.get("@id"), _FILE_ID_IN_URL) if isinstance(preview, dict) else None,
            "files": files,
        }

    async def get_file(self, model_id: str, file_id: str) -> dict[str, Any]:
        """One file's details, with the raw-download path Manyfold gives for it."""
        model_id, file_id = valid_id(model_id), valid_id(file_id)
        body = await self._get_json(f"/models/{model_id}/model_files/{file_id}", "this file")
        content_path = urlsplit(str(body.get("contentUrl") or "")).path
        marker = f"/models/{model_id}/raw/"
        position = content_path.find(marker)
        tail = content_path[position + len(marker) :] if position >= 0 else ""
        segments = unquote(tail).split("/")
        if not tail or any(segment in ("", ".", "..") for segment in segments):
            raise ManyfoldUnavailableError("Manyfold sent no usable download link for this file")
        size = body.get("contentSize")
        return {
            "id": file_id,
            "model_id": model_id,
            "name": str(body.get("name") or file_id),
            "filename": os.path.basename(unquote(tail)),
            "mime": str(body.get("encodingFormat") or ""),
            "size": size if isinstance(size, int) else None,
            "raw_path": f"/models/{model_id}/raw/{tail}",
        }

    async def check(self) -> int:
        """Sign in and read the first page of models; returns the model count."""
        return int((await self.list_models())["total"])

    # ---- files and previews ------------------------------------------

    async def download_file(self, file: dict[str, Any]) -> bytes:
        response = await self._send(f"{self._base}{file['raw_path']}", follow_redirects=True)
        try:
            self._raise_for(response, "this file")
            return await self._read_capped(response, MAX_FILE_BYTES, "The file")
        finally:
            await response.aclose()

    async def fetch_preview(self, model_id: str) -> tuple[bytes, str]:
        """The model's preview as an image, from Manyfold's own derivatives.

        An image preview comes as Manyfold's small "preview" copy, a 3D file
        as its rendered picture when Manyfold made one. Without a picture
        Manyfold sends the original file instead, so anything that doesn't
        start like an image is refused after its first bytes.
        """
        model = await self.get_model(model_id)
        file_id = model["preview_file_id"]
        if not file_id:
            raise ManyfoldNotFoundError("This model has no preview")
        file = await self.get_file(model["id"], file_id)
        extension = os.path.splitext(file["filename"])[1].lower()
        mime = file["mime"]
        if mime.startswith("image/"):
            derivative = "preview"
        elif mime.startswith("model/"):
            derivative = "render"
        else:
            raise ManyfoldNotFoundError("This model has no preview")
        if not re.fullmatch(r"\.[a-z0-9]{1,10}", extension):
            raise ManyfoldNotFoundError("This model has no preview")
        response = await self._send(
            f"{self._base}/models/{model['id']}/model_files/{file_id}{extension}",
            params={"derivative": derivative},
            follow_redirects=True,
        )
        try:
            self._raise_for(response, "the preview")
            data = bytearray()
            content_type: str | None = None
            async for chunk in response.aiter_bytes():
                data += chunk
                if content_type is None and len(data) >= 12:
                    content_type = _image_type(bytes(data[:12]))
                    if content_type is None:
                        raise ManyfoldNotFoundError("This model has no preview")
                if len(data) > MAX_PREVIEW_BYTES:
                    raise ManyfoldNotFoundError("This model's preview is too large")
            content_type = content_type or _image_type(bytes(data[:12]))
            if content_type is None:
                raise ManyfoldNotFoundError("This model has no preview")
            return bytes(data), content_type
        finally:
            await response.aclose()

    # ---- ProviderService -----------------------------------------------

    async def get_status(self, db) -> ProviderStatus:
        configured = self._config.configured
        return ProviderStatus(authenticated=configured, can_download=configured)

    async def resolve(self, ref: ProviderResourceRef) -> ProviderResolvedModel:
        model = await self.get_model(ref.external_id)
        return ProviderResolvedModel(ref=ref, design=model, instances=model["files"])

    async def get_download(self, ref: ProviderResourceRef) -> ProviderDownloadInfo:
        if not ref.sub_id:
            raise ManyfoldNotFoundError("Name the file to download")
        file = await self.get_file(ref.external_id, ref.sub_id)
        return ProviderDownloadInfo(ref=ref, url=f"{self._base}{file['raw_path']}", suggested_filename=file["filename"])

    async def download(self, info: ProviderDownloadInfo) -> ProviderDownload:
        if (urlsplit(info.url).hostname or "").lower() != self._host or not info.url.startswith(self._base + "/"):
            raise ManyfoldUnavailableError("Refusing to download from outside the Manyfold install")
        data = await self.download_file({"raw_path": info.url[len(self._base) :]})
        return ProviderDownload(file_bytes=data, filename=info.suggested_filename)

    async def fetch_thumbnail(self, url: str) -> tuple[bytes, str]:
        model_id = _id_from(url, _MODEL_ID_IN_URL)
        if (urlsplit(url).hostname or "").lower() != self._host or not model_id:
            raise ManyfoldUnavailableError("Refusing to fetch an image from outside the Manyfold install")
        return await self.fetch_preview(model_id)