# -*- coding: utf-8 -*- """Rule checks of one played game: the AI answers, the game state and the speed. `analyze(events, steps, settings, source)` -> dict with rounds per round: question (raw, with audio tags), chosen answer, feedback, tip, timings llm every LLM call (purpose, round, ms, tokens, cost) totals tokens, cost, LLM time, HTTP calls issues [{"severity": "error" | "warn" | "info", "code", "round", "message"}] """ import re import unicodedata LETTERS = 'ABCD' MAX_OPTION_AI = 25 # what the prompt asks for MAX_OPTION_UI = 30 # what the answer buttons show (the demo pack uses up to 30) SLOW_LLM_MS = 20000 SLOW_WAIT_MS = {'question': 4000, 'answer': 10000, 'start': 15000} # answer: 3 s countdown + 5 s result by design AUDIO_TAG = re.compile(r'\[[a-z][a-z \-]{1,24}\]') TEXT_FIELDS = ('moderator_welcome', 'rules_summary', 'moderator_intro', 'result_announcement', 'transition_text', 'explanation') # Text boxes of the modern card: (characters per line, lines) measured on Kodi frames (font27 ~12.7 px per character # and 37 px per line, font32 ~15 px and 44 px): what fits without scrolling. FIT = {'intro': (64, 7), 'feedback': (70, 5), 'explanation': (72, 3)} STOP = set('a i o u v k s z na je se to ten ta ve od do za po při pro jak kdo co který která které jaký jaká jaké ' 'filmu film seriálu seriál v kterém kterou jakou jaký byl byla bylo jsou the of in'.split()) def _norm(text): text = unicodedata.normalize('NFKD', str(text or '').lower()) return ''.join(c for c in text if not unicodedata.combining(c)) def _words(text): return {w for w in re.findall(r'[a-z0-9]+', _norm(text)) if len(w) > 2 and w not in STOP} def _stems(text): """Word beginnings (Czech inflects word endings: Wachowští / Wachowski, tajemství / tajemstvím).""" return {w[:5] for w in _words(text) if len(w) >= 4} def _jaccard(a, b): return len(a & b) / float(len(a | b)) if a and b else 0.0 def _usage_tokens(usage): usage = usage or {} tokens_in = usage.get('input_tokens', usage.get('prompt_tokens', 0)) or 0 tokens_out = usage.get('output_tokens', usage.get('completion_tokens', 0)) or 0 cached = (usage.get('input_tokens_details') or usage.get('prompt_tokens_details') or {}).get('cached_tokens', 0) \ or usage.get('cached_input_tokens', 0) or 0 return tokens_in, tokens_out, cached def _cost(provider, model, usage): try: from wcs.ai.UsageTracker import calculate_cost return calculate_cost(provider, model, usage or {}).get('total_cost', 0.0) except Exception: return 0.0 def _audio_tags_allowed(): try: from wcs.games.millionaire_tts import is_elevenlabs_v3_active return bool(is_elevenlabs_v3_active()) except Exception: return False def analyze(events, steps, settings, source='ai', realtime=False): """realtime: the game ran with real countdowns, so the player's waits are judged too.""" issues = [] def issue(severity, code, message, round_num=None): issues.append({'severity': severity, 'code': code, 'round': round_num, 'message': message}) rounds, welcome, llm, http = {}, None, [], [] calls = {} for event in events: kind = event.get('kind') if kind == 'content.call': calls[event['call']] = event elif kind == 'content.result': call = calls.get(event['call'], {}) args = call.get('args') or [] method = event.get('method') if method == 'welcome': welcome = event.get('result') or {} elif method == 'question' and args: rounds.setdefault(args[0], {})['question'] = event.get('result') or {} rounds[args[0]]['question_ms'] = event.get('ms') rounds[args[0]]['user_message'] = event.get('user_message', '') elif method == 'feedback' and args: entry = rounds.setdefault(args[0], {}) entry['feedback'] = event.get('result') or {} entry['feedback_ms'] = event.get('ms') entry['chosen'] = args[1] if len(args) > 1 else None entry['is_correct'] = args[2] if len(args) > 2 else None elif method == 'tip': question = (args[0] if args else {}) or {} for number, entry in rounds.items(): if (entry.get('question') or {}).get('question') == question.get('question'): entry['tip'] = event.get('result') entry['tip_ms'] = event.get('ms') elif kind == 'content.error': call = calls.get(event['call'], {}) args = call.get('args') or [] issue('error', 'content_error', '{}: {}'.format(event.get('method'), event.get('error')), args[0] if args and isinstance(args[0], int) else None) elif kind == 'llm.request': llm.append({'call': event['call'], 'seq': event['seq'], 'purpose': event.get('purpose'), 'provider': event.get('provider'), 'model': event.get('model'), 'messages': len(event.get('messages') or []), 'last_user': ((event.get('messages') or [{}])[-1].get('content', '') if event.get('messages') else '')}) elif kind in ('llm.response', 'llm.error'): entry = next((x for x in llm if x['call'] == event['call']), None) if entry is None: continue entry['ms'] = event.get('ms') entry['end_seq'] = event['seq'] if kind == 'llm.error': entry['error'] = event.get('error') issue('error', 'llm_error', '{} ({} ms): {}'.format(entry['purpose'], event.get('ms'), event.get('error'))) else: usage = event.get('usage') or _http_usage(events, entry['seq'], event['seq']) tokens_in, tokens_out, cached = _usage_tokens(usage) entry.update({'in': tokens_in, 'out': tokens_out, 'cached': cached, 'cost': _cost(entry['provider'], entry['model'], usage)}) elif kind == 'http': http.append(event) elif kind == 'kodi.log' and event.get('level', 0) >= 3: issue('error', 'kodi_log', event.get('message', '')[:300]) elif kind == 'probe.error': issue('error', 'probe', event.get('error', '')) from wcs.games.millionaire_prompts import MONEY_AMOUNTS by_money = {money: number for number, money in MONEY_AMOUNTS.items()} last_question_round = None for entry in llm: last_user = entry.get('last_user', '') match = re.search(r'Kolo (\d+)/9', last_user) entry['round'] = int(match.group(1)) if match else None if entry['purpose'] != 'chat': entry['kind'] = entry['purpose'] elif 'Hráč vybral' in last_user: entry['kind'] = 'feedback' if not match: # older prompts named the amount instead of the round money = re.search(r'Částka: ([0-9 ]+ Kč)', last_user) entry['round'] = by_money.get(money.group(1)) if money else None else: entry['kind'] = 'question' if match else 'welcome' if entry['kind'] == 'question': last_question_round = entry['round'] elif entry['kind'] == 'tip': entry['round'] = last_question_round if entry.get('ms', 0) > SLOW_LLM_MS: issue('warn', 'slow_llm', '{} trvalo {:.1f} s'.format(entry['kind'], entry['ms'] / 1000.0), entry.get('round')) played = {} for step in steps: snap = step.get('snapshot') or {} if step['step'].startswith('answer-'): played[snap.get('round')] = step tags_ok = _audio_tags_allowed() max_option = MAX_OPTION_AI if source == 'ai' else _demo_max_option() # -- welcome -------------------------------------------------------------- if source == 'ai' and welcome is not None: for key in ('moderator_welcome', 'rules_summary'): if key not in welcome: issue('error', 'schema', 'welcome bez pole {}'.format(key)) _check_tags(welcome, tags_ok, issue, None) # -- questions ------------------------------------------------------------ seen = [] answers = [] letters = [] for number in sorted(rounds): entry = rounds[number] question = entry.get('question') if question is None: continue entry['used'] = number in played or any((s.get('snapshot') or {}).get('round') == number for s in steps if s['step'].startswith('question-')) missing = [k for k in ('question', 'options', 'correct', 'explanation') if k not in question] if missing: issue('error', 'schema', 'otázce chybí {}'.format(', '.join(missing)), number) options = question.get('options') or {} if isinstance(options, list): options = dict(zip(LETTERS, options)) texts = [str(options.get(letter, '') or '') for letter in LETTERS] if sorted(options) != list(LETTERS) or not all(t.strip() for t in texts): issue('error', 'options', 'možnosti nejsou přesně A-D neprázdné: {}'.format(options), number) for letter, text in zip(LETTERS, texts): if len(text) > max(max_option, MAX_OPTION_UI): issue('error', 'option_length', '{}: {} znaků > {} ({})'.format(letter, len(text), MAX_OPTION_UI, text), number) elif len(text) > max_option: issue('info', 'option_length', '{}: {} znaků > {} z promptu ({})'.format(letter, len(text), max_option, text), number) if len({_norm(t).strip() for t in texts}) < 4: issue('error', 'options_unique', 'duplicitní možnosti: {}'.format(texts), number) else: tokens = [set(re.findall(r'[a-z0-9]+', _norm(t))) for t in texts] for i in range(4): for j in range(i + 1, 4): a, b = tokens[i], tokens[j] # whole words: Heisenberg / Eisenberg is a fair trap if a and b and (a <= b or b <= a): issue('warn', 'options_ambiguous', 'skoro stejné možnosti: {} / {}'.format(texts[i], texts[j]), number) correct = str(question.get('correct', '')).strip().upper() if correct not in LETTERS: issue('error', 'correct', 'správná odpověď {!r} není A-D'.format(question.get('correct')), number) else: letters.append(correct) right = texts[LETTERS.index(correct)] if len(right) >= 4 and _norm(right) in _norm(question.get('question', '')): issue('warn', 'answer_leak', 'text správné odpovědi je v otázce ({})'.format(right), number) explanation = question.get('explanation', '') if explanation and _stems(right) and not (_stems(right) & _stems(explanation)) and source == 'ai': issue('info', 'explanation', 'vysvětlení nezmiňuje správnou odpověď ({})'.format(right), number) if correct in LETTERS: right_norm = _norm(texts[LETTERS.index(correct)]).strip() for other_number, other_right in answers: if right_norm and right_norm == other_right: issue('warn', 'repeated_answer', 'stejná správná odpověď jako v kole {} ({})'.format( other_number, texts[LETTERS.index(correct)]), number) answers.append((number, right_norm)) words = _words(question.get('question', '')) for other_number, other_words, other_text in seen: score = _jaccard(words, other_words) if _norm(question.get('question')) == _norm(other_text): issue('error', 'duplicate', 'stejná otázka jako v kole {}'.format(other_number), number) elif score >= 0.6: issue('warn', 'similar', 'podobná otázce z kola {} ({:.0%})'.format(other_number, score), number) seen.append((number, words, question.get('question', ''))) for key in ('question',): if AUDIO_TAG.search(question.get(key, '')): issue('error', 'audio_tag', 'audio tag v textu otázky', number) if any(AUDIO_TAG.search(t) for t in texts): issue('error', 'audio_tag', 'audio tag v možnostech', number) _check_tags(question, tags_ok, issue, number) intro = question.get('moderator_intro', '') if source == 'ai' and settings.get('milionar_len_intro', '1') != '0' and not intro.strip(): issue('warn', 'empty', 'prázdné moderator_intro', number) if source == 'ai' and settings.get('milionar_len_explanation', '1') != '0' and \ not question.get('explanation', '').strip(): issue('warn', 'empty', 'prázdné explanation', number) if len(letters) >= 6: counts = {letter: letters.count(letter) for letter in LETTERS} if max(counts.values()) > len(letters) * 0.55: issue('warn', 'letter_bias', 'správné odpovědi nerovnoměrně: {}'.format(counts)) # -- feedback ------------------------------------------------------------- for number in sorted(rounds): entry = rounds[number] feedback = entry.get('feedback') if feedback is None: continue if source == 'ai': for key in ('result_announcement', 'transition_text'): if key not in feedback: issue('error', 'schema', 'feedbacku chybí {}'.format(key), number) result = _norm(re.split(r'[.!?]', feedback.get('result_announcement', '') or '', 1)[0]) # the verdict sentence if entry.get('is_correct') is False and re.search(r'\bspravne\b', result) and \ not re.search(r'nespravn|spatn|bohuzel|lituji', result): issue('warn', 'feedback_tone', 'špatná odpověď, ale oznámení zní jako správná', number) if entry.get('is_correct') is True and re.search(r'\bspatne\b|bohuzel|nespravn', result): issue('warn', 'feedback_tone', 'správná odpověď, ale oznámení zní jako špatná', number) _check_tags(feedback, tags_ok, issue, number) # -- state the player saw --------------------------------------------------- from wcs.games.millionaire_prompts import MONEY_AMOUNTS for step in steps: snap = step.get('snapshot') or {} view = snap.get('view') or {} question = snap.get('question') or {} number = snap.get('round') screen = snap.get('screen') if screen == 'question' and question: options = question.get('options') or [] shown = [view.get('Opt' + letter, '') for letter in LETTERS] if shown != list(options): issue('error', 'view', 'tlačítka {} != otázka {}'.format(shown, options), number) if view.get('QuestionText') != question.get('question'): issue('error', 'view', 'QuestionText neodpovídá otázce', number) if view.get('Money') != MONEY_AMOUNTS.get(number): issue('error', 'view', 'částka {} != {}'.format(view.get('Money'), MONEY_AMOUNTS.get(number)), number) done = sum(1 for n in range(1, 10) if view.get('Ladder.{}.State'.format(n)) == 'done') if done != number - 1 or view.get('Ladder.{}.State'.format(number)) != 'current': issue('error', 'view', 'žebříček neodpovídá kolu', number) if screen in ('feedback', 'gameover') and question: if view.get('RightLetter') != question.get('correct_letter'): issue('error', 'view', 'RightLetter {} != {}'.format(view.get('RightLetter'), question.get('correct_letter')), number) if not view.get('Result') and source == 'ai': issue('warn', 'view', 'prázdné oznámení výsledku na obrazovce', number) if screen == 'error': issue('error', 'error_screen', 'hra skončila na obrazovce chyby', number) kind = step['step'].split('-')[0] limit = SLOW_WAIT_MS.get(kind) if realtime and limit and step.get('wait_ms', 0) > limit and step['step'] != 'question-1': issue('warn', 'slow_wait', '{} čekal {:.1f} s'.format(step['step'], step['wait_ms'] / 1000.0), number) # -- HTTP ------------------------------------------------------------------- checks_http = [e for e in http if e.get('method') == 'GET' and e.get('url', '').rstrip('/').endswith('/models')] if checks_http: issue('info', 'key_check', '{}× ověření API klíče (GET /models) před voláním, celkem {} ms'.format( len(checks_http), sum(e.get('ms', 0) for e in checks_http))) for event in http: if event.get('status', 200) >= 400 or event.get('error'): issue('error', 'http', '{} {} -> {} {}'.format(event.get('method'), event.get('url'), event.get('status', ''), event.get('error', ''))) totals = { 'llm_calls': len(llm), 'tokens_in': sum(x.get('in', 0) for x in llm), 'tokens_out': sum(x.get('out', 0) for x in llm), 'tokens_cached': sum(x.get('cached', 0) for x in llm), 'cost_usd': round(sum(x.get('cost', 0.0) for x in llm), 6), 'llm_ms': sum(x.get('ms', 0) or 0 for x in llm), 'http_calls': len(http), 'http_ms': sum(e.get('ms', 0) for e in http), 'errors': sum(1 for i in issues if i['severity'] == 'error'), 'warnings': sum(1 for i in issues if i['severity'] == 'warn'), } fit = text_fit(welcome, rounds, issue) totals['warnings'] = sum(1 for i in issues if i['severity'] == 'warn') return {'welcome': welcome, 'rounds': rounds, 'llm': llm, 'totals': totals, 'issues': issues, 'letters': letters, 'fit': fit} def _http_usage(events, first_seq, last_seq): """Usage from the raw provider answer (for LLM calls whose code does not return it, e.g. the tip).""" for event in events: if event.get('kind') == 'http' and first_seq < event['seq'] < last_seq and event.get('method') == 'POST': response = event.get('response') if isinstance(response, dict): return response.get('usage') or response.get('usageMetadata') and { 'input_tokens': response['usageMetadata'].get('promptTokenCount', 0), 'output_tokens': response['usageMetadata'].get('candidatesTokenCount', 0)} or {} return {} def lines(text, width): """Lines a text takes when wrapped at `width` characters (paragraphs kept).""" import textwrap return sum(max(1, len(textwrap.wrap(part, width))) for part in str(text or '').split('\n') if part.strip()) def text_fit(welcome, rounds, issue): """Estimated lines of the texts on their cards; warns when one would have to scroll. Returns the measurements.""" from wcs.games.millionaire_tts import strip_audio_tags sizes = {'intro': [], 'feedback': [], 'explanation': []} if welcome: width, room = FIT['intro'] used = (lines(strip_audio_tags(welcome.get('moderator_welcome', '')), width) + 1 + lines(strip_audio_tags(welcome.get('rules_summary', '')), width)) sizes['intro'].append(used) if used > room: issue('warn', 'text_fit', 'úvod má ~{} řádků, karta pojme {}'.format(used, room)) for number in sorted(rounds): entry = rounds[number] feedback, question = entry.get('feedback'), entry.get('question') or {} if feedback: width, room = FIT['feedback'] used = (lines(strip_audio_tags(feedback.get('result_announcement', '')), width) + lines(strip_audio_tags(feedback.get('transition_text', '')), width)) sizes['feedback'].append(used) if used > room and entry.get('is_correct'): issue('warn', 'text_fit', 'moderátor po odpovědi ~{} řádků, karta pojme {}'.format(used, room), number) if question.get('explanation') and entry.get('used', True): width, room = FIT['explanation'] used = lines(strip_audio_tags(question['explanation']), width) sizes['explanation'].append(used) if used > room: issue('warn', 'text_fit', 'vysvětlení ~{} řádků, karta pojme {}'.format(used, room), number) return sizes def _check_tags(data, tags_ok, issue, number): if tags_ok: return for key in TEXT_FIELDS: if AUDIO_TAG.search(str(data.get(key, '') or '')): issue('warn', 'audio_tag', 'audio tag v {} bez ElevenLabs v3'.format(key), number) def _demo_max_option(): from wcs.games.demo import catalog return catalog.MAX_OPTION