"""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.")