#!/usr/bin/env python3 import json import os import re import subprocess import sys import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen API_HOST = os.environ.get("TRIVIAI_API_HOST", "127.0.0.1") API_PORT = int(os.environ.get("TRIVIAI_API_PORT", "19194")) TIMEOUT = float(os.environ.get("TRIVIAI_TIMEOUT", "25")) GEMINI_KEY_FILE = Path(os.environ.get("TRIVIAI_GEMINI_KEY_FILE", "/home/ovidiu/~devops-center/secrets/gemini-api-key.txt")) OPENROUTER_KEY_FILE = Path(os.environ.get("TRIVIAI_OPENROUTER_KEY_FILE", "/home/ovidiu/~devops-center/secrets/openrouter")) GEMINI_MODELS = [ item.strip() for item in os.environ.get("TRIVIAI_GEMINI_MODELS", "gemini-2.5-flash,gemini-2.5-flash-lite").split(",") if item.strip() ] OPENROUTER_MODELS = [ item.strip() for item in os.environ.get("TRIVIAI_OPENROUTER_MODELS", "openrouter/auto").split(",") if item.strip() ] LIVE_MODEL = os.environ.get("TRIVIAI_LIVE_MODEL", "gemini-2.5-flash-native-audio-preview-12-2025") TOKEN_WINDOW_SECONDS = int(os.environ.get("TRIVIAI_TOKEN_WINDOW_SECONDS", "10")) _token_rate_limit: dict[str, float] = {} def debug_log(message: str) -> None: print(f"[triviai] {message}", file=sys.stderr, flush=True) def load_api_key() -> str | None: env_value = os.environ.get("GEMINI_API_KEY", "").strip() if env_value: return env_value if GEMINI_KEY_FILE.exists(): return GEMINI_KEY_FILE.read_text(encoding="utf-8").strip() return None def load_openrouter_api_key() -> str | None: env_value = os.environ.get("OPENROUTER_API_KEY", "").strip() if env_value: return env_value if OPENROUTER_KEY_FILE.exists(): return OPENROUTER_KEY_FILE.read_text(encoding="utf-8").strip() return None def safe_text(value: Any, default: str = "") -> str: if not isinstance(value, str): return default return re.sub(r"\s+", " ", value).strip() def looks_like_quota_issue(message: str) -> bool: normalized = safe_text(message).lower() return any( marker in normalized for marker in [ "resource_exhausted", "quota", "rate limit", "429", "generaterequestsperdayperprojectperformodel-freetier", ] ) def read_json_body(handler: BaseHTTPRequestHandler) -> dict[str, Any]: length = int(handler.headers.get("Content-Length", "0")) raw = handler.rfile.read(length) if length else b"{}" return json.loads(raw.decode("utf-8")) def json_response(handler: BaseHTTPRequestHandler, code: int, payload: dict[str, Any]) -> None: body = json.dumps(payload).encode("utf-8") handler.send_response(code) handler.send_header("Content-Type", "application/json") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def parse_model_json(raw_text: str) -> dict[str, Any]: text = raw_text.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text) start = text.find("{") end = text.rfind("}") if start == -1 or end == -1 or end <= start: raise ValueError("no JSON object found in Gemini response") return json.loads(text[start:end + 1]) def extract_text_from_gemini(payload: dict[str, Any]) -> str: candidates = payload.get("candidates") if not isinstance(candidates, list) or not candidates: raise ValueError("no candidates returned") parts = candidates[0].get("content", {}).get("parts", []) if not isinstance(parts, list): raise ValueError("invalid candidate format") chunks = [part.get("text", "") for part in parts if isinstance(part, dict)] text = "".join(chunks).strip() if not text: raise ValueError("empty candidate text") return text def gemini_generate(prompt: str) -> dict[str, Any]: api_key = load_api_key() if not api_key: raise RuntimeError("Gemini API key is not configured on the server") body = json.dumps({ "contents": [{"role": "user", "parts": [{"text": prompt}]}], "generationConfig": { "temperature": 0.7, "responseMimeType": "application/json", }, }).encode("utf-8") failures: list[str] = [] for model_name in GEMINI_MODELS: request = Request( url=f"https://generativelanguage.googleapis.com/v1beta/models/{model_name}:generateContent", data=body, method="POST", headers={ "Content-Type": "application/json", "x-goog-api-key": api_key, }, ) try: with urlopen(request, timeout=TIMEOUT) as response: payload = json.loads(response.read().decode("utf-8")) parsed = parse_model_json(extract_text_from_gemini(payload)) parsed["_geminiModel"] = model_name return parsed except HTTPError as exc: detail = exc.read().decode("utf-8", errors="ignore") failures.append(f"{model_name}: HTTP {exc.code}") debug_log(f"Gemini model {model_name} failed with HTTP {exc.code}: {detail[:220]}") except URLError as exc: failures.append(f"{model_name}: network error") debug_log(f"Gemini model {model_name} network error: {exc.reason}") except Exception as exc: failures.append(f"{model_name}: {safe_text(str(exc), 'unknown error')}") debug_log(f"Gemini model {model_name} parse/runtime error: {exc}") raise RuntimeError("All Gemini models failed: " + "; ".join(failures)) def extract_openrouter_text(payload: dict[str, Any]) -> str: choices = payload.get("choices") if not isinstance(choices, list) or not choices: raise ValueError("no choices returned") message = choices[0].get("message", {}) if not isinstance(message, dict): raise ValueError("invalid choice format") text = safe_text(message.get("content")) if not text: raise ValueError("empty choice content") return text def openrouter_generate(prompt: str) -> dict[str, Any]: api_key = load_openrouter_api_key() if not api_key: raise RuntimeError("OpenRouter API key is not configured on the server") failures: list[str] = [] for model_name in OPENROUTER_MODELS: body = json.dumps({ "model": model_name, "messages": [{"role": "user", "content": prompt}], "temperature": 0.7, "response_format": {"type": "json_object"}, }).encode("utf-8") request = Request( url="https://openrouter.ai/api/v1/chat/completions", data=body, method="POST", headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", "HTTP-Referer": "https://brainstorming.ovidiutm.com/triviai/", "X-Title": "TriviAI", }, ) try: with urlopen(request, timeout=TIMEOUT) as response: payload = json.loads(response.read().decode("utf-8")) parsed = parse_model_json(extract_openrouter_text(payload)) parsed["_openrouterModel"] = safe_text(payload.get("model"), model_name) return parsed except HTTPError as exc: detail = exc.read().decode("utf-8", errors="ignore") failures.append(f"{model_name}: HTTP {exc.code}") debug_log(f"OpenRouter model {model_name} failed with HTTP {exc.code}: {detail[:220]}") except URLError as exc: failures.append(f"{model_name}: network error") debug_log(f"OpenRouter model {model_name} network error: {exc.reason}") except Exception as exc: failures.append(f"{model_name}: {safe_text(str(exc), 'unknown error')}") debug_log(f"OpenRouter model {model_name} parse/runtime error: {exc}") raise RuntimeError("All OpenRouter models failed: " + "; ".join(failures)) def issue_live_token() -> dict[str, Any]: api_key = load_api_key() if not api_key: raise RuntimeError("Gemini API key is not configured on the server") completed = subprocess.run( ["node", "/home/ovidiu/~devops-center/brainstorming/TriviAI/create-live-token.mjs"], capture_output=True, text=True, timeout=TIMEOUT, env={ **os.environ, "GEMINI_API_KEY": api_key, "TRIVIAI_LIVE_MODEL": LIVE_MODEL, }, check=False, ) if completed.returncode != 0: stderr = safe_text(completed.stderr, "token generation failed") raise RuntimeError(stderr) payload = json.loads(completed.stdout) token = safe_text(payload.get("token")) if not token: raise RuntimeError("Gemini did not return an ephemeral token") return payload def fallback_question(topic: str, personality: str) -> dict[str, Any]: topic_key = topic.lower().strip() bank = { "space exploration": { "question": "Which planet in our solar system is famous for its prominent ring system?", "options": ["Mars", "Saturn", "Mercury", "Venus"], "correctAnswer": "Saturn", "hostCommentary": f"{personality} takes the mic. We are opening with a friendly orbital warm-up.", }, "google ai ecosystem": { "question": "Which Google Cloud product is the main enterprise platform for building and deploying AI systems?", "options": ["Vertex AI", "NotebookLM", "Google Photos", "Chrome Sync"], "correctAnswer": "Vertex AI", "hostCommentary": f"{personality} grins. Time for a platform-layer question with just enough bite.", }, } return bank.get( topic_key, { "question": f"Which answer is the safest default when a trivia host asks about {topic} without extra context?", "options": ["Ask for evidence", "Guess wildly", "Ignore the topic", "Delete the scoreboard"], "correctAnswer": "Ask for evidence", "hostCommentary": f"{personality} steps in with a fallback round while Gemini catches its breath.", }, ) def build_question_prompt(topic: str, personality: str, exclude_questions: list[str] | None = None) -> str: exclude_block = "" cleaned_excludes = [safe_text(item) for item in (exclude_questions or []) if safe_text(item)] if cleaned_excludes: joined = "\n".join(f"- {item}" for item in cleaned_excludes[:6]) exclude_block = f""" Do NOT reuse or closely paraphrase any of these previously used questions: {joined} """ return f""" You are generating one multiple-choice trivia question for a web app. Topic: - {topic} Host personality: - {personality} Return ONLY valid JSON with this exact shape: {{ "hostCommentary": "1 short opening line in the host personality", "question": "question text", "options": ["option 1", "option 2", "option 3", "option 4"], "correctAnswer": "exact text of the correct option" }} Rules: - exactly 4 options - only 1 correct option - keep facts broadly reliable and non-controversial - choose a different angle from previous rounds for the same topic - prefer high-confidence, widely known facts over obscure trivia - if the topic is a city, region, or country, prefer landmarks, geography, public institutions, sports clubs, museums, rivers, parks, counties/regions, or well-known historical sites - avoid biography questions unless you are highly certain the person is directly tied to the place - avoid uncertain or arguable claims - no markdown - no code fences - no text outside the JSON {exclude_block} """.strip() def build_answer_prompt(question: str, correct_answer: str, user_answer: str, personality: str) -> str: return f""" You are a trivia host reacting to an answer. Question: - {question} Correct answer: - {correct_answer} User answer: - {user_answer} Host personality: - {personality} Return ONLY valid JSON: {{ "feedback": "1 or 2 short sentences saying whether the user was right or wrong in character" }} Rules: - be lively and concise - no markdown - no code fences - no text outside the JSON """.strip() def looks_like_place_topic(topic: str) -> bool: topic_lower = topic.lower() return any( marker in topic_lower for marker in [ "romania", "city", "county", "region", "village", "town", "country", ",", ] ) def question_seems_risky(topic: str, question: str) -> bool: normalized = question.lower() if looks_like_place_topic(topic): risky_phrases = [ "who was born", "was born in", "which famous romanian writer", "which famous writer", "which poet", "which novelist", "which playwright", "who founded", "who discovered", ] return any(phrase in normalized for phrase in risky_phrases) return False def generate_question(topic: str, personality: str, exclude_questions: list[str] | None = None) -> dict[str, Any]: provider_message = "" try: raw = gemini_generate(build_question_prompt(topic, personality, exclude_questions)) source = "gemini" model = safe_text(raw.get("_geminiModel"), GEMINI_MODELS[0]) provider_message = f"Generated with {model}." except Exception as gemini_exc: debug_log(f"Gemini question generation failed, trying OpenRouter: {gemini_exc}") try: raw = openrouter_generate(build_question_prompt(topic, personality, exclude_questions)) source = "openrouter" model = safe_text(raw.get("_openrouterModel"), OPENROUTER_MODELS[0]) if looks_like_quota_issue(str(gemini_exc)): provider_message = f"Used OpenRouter fallback ({model}) because Gemini free-tier quota was exhausted." else: provider_message = f"Used OpenRouter fallback ({model}) because Gemini was unavailable." except Exception as openrouter_exc: debug_log(f"OpenRouter question generation failed after Gemini: {openrouter_exc}") payload = fallback_question(topic, personality) payload["source"] = "fallback" payload["fallbackReason"] = ( f"Gemini failed: {safe_text(str(gemini_exc), 'unknown error')}; " f"OpenRouter failed: {safe_text(str(openrouter_exc), 'unknown error')}" ) return payload try: options = raw.get("options") if not isinstance(options, list) or len(options) < 4: raise ValueError("model did not return four options") cleaned_options = [safe_text(item) for item in options[:4] if safe_text(item)] if len(cleaned_options) < 4: raise ValueError("model returned empty option values") question_text = safe_text(raw.get("question"), f"Which answer best fits the topic {topic}?") normalized_excludes = { safe_text(item).lower() for item in (exclude_questions or []) if safe_text(item) } normalized_question = question_text.lower() if normalized_question in normalized_excludes: raise ValueError("model repeated a recently used question") if question_seems_risky(topic, question_text): raise ValueError("model produced a risky low-confidence place/person question") correct = safe_text(raw.get("correctAnswer")) if correct not in cleaned_options: correct = cleaned_options[0] return { "hostCommentary": safe_text(raw.get("hostCommentary"), f"{personality} is ready with a fresh question."), "question": question_text, "options": cleaned_options, "correctAnswer": correct, "source": source, "model": model, "providerMessage": provider_message, } except Exception as exc: debug_log(f"Question payload normalization fell back locally: {exc}") payload = fallback_question(topic, personality) payload["source"] = "fallback" payload["fallbackReason"] = safe_text(str(exc), "question generation failed") return payload def generate_feedback(question: str, correct_answer: str, user_answer: str, personality: str) -> dict[str, str]: try: raw = gemini_generate(build_answer_prompt(question, correct_answer, user_answer, personality)) except Exception as gemini_exc: debug_log(f"Gemini answer generation failed, trying OpenRouter: {gemini_exc}") try: raw = openrouter_generate(build_answer_prompt(question, correct_answer, user_answer, personality)) except Exception as openrouter_exc: debug_log(f"OpenRouter answer generation failed after Gemini: {openrouter_exc}") if safe_text(user_answer) == safe_text(correct_answer): return {"feedback": f"{personality} lights up. Correct answer. Clean hit."} return {"feedback": f"{personality} winces politely. Not quite. The right answer was {correct_answer}."} try: feedback = safe_text(raw.get("feedback")) if not feedback: raise ValueError("feedback missing") return {"feedback": feedback} except Exception as exc: debug_log(f"Answer payload normalization fell back locally: {exc}") if safe_text(user_answer) == safe_text(correct_answer): return {"feedback": f"{personality} lights up. Correct answer. Clean hit."} return {"feedback": f"{personality} winces politely. Not quite. The right answer was {correct_answer}."} def generate_speech(text: str, voice: str = "Puck") -> dict[str, Any]: api_key = load_api_key() if not api_key: raise RuntimeError("Gemini API key is not configured on the server") completed = subprocess.run( ["node", "/home/ovidiu/~devops-center/brainstorming/TriviAI/create-speech.mjs"], capture_output=True, text=True, timeout=TIMEOUT, env={ **os.environ, "GEMINI_API_KEY": api_key, "TRIVIAI_TTS_TEXT": text, "TRIVIAI_TTS_VOICE": voice, }, check=False, ) if completed.returncode != 0: stderr = safe_text(completed.stderr, "speech generation failed") raise RuntimeError(stderr) payload = json.loads(completed.stdout) if not safe_text(payload.get("audio")): raise RuntimeError("Speech payload did not include audio") return payload class TriviAIHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: if self.path == "/health": json_response(self, 200, {"ok": True, "service": "triviai"}) return json_response(self, 404, {"error": "Not found"}) def do_POST(self) -> None: if self.path == "/live-token": self.handle_live_token() return if self.path == "/question": self.handle_question() return if self.path == "/answer": self.handle_answer() return if self.path == "/speak": self.handle_speak() return json_response(self, 404, {"error": "Not found"}) def handle_question(self) -> None: try: payload = read_json_body(self) topic = safe_text(payload.get("topic"), "General knowledge") personality = safe_text(payload.get("personality"), "Classic Game Show Host") exclude_questions_raw = payload.get("excludeQuestions", []) exclude_questions = exclude_questions_raw if isinstance(exclude_questions_raw, list) else [] json_response(self, 200, generate_question(topic, personality, exclude_questions)) except Exception as exc: json_response(self, 500, {"error": safe_text(str(exc), "question generation failed")}) def handle_live_token(self) -> None: client_ip = self.client_address[0] if self.client_address else "unknown" last_issue = _token_rate_limit.get(client_ip, 0.0) now = time.time() if now - last_issue < TOKEN_WINDOW_SECONDS: json_response(self, 429, {"error": "Please wait a few seconds before starting another live voice session."}) return try: _token_rate_limit[client_ip] = now json_response(self, 200, issue_live_token()) except Exception as exc: debug_log(f"Live token request failed: {exc}") json_response(self, 500, {"error": safe_text(str(exc), "live token creation failed")}) def handle_answer(self) -> None: try: payload = read_json_body(self) question = safe_text(payload.get("question"), "Unknown question") correct_answer = safe_text(payload.get("correctAnswer"), "") user_answer = safe_text(payload.get("userAnswer"), "") personality = safe_text(payload.get("personality"), "Classic Game Show Host") json_response(self, 200, generate_feedback(question, correct_answer, user_answer, personality)) except Exception as exc: json_response(self, 500, {"error": safe_text(str(exc), "answer evaluation failed")}) def handle_speak(self) -> None: try: payload = read_json_body(self) text = safe_text(payload.get("text")) voice = safe_text(payload.get("voice"), "Puck") if not text: json_response(self, 400, {"error": "text is required"}) return json_response(self, 200, generate_speech(text, voice)) except Exception as exc: debug_log(f"Speech generation failed: {exc}") json_response(self, 500, {"error": safe_text(str(exc), "speech generation failed")}) def main() -> None: server = ThreadingHTTPServer((API_HOST, API_PORT), TriviAIHandler) debug_log(f"Starting service on {API_HOST}:{API_PORT}") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()