""" AI usage tracking: the price of every AI call and its record in the usage history (wcs/ai/usage_store.py). The calls are recorded in one place, AIProvider.send_prompt (a failed call too); `describe_session` gives a game's calls the title of the game. The "Ceny AI" dialog (wcs/ai/DialogAICosts.py) reads the history. """ import datetime import json import os import xbmc from wcs.ai import usage_store from wcs.ai.model_catalog import price_for from wcs.user_data import _profile # Cenové tabulky pro jednotlivé poskytovatele ($/1M tokenů) -- jen historické modely pro starší statistiky; # aktuální modely a ceny drží wcs/ai/model_catalog.py (čte je get_pricing) PRICING_TABLES = { "openai": { "gpt-5.2": {"input": 1.75, "cached_input": 0.175, "output": 14.00}, "gpt-5.1": {"input": 1.25, "cached_input": 0.125, "output": 10.00}, "gpt-5-mini": {"input": 0.25, "cached_input": 0.025, "output": 2.00}, "gpt-4.1": {"input": 2.00, "cached_input": 1.00, "output": 8.00}, "gpt-4.1-mini": {"input": 0.40, "cached_input": 0.20, "output": 1.60}, "gpt-4.1-nano": {"input": 0.10, "cached_input": 0.05, "output": 0.40}, "gpt-4o": {"input": 2.50, "cached_input": 1.25, "output": 10.00}, "gpt-4o-latest": {"input": 2.50, "cached_input": 1.25, "output": 10.00}, "chatgpt-4o-latest": {"input": 2.50, "cached_input": 1.25, "output": 10.00}, "gpt-4o-mini": {"input": 0.15, "cached_input": 0.075, "output": 0.30}, "o3": {"input": 1.00, "output": 4.00}, "o4-mini": {"input": 0.60, "output": 2.40}, "o3-mini": {"input": 0.15, "cached_input": 0.075, "output": 0.60}, "o1": {"input": 15.00, "cached_input": 7.50, "output": 60.00}, "o1-mini": {"input": 3.00, "cached_input": 1.50, "output": 12.00}, "gpt-35-turbo": {"input": 0.50, "output": 1.50} }, "anthropic": { "claude-opus-4-20250514": {"input": 15.00, "cached_input": 0.30, "output": 75.00}, "claude-sonnet-4-20250514": {"input": 3.00, "cached_input": 0.30, "output": 15.00}, "claude-3-5-sonnet-20241022": {"input": 3.00, "cached_input": 0.30, "output": 15.00}, "claude-3-5-haiku-20241022": {"input": 0.80, "cached_input": 0.08, "output": 4.00}, "claude-3-opus-20240229": {"input": 15.00, "cached_input": 1.50, "output": 75.00}, "claude-3-sonnet-20240229": {"input": 3.00, "cached_input": 0.30, "output": 15.00}, "claude-3-haiku-20240307": {"input": 0.25, "cached_input": 0.025, "output": 1.25} }, "google": { "gemini-2.5-pro": {"input": 1.25, "cached_input": 0.31, "output": 10.00}, "gemini-2.5-flash": {"input": 0.30, "cached_input": 0.075, "output": 2.50}, "gemini-2.5-flash-lite": {"input": 0.10, "cached_input": 0.025, "output": 0.40}, "gemini-2.0-flash": {"input": 0.15, "cached_input": 0.0375, "output": 0.60}, "gemini-2.0-flash-lite": {"input": 0.075, "cached_input": 0.0188, "output": 0.30}, "gemini-1.5-pro": {"input": 1.25, "cached_input": 0.31, "output": 5.00}, "gemini-1.5-flash": {"input": 0.075, "cached_input": 0.0188, "output": 0.30}, "gemini-1.5-flash-8b": {"input": 0.0375, "cached_input": 0.0094, "output": 0.15} }, "mistral": { "mistral-large-latest": {"input": 2.00, "output": 6.00}, "magistral-medium-latest": {"input": 2.00, "output": 5.00}, "mistral-medium-latest": {"input": 0.40, "output": 2.00}, "magistral-small-latest": {"input": 0.10, "output": 0.30}, "mistral-small-latest": {"input": 0.10, "output": 0.30}, "codestral-latest": {"input": 0.30, "output": 0.90}, "devstral-small-latest": {"input": 0.10, "output": 0.30}, "pixtral-large-latest": {"input": 2.00, "output": 6.00}, "ministral-8b-latest": {"input": 0.10, "output": 0.10}, "ministral-3b-latest": {"input": 0.04, "output": 0.04}, "open-mistral-nemo": {"input": 0.10, "output": 0.30}, "mistral-saba-latest": {"input": 0.10, "output": 0.30} }, "groq": { "meta-llama/llama-4-maverick-17b-128e-instruct": {"input": 0.50, "output": 0.77}, "meta-llama/llama-4-scout-17b-16e-instruct": {"input": 0.11, "output": 0.34}, "llama-3.3-70b-versatile": {"input": 0.59, "output": 0.79}, "llama-3.1-8b-instant": {"input": 0.05, "output": 0.08}, "openai/gpt-oss-120b": {"input": 0.15, "cached_input": 0.075, "output": 0.60}, "openai/gpt-oss-20b": {"input": 0.075, "output": 0.30}, "groq/compound": {"input": 0.30, "output": 0.90}, "groq/compound-mini": {"input": 0.10, "output": 0.30}, "moonshotai/kimi-k2-instruct-0905": {"input": 1.00, "cached_input": 0.50, "output": 3.00}, "qwen/qwen3-32b": {"input": 0.29, "output": 0.59} } } def get_pricing(provider, model, today=None): """Ceník modelu ($/1M tokenů): aktuální model z katalogu (s ohlášenými změnami cen k danému dni), jinak historický.""" return price_for(provider, model, today) or PRICING_TABLES.get(provider, {}).get(model, {}) def usage_from_google(meta): """Gemini usageMetadata na standardní usage; thinking tokeny se účtují jako výstup, cached jsou součástí promptu.""" return { "prompt_tokens": meta.get("promptTokenCount", 0), "completion_tokens": meta.get("candidatesTokenCount", 0) + meta.get("thoughtsTokenCount", 0), "prompt_tokens_details": {"cached_tokens": meta.get("cachedContentTokenCount", 0)}, "total_tokens": meta.get("totalTokenCount", 0), } def _token_counts(usage_data): """(input, cached, cache_write, output) tokeny z usage libovolného poskytovatele; input neobsahuje cached ani zápisy do cache.""" if "input_tokens" in usage_data: # OpenAI Responses a Anthropic input_tokens = usage_data.get("input_tokens", 0) output_tokens = usage_data.get("output_tokens", 0) if "cached_input_tokens" in usage_data or "cache_read_input_tokens" in usage_data: # Anthropic (naše usage i syrová odpověď): input_tokens už neobsahují čtení ani zápisy cache return (input_tokens, usage_data.get("cached_input_tokens", usage_data.get("cache_read_input_tokens", 0)), usage_data.get("cache_creation_input_tokens", 0), output_tokens) details = usage_data.get("input_tokens_details") or {} else: # Chat Completions (Mistral, Groq) a Google převedený na tento tvar input_tokens = usage_data.get("prompt_tokens", 0) output_tokens = usage_data.get("completion_tokens", 0) details = usage_data.get("prompt_tokens_details") or {} cached_tokens = details.get("cached_tokens", 0) return max(0, input_tokens - cached_tokens), cached_tokens, 0, output_tokens def calculate_cost(provider, model, usage_data): """ Vypočítá náklady pro daný provider/model/usage. Args: provider (str): AI poskytovatel (openai, anthropic, google, mistral, groq) model (str): Název modelu usage_data (dict): Usage data s tokeny Returns: dict: náklady (input_cost, cached_input_cost, output_cost, total_cost), úspora díky cache (cache_savings) a tokeny """ input_tokens, cached_tokens, cache_write_tokens, output_tokens = _token_counts(usage_data) pricing = get_pricing(provider, model) if not pricing: xbmc.log(f'[wcs] No pricing data for {provider}/{model}, returning 0 cost but tracked tokens', xbmc.LOGWARNING) return { "input_cost": 0.0, "cached_input_cost": 0.0, "output_cost": 0.0, "total_cost": 0.0, "cache_savings": 0.0, "input_tokens": input_tokens + cache_write_tokens, "cached_tokens": cached_tokens, "output_tokens": output_tokens } # Dlouhý kontext se účtuje celý jinou sazbou long_context = pricing.get("long_context") if long_context and input_tokens + cached_tokens + cache_write_tokens > long_context["above"]: pricing = {**pricing, **long_context} # Ceny jsou za 1M tokenů; model bez ceny za cache čte cache za cenu běžného vstupu, zápis do cache také input_price = pricing.get("input", 0.0) input_cost = (input_tokens * input_price + cache_write_tokens * pricing.get("cache_write", input_price)) / 1_000_000 cached_price = pricing.get("cached_input", input_price) cached_input_cost = cached_tokens * cached_price / 1_000_000 cache_savings = cached_tokens * (input_price - cached_price) / 1_000_000 output_cost = output_tokens * pricing.get("output", 0.0) / 1_000_000 total_cost = input_cost + cached_input_cost + output_cost xbmc.log(f'[wcs] Cost calculation: input={input_tokens + cache_write_tokens}, cached={cached_tokens}, output={output_tokens}, total_cost=${total_cost:.6f}', xbmc.LOGINFO) return { "input_cost": input_cost, "cached_input_cost": cached_input_cost, "output_cost": output_cost, "total_cost": total_cost, "cache_savings": cache_savings, "input_tokens": input_tokens + cache_write_tokens, "cached_tokens": cached_tokens, "output_tokens": output_tokens } _session_titles = {} def describe_session(session_id, title): """Names a game's calls (the calls of one session id) in the history.""" if session_id and title: _session_titles[session_id] = title def track_ai_usage(provider, model, usage_data, request_type="other", session_id=None, title=None): """ Zapíše jedno AI volání do historie i s cenou. Volá se z AIProvider.send_prompt, jinde ne. Args: provider (str): AI poskytovatel model (str): Použitý model usage_data (dict): Usage data z API response (volitelně s latency_ms) request_type (str): K čemu volání sloužilo (millionaire_chat, director_tip, quiz, ...) session_id (str): ID herní session; volání jedné hry se v historii spojí title (str): Název hry; jinak ten, který zadal describe_session """ try: usage_data = usage_data or {} cost_info = calculate_cost(provider, model, usage_data) usage_store.record_call( provider, model, request_type, session_id=session_id or '', title=title or _session_titles.get(session_id, ''), latency_ms=usage_data.get('latency_ms', 0), input_tokens=cost_info["input_tokens"], cached_tokens=cost_info["cached_tokens"], output_tokens=cost_info["output_tokens"], input_cost=cost_info["input_cost"], cached_cost=cost_info["cached_input_cost"], output_cost=cost_info["output_cost"], savings=cost_info["cache_savings"]) usage_store.prune_if_due() except Exception as e: xbmc.log(f'[wcs] Error recording AI usage: {e}', xbmc.LOGERROR) def track_ai_failure(provider, model, request_type, error, session_id=None, latency_ms=0): """Zapíše volání, které selhalo (bez tokenů a ceny), i s důvodem.""" try: usage_store.record_call(provider, model, request_type or 'other', session_id=session_id or '', title=_session_titles.get(session_id, ''), status='error', error=str(error), latency_ms=latency_ms) except Exception as e: xbmc.log(f'[wcs] Error recording AI failure: {e}', xbmc.LOGERROR) # Kurz koruny k dolaru: denní kurz ČNB, záložní hodnota platí, dokud se kurz nepodaří stáhnout USD_TO_CZK_FALLBACK = 21.647 # ČNB 2026-10-06 _CNB_DAILY_URL = ('https://www.cnb.cz/en/financial-markets/foreign-exchange-market/' 'central-bank-exchange-rate-fixing/central-bank-exchange-rate-fixing/daily.txt') _RATE_FILE = os.path.join(_profile, 'usd_czk_rate.json') _rate_memo = {} def _fetch_cnb_usd_rate(): """USD/CZK z denního kurzovního lístku ČNB (řádek 'USA|dollar|1|USD|21.647'), nebo None.""" import requests response = requests.get(_CNB_DAILY_URL, timeout=3) response.raise_for_status() for line in response.text.splitlines(): parts = line.split('|') if len(parts) == 5 and parts[3] == 'USD': return float(parts[4].replace(',', '.')) / float(parts[2]) return None def usd_to_czk(): """Kurz USD/CZK; stahuje se nejvýše jednou denně, při výpadku platí poslední známý kurz (nebo záložní).""" today = datetime.date.today().isoformat() if _rate_memo.get('day') == today: return _rate_memo['rate'] state = {} try: with open(_RATE_FILE, encoding='utf-8') as handle: state = json.load(handle) except (OSError, ValueError): pass rate = state.get('rate') or USD_TO_CZK_FALLBACK if state.get('checked') != today: try: rate = _fetch_cnb_usd_rate() or rate except Exception as e: xbmc.log(f'[wcs] USD/CZK rate refresh failed, using {rate}: {e}', xbmc.LOGWARNING) try: os.makedirs(os.path.dirname(_RATE_FILE), exist_ok=True) with open(_RATE_FILE, 'w', encoding='utf-8') as handle: json.dump({'rate': rate, 'checked': today}, handle) except OSError: pass _rate_memo.update(day=today, rate=rate) return rate def _usd_text(cost): if cost == 0: return "$0" return f"${cost:.2f}" if cost >= 1.0 else f"${cost:.3f}" if cost >= 0.01 else f"${cost:.6f}" def _czk_text(czk_cost): if czk_cost == 0: return "0 Kč" return f"{czk_cost:.0f} Kč" if czk_cost >= 10.0 else f"{czk_cost:.1f} Kč" if czk_cost >= 0.1 else f"{czk_cost:.2f} Kč" def cost_parts(cost, addon=None): """(USD text, CZK text) of a cost; the part the user's currency setting (ai_cost_currency) hides is ''.""" currency_setting = "USD(CZK)" if addon: try: currency_setting = {"0": "USD", "1": "CZK", "2": "USD(CZK)"}.get(addon.getSetting("ai_cost_currency"), addon.getSetting("ai_cost_currency") or "USD(CZK)") except Exception: currency_setting = "USD(CZK)" usd = _usd_text(cost) if currency_setting != "CZK" else "" czk = _czk_text(cost * usd_to_czk()) if currency_setting != "USD" else "" return usd, czk def format_cost(cost, addon=None): """Formátuje náklady pro zobrazení podle uživatelského nastavení.""" usd, czk = cost_parts(cost, addon) return usd + (f" ({czk})" if usd and czk else czk) def format_tokens(tokens): """Formátuje počet tokenů pro zobrazení.""" if tokens >= 1_000_000: return f"{tokens/1_000_000:.1f}M" elif tokens >= 1_000: return f"{tokens/1_000:.1f}K" else: return str(tokens)