""" Service AI Vision — description, classification et tags via Google Gemini ou OpenRouter. Résilience : - Timeout configurable (AI_REQUEST_TIMEOUT) - Retry avec backoff exponentiel (AI_MAX_RETRIES) - Client Gemini non-singleton (recréé si la clé change) """ import asyncio import json import logging import re import base64 import time import httpx from pathlib import Path from typing import Optional, Tuple from google import genai from google.genai import types from app.config import settings from app.services.storage_backend import get_storage_backend from app.services.runtime_config import runtime_ai_config logger = logging.getLogger(__name__) # Cache du client Gemini avec TTL _client: Optional[genai.Client] = None _client_api_key: Optional[str] = None def _get_client() -> genai.Client: global _client, _client_api_key current_key = settings.GEMINI_API_KEY if _client is None or _client_api_key != current_key: _client = genai.Client(api_key=current_key) _client_api_key = current_key return _client async def _read_image(file_path: str) -> tuple[bytes, str]: """Lit l'image via le StorageBackend et la redimensionne pour l'AI (max 1024px).""" path = Path(file_path) suffix = path.suffix.lower() backend = get_storage_backend() data = await backend.get_bytes(file_path) # Redimensionner pour éviter les timeouts AI sur les grandes images try: from PIL import Image as PILImage import io as pil_io original_size = len(data) img = PILImage.open(pil_io.BytesIO(data)) max_dim = 1024 if img.width > max_dim or img.height > max_dim: img.thumbnail((max_dim, max_dim), PILImage.Resampling.LANCZOS) buf = pil_io.BytesIO() if img.mode in ("RGBA", "P"): img = img.convert("RGB") img.save(buf, format="JPEG", quality=85) data = buf.getvalue() logger.info("ai.image_resized", extra={ "file": path.name, "original_bytes": original_size, "resized_bytes": len(data), }) return data, "image/jpeg" except Exception: pass # Si Pillow échoue, on utilise l'image originale mime_map = { ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", } media_type = mime_map.get(suffix, "image/jpeg") return data, media_type def _extract_json(text: str) -> Optional[dict]: cleaned = re.sub(r"```json\s*|```\s*", "", (text or "")).strip() json_match = re.search(r"\{.*\}", cleaned, re.DOTALL) if not json_match: return None try: return json.loads(json_match.group()) except json.JSONDecodeError: return None def _usage_tokens_gemini(response) -> tuple[Optional[int], Optional[int]]: usage = getattr(response, "usage_metadata", None) if not usage: return None, None prompt_tokens = getattr(usage, "prompt_token_count", None) output_tokens = getattr(usage, "candidates_token_count", None) return prompt_tokens, output_tokens async def _retry_with_backoff(fn, *args, max_retries=None, **kwargs): retries = max_retries if max_retries is not None else settings.AI_MAX_RETRIES last_error = None for attempt in range(retries + 1): try: return await fn(*args, **kwargs) except asyncio.TimeoutError: last_error = "timeout" wait = 2 ** attempt logger.warning("ai.retry.timeout", extra={ "attempt": attempt + 1, "max_retries": retries, "wait_s": wait, }) except Exception as e: last_error = str(e)[:200] wait = 2 ** attempt logger.warning("ai.retry.error", extra={ "attempt": attempt + 1, "max_retries": retries, "wait_s": wait, "error": str(e)[:200], }) if attempt < retries: await asyncio.sleep(wait) raise Exception(f"AI request failed after {retries + 1} attempts: {last_error}") # ───────────────────────────────────────────────────────────── # Google Gemini # ───────────────────────────────────────────────────────────── async def _generate_gemini( prompt: str, image_bytes: Optional[bytes] = None, media_type: Optional[str] = None, max_tokens: int = 1024, model_override: Optional[str] = None, ) -> dict: if not settings.GEMINI_API_KEY: return {"text": None, "usage": (None, None)} client = _get_client() contents = [] if image_bytes and media_type: contents.append(types.Part.from_bytes(data=image_bytes, mime_type=media_type)) contents.append(prompt) model = model_override or runtime_ai_config.model or settings.GEMINI_MODEL async def _call(): return await asyncio.wait_for( asyncio.to_thread( client.models.generate_content, model=model, contents=contents, config=types.GenerateContentConfig( max_output_tokens=max_tokens, response_mime_type="application/json", ), ), timeout=settings.AI_REQUEST_TIMEOUT, ) try: response = await _retry_with_backoff(_call) usage = _usage_tokens_gemini(response) return {"text": getattr(response, "text", ""), "usage": usage} except Exception as e: logger.error("ai.gemini.error", extra={ "model": model, "error_type": type(e).__name__, "error": str(e)[:300], }) return {"text": None, "usage": (None, None), "error": str(e)[:300]} # ───────────────────────────────────────────────────────────── # OpenRouter # ───────────────────────────────────────────────────────────── async def _generate_openrouter( prompt: str, image_bytes: Optional[bytes] = None, media_type: Optional[str] = None, max_tokens: int = 1024, model_override: Optional[str] = None, ) -> dict: if not settings.OPENROUTER_API_KEY: return {"text": None, "usage": (None, None)} model = model_override or runtime_ai_config.model or settings.OPENROUTER_MODEL headers = { "Authorization": f"Bearer {settings.OPENROUTER_API_KEY}", "Content-Type": "application/json", "HTTP-Referer": "imago-pipeline", "X-Title": settings.APP_NAME, } messages = [] content_payload = [{"type": "text", "text": prompt}] if image_bytes and media_type: b64_img = base64.b64encode(image_bytes).decode("utf-8") content_payload.append({ "type": "image_url", "image_url": {"url": f"data:{media_type};base64,{b64_img}"} }) messages.append({"role": "user", "content": content_payload}) payload = { "model": model, "messages": messages, "max_tokens": max_tokens, "response_format": {"type": "json_object"} } timeout_secs = settings.AI_REQUEST_TIMEOUT logger.info("ai.openrouter.request", extra={ "model": model, "timeout_s": timeout_secs, "image_bytes": len(image_bytes) if image_bytes else 0, }) async def _call(): t0 = time.time() try: async with httpx.AsyncClient( timeout=httpx.Timeout(timeout_secs, connect=15.0) ) as client: response = await client.post( "https://openrouter.ai/api/v1/chat/completions", json=payload, headers=headers, ) elapsed = time.time() - t0 logger.info("ai.openrouter.response", extra={ "status": response.status_code, "elapsed_s": round(elapsed, 1), }) response.raise_for_status() return response.json() except httpx.TimeoutException as e: elapsed = time.time() - t0 logger.error("ai.openrouter.timeout", extra={ "elapsed_s": round(elapsed, 1), "timeout_setting": timeout_secs, "detail": str(e), }) raise try: data = await _retry_with_backoff(_call) text = "" if "choices" in data and len(data["choices"]) > 0: text = data["choices"][0]["message"]["content"] usage_data = data.get("usage", {}) prompt_tokens = usage_data.get("prompt_tokens") output_tokens = usage_data.get("completion_tokens") return {"text": text, "usage": (prompt_tokens, output_tokens)} except Exception as e: logger.error("ai.openrouter.final_error", extra={ "model": model, "error_type": type(e).__name__, "error": str(e)[:300], }) return {"text": None, "usage": (None, None), "error": str(e)[:300]} # ───────────────────────────────────────────────────────────── # Dispatcher # ───────────────────────────────────────────────────────────── async def _generate( prompt: str, image_bytes: Optional[bytes] = None, media_type: Optional[str] = None, max_tokens: int = 1024, provider_override: Optional[str] = None, model_override: Optional[str] = None, ) -> dict: provider = provider_override or runtime_ai_config.provider or settings.AI_PROVIDER.lower() logger.info("ai.generate", extra={"provider": provider}) if provider == "openrouter": return await _generate_openrouter( prompt, image_bytes, media_type, max_tokens, model_override=model_override, ) else: return await _generate_gemini( prompt, image_bytes, media_type, max_tokens, model_override=model_override, ) # ───────────────────────────────────────────────────────────── # Fonctions publiques # ───────────────────────────────────────────────────────────── def _build_prompt(ocr_hint: Optional[str], language: str) -> str: ocr_section = "" if ocr_hint and len(ocr_hint.strip()) > 5: ocr_section = f""" Texte détecté dans l'image par OCR (utilise-le pour enrichir ta réponse) : \"\"\" {ocr_hint[:500]} \"\"\" """ return f"""Analyse cette image avec précision et retourne UNIQUEMENT un objet JSON valide avec ces champs : {{ "description": "Description complète et détaillée en {language}, 2-4 phrases. Décris le sujet principal, le contexte, les couleurs, l'ambiance.", "tags": ["tag1", "tag2", "tag3"], "confidence": 0.95 }} Règles pour les tags : - Entre {settings.AI_TAGS_MIN} et {settings.AI_TAGS_MAX} tags - En minuscules, sans espaces (utiliser des tirets si nécessaire) - Couvrir : sujet principal, type d'image, couleurs dominantes, style, contexte - Exemples : portrait, paysage, architecture, nature, nourriture, texte, document, animal, sport, technologie, intérieur, extérieur {ocr_section} Réponds UNIQUEMENT avec le JSON, sans texte avant ou après, sans balises markdown.""" async def analyze_image( file_path: str, ocr_hint: Optional[str] = None, language: str = "français", provider_override: Optional[str] = None, model_override: Optional[str] = None, ) -> dict: if not settings.AI_ENABLED: return {} active_provider = provider_override or runtime_ai_config.provider or settings.AI_PROVIDER active_model = model_override or runtime_ai_config.model or ( settings.OPENROUTER_MODEL if active_provider == "openrouter" else settings.GEMINI_MODEL ) result = { "description": None, "tags": [], "confidence": None, "model": active_model, "prompt_tokens": None, "output_tokens": None, } try: image_bytes, media_type = await _read_image(file_path) prompt = _build_prompt(ocr_hint, language) response = await _generate( prompt=prompt, image_bytes=image_bytes, media_type=media_type, max_tokens=settings.GEMINI_MAX_TOKENS, provider_override=provider_override, model_override=model_override, ) text = response.get("text") result["prompt_tokens"], result["output_tokens"] = response.get("usage") if text: parsed = _extract_json(text) if parsed: result["description"] = parsed.get("description") result["tags"] = parsed.get("tags", []) result["confidence"] = parsed.get("confidence") else: logger.warning("ai.vision.json_parse_failed", extra={"raw": text[:100]}) if response.get("error"): logger.error("ai.vision.provider_error", extra={"error": response["error"]}) except Exception as e: logger.error("ai.vision.unexpected_error", extra={"error": str(e)}) return result async def extract_text_with_ai(file_path: str) -> dict: result = { "text": None, "has_text": False, "language": "unknown", "confidence": 0.0, "method": f"ai-{settings.AI_PROVIDER}" } if not settings.AI_ENABLED: return result logger.info("ai.ocr.fallback_start", extra={"file": Path(file_path).name}) try: image_bytes, media_type = await _read_image(file_path) prompt = """Agis comme un moteur OCR avancé. Extrais TOUT le texte visible dans cette image. Retourne UNIQUEMENT un objet JSON : { "text": "Le texte complet extrait ici...", "language": "fr" (code langue ISO 2 lettres, ex: fr, en, es), "confidence": 0.9 (estimation confiance 0.0 à 1.0) } Si aucun texte n'est visible, retourne : {"text": "", "has_text": false} """ response = await _generate( prompt=prompt, image_bytes=image_bytes, media_type=media_type, max_tokens=1024 ) text = response.get("text") if text: parsed = _extract_json(text) if parsed: extracted = parsed.get("text", "").strip() result["text"] = extracted result["has_text"] = bool(extracted) or parsed.get("has_text", False) result["language"] = parsed.get("language", "unknown") result["confidence"] = parsed.get("confidence", 0.0) logger.info("ai.ocr.success", extra={"chars": len(extracted)}) else: logger.warning("ai.ocr.json_parse_failed") else: logger.info("ai.ocr.empty_response") except Exception as e: logger.error("ai.ocr.error", extra={"error": str(e)}) return result async def summarize_url(url: str, content: str, language: str = "français") -> dict: result = { "summary": "", "tags": [], "model": settings.AI_PROVIDER, "prompt_tokens": None, "output_tokens": None, } if not settings.AI_ENABLED: return result prompt = f"""Tu reçois le contenu d'une page web. Génère un résumé et des tags en {language}. URL : {url} Contenu : \"\"\" {content[:3000]} \"\"\" Retourne UNIQUEMENT ce JSON : {{ "summary": "Résumé clair en 3-5 phrases en {language}", "tags": ["tag1", "tag2", "tag3"] }}""" try: response = await _generate(prompt=prompt, max_tokens=settings.GEMINI_MAX_TOKENS) text = response.get("text") if text: parsed = _extract_json(text) if parsed: result["summary"] = parsed.get("summary", "") result["tags"] = parsed.get("tags", []) result["prompt_tokens"], result["output_tokens"] = response.get("usage") except Exception as e: logger.error("ai.summarize_url.error", extra={"error": str(e)}) return result async def draft_task(description: str, context: Optional[str], language: str = "français") -> dict: result = { "title": "", "description": "", "steps": [], "estimated_time": None, "priority": None, "model": settings.AI_PROVIDER, "prompt_tokens": None, "output_tokens": None, } if not settings.AI_ENABLED: return result ctx_section = f"\nContexte : {context}" if context else "" prompt = f"""Tu es un assistant de gestion de tâches. Génère une tâche structurée en {language}. Description : {description}{ctx_section} Retourne UNIQUEMENT ce JSON : {{ "title": "Titre court et actionnable", "description": "Description complète de la tâche", "steps": ["Étape 1", "Étape 2", "Étape 3"], "estimated_time": "30 minutes", "priority": "haute|moyenne|basse" }}""" try: response = await _generate(prompt=prompt, max_tokens=settings.GEMINI_MAX_TOKENS) text = response.get("text") if text: parsed = _extract_json(text) if parsed: result.update(parsed) result["prompt_tokens"], result["output_tokens"] = response.get("usage") except Exception as e: logger.error("ai.draft_task.error", extra={"error": str(e)}) return result