""" AI Provider abstraction layer pro různé poskytovatele AI. Podporuje OpenAI, Anthropic Claude a Google AI. """ import time import xbmcaddon import xbmc from wcs.ai import accounts from wcs.ai.OpenAIClient import send_prompt as openai_send_prompt from wcs.ai.UsageTracker import track_ai_failure, track_ai_usage # (provider, klíč) dvojice, které v tomto procesu prošly ověřením _validated_keys = set() def check_ai_api_key_and_handle_error(provider=None): """ Kontroluje AI API klíč a v případě chyby zobrazí dialog. Vrací True pokud je klíč v pořádku, False pokud nastala chyba. Args: provider: Volitelný parametr pro specifikaci poskytovatele """ try: from wcs.login.DialogAILogin import show_ai_login_dialog current_provider = accounts.resolve(provider) provider_name = accounts.name(current_provider) # Klíč není nastaven - zobrazit dialog s poskytovatelem if not accounts.has_key(current_provider): return show_ai_login_dialog(provider=current_provider) # Ověření platnosti klíče (jednou za proces: změněný klíč je jiná dvojice, neplatný klíč selže u skutečného požadavku) checked = (current_provider, accounts.get_key(current_provider)) if checked in _validated_keys: return True try: if accounts.validate(current_provider): _validated_keys.add(checked) return True # Klíč je neplatný - zobrazit dialog pro nový klíč xbmc.log(f"[plugin.video.milionar] {provider_name} API klíč je neplatný", level=xbmc.LOGWARNING) except Exception as e: xbmc.log(f"[plugin.video.milionar] Chyba při ověřování {provider_name} API klíče: {e}", level=xbmc.LOGWARNING) return show_ai_login_dialog(provider=current_provider) except Exception as e: xbmc.log(f"[plugin.video.milionar] Error checking AI API key: {e}", level=xbmc.LOGERROR) return False class AIProvider: """Základní třída pro AI poskytovatele.""" def __init__(self, provider=None): self.addon = xbmcaddon.Addon() # Použij poskytovatele z parametru nebo z nastavení jako fallback self.current_provider = provider or self.addon.getSetting("ai_provider") or "mistral" # Normalizace názvu poskytovatele if "mistral" in self.current_provider.lower(): self.current_provider = "mistral" def send_prompt(self, prompt: str, model: str = None, use_tools: bool = False, usage_type: str = "other", session_id: str = None, **kwargs) -> tuple: """ Odešle prompt podle vybraného poskytovatele. Automaticky kontroluje API klíč před odesláním a zapíše volání (i selhané) do historie nákladů. Args: prompt (str): Text promptu model (str, optional): Specifický model (pokud None, použije se z nastavení) use_tools (bool): Použít tools jako web search (pouze pro OpenAI) usage_type (str): K čemu volání slouží (millionaire_chat, quiz, ...); řídí třídění v Cenách AI session_id (str, optional): ID hry; volání jedné hry se v historii spojí Returns: tuple: (response_text, usage_data) """ # Kontrola API klíče před odesláním if not check_ai_api_key_and_handle_error(provider=self.current_provider): raise ValueError("AI API klíč není nastaven nebo dialog byl zrušen.") sender = {"openai": lambda: self._send_openai_prompt(prompt, model, use_tools, **kwargs), "anthropic": lambda: self._send_anthropic_prompt(prompt, model, **kwargs), "google": lambda: self._send_google_prompt(prompt, model, **kwargs), "mistral": lambda: self._send_mistral_prompt(prompt, model, **kwargs), "groq": lambda: self._send_groq_prompt(prompt, model, **kwargs)}.get(self.current_provider) if sender is None: raise ValueError(f"Nepodporovaný AI poskytovatel: {self.current_provider}") used_model = model or self.get_current_model() started = time.monotonic() try: result = sender() except Exception as error: track_ai_failure(self.current_provider, used_model, usage_type, error, session_id, int((time.monotonic() - started) * 1000)) raise usage = result[1] if isinstance(result, tuple) and len(result) > 1 and isinstance(result[1], dict) else {} usage["latency_ms"] = int((time.monotonic() - started) * 1000) track_ai_usage(self.current_provider, used_model, usage, usage_type, session_id=session_id) return result def _send_openai_prompt(self, prompt: str, model: str = None, use_tools: bool = False, **kwargs) -> tuple: """Odešle prompt přes OpenAI API.""" return openai_send_prompt(prompt, model, use_tools, **kwargs) def _send_anthropic_prompt(self, prompt: str, model: str = None, **kwargs) -> tuple: """Odešle prompt přes Anthropic Claude API.""" from wcs.ai.AnthropicClient import send_prompt as anthropic_send_prompt return anthropic_send_prompt(prompt, model, **kwargs) def _send_google_prompt(self, prompt: str, model: str = None, **kwargs) -> tuple: """Odešle prompt přes Google AI API.""" from wcs.ai.GoogleClient import send_prompt as google_send_prompt return google_send_prompt(prompt, model, **kwargs) def _send_mistral_prompt(self, prompt: str, model: str = None, **kwargs) -> tuple: """Odešle prompt přes Mistral AI API.""" from wcs.ai.MistralClient import send_prompt as mistral_send_prompt return mistral_send_prompt(prompt, model, **kwargs) def _send_groq_prompt(self, prompt: str, model: str = None, **kwargs) -> tuple: """Odešle prompt přes Groq API.""" from wcs.ai.GroqClient import send_prompt as groq_send_prompt return groq_send_prompt(prompt, model, **kwargs) def get_current_provider(self) -> str: """Vrací aktuálně vybraný poskytovatel.""" return self.current_provider def get_available_providers(self) -> list: """Vrací seznam dostupných poskytovatelů.""" return ["openai", "anthropic", "google", "mistral", "groq"] def get_current_model(self) -> str: """Vrací aktuálně vybraný model podle poskytovatele.""" if self.current_provider == "openai": from wcs.ai.OpenAIClient import get_current_model return get_current_model() elif self.current_provider == "anthropic": from wcs.ai.AnthropicClient import get_current_model return get_current_model() elif self.current_provider == "google": from wcs.ai.GoogleClient import get_current_model return get_current_model() elif self.current_provider == "mistral": from wcs.ai.MistralClient import get_current_model return get_current_model() elif self.current_provider == "groq": from wcs.ai.GroqClient import get_current_model return get_current_model() else: return "neznámý" def get_model_info_string(self) -> str: """Vrací formátovaný řetězec s informací o aktuálním modelu a poskytovateli.""" provider_names = { "openai": "OpenAI", "anthropic": "Anthropic", "google": "Google AI", "mistral": "Mistral AI", "groq": "Groq" } provider_display = provider_names.get(self.current_provider, self.current_provider) model = self.get_current_model() return f"AI: {provider_display} - {model}" # Globální instance pro snadné použití ai_provider = AIProvider() def send_prompt(prompt: str, model: str = None, use_tools: bool = False, provider: str = None, **kwargs) -> tuple: """ Wrapper funkce pro posílání promptů. Automaticky používá aktuálně vybraný AI poskytovatel. Automaticky kontroluje API klíč před odesláním. Args: prompt (str): Text promptu model (str, optional): Specifický model use_tools (bool): Použít tools jako web search (pouze pro OpenAI) provider (str, optional): Specifický poskytovatel Returns: tuple: (response_text, usage_data) """ # Pokud je specifikován provider, vytvoř novou instanci if provider: temp_provider = AIProvider(provider=provider) return temp_provider.send_prompt(prompt, model, use_tools, **kwargs) else: return ai_provider.send_prompt(prompt, model, use_tools, **kwargs)