import time import logging import random import os import re from functools import wraps from typing import Optional, Union, List # Versuche neue Google GenAI Lib (v1.0+) try: from google import genai from google.genai import types HAS_NEW_GENAI = True except ImportError: HAS_NEW_GENAI = False # Fallback auf alte Lib try: import google.generativeai as old_genai HAS_OLD_GENAI = True except ImportError: HAS_OLD_GENAI = False from ..config import settings logger = logging.getLogger(__name__) # ============================================================================== # 1. DECORATORS # ============================================================================== def retry_on_failure(max_retries: int = 3, delay: float = 2.0): """ Decorator for retrying functions with exponential backoff. """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: last_exception = e # Don't retry on certain fatal errors (can be extended) if isinstance(e, ValueError) and "API Key" in str(e): raise e wait_time = delay * (2 ** attempt) + random.uniform(0, 1) logger.warning(f"Retry {attempt + 1}/{max_retries} for '{func.__name__}' after error: {e}. Waiting {wait_time:.1f}s") time.sleep(wait_time) logger.error(f"Function '{func.__name__}' failed after {max_retries} attempts.") raise last_exception return wrapper return decorator # ============================================================================== # 2. TEXT TOOLS # ============================================================================== def clean_text(text: str) -> str: """Removes excess whitespace and control characters.""" if not text: return "" text = str(text).strip() text = re.sub(r'\s+', ' ', text) return text def normalize_string(s: str) -> str: """Basic normalization (lowercase, stripped).""" return s.lower().strip() if s else "" # ============================================================================== # 3. LLM WRAPPER (GEMINI) # ============================================================================== @retry_on_failure(max_retries=3) def call_gemini( prompt: Union[str, List[str]], model_name: str = "gemini-2.0-flash", temperature: float = 0.3, json_mode: bool = False, system_instruction: Optional[str] = None ) -> str: """ Unified caller for Gemini API. Prefers new `google.genai` library. """ api_key = settings.GEMINI_API_KEY if not api_key: raise ValueError("GEMINI_API_KEY is missing in configuration.") # Option A: New Library (google-genai) if HAS_NEW_GENAI: try: client = genai.Client(api_key=api_key) config = { "temperature": temperature, "top_p": 0.95, "top_k": 40, "max_output_tokens": 8192, } if json_mode: config["response_mime_type"] = "application/json" response = client.models.generate_content( model=model_name, contents=[prompt] if isinstance(prompt, str) else prompt, config=config, ) if not response.text: raise ValueError("Empty response from Gemini") return response.text.strip() except Exception as e: logger.error(f"Error with google-genai lib: {e}") if not HAS_OLD_GENAI: raise e # Fallthrough to Option B # Option B: Old Library (google-generativeai) if HAS_OLD_GENAI: try: old_genai.configure(api_key=api_key) generation_config = { "temperature": temperature, "top_p": 0.95, "top_k": 40, "max_output_tokens": 8192, } if json_mode: generation_config["response_mime_type"] = "application/json" model = old_genai.GenerativeModel( model_name=model_name, generation_config=generation_config, system_instruction=system_instruction ) response = model.generate_content(prompt) return response.text.strip() except Exception as e: logger.error(f"Error with google-generativeai lib: {e}") raise e raise ImportError("No Google GenAI library installed (neither google-genai nor google-generativeai).")