Brancheneinstufung2/company-explorer/backend/lib/core_utils.py

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