api/image_provider.py

"""Adapter server-side per la generazione e modifica immagini.

Le credenziali del provider restano esclusivamente in ``POLLINATIONS_API_KEY``.
Il modulo restituisce solo URL HTTPS Pollinations validati: nessun token o byte immagine
viene inoltrato attraverso SSE.
"""
from __future__ import annotations

import asyncio
import os
import random
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse

import httpx

POLLINATIONS_BASE_URL = "https://gen.pollinations.ai"
_ALLOWED_HOSTS = {"gen.pollinations.ai", "image.pollinations.ai", "media.pollinations.ai"}
_TRANSIENT_STATUS_CODES = {429, 500, 502, 503, 504}
_MAX_PROMPT_CHARS = 1_200


@dataclass(frozen=True)
class RemoteImage:
    """Riferimento remoto sicuro da materializzare lato client nel VFS."""

    url: str
    mime_type: str
    provider: str = "pollinations"
    revised_prompt: str | None = None


class ImageProviderError(RuntimeError):
    """Errore sicuro da esporre al chiamante senza leak di segreti provider."""

    def __init__(self, message: str, *, status_code: int | None = None, retry_after: float | None = None) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after


def is_pollinations_remote_url(value: str) -> bool:
    """Accetta solo URL HTTPS del dominio Pollinations controllato dal provider."""
    try:
        parsed = urlparse(value)
    except ValueError:
        return False
    host = (parsed.hostname or "").lower().rstrip(".")
    return (
        parsed.scheme == "https"
        and bool(parsed.path)
        and host in _ALLOWED_HOSTS
    )


def _provider_key() -> str:
    value = os.getenv("POLLINATIONS_API_KEY", "").strip()
    if not value:
        raise ImageProviderError("Il provider immagini server-side non è configurato.")
    return value


def _bounded_prompt(value: str) -> str:
    prompt = value.strip()
    if not prompt:
        raise ImageProviderError("Il prompt dell’immagine è obbligatorio.", status_code=400)
    return prompt[:_MAX_PROMPT_CHARS]


def _retry_after(response: httpx.Response) -> float | None:
    raw = response.headers.get("retry-after", "").strip()
    try:
        seconds = float(raw)
    except ValueError:
        return None
    return max(0.0, min(seconds, 30.0))


def _read_remote_image(payload: dict[str, Any]) -> RemoteImage:
    items = payload.get("data")
    first = items[0] if isinstance(items, list) and items else None
    url = first.get("url") if isinstance(first, dict) else None
    if not isinstance(url, str) or not is_pollinations_remote_url(url):
        raise ImageProviderError("Il provider non ha restituito un URL immagine sicuro.")
    mime = first.get("media_type") if isinstance(first, dict) else None
    revised_prompt = first.get("revised_prompt") if isinstance(first, dict) else None
    return RemoteImage(
        url=url,
        mime_type=mime if isinstance(mime, str) and mime.startswith("image/") else "image/jpeg",
        revised_prompt=revised_prompt if isinstance(revised_prompt, str) else None,
    )


async def _post_image_operation(path: str, payload: dict[str, Any], *, timeout_seconds: float) -> RemoteImage:
    """Esegue una richiesta provider con un solo retry per errori realmente transitori."""
    headers = {
        "Authorization": f"Bearer {_provider_key()}",
        "Content-Type": "application/json",
        "Accept": "application/json",
        "User-Agent": "AgenteAI/3.4 image-provider",
    }
    async with httpx.AsyncClient(timeout=httpx.Timeout(timeout_seconds), follow_redirects=False) as client:
        for attempt in range(2):
            try:
                response = await client.post(f"{POLLINATIONS_BASE_URL}{path}", headers=headers, json=payload)
            except httpx.TimeoutException as exc:
                if attempt == 0:
                    await asyncio.sleep(0.5 + random.uniform(0.0, 0.25))
                    continue
                raise ImageProviderError("Il provider immagini non ha risposto entro il tempo previsto.") from exc
            except httpx.HTTPError as exc:
                if attempt == 0:
                    await asyncio.sleep(0.5 + random.uniform(0.0, 0.25))
                    continue
                raise ImageProviderError("Il provider immagini non è raggiungibile.") from exc

            if response.status_code == 200:
                try:
                    return _read_remote_image(response.json())
                except ValueError as exc:
                    raise ImageProviderError("Il provider ha restituito una risposta immagine non valida.") from exc

            retry_after = _retry_after(response)
            if response.status_code in _TRANSIENT_STATUS_CODES and attempt == 0:
                await asyncio.sleep(retry_after if retry_after is not None else 0.75 + random.uniform(0.0, 0.25))
                continue

            message_by_status = {
                400: "La richiesta immagine non è valida.",
                401: "Il provider immagini non è autenticato correttamente.",
                402: "Il credito del provider immagini non è disponibile.",
                403: "Il provider immagini non autorizza questo modello o questa operazione.",
                429: "Il provider immagini è temporaneamente soggetto a rate limit.",
            }
            raise ImageProviderError(
                message_by_status.get(response.status_code, "Il provider immagini ha restituito un errore."),
                status_code=response.status_code,
                retry_after=retry_after,
            )

    raise ImageProviderError("Il provider immagini non ha prodotto alcun risultato.")


async def generate_pollinations_image(prompt: str, *, width: int = 1024, height: int = 1024) -> RemoteImage:
    """Genera una singola immagine remota, senza serializzare l’immagine nella chat/SSE."""
    safe_width = min(max(int(width), 256), 1024)
    safe_height = min(max(int(height), 256), 1024)
    return await _post_image_operation(
        "/v1/images/generations",
        {
            "prompt": _bounded_prompt(prompt),
            "model": "flux",
            "n": 1,
            "size": f"{safe_width}x{safe_height}",
            "quality": "medium",
            "response_format": "url",
            "safe": True,
        },
        timeout_seconds=55,
    )


async def edit_pollinations_image(
    prompt: str,
    *,
    source_bytes: bytes,
    source_mime: str = "image/jpeg",
) -> RemoteImage:
    """Modifica un artefatto VFS inviandolo al provider come multipart/form-data."""
    if not source_bytes or len(source_bytes) > 5 * 1024 * 1024:
        raise ImageProviderError("L’immagine di origine deve avere una dimensione compresa tra 1 byte e 5 MB.", status_code=400)
    if source_mime not in {"image/jpeg", "image/png", "image/webp"}:
        raise ImageProviderError("Il formato dell’immagine di origine non è supportato.", status_code=400)

    extension = {"image/jpeg": "jpg", "image/png": "png", "image/webp": "webp"}[source_mime]
    data = {
        "prompt": _bounded_prompt(prompt),
        "model": "flux",
        "n": "1",
        "response_format": "url",
        "safe": "true",
    }
    headers = {
        "Authorization": f"Bearer {_provider_key()}",
        "Accept": "application/json",
        "User-Agent": "AgenteAI/3.4 image-provider",
    }
    async with httpx.AsyncClient(timeout=httpx.Timeout(70), follow_redirects=False) as client:
        for attempt in range(2):
            try:
                response = await client.post(
                    f"{POLLINATIONS_BASE_URL}/v1/images/edits",
                    headers=headers,
                    data=data,
                    files={"image": (f"source.{extension}", source_bytes, source_mime)},
                )
            except httpx.TimeoutException as exc:
                if attempt == 0:
                    await asyncio.sleep(0.5 + random.uniform(0.0, 0.25))
                    continue
                raise ImageProviderError("Il provider immagini non ha risposto entro il tempo previsto.") from exc
            except httpx.HTTPError as exc:
                if attempt == 0:
                    await asyncio.sleep(0.5 + random.uniform(0.0, 0.25))
                    continue
                raise ImageProviderError("Il provider immagini non è raggiungibile.") from exc

            if response.status_code == 200:
                try:
                    return _read_remote_image(response.json())
                except ValueError as exc:
                    raise ImageProviderError("Il provider ha restituito una risposta immagine non valida.") from exc
            retry_after = _retry_after(response)
            if response.status_code in _TRANSIENT_STATUS_CODES and attempt == 0:
                await asyncio.sleep(retry_after if retry_after is not None else 0.75 + random.uniform(0.0, 0.25))
                continue
            message_by_status = {
                400: "La richiesta di modifica immagine non è valida.",
                401: "Il provider immagini non è autenticato correttamente.",
                402: "Il credito del provider immagini non è disponibile.",
                403: "Il provider immagini non autorizza questa modifica.",
                429: "Il provider immagini è temporaneamente soggetto a rate limit.",
            }
            raise ImageProviderError(message_by_status.get(response.status_code, "Il provider immagini ha restituito un errore."), status_code=response.status_code, retry_after=retry_after)
    raise ImageProviderError("Il provider immagini non ha prodotto alcun risultato.")