Spaces:
Running
Running
File size: 9,680 Bytes
5094e70 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 | """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.")
|