""" Mistral AI API handler Kompletní implementace pro Mistral AI API podle oficiální dokumentace """ import os import requests import xbmcaddon import xbmc import json import time from wcs.ai.DialogAIDebugRequest import show_api_request_debug, show_api_response_debug, show_api_error_debug from wcs.ai.model_catalog import DEFAULT_MODEL, model_or_default # Načtení nastavení addonu addon = xbmcaddon.Addon() API_BASE_URL = "https://api.mistral.ai/v1" def get_current_model(): """Vrací aktuálně vybraný Mistral model z nastavení.""" addon = xbmcaddon.Addon() return model_or_default('mistral', addon.getSetting("mistral_model") or DEFAULT_MODEL['mistral']) def _message_text(content): """Reasoning-capable models may return content as chunks (thinking + text).""" if isinstance(content, list): return "".join(chunk.get("text", "") for chunk in content if isinstance(chunk, dict) and chunk.get("type") == "text") return content or "" def send_prompt(prompt: str, model: str = None, **kwargs) -> tuple: """ Odešle prompt na Mistral AI API. Args: prompt (str): Text promptu model (str, optional): Specifický model (pokud None, použije se z nastavení) Returns: tuple: (response_text, usage_data) """ addon = xbmcaddon.Addon() API_KEY = addon.getSetting("mistral_api_key") if not API_KEY: raise ValueError("Mistral API klíč není nastaven.") # Extra data pro debug context_titles = kwargs.get('context_titles') context_limit = kwargs.get('context_limit') system_prompt = kwargs.get('system_prompt') # Pokud není model specifikován, použij model z nastavení if model is None: model = get_current_model() url = f"{API_BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", "Accept": "application/json" } # Načtení parametrů z nastavení temperature = float(addon.getSetting("mistral_temperature") or "0.7") top_p = float(addon.getSetting("mistral_top_p") or "1.0") max_tokens = int(kwargs.get("max_output_tokens") or addon.getSetting("mistral_max_tokens") or "4096") random_seed = int(addon.getSetting("mistral_random_seed") or "0") # Sestavení messages messages = [] # Podpora multi-turn messages messages_list = kwargs.get('messages', None) if messages_list: # Zkontrolovat jestli už messages obsahují system roli has_system = any(msg.get('role') == 'system' for msg in messages_list) # Přidat system prompt jako první, pokud není v messages a máme ho if system_prompt and not has_system: messages.append({"role": "system", "content": system_prompt}) # Přidat multi-turn messages for msg in messages_list: messages.append({"role": msg['role'], "content": msg['content']}) else: # Jednoduchý prompt bez multi-turn if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) payload = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "top_p": top_p } # Přidej random_seed pouze pokud není 0 (default) if random_seed != 0: payload["random_seed"] = random_seed # JSON mode - vynutit JSON response formát json_mode = kwargs.get('json_mode', False) if json_mode: payload["response_format"] = {"type": "json_object"} try: xbmc.log(f"[Mistral API] Odesílám prompt na model {model}", xbmc.LOGINFO) # === DEBUG PAYLOAD - messages už obsahují system pokud existuje === debug_payload = { "messages": messages, "model": model, "temperature": temperature, "max_tokens": max_tokens } # Debug: Zobrazí standardní request před odesláním if not kwargs.get('private_content', False) and not show_api_request_debug("Mistral", model, debug_payload, context_titles=context_titles, context_limit=context_limit): raise Exception("Odeslání API requestu bylo zrušeno uživatelem.") start_time = time.time() response = requests.post(url, headers=headers, json=payload, timeout=120) end_time = time.time() latency = end_time - start_time response.raise_for_status() data = response.json() # Debug: Zobrazí raw response po obdržení # Extrakce odpovědi z Mistral API formátu if 'choices' in data and len(data['choices']) > 0: response_text = _message_text(data['choices'][0]['message'].get('content')) else: response_text = "Chyba: Neplatná odpověď z Mistral API" # Debug: Zobrazí response s extrahovaným textem if not kwargs.get('private_content', False): show_api_response_debug("Mistral AI", model, data, response_text, latency=latency) # Extrakce usage informací (pokud jsou dostupné) usage_data = {} if 'usage' in data: usage_data = { 'prompt_tokens': data['usage'].get('prompt_tokens', 0), 'completion_tokens': data['usage'].get('completion_tokens', 0), 'total_tokens': data['usage'].get('total_tokens', 0) } if data['usage'].get('prompt_tokens_details'): usage_data['prompt_tokens_details'] = data['usage']['prompt_tokens_details'] xbmc.log(f"[Mistral API] Úspěšně získána odpověď: {len(response_text)} znaků", xbmc.LOGINFO) if kwargs.get('include_raw', False): return response_text, usage_data, data, latency return response_text, usage_data except requests.exceptions.RequestException as e: error_msg = f"Chyba při komunikaci s Mistral API: {str(e)}" xbmc.log(f"[Mistral API] {error_msg}", xbmc.LOGERROR) # Debug: Zobrazí chybu show_api_error_debug("Mistral AI", model, error_msg, {"exception": str(e)}) raise Exception(error_msg) except Exception as e: error_msg = f"Neočekávaná chyba v Mistral API: {str(e)}" xbmc.log(f"[Mistral API] {error_msg}", xbmc.LOGERROR) # Debug: Zobrazí chybu show_api_error_debug("Mistral AI", model, error_msg, {"exception": str(e)}) raise Exception(error_msg) def validate_api_key(): """ Ověří platnost API klíče. Returns: bool: True pokud je klíč platný, False jinak """ addon = xbmcaddon.Addon() API_KEY = addon.getSetting("mistral_api_key") if not API_KEY: return False try: # Skutečné API volání pro ověření klíče url = f"{API_BASE_URL}/models" headers = { "Authorization": f"Bearer {API_KEY}", "Accept": "application/json" } response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() return True except requests.exceptions.HTTPError as e: if e.response.status_code == 401: xbmc.log("[wcs][mistral_api] validate_api_key: Neplatný API klíč (401)", xbmc.LOGWARNING) return False else: xbmc.log(f"[wcs][mistral_api] validate_api_key: HTTP chyba {e.response.status_code}", xbmc.LOGWARNING) return False except requests.exceptions.RequestException as e: xbmc.log(f"[wcs][mistral_api] validate_api_key: Síťová chyba - {e}", xbmc.LOGWARNING) return False except Exception as e: xbmc.log(f"[wcs][mistral_api] validate_api_key: Neočekávaná chyba - {e}", xbmc.LOGWARNING) return False