# ============================================================ # core/balancer.py — Titan Gateway V16.2 "Truly Free" # ============================================================ # CHANGELOG V16.2: # [AUDITOR] _audit_story_gemini() method DELETED entirely. # Step 2b removed from generate_immersion_story(). # Story flows directly: Narrator → Scholars. # Fixes AttributeError on prompts.AUDITOR_* attributes. # [PIPELINE] Version tag → "V16.2-TrulyFree" # [PRESERVED] All V16.0 / V15.0 logic unchanged: # • roll_historical_dice() — History Dice topic selection # • lang_focus tolerance — invalid input defaults to 'world' # • number_rule injection — all translation prompts # • _pick_category() — fallback when Dice fails # ============================================================ import asyncio import copy import datetime import io import json import logging import os import random import re import tempfile import time import uuid from typing import Any, Dict, List, Optional, Tuple import httpx import uvicorn from fastapi import FastAPI, File, HTTPException, Request, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from google import genai from groq import AsyncGroq, RateLimitError as GroqRateLimitError from pydub import AudioSegment # Import refactored modules from core import prompts from core import api_clients # ────────────────────────────────────────────────────────────────── # Logging # ────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)-8s | %(name)s — %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("titan.gateway") # ────────────────────────────────────────────────────────────────── # ANSI console helpers # ────────────────────────────────────────────────────────────────── _RED = "\033[91m\033[1m" _YELLOW = "\033[93m\033[1m" _CYAN = "\033[96m" _GREEN = "\033[92m\033[1m" _RESET = "\033[0m" def _critical(msg: str) -> None: print(f"{_RED}[CRITICAL] {msg}{_RESET}", flush=True) logger.critical(msg) def _warn(msg: str) -> None: print(f"{_YELLOW}[WARNING] {msg}{_RESET}", flush=True) logger.warning(msg) def _info(msg: str) -> None: print(f"{_CYAN}[INFO] {msg}{_RESET}", flush=True) logger.info(msg) def _ok(msg: str) -> None: print(f"{_GREEN}[OK] {msg}{_RESET}", flush=True) logger.info(msg) # ────────────────────────────────────────────────────────────────── # API Key pools — 8 Groq + 5 Gemini (total 13) # ────────────────────────────────────────────────────────────────── _GROQ_ENV_NAMES: List[str] = [f"GROQ_KEY_{i}" for i in range(1, 9)] _GEMINI_ENV_NAMES: List[str] = [f"GOOGLE_KEY_{i}" for i in range(1, 6)] GROQ_API_KEYS: List[str] = [k for k in (os.environ.get(n, "") for n in _GROQ_ENV_NAMES) if k] GEMINI_API_KEYS: List[str] = [k for k in (os.environ.get(n, "") for n in _GEMINI_ENV_NAMES) if k] GROQ_KEY_NAMES: Dict[str, str] = {os.environ.get(n, ""): n for n in _GROQ_ENV_NAMES if os.environ.get(n)} GEMINI_KEY_NAMES: Dict[str, str] = {os.environ.get(n, ""): n for n in _GEMINI_ENV_NAMES if os.environ.get(n)} GEMMA_MODEL_ID: str = os.environ.get("GEMINI_MODEL", "gemma-3-27b-it") # ────────────────────────────────────────────────────────────────── # V14.0 — Hugging Face Infrastructure # ────────────────────────────────────────────────────────────────── HF_TOKEN: str = os.environ.get("HF_TOKEN", "").strip() # HF text router removed (Qwen/novita retired) — Narrator now runs on Groq GPT OSS 120B # AudioLDM 2 ambient sound generation HF_AUDIOLDM2_URL: str = "https://api-inference.huggingface.co/models/cvssp/audioldm2" HF_AUDIO_DURATION: float = 10.0 # seconds of generated audio HF_AUDIO_GUIDANCE: float = 3.5 # guidance_scale for AudioLDM 2 HF_AUDIO_RETRIES: int = 3 HF_AUDIO_TIMEOUT: float = 90.0 # seconds — model cold-start can be slow # HF sequential lock — ONE token, all HF calls must be strictly serialised # Instantiated in SmartKeyManager.__init__ _HF_SEQUENTIAL_LOCK: Optional[asyncio.Lock] = None # ────────────────────────────────────────────────────────────────── # V10.1 — Key Role Ranges (0-indexed slices into the key lists) # ────────────────────────────────────────────────────────────────── # # GROQ (generation-only — Groq NEVER touches non-English output): # generation (0,8) — ALL 8 keys for English story writing # "translation" role DELETED from Groq: LLaMA hallucinates on # Arabic/German/Ukrainian, producing Chinese characters and # mixed-script garbage that corrupts the story JSON. # # GEMINI (translation + enrichment + validation): # translation (0,5) — ALL 5 keys for AR/ES/DE/UK story translation # enrichment (0,3) — GOOGLE_KEY_1-3 for Scholar / Historian # validation (3,5) — GOOGLE_KEY_4-5 for Architect + pre-flight # # Automatic fallback to the full provider pool when role keys are # all on cooldown is preserved in _acquire_groq_role / # _acquire_gemini_role. # ────────────────────────────────────────────────────────────────── _GROQ_ROLE_RANGES: Dict[str, Tuple[int, int]] = { "maestro": (0, 7), # ALL 7 keys — dedicated to IMMERSION_MAESTRO_MODEL (Narrator) "generation": (0, 8), # ALL Groq keys — English writing (architect helper, etc.) "any": (0, 8), # full pool fallback (also Whisper transcription) # "translation" intentionally absent: Groq banned from translation } _GEMINI_ROLE_RANGES: Dict[str, Tuple[int, int]] = { "translation": (0, 5), # ALL Gemini keys — AR/ES/DE/UK translation "enrichment": (0, 3), # GOOGLE_KEY_1-3 — Scholar / Historian "validation": (3, 5), # GOOGLE_KEY_4-5 — Architect / pre-flight "any": (0, 5), # full pool fallback } # ────────────────────────────────────────────────────────────────── # Runtime constants # ────────────────────────────────────────────────────────────────── RETRY_BACKOFF_SECONDS: float = 1.5 MAX_RETRIES: int = 4 RATE_LIMIT_COOLDOWN_S: float = 65.0 # Marks a key on a brief cooldown after a server-side 500 error # (not the same as rate-limiting, recovers fast) SERVER_ERROR_COOLDOWN_S: float = 20.0 CHUNK_DURATION_MS: int = 5 * 60 * 1000 CHUNK_EXPORT_BITRATE: str = "64k" CHUNK_EXPORT_FORMAT: str = "mp3" TRANSLATION_BATCH_SIZE: int = 55 TRANSLATION_TEMPERATURE: float = 0.10 IMMERSION_AUTHOR_MODEL: str = "llama-3.3-70b-versatile" # Director + Dice — json_object support IMMERSION_TRANS_GROQ_MODEL: str = "llama-3.3-70b-versatile" # V18.0 — Narrator Model: GPT OSS 120B (Reasoning category on Groq) # Replaces Llama 3.3 70B. Larger reasoning capacity handles the Iron Standard # V18.0 prompt (11 rules, Anchor Test, Primary Source Test, Last-Line Contract) # without losing rule discipline mid-generation. # Keys 0-1 (GROQ_KEY_1, GROQ_KEY_2) reserved exclusively for this model. IMMERSION_MAESTRO_MODEL: str = "llama-3.3-70b-versatile" # Narrator — chat model, better long-form prose # Recursive translation chunking thresholds RECURSIVE_TRANSLATE_THRESHOLD: int = 1_200 # words — above this, chunk the chapter text RECURSIVE_TRANSLATE_CHUNK_WORDS: int = 600 # target words per sub-chunk # Scholar per-chapter max input length (characters — truncate to prevent overflow) SCHOLAR_CHAPTER_MAX_CHARS: int = 4_000 # ~900-1000 words # ── 12 story categories ────────────────────────────────────────── STORY_CATEGORIES: List[Tuple[int, str, str]] = [ ( 0, "Wars & Battles", "Choose a famous battle or military campaign — the strategy, the turning point, the human cost."), ( 1, "Scientific Discoveries", "Choose a groundbreaking scientific discovery — the Eureka moment, the struggle against skepticism."), ( 2, "Ancient Civilizations", "Choose a specific moment from an ancient civilization (Egypt, Rome, Greece, Mesopotamia, Maya, China)."), ( 3, "Trade Routes & Exploration", "Choose a famous voyage of discovery or legendary trade route — the Silk Road, Age of Exploration."), ( 4, "Art, Architecture & Culture", "Choose the creation of a famous artwork or monument — Sistine Chapel, Alhambra, Great Wall."), ( 5, "Revolutions & Independence", "Choose a political revolution (NOT modern social movements). French/American Revolution."), ( 6, "Natural Disasters & Survival", "Choose a famous historical natural disaster — Pompeii, the Black Death, Great Fire of London."), ( 7, "Mystery, Mythology & Legend", "Choose a historical mystery or legendary event — Library of Alexandria, lost cities, Trojan War."), ( 8, "Maritime & Naval History", "Choose a famous naval battle or legendary sea voyage — Spanish Armada, Viking raids."), ( 9, "Invention & Technology", "Choose the invention of a world-changing technology — printing press, steam engine, gunpowder."), (10, "Political Intrigue & Espionage", "Choose a famous conspiracy or assassination — Julius Caesar, Gunpowder Plot, Cold War Berlin."), (11, "Religion, Philosophy & Ideas", "Choose a philosophical movement or religious founding moment that shaped civilizations."), ] SLOT_TO_OFFSET: Dict[int, int] = {0: 0, 6: 1, 12: 2, 18: 3} def _pick_category(hour_slot: int, day_of_year: int) -> Tuple[int, str, str]: """ Fallback deterministic category picker. Used when roll_historical_dice() fails (e.g. no API key available yet). """ offset = SLOT_TO_OFFSET.get(hour_slot, 0) idx = (day_of_year * 4 + offset) % len(STORY_CATEGORIES) cat = STORY_CATEGORIES[idx] return cat[0], cat[1], cat[2] async def roll_historical_dice(api_key: str, today_month_day: str) -> Optional[Dict[str, str]]: """ V16.0 — History Dice. Asks the Maestro LLM to list 5 real historical events that occurred on `today_month_day`, then picks one at random via random.choice(). Returns a dict {"event": "...", "year": "...", "category": "..."} on success, or None if the call fails (caller falls back to _pick_category). Design notes: • Uses call_groq_json so Groq is the sole provider (no Gemini spend). • Temperature 0.3 keeps answers grounded while allowing variety. • Single attempt only — failure is non-fatal; the pipeline continues. """ prompt = prompts.DICE_PROMPT_TEMPLATE.format(today_month_day=today_month_day) try: data = await api_clients.call_groq_json( api_key=api_key, prompt=prompt, caller="HistoryDice", max_tokens=800, temperature=0.3, ) events = data.get("events", []) if not events: _warn("[HistoryDice] LLM returned empty events list — using fallback category") return None # V18.1 — Quality filter: reject events without a verified year verified_events = [ e for e in events if e.get("year", "").strip() and e.get("year", "").strip().lower() not in ("unknown", "n/a", "circa", "") ] if verified_events: events = verified_events chosen = random.choice(events) _ok( f"[HistoryDice] Rolled: '{chosen.get('event','?')}' " f"({chosen.get('year','?')}) — category: {chosen.get('category','?')}" ) return chosen except Exception as exc: _warn(f"[HistoryDice] Failed: {exc} — using fallback category") return None # ────────────────────────────────────────────────────────────────── # V10.0 — Text Contamination Detection # ────────────────────────────────────────────────────────────────── def _text_is_contaminated(text: str) -> bool: """Return True if the text contains technical artifacts that must not enter the story.""" return api_clients.text_is_contaminated(text) # ══════════════════════════════════════════════════════════════════ # SmartKeyManager — V10.0 (refactored to use api_clients) # ══════════════════════════════════════════════════════════════════ class SmartKeyManager: """ Manages 8 Groq + 5 Gemini API keys. V10.1 Role assignment — Groq is English generation ONLY: Groq/generation → ALL 8 Groq keys — English story writing Groq/transcription→ Groq (Whisper) — audio, language-agnostic Gemini/translation → ALL 5 Gemini keys — AR/ES/DE/UK translation Gemini/enrichment → GOOGLE_KEY_1-3 — Scholar / Historian Gemini/validation → GOOGLE_KEY_4-5 — Architect / pre-flight Groq NEVER produces non-English text: LLaMA hallucinates Chinese characters and mixed-script output during Arabic translation. All multilingual story translation uses Gemini exclusively. Features (unchanged from V10.0): • Per-key cooldown: rate limits + server errors tracked separately • Role-aware acquisition with automatic fallback to the full pool • ANSI RED [CRITICAL] alerts naming the failing key and backup • Never crashes the pipeline """ def __init__(self) -> None: global _HF_SEQUENTIAL_LOCK self._groq_keys: List[str] = list(GROQ_API_KEYS) self._gemini_keys: List[str] = list(GEMINI_API_KEYS) self._groq_cooldown: Dict[str, float] = {} self._gemini_cooldown: Dict[str, float] = {} self._groq_cursor: int = 0 self._gemini_cursor: int = 0 self._groq_lock = asyncio.Lock() self._gemini_lock = asyncio.Lock() # V14.0 — HF sequential lock: only one HF request at a time # (one token → no parallel HF calls allowed) _HF_SEQUENTIAL_LOCK = asyncio.Lock() self._hf_lock = _HF_SEQUENTIAL_LOCK # V18.0 — HF text client removed (Qwen/novita retired). # Narrator runs on Llama 3.3 70B — GPT OSS 120B is a reasoning model (content=None). # HF_TOKEN is still required by immersion.py for image generation (FLUX/Pollinations). self._hf_text_client = None if not HF_TOKEN: _info("HF_TOKEN not set — image generation (immersion.py) will use Pollinations fallback") _info( f"SmartKeyManager V18.0 ready — " f"{len(self._groq_keys)} Groq keys " f"(Narrator: {IMMERSION_MAESTRO_MODEL}) | " f"(Director+Dice: {IMMERSION_AUTHOR_MODEL}) | " f"{len(self._gemini_keys)} Gemini keys " f"(model: {GEMMA_MODEL_ID} — translation/enrichment/validation) | " f"HF AudioLDM2: sequential lock active" ) # ────────────────────────────────────────────────────────────── # Core key acquisition (unchanged from V9.2.3) # ────────────────────────────────────────────────────────────── async def _acquire_groq(self, caller: str = "") -> str: while True: async with self._groq_lock: now = time.time() available = [k for k in self._groq_keys if now >= self._groq_cooldown.get(k, 0.0)] if available: key = available[self._groq_cursor % len(available)] self._groq_cursor += 1 return key min_wait = min( max(0.1, self._groq_cooldown.get(k, 0.0) - now) for k in self._groq_keys ) _warn(f"[{caller}] ALL {len(self._groq_keys)} Groq keys on cooldown. " f"Waiting {min_wait:.1f}s…") await asyncio.sleep(min_wait + 0.2) async def _acquire_gemini(self, caller: str = "") -> str: while True: async with self._gemini_lock: now = time.time() available = [k for k in self._gemini_keys if now >= self._gemini_cooldown.get(k, 0.0)] if available: key = available[self._gemini_cursor % len(available)] self._gemini_cursor += 1 return key min_wait = min( max(0.1, self._gemini_cooldown.get(k, 0.0) - now) for k in self._gemini_keys ) _warn(f"[{caller}] ALL {len(self._gemini_keys)} Gemini keys on cooldown. " f"Waiting {min_wait:.1f}s…") await asyncio.sleep(min_wait + 0.2) # ────────────────────────────────────────────────────────────── # V10.0 — Role-aware key acquisition # ────────────────────────────────────────────────────────────── async def _acquire_groq_role(self, role: str = "any", caller: str = "") -> str: """ Acquire a Groq key from the specified role pool. Falls back to the full Groq pool if all role-specific keys are cooling down. """ lo, hi = _GROQ_ROLE_RANGES.get(role, (0, len(self._groq_keys))) lo = min(lo, len(self._groq_keys)) hi = min(hi, len(self._groq_keys)) role_keys = self._groq_keys[lo:hi] if role_keys: async with self._groq_lock: now = time.time() available = [k for k in role_keys if now >= self._groq_cooldown.get(k, 0.0)] if available: key = available[self._groq_cursor % len(available)] self._groq_cursor += 1 return key _warn( f"[{caller}] Role '{role}' Groq keys all on cooldown — " f"falling back to full pool" ) return await self._acquire_groq(caller) async def _acquire_gemini_role(self, role: str = "any", caller: str = "") -> str: """ Acquire a Gemini key from the specified role pool. Falls back to the full Gemini pool if all role-specific keys are cooling down. """ lo, hi = _GEMINI_ROLE_RANGES.get(role, (0, len(self._gemini_keys))) lo = min(lo, len(self._gemini_keys)) hi = min(hi, len(self._gemini_keys)) role_keys = self._gemini_keys[lo:hi] if role_keys: async with self._gemini_lock: now = time.time() available = [k for k in role_keys if now >= self._gemini_cooldown.get(k, 0.0)] if available: key = available[self._gemini_cursor % len(available)] self._gemini_cursor += 1 return key _warn( f"[{caller}] Role '{role}' Gemini keys all on cooldown — " f"falling back to full pool" ) return await self._acquire_gemini(caller) # ── Rate-limit marking ───────────────────────────────────────── async def _mark_groq_rl(self, key: str, caller: str = "", cooldown: float = RATE_LIMIT_COOLDOWN_S) -> None: async with self._groq_lock: self._groq_cooldown[key] = time.time() + cooldown key_name = GROQ_KEY_NAMES.get(key, f"GROQ_KEY[…{key[-4:]}]") now = time.time() backup = next( (GROQ_KEY_NAMES.get(k, f"…{k[-4:]}") for k in self._groq_keys if k != key and now >= self._groq_cooldown.get(k, 0.0)), "ALL_ON_COOLDOWN", ) _critical( f"{key_name} failed (RateLimit) [{caller}]. " f"Cooldown {cooldown:.0f}s. Switching to backup → {backup}" ) async def _mark_gemini_rl(self, key: str, caller: str = "", cooldown: float = RATE_LIMIT_COOLDOWN_S) -> None: async with self._gemini_lock: self._gemini_cooldown[key] = time.time() + cooldown key_name = GEMINI_KEY_NAMES.get(key, f"GEMINI_KEY[…{key[-4:]}]") now = time.time() backup = next( (GEMINI_KEY_NAMES.get(k, f"…{k[-4:]}") for k in self._gemini_keys if k != key and now >= self._gemini_cooldown.get(k, 0.0)), "ALL_ON_COOLDOWN", ) _critical( f"{key_name} failed (RateLimit) [{caller}]. " f"Cooldown {cooldown:.0f}s. Switching to backup → {backup}" ) # V10.0 — Brief server-error cooldown (not a rate limit — recovers fast) async def _mark_groq_server_error(self, key: str, caller: str = "") -> None: async with self._groq_lock: self._groq_cooldown[key] = time.time() + SERVER_ERROR_COOLDOWN_S key_name = GROQ_KEY_NAMES.get(key, f"GROQ_KEY[…{key[-4:]}]") _warn( f"{key_name} returned server error [{caller}]. " f"Brief cooldown {SERVER_ERROR_COOLDOWN_S:.0f}s — switching key." ) async def _mark_gemini_server_error(self, key: str, caller: str = "") -> None: async with self._gemini_lock: self._gemini_cooldown[key] = time.time() + SERVER_ERROR_COOLDOWN_S key_name = GEMINI_KEY_NAMES.get(key, f"GEMINI_KEY[…{key[-4:]}]") _warn( f"{key_name} returned server error [{caller}]. " f"Brief cooldown {SERVER_ERROR_COOLDOWN_S:.0f}s — switching key." ) # ── JSON fence cleaner ───────────────────────────────────────── @staticmethod def _clean_json(raw: str) -> str: return api_clients.clean_json(raw) # ══════════════════════════════════════════════════════════════ # V14.0 — HF AudioLDM 2 (sequential, one token) # ══════════════════════════════════════════════════════════════ async def _generate_audio_ldm2( self, audio_prompt: str, out_path: str, ) -> bool: """ Generate a 10-second ambient audio clip via HF AudioLDM 2. Strictly serialised: acquires self._hf_lock before every call. """ if not HF_TOKEN: _warn("[AudioLDM2] HF_TOKEN not set — skipping audio generation") return False # We'll use a low-level single-attempt function inside the lock async with self._hf_lock: # Single attempt, but we still retry if needed? The low-level # function should be a single attempt; we handle retries here. for attempt in range(1, HF_AUDIO_RETRIES + 1): try: success = await api_clients.generate_audio_ldm2_single( hf_token=HF_TOKEN, audio_prompt=audio_prompt, out_path=out_path, timeout=HF_AUDIO_TIMEOUT ) if success: return True except Exception as exc: logger.error(f"[AudioLDM2] attempt {attempt} failed: {exc}") await asyncio.sleep(10.0 * attempt) # backoff logger.error("[AudioLDM2] ❌ Failed after %d attempts", HF_AUDIO_RETRIES) return False # ══════════════════════════════════════════════════════════════ # CINEMA PIPELINE — Transcription & subtitle translation # (Preserved from V7.1 — unchanged, but using api_clients) # ══════════════════════════════════════════════════════════════ async def _dispatch_transcription( self, audio_bytes: bytes, filename: str, language: Optional[str] = None, prompt: Optional[str] = None, ) -> Dict[str, Any]: last_error = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_groq("Cinema/Transcribe") try: return await api_clients.dispatch_transcription( api_key=key, audio_bytes=audio_bytes, filename=filename, language=language, prompt=prompt, ) except GroqRateLimitError: await self._mark_groq_rl(key, "Cinema/Transcribe") last_error = "RateLimit" except Exception as exc: logger.error("Transcribe error attempt %d: %s", attempt, exc) last_error = exc await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) raise Exception(f"All transcription retries exhausted: {last_error}") def _validate_and_clean_segments( self, segments: List[Dict[str, Any]] ) -> List[Dict[str, Any]]: # This method is unchanged; it doesn't call external APIs. if not segments: return [] segs = sorted(segments, key=lambda s: float(s.get("start", 0))) cleaned: List[Dict[str, Any]] = [] seen: Dict[str, float] = {} for seg in segs: text = seg.get("text", "").strip() if not text: continue start = float(seg.get("start", 0)) end = float(seg.get("end", 0)) if start > end: start, end = end, start if end - start < 0.08: end = start + 0.5 if end - start > 25.0: end = start + 25.0 key_txt = re.sub(r"\s+", " ", text.lower()) prev = seen.get(key_txt) if prev is not None and (start - prev) < 4.0: continue seen[key_txt] = start cleaned.append({**seg, "start": round(start, 3), "end": round(end, 3), "text": text}) for i in range(1, len(cleaned)): if cleaned[i]["start"] < cleaned[i - 1]["end"]: ns = round(cleaned[i - 1]["end"] + 0.05, 3) cleaned[i] = {**cleaned[i], "start": ns} if cleaned[i]["start"] >= cleaned[i]["end"]: cleaned[i] = {**cleaned[i], "end": round(cleaned[i]["start"] + 0.5, 3)} return cleaned async def transcribe_audio( self, audio_bytes: bytes, filename: str = "audio.webm", language: Optional[str] = None, ) -> Dict[str, Any]: # This method is unchanged (it uses _dispatch_transcription) ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "webm" temp_paths: List[str] = [] try: with tempfile.NamedTemporaryFile(delete=False, suffix=f".{ext}") as tf: tf.write(audio_bytes) src = tf.name temp_paths.append(src) del audio_bytes try: audio = await asyncio.to_thread(AudioSegment.from_file, src, format=ext) except Exception as exc: raise HTTPException(400, f"Bad audio '{ext}': {exc}") total_ms = len(audio) total_chunks = max(1, -(-total_ms // CHUNK_DURATION_MS)) chunk_meta: List[Dict] = [] for i in range(total_chunks): s_ms, e_ms = i * CHUNK_DURATION_MS, min((i + 1) * CHUNK_DURATION_MS, total_ms) chunk_seg = audio[s_ms:e_ms] with tempfile.NamedTemporaryFile( delete=False, suffix=f".{CHUNK_EXPORT_FORMAT}", prefix=f"titan_chunk{i}_" ) as cf: cpath = cf.name temp_paths.append(cpath) await asyncio.to_thread( chunk_seg.export, cpath, format=CHUNK_EXPORT_FORMAT, bitrate=CHUNK_EXPORT_BITRATE, ) chunk_meta.append({"path": cpath, "offset_ms": s_ms}) merged_text = "" merged_segs: List[Dict] = [] merged_lang = "unknown" g_id = 0 rolling_prompt = "" for meta in chunk_meta: result = await self._dispatch_transcription( open(meta["path"], "rb").read(), filename=f"chunk_{meta['offset_ms']}.{CHUNK_EXPORT_FORMAT}", language=language, prompt=rolling_prompt or None, ) merged_text += (" " if merged_text else "") + result.get("text", "") merged_lang = result.get("language", merged_lang) rolling_prompt = merged_text[-200:] offset = meta["offset_ms"] / 1000.0 for seg in result.get("segments", []): adj = {**seg, "start": round(seg.get("start", 0.0) + offset, 3), "end": round(seg.get("end", 0.0) + offset, 3), "id": str(g_id)} g_id += 1 merged_segs.append(adj) clean_segs = self._validate_and_clean_segments(merged_segs) for new_id, seg in enumerate(clean_segs): seg["id"] = str(new_id) return {"text": merged_text, "segments": clean_segs, "language": merged_lang} finally: for p in temp_paths: try: if os.path.exists(p): os.remove(p) except OSError: pass async def translate_segments( self, segments: List[Dict[str, Any]], target_language: str = "Arabic", ) -> Dict[str, str]: if not segments: return {} batches = [segments[i: i + TRANSLATION_BATCH_SIZE] for i in range(0, len(segments), TRANSLATION_BATCH_SIZE)] all_results: Dict[str, str] = {} context_hint: str = "" for b_idx, batch in enumerate(batches): numbered = "\n".join(f"[{seg['id']}] {seg['text']}" for seg in batch) prompt = prompts.build_subtitle_translation_prompt( numbered, target_language, context_hint, b_idx + 1, len(batches) ) batch_result: Dict[str, str] = {} for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_gemini("Cinema/Translate") try: parsed = await api_clients.call_gemini_translation_batch(key, prompt) missing = {seg["id"] for seg in batch} - parsed.keys() if missing: orig = {seg["id"]: seg["text"] for seg in batch} for mid in missing: parsed[mid] = orig[mid] batch_result = parsed break except Exception as exc: es = str(exc).lower() if "429" in str(exc) or "rate" in es or "quota" in es: await self._mark_gemini_rl(key, "Cinema/Translate") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) if not batch_result: batch_result = {seg["id"]: seg["text"] for seg in batch} all_results.update(batch_result) context_hint = "\n".join(f"[{k}] → {v}" for k, v in list(batch_result.items())[-3:]) return all_results # ══════════════════════════════════════════════════════════════ # Generic JSON callers — V10.0 (role-aware + server-error handling) # ══════════════════════════════════════════════════════════════ async def _call_groq_json( self, prompt: str, caller: str, max_tokens: int = 4000, temperature: float = 0.4, role: str = "any", model: str = IMMERSION_MAESTRO_MODEL, # V18.0: GPT OSS 120B default ) -> Dict[str, Any]: """ Call Groq with JSON-mode response. Uses api_clients.call_groq_json for the actual HTTP call. V18.2: defaults to IMMERSION_MAESTRO_MODEL (openai/gpt-oss-120b). Director uses IMMERSION_AUTHOR_MODEL (llama-3.3-70b-versatile) via explicit override. """ last_error = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_groq_role(role, caller) try: _info(f"[{caller}] Groq attempt {attempt}/{MAX_RETRIES} | role={role} | model={model} | key=…{key[-4:]}") data = await api_clients.call_groq_json( api_key=key, prompt=prompt, caller=caller, max_tokens=max_tokens, temperature=temperature, model=model, ) return data except GroqRateLimitError: await self._mark_groq_rl(key, caller) last_error = "RateLimit" except Exception as exc: exc_str = str(exc) # V10.0: Detect HTTP 500 / server errors → brief cooldown, retry if "500" in exc_str or "502" in exc_str or "503" in exc_str or "server_error" in exc_str.lower(): await self._mark_groq_server_error(key, caller) last_error = f"ServerError({exc_str[:80]})" else: logger.error(f"[{caller}] Error attempt={attempt}: {exc}") last_error = exc await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) raise HTTPException(502, f"{caller} failed after {MAX_RETRIES} attempts: {last_error}") async def _call_gemini_json( self, prompt: str, caller: str = "", max_tokens: int = 4000, temperature: float = 0.25, role: str = "any", ) -> Dict[str, Any]: """ Call Gemini with JSON-expected response. Uses api_clients.call_gemini_json for the actual HTTP call. """ last_error = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_gemini_role(role, caller) try: _info(f"[{caller}] Gemini attempt {attempt}/{MAX_RETRIES} | role={role} | key=…{key[-4:]}") data = await api_clients.call_gemini_json( api_key=key, prompt=prompt, caller=caller, max_tokens=max_tokens, temperature=temperature, ) return data except Exception as exc: exc_str = str(exc) es = exc_str.lower() if "429" in exc_str or "rate" in es or "quota" in es: await self._mark_gemini_rl(key, caller) elif "500" in exc_str or "502" in exc_str or "503" in exc_str or "server_error" in es: await self._mark_gemini_server_error(key, caller) last_error = f"ServerError({exc_str[:80]})" else: logger.error("[%s] Gemini error attempt=%d: %s", caller, attempt, exc) await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) last_error = exc raise RuntimeError(f"[{caller}] All {MAX_RETRIES} Gemini retries exhausted: {last_error}") # ══════════════════════════════════════════════════════════════ # STEP 1 – Architect (Gemini/validation keys) – Screenplay Outline # ══════════════════════════════════════════════════════════════ async def _generate_screenplay_outline( self, category: str, cat_instruction: str, lang_focus: str, date_anchor: str, today_month_day: str, ) -> Dict[str, Any]: focus_map = { "ar": "Focus on events from Arab, Islamic, Berber, or broader Middle Eastern history.", # القوزاق = Cossacks (V15.0 terminology standard) "uk": "Focus on events from Ukrainian, Slavic, القوزاق (Cossack), or Eastern European history.", "world": "Choose from any civilization on Earth — be global, surprising, and varied.", } focus_instruction = focus_map.get(lang_focus, focus_map["world"]) prompt = prompts.ARCHITECT_PROMPT_TEMPLATE.format( today_month_day=today_month_day, category=category, cat_instruction=cat_instruction, focus_instruction=focus_instruction, ) # V18.5: Architect on GPT OSS 120B (Groq/maestro) — # stronger reasoning = better JSON schema + fewer hallucinated links. # V18.6: max_tokens raised 8000 → 16000 — gpt-oss-120b is a reasoning # model that consumes tokens on internal chain-of-thought before output. # The large Architect schema (characters + 3 chapters × 10 action_points) # was hitting the 8k limit mid-JSON, producing malformed output and # triggering json_validate_failed (HTTP 400) on the first 1-2 attempts. return await self._call_groq_json( prompt, "Architect", max_tokens=16000, temperature=0.5, role="maestro", model="openai/gpt-oss-120b" ) async def _write_chapter_novelist( self, outline: dict, chapter_num: int, lang_focus: str = "", character_descs: str = "", continuity_tail: str = "", **kwargs ) -> str: """ V15.0 Documentary Narrator — Powered by Groq Direct. Changes from V14.1: • Passes act_title from the outline into the user prompt so the Narrator knows whether it is writing Act I (Setup/Shock), Act II (Technical Detail), or Act III (Verified Aftermath/Conclusion). • Chapter 3 receives an explicit definitive-ending instruction. • Uses role-aware key acquisition (_acquire_groq_role "maestro") so long chapter writes are routed to GROQ_KEY_1-2 first. • 4 retry attempts with exponential back-off; 429 triggers per-key cooldown via _mark_groq_rl. • frequency_penalty and presence_penalty are enforced inside api_clients.call_groq_chat_completion (0.9 / 0.7). """ ch_outline = next( (ch for ch in outline.get("chapters", []) if ch.get("chapter_number") == chapter_num), None ) if not ch_outline: raise ValueError(f"Chapter {chapter_num} not found in outline") # ── Build character block ───────────────────────────────── characters = outline.get("characters", []) chars_block = "\n".join( f"{c.get('name', 'Unknown')}: {c.get('description', '')}" for c in characters ) # ── Build action points block ───────────────────────────── action_points = ch_outline.get("action_points", []) ap_text = "\n".join(f" {ap}" for ap in action_points) if action_points else "(none)" # ── Build continuity block ──────────────────────────────── continuity_block = "" if continuity_tail.strip(): continuity_block = prompts.CONTINUITY_BLOCK_TEMPLATE.format( prev_chapter_num=chapter_num - 1, continuity_tail=continuity_tail, ) # ── Act title (V15.0 — from outline, falls back gracefully) ── act_title = ch_outline.get( "act_title", {1: "Act I — The Setup / The Shock", 2: "Act II — Technical & Military Detail", 3: "Act III — The Verified Aftermath"}.get(chapter_num, f"Chapter {chapter_num}") ) # ── Chapter 3: definitive ending instruction ────────────── ch3_close = prompts.CHAPTER3_CLOSE_TEMPLATE if chapter_num == 3 else "" # ── Word count target ────────────────────────────────────── # Ch3 is tighter (650-800) to enforce a clean conclusion. word_count = ( "650-800 words — write the COMPLETE conclusion. " "Count your words. If below 650 — expand each action point with one additional " "documented detail. Do not stop early. No open threads. No truncation." if chapter_num == 3 else "Write EXACTLY 800 to 950 words. Count your words before finishing. " "If below 800 — expand each action point with one additional documented detail from the era. " "Do not pad. Do not repeat. Expand with verified facts only." ) system_prompt = prompts.NOVELIST_SYSTEM_PROMPT user_prompt = prompts.NOVELIST_USER_PROMPT_TEMPLATE.format( title=outline.get("title_en", "Untitled"), era=outline.get("historical_era", ""), characters_block=chars_block, chapter_num=chapter_num, act_title=act_title, summary=ch_outline.get("summary", ""), location=ch_outline.get("location", ""), sensory_cue=ch_outline.get("sensory_cue", ""), continuity_block=continuity_block, action_points_text=ap_text, word_count=word_count, chapter3_close=ch3_close, ) # V18.5: Llama 3.3 70B chat model — no reasoning phase. # 4000 tokens = ~3000 words, more than enough for 800-950 target. max_tokens_map = {1: 4000, 2: 4000, 3: 4000} max_tok = max_tokens_map.get(chapter_num, 3000) last_error: Optional[Exception] = None # 4-attempt loop with role-aware key rotation and per-key cooldown for attempt in range(1, MAX_RETRIES + 1): # Use "maestro" role (keys 0-1) first; falls back to full pool automatically api_key = await self._acquire_groq_role("maestro", f"Novelist/Ch{chapter_num}") try: _info( f"[Novelist/Ch{chapter_num}] attempt {attempt}/{MAX_RETRIES} | " f"act='{act_title}' | key=…{api_key[-4:]}" ) # frequency_penalty=0.9, presence_penalty=0.7 enforced inside api_clients content = await api_clients.call_groq_chat_completion( api_key=api_key, system_prompt=system_prompt, user_prompt=user_prompt, model=IMMERSION_MAESTRO_MODEL, max_tokens=max_tok, temperature=0.5, ) if not content or len(content.split()) < 100: raise ValueError( f"Response too short ({len(content.split()) if content else 0} words)" ) # V18.5 — Fix broken number format: ( ,000) → remove artifact import re as _re content = _re.sub(r'\(\ *,\ *\d*\)', '', content) content = content.strip() # V18.1 — Ch3 minimum word guard: retry if under 400 words if chapter_num == 3 and len(content.split()) < 400: raise ValueError( f"Ch3 too short ({len(content.split())} words — minimum 400) — forcing retry" ) _ok( f"[Novelist/Ch{chapter_num}] ✅ ~{len(content.split())} words | " f"act='{act_title}'" ) return content.strip() except GroqRateLimitError: await self._mark_groq_rl(api_key, f"Novelist/Ch{chapter_num}") last_error = Exception("RateLimit") # No extra sleep — _acquire_groq_role will wait on the next attempt except Exception as exc: exc_str = str(exc) logger.error( "[Novelist/Ch%d] attempt %d failed: %s", chapter_num, attempt, exc_str ) last_error = exc if "429" in exc_str or "rate" in exc_str.lower(): await self._mark_groq_rl(api_key, f"Novelist/Ch{chapter_num}") elif "500" in exc_str or "502" in exc_str or "503" in exc_str: await self._mark_groq_server_error(api_key, f"Novelist/Ch{chapter_num}") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * (2 ** (attempt - 1))) raise HTTPException( 502, f"[Novelist] Chapter {chapter_num} ({act_title}) failed after " f"{MAX_RETRIES} attempts: {last_error}" ) async def _extract_chapter_metadata_director( self, chapter_num: int, chapter_text: str, ch_outline: Dict[str, Any], ) -> Dict[str, Any]: """ Agent 3 — The Director. Analyzes a finished chapter text using Groq (fast, cheap, no HF token). Extracts: - cinematic_image_prompt: photorealistic still description - audio_environment: {primary_ambience, intensity, dynamic_cues, audio_prompt} where audio_prompt is a rich descriptive string for AudioLDM 2. Returns a dict that is merged into the chapter object. """ ambiance = ch_outline.get("ambiance", "battlefield") excerpt = chapter_text[:2000] # Director only needs to read the chapter, not the whole thing prompt = prompts.DIRECTOR_PROMPT_TEMPLATE.format( chapter_num=chapter_num, excerpt=excerpt, ambiance=ambiance, ) last_error: Optional[Exception] = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_groq_role("generation", f"Director/Ch{chapter_num}") try: _info(f"[Director/Ch{chapter_num}] attempt {attempt} | key=…{key[-4:]}") data = await api_clients.call_groq_json( api_key=key, prompt=prompt, caller=f"Director/Ch{chapter_num}", max_tokens=700, temperature=0.4, model="llama-3.3-70b-versatile", # Director: Llama 70B — supports json_object ) _ok(f"[Director/Ch{chapter_num}] metadata extracted") return data except GroqRateLimitError: await self._mark_groq_rl(key, f"Director/Ch{chapter_num}") last_error = Exception("RateLimit") except Exception as exc: exc_str = str(exc) if "500" in exc_str or "502" in exc_str or "503" in exc_str: await self._mark_groq_server_error(key, f"Director/Ch{chapter_num}") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) last_error = exc _warn(f"[Director/Ch{chapter_num}] All attempts failed — using outline defaults") return { "cinematic_image_prompt": ch_outline.get("summary", "")[:120], "audio_environment": { "primary_ambience": ch_outline.get("ambiance", "battlefield"), "intensity": "medium", "dynamic_cues": ch_outline.get("sensory_cue", ""), "audio_prompt": ch_outline.get("sensory_cue", "ambient historical sounds"), }, } async def _write_chapter( self, chapter_num: int, outline: Dict[str, Any], previous_chapters: List[Dict[str, Any]], character_descs: List[Dict[str, str]], ) -> Dict[str, Any]: """ Orchestrates The Narrator + The Director for one chapter. Returns a chapter dict compatible with the V15.0 schema. 3-Act enforcement (V15.0): Ch1 — Act I (The Setup / The Shock) Ch2 — Act II (Technical & Military Detail) Ch3 — Act III (The Verified Aftermath — DEFINITIVE CONCLUSION) """ ch_outline = next( (ch for ch in outline["chapters"] if ch["chapter_number"] == chapter_num), None ) if not ch_outline: raise ValueError(f"Chapter {chapter_num} not found in outline") # Act label — from outline if present, otherwise use defaults act_title = ch_outline.get( "act_title", {1: "Act I — The Setup / The Shock", 2: "Act II — Technical & Military Detail", 3: "Act III — The Verified Aftermath"}.get(chapter_num, f"Chapter {chapter_num}") ) # Build continuity tail — last 300 words of the immediately preceding chapter continuity_tail = "" if previous_chapters: prev_text = previous_chapters[-1].get("text_en", "") words = prev_text.split() continuity_tail = " ".join(words[-300:]) if len(words) > 300 else prev_text _info( f"[Pipeline] Ch{chapter_num} — '{act_title}': " f"Narrator (Groq/maestro) → Director (Groq)" ) # --- Agent 2: The Narrator (Groq, role-aware) --- t_nov = time.monotonic() text_en = await self._write_chapter_novelist( chapter_num=chapter_num, outline=outline, continuity_tail=continuity_tail, character_descs=character_descs, ) _ok( f"[Narrator/Ch{chapter_num}] '{act_title}' done " f"({time.monotonic()-t_nov:.1f}s) — {len(text_en.split())} words" ) # --- Agent 3: The Director (Groq, fast metadata extraction) --- t_dir = time.monotonic() metadata = await self._extract_chapter_metadata_director( chapter_num=chapter_num, chapter_text=text_en, ch_outline=ch_outline, ) _ok(f"[Director/Ch{chapter_num}] done ({time.monotonic()-t_dir:.1f}s)") return { "chapter_number": chapter_num, "act_title": act_title, "text_en": text_en, "cinematic_image_prompt": metadata.get("cinematic_image_prompt", ""), "audio_environment": metadata.get("audio_environment", { "primary_ambience": ch_outline.get("ambiance", "battlefield"), "intensity": "medium", "dynamic_cues": "", "audio_prompt": "", }), } # ══════════════════════════════════════════════════════════════ # STEP 3 – The Scholar V10.0 — Per-Chapter Parallelism # ══════════════════════════════════════════════════════════════ async def _scholar_chapter_content( self, chapter_num: int, chapter_text: str, story_context: str, ) -> Dict[str, Any]: """ Generate vocabulary (5 items) + quiz (1 MCQ) for a single chapter. Uses the enrichment Gemini key pool. Input text is truncated to SCHOLAR_CHAPTER_MAX_CHARS to prevent overflow. """ # Truncate safely — never pass the full chapter text if it's very long safe_text = chapter_text[:SCHOLAR_CHAPTER_MAX_CHARS] # Trim story_context to a headline context_head = story_context[:350].strip() # V15.0: use build_scholar_chapter_prompt() which randomises the # correct-answer slot (A/B/C/D) at build time, eliminating "always A" bias. prompt = prompts.build_scholar_chapter_prompt( context_head=context_head, chapter_num=chapter_num, safe_text=safe_text, ) result = await self._call_gemini_json( prompt, f"Scholar/Ch{chapter_num}", max_tokens=2000, temperature=0.35, role="enrichment" ) # Validate and normalise vocabulary vocab = result.get("vocabulary", []) if len(vocab) != 5: _warn(f"[Scholar/Ch{chapter_num}] Got {len(vocab)} vocab items, expected 5. Padding.") _default_v = {"word": "", "meaning_en": "", "example": ""} vocab = (vocab + [_default_v] * 5)[:5] # Validate and normalise quiz quiz = result.get("quiz", {}) if not quiz or not quiz.get("question_en"): _warn(f"[Scholar/Ch{chapter_num}] Quiz missing — using placeholder.") quiz = { "question_en": f"What was the central conflict in Chapter {chapter_num}?", "options_en": ["A. Option 1", "B. Option 2", "C. Option 3", "D. Option 4"], "correct_answer_en": "A. Option 1", } if len(quiz.get("options_en", [])) != 4: quiz["options_en"] = (quiz.get("options_en", []) + ["", "", "", ""])[:4] return {"vocabulary": vocab, "quiz": quiz} async def _historian_trivia_content( self, story_summary: str, ) -> List[Dict[str, Any]]: """ Generate 3 global trivia facts from the story premise + historical context. Input is intentionally kept short (~200 words) for reliability. Uses the enrichment Gemini key pool. """ # Keep the input very lean — trivia needs broad context, not full chapter text summary_head = story_summary[:700].strip() prompt = prompts.HISTORIAN_TRIVIA_PROMPT_TEMPLATE.format( summary_head=summary_head, ) result = await self._call_gemini_json( prompt, "Historian", max_tokens=1000, temperature=0.5, role="enrichment" ) trivia = result.get("trivia", []) if len(trivia) != 3: _warn(f"[Historian] Got {len(trivia)} trivia facts, expected 3. Padding.") _default_t = {"fact_en": "", "period": ""} trivia = (trivia + [_default_t] * 3)[:3] return trivia async def _generate_scholar_content_v10( self, chapters: List[Dict[str, Any]], historical_context: str, story_premise: str, ) -> Tuple[List[List[Dict]], List[Dict], List[Dict]]: """ V10.0 Scholar orchestrator — fires 4 parallel tasks: • _scholar_chapter_content × 3 (per-chapter vocab + quiz) • _historian_trivia_content × 1 (global trivia) Returns: vocab_per_chapter : List of 3 × 5-item lists trivia_list : List of 3 trivia dicts quiz_per_chapter : List of 3 quiz dicts """ story_context = f"{historical_context} {story_premise}".strip() # Build 3 chapter tasks + 1 trivia task chapter_tasks = [ self._scholar_chapter_content( chapter_num=ch["chapter_number"], chapter_text=ch.get("text_en", ""), story_context=story_context, ) for ch in chapters ] trivia_task = self._historian_trivia_content(story_summary=story_context) # Run all 4 in parallel all_results = await asyncio.gather(*chapter_tasks, trivia_task, return_exceptions=True) chapter_results = all_results[:-1] trivia_result = all_results[-1] vocab_per_chapter: List[List[Dict]] = [] quiz_per_chapter: List[Dict] = [] _default_quiz_item = { "question_en": "", "options_en": ["", "", "", ""], "correct_answer_en": "", } _default_vocab_item = {"word": "", "meaning_en": "", "example": ""} for i, ch in enumerate(chapters): res = chapter_results[i] if isinstance(res, Exception): _warn(f"[Scholar/Ch{ch['chapter_number']}] Parallel task failed: {res}. Using defaults.") vocab_per_chapter.append([dict(_default_vocab_item)] * 5) quiz_per_chapter.append(dict(_default_quiz_item)) else: vocab_per_chapter.append(res.get("vocabulary", [dict(_default_vocab_item)] * 5)) quiz_per_chapter.append(res.get("quiz", dict(_default_quiz_item))) if isinstance(trivia_result, Exception): _warn(f"[Historian] Parallel task failed: {trivia_result}. Using defaults.") trivia_list: List[Dict] = [{"fact_en": "", "period": ""}] * 3 else: trivia_list = trivia_result # type: ignore[assignment] _ok(f"[Scholar V10] Done — {len(vocab_per_chapter)} vocab sets, " f"{len(trivia_list)} trivia facts, {len(quiz_per_chapter)} quizzes") return vocab_per_chapter, trivia_list, quiz_per_chapter # ══════════════════════════════════════════════════════════════ # STEP 4 – Per-Chapter Parallel Translation (V11.0) # ══════════════════════════════════════════════════════════════ async def _translate_story_meta_gemini( self, master_en: Dict[str, Any], target_language: str, lang_code: str, ) -> Dict[str, Any]: """ Translate ONLY the short metadata fields and historical trivia. Output is ~400 tokens — fast and never truncated. Returns a dict with keys: title_{lc}, story_premise_{lc}, historical_context_{lc}, learning_goal_{lc}, historical_trivia (list of fact_{lc} dicts) """ extra_ar = ( "\nSTRICT RULE: Use Grand Literary Modern Standard Arabic " "(الفصحى التراثية). Every phrase must match the English in " "descriptive richness.\n" if lang_code == "ar" else "" ) # V16.0: inject number-formatting rule for all languages number_rule = prompts.get_number_rule(lang_code) trivia_items = master_en.get("historical_trivia", []) trivia_json = json.dumps( [{"fact_en": t.get("fact_en", "")} for t in trivia_items], ensure_ascii=False, ) # Build the prompt using the base template (we'll format with json.dumps values) prompt = prompts.META_TRANSLATION_PROMPT_BASE.format( target_language=target_language, extra_ar=extra_ar, number_rule=number_rule, title_en_json=json.dumps(master_en.get("title_en",""), ensure_ascii=False), story_premise_json=json.dumps(master_en.get("story_premise",""), ensure_ascii=False), historical_context_json=json.dumps(master_en.get("historical_context_en",""), ensure_ascii=False), learning_goal_json=json.dumps(master_en.get("learning_goal_en",""), ensure_ascii=False), trivia_json=trivia_json, lang_code=lang_code, ) last_error = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_gemini_role("translation", f"MetaTrans/{lang_code.upper()}") try: _info(f"[MetaTrans/{lang_code.upper()}] attempt {attempt} | key=…{key[-4:]}") data = await api_clients.call_gemini_json( api_key=key, prompt=prompt, caller=f"MetaTrans/{lang_code.upper()}", max_tokens=1500, temperature=0.15, ) _ok(f"[MetaTrans/{lang_code.upper()}] done") return data except (json.JSONDecodeError, ValueError) as exc: logger.error("[MetaTrans/%s] JSON error attempt=%d: %s", lang_code.upper(), attempt, exc) last_error = str(exc) await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) except Exception as exc: exc_str = str(exc) es = exc_str.lower() if "429" in exc_str or "rate" in es or "quota" in es: await self._mark_gemini_rl(key, f"MetaTrans/{lang_code.upper()}") elif "500" in exc_str or "502" in exc_str or "503" in exc_str or "server_error" in es: await self._mark_gemini_server_error(key, f"MetaTrans/{lang_code.upper()}") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) last_error = exc_str _warn(f"[MetaTrans/{lang_code.upper()}] All attempts failed — returning empty meta.") return {} async def _translate_single_chapter_gemini( self, chapter: Dict[str, Any], target_language: str, lang_code: str, ) -> Dict[str, Any]: """ Translate ONE chapter: story text + 5 vocabulary items + 1 quiz question. Input budget: ~850 words chapter + 5 vocab + 1 quiz ≈ 1400 tokens in Output budget: ~1400 tokens translated — safely within Gemini limits. Returns a dict matching the chapter structure expected by _merge_into_mega_json(): chapter_number, text_{lc}, vocabulary (list), quiz_{lc} """ chapter_num = chapter.get("chapter_number", 0) text_en = chapter.get("text_en", "") vocab_en = chapter.get("vocabulary", []) quiz_en = chapter.get("quiz", {}) extra_ar = ( "\nSTRICT ARABIC RULE: Match the English word count and descriptive " "density. Use Grand Literary Modern Standard Arabic (الفصحى التراثية). " "Translate every sensory detail — do NOT summarise.\n" if lang_code == "ar" else "" ) # V16.0: inject number-formatting rule number_rule = prompts.get_number_rule(lang_code) # Compact input for this single chapter chapter_input = { "chapter_number": chapter_num, "text_en": text_en, "vocabulary": [ {"word": v.get("word",""), "meaning_en": v.get("meaning_en",""), "example": v.get("example","")} for v in vocab_en ], "quiz": { "question_en": quiz_en.get("question_en", ""), "options_en": quiz_en.get("options_en", []), "correct_answer_en":quiz_en.get("correct_answer_en", ""), }, } compact = json.dumps(chapter_input, ensure_ascii=False, separators=(",", ":")) prompt = prompts.SINGLE_CHAPTER_TRANSLATION_PROMPT_BASE.format( target_language=target_language, extra_ar=extra_ar, number_rule=number_rule, lang_code=lang_code, chapter_input_json=compact, chapter_num=chapter_num, ) last_error = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_gemini_role( "translation", f"ChTrans/{lang_code.upper()}/Ch{chapter_num}" ) try: _info( f"[ChTrans/{lang_code.upper()}/Ch{chapter_num}] " f"attempt {attempt} | key=…{key[-4:]}" ) raw_content = await api_clients.call_gemini_json( api_key=key, prompt=prompt, caller=f"ChTrans/{lang_code.upper()}/Ch{chapter_num}", max_tokens=3000, temperature=0.20, ) # Note: call_gemini_json already returns parsed JSON, # so we don't need to parse again. data = raw_content # Quality gate: translated text should be reasonably long t_text = data.get(f"text_{lang_code}", "") t_words = len(t_text.split()) en_words = len(text_en.split()) if t_words < en_words * 0.35: _warn( f"[ChTrans/{lang_code.upper()}/Ch{chapter_num}] " f"Translated text only {t_words} words " f"(expected ~{en_words}) — retrying with fresh key." ) await self._mark_gemini_server_error( key, f"ChTrans/{lang_code.upper()}/Ch{chapter_num}" ) last_error = "TruncatedTranslation" continue _ok( f"[ChTrans/{lang_code.upper()}/Ch{chapter_num}] " f"done — {t_words} words" ) return data except (json.JSONDecodeError, ValueError) as exc: logger.error( "[ChTrans/%s/Ch%d] JSON error attempt=%d: %s", lang_code.upper(), chapter_num, attempt, exc, ) last_error = str(exc) await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) except Exception as exc: exc_str = str(exc) es = exc_str.lower() if "429" in exc_str or "rate" in es or "quota" in es: await self._mark_gemini_rl(key, f"ChTrans/{lang_code.upper()}/Ch{chapter_num}") elif "500" in exc_str or "502" in exc_str or "503" in exc_str or "server_error" in es: await self._mark_gemini_server_error(key, f"ChTrans/{lang_code.upper()}/Ch{chapter_num}") else: logger.error( "[ChTrans/%s/Ch%d] Error attempt=%d: %s", lang_code.upper(), chapter_num, attempt, exc, ) await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) last_error = exc_str _warn( f"[ChTrans/{lang_code.upper()}/Ch{chapter_num}] " f"All attempts failed — returning empty chapter translation." ) return {"chapter_number": chapter_num} async def _translate_story_per_chapter_v11( self, master_en: Dict[str, Any], target_language: str, lang_code: str, ) -> Optional[Dict[str, Any]]: """ V11.0 Per-Chapter Parallel Translation orchestrator. Replaces the monolithic _translate_story_via_gemini_v11 which caused output-token truncation (only Chapter 1 was translated). Fires asyncio.gather() over: • 1× _translate_story_meta_gemini() — metadata + trivia • N× _translate_single_chapter_gemini() — one per chapter All tasks run concurrently using the Gemini "translation" role key pool. Total output tokens per language: ~4600 (vs ~8000+ for the monolithic approach that caused truncation). Assembles results into the same dict structure consumed by _merge_into_mega_json() — downstream code is unchanged. """ chapters = master_en.get("chapters", []) if not chapters: return None _info( f"[Trans/{lang_code.upper()}] V11 per-chapter parallel — " f"{len(chapters)} chapter task(s) + 1 meta task" ) # Build all tasks meta_task = self._translate_story_meta_gemini(master_en, target_language, lang_code) chapter_tasks = [ self._translate_single_chapter_gemini(ch, target_language, lang_code) for ch in chapters ] # Run all in parallel all_results = await asyncio.gather( meta_task, *chapter_tasks, return_exceptions=True ) meta_result = all_results[0] chapter_results = all_results[1:] # Handle meta failure if isinstance(meta_result, Exception): _warn(f"[Trans/{lang_code.upper()}] Meta translation failed: {meta_result}") meta_result = {} # Assemble into the structure _merge_into_mega_json() expects assembled: Dict[str, Any] = { f"title_{lang_code}": meta_result.get(f"title_{lang_code}", master_en.get("title_en", "")), f"story_premise_{lang_code}": meta_result.get(f"story_premise_{lang_code}", master_en.get("story_premise", "")), f"historical_context_{lang_code}": meta_result.get(f"historical_context_{lang_code}", master_en.get("historical_context_en", "")), f"learning_goal_{lang_code}": meta_result.get(f"learning_goal_{lang_code}", master_en.get("learning_goal_en", "")), "historical_trivia": meta_result.get("historical_trivia", []), "chapters": [], } for i, ch_result in enumerate(chapter_results): src_ch = chapters[i] ch_num = src_ch.get("chapter_number", i + 1) if isinstance(ch_result, Exception): _warn( f"[Trans/{lang_code.upper()}] Chapter {ch_num} task " f"raised exception: {ch_result} — using EN fallback." ) ch_result = {"chapter_number": ch_num} # Ensure chapter_number is present ch_result.setdefault("chapter_number", ch_num) # Fallback: if translated text is missing or suspiciously short, # run a direct chunked translation as a last resort t_text = ch_result.get(f"text_{lang_code}", "") en_text = src_ch.get("text_en", "") en_words = len(en_text.split()) if not t_text or (en_words > 100 and len(t_text.split()) < en_words * 0.35): _warn( f"[Trans/{lang_code.upper()}] Chapter {ch_num} text " f"missing or too short — running chunked fallback." ) try: rechunked = await self._translate_chapter_text_chunked_gemini( en_text, target_language, lang_code, ch_num ) ch_result[f"text_{lang_code}"] = rechunked except Exception as exc: _warn(f"[Trans/{lang_code.upper()}] Chunked fallback failed: {exc}") assembled["chapters"].append(ch_result) ok_chapters = [ c.get("chapter_number") for c in assembled["chapters"] if c.get(f"text_{lang_code}") ] _ok( f"[Trans/{lang_code.upper()}] V11 parallel complete — " f"translated chapters: {ok_chapters}" ) return assembled async def _translate_chapter_text_chunked_gemini( self, text: str, target_language: str, lang_code: str, chapter_num: int, ) -> str: """ V10.1 — Recursive chunking for long chapter texts via Gemini. Replaces the V10.0 Groq-based version. Splits the source text at sentence boundaries into ~RECURSIVE_TRANSLATE_CHUNK_WORDS-word sub-chunks, translates each with a Gemini "translation" role key, then rejoins into a single translated chapter string. Called when _translate_story_via_gemini_v11 detects that a chapter's translated text is missing, contaminated, or < 40% the expected word count relative to the English source. """ extra_ar = ( "Use Grand Literary Modern Standard Arabic (الفصحى التراثية). " "Match the source word count and descriptive density. " if lang_code == "ar" else "" ) # V16.0: number-formatting rule for all languages number_rule = prompts.get_number_rule(lang_code) # Gemini doesn't use a separate system message; merge into prompt. prefix = prompts.CHUNKED_TRANSLATION_PREFIX_TEMPLATE.format( target_language=target_language, extra_ar=extra_ar, number_rule=number_rule, ) # Split at sentence boundaries into ~600-word sub-chunks. sentences: List[str] = re.split(r"(?<=[.!?])\s+", text) chunks: List[str] = [] current: List[str] = [] cur_words: int = 0 for sent in sentences: sw = len(sent.split()) if cur_words + sw > RECURSIVE_TRANSLATE_CHUNK_WORDS and current: chunks.append(" ".join(current)) current = [sent] cur_words = sw else: current.append(sent) cur_words += sw if current: chunks.append(" ".join(current)) _info( f"[GeminiChunkTrans/Ch{chapter_num}/{lang_code.upper()}] " f"{len(text.split())} words → {len(chunks)} sub-chunks" ) translated_chunks: List[str] = [] for chunk_idx, chunk in enumerate(chunks): last_err = None for attempt in range(1, MAX_RETRIES + 1): key = await self._acquire_gemini_role( "translation", f"GeminiChunkTrans/Ch{chapter_num}/{lang_code.upper()}/c{chunk_idx}", ) try: _info( f"[GeminiChunkTrans/Ch{chapter_num}/{lang_code.upper()}] " f"chunk={chunk_idx} attempt={attempt} key=…{key[-4:]}" ) result = await api_clients.call_gemini_chunk_translation( api_key=key, prefix=prefix, chunk=chunk, ) # Reject contaminated output and retry with a fresh key. if _text_is_contaminated(result): raise ValueError( f"Translated chunk appears contaminated: {result[:100]}" ) translated_chunks.append(result) break except Exception as exc: exc_str = str(exc) es = exc_str.lower() if "429" in exc_str or "rate" in es or "quota" in es: await self._mark_gemini_rl( key, f"GeminiChunkTrans/Ch{chapter_num}/{lang_code.upper()}", ) last_err = "RateLimit" elif "500" in exc_str or "502" in exc_str or "503" in exc_str or "server_error" in es: await self._mark_gemini_server_error( key, f"GeminiChunkTrans/Ch{chapter_num}/{lang_code.upper()}", ) last_err = f"ServerError({exc_str[:60]})" else: logger.error( "[GeminiChunkTrans/Ch%d/%s] chunk=%d attempt=%d: %s", chapter_num, lang_code.upper(), chunk_idx, attempt, exc, ) last_err = exc await asyncio.sleep(RETRY_BACKOFF_SECONDS * attempt) else: # All retries exhausted — keep English text for this chunk. _warn( f"[GeminiChunkTrans/Ch{chapter_num}/{lang_code.upper()}] " f"Sub-chunk {chunk_idx} failed ({last_err}) — English fallback." ) translated_chunks.append(chunk) return " ".join(translated_chunks) # ══════════════════════════════════════════════════════════════ # V15.0 — Quiz Option Shuffler (100% eliminates "always A" bias) # ══════════════════════════════════════════════════════════════ @staticmethod def _shuffle_quiz_options(quiz: Dict[str, Any]) -> Dict[str, Any]: """ Shuffle quiz answer options for all language variants in a quiz dict while keeping every correct_answer_* pointer accurate after the shuffle. Algorithm (per language that has options): 1. Record which option text is the correct answer. 2. Create a list of (index, option_text) pairs and shuffle them. 3. Rewrite options_* in the new shuffled order. 4. Find the new position of the previously-correct option text. 5. Rewrite correct_answer_* to that new position's letter prefix. The English shuffle is the canonical shuffle; all other languages apply the SAME permutation so every language's "position B" refers to the same factual answer across all locales. Returns a new quiz dict — the input is not mutated. """ quiz = dict(quiz) # shallow copy — we'll replace list values below # Build the canonical permutation from the English options en_options = quiz.get("options_en", []) if len(en_options) != 4: # Guard: if options are malformed, return unchanged return quiz # Shuffle indices indices = list(range(4)) random.shuffle(indices) # Map old letter → new letter for correct_answer rewriting letters = ["A", "B", "C", "D"] old_to_new_letter: Dict[str, str] = { letters[old_i]: letters[new_i] for new_i, old_i in enumerate(indices) } def _reorder(options: List[str]) -> List[str]: return [options[i] for i in indices] def _remap_correct(correct: str, options_before: List[str], options_after: List[str]) -> str: """ Find the text of the correct answer in options_before, locate it in options_after, and return the matching 'X. ...' string. Falls back to old_to_new_letter prefix remapping if text match fails. """ # Primary: match by full text for opt in options_after: if opt == correct: return opt # Secondary: strip "A. " prefix and match bare text correct_bare = re.sub(r"^[A-D]\.\s*", "", correct).strip() for opt in options_after: opt_bare = re.sub(r"^[A-D]\.\s*", "", opt).strip() if opt_bare == correct_bare: return opt # Tertiary: remap the letter prefix m = re.match(r"^([A-D])\.\s*(.*)", correct) if m: new_letter = old_to_new_letter.get(m.group(1), m.group(1)) return f"{new_letter}. {m.group(2)}" return correct # Apply to English en_before = quiz["options_en"] en_after = _reorder(en_before) en_correct = quiz.get("correct_answer_en", "") quiz["options_en"] = en_after quiz["correct_answer_en"] = _remap_correct(en_correct, en_before, en_after) # Apply the SAME permutation to every other language variant for lc in ["ar", "uk", "es", "de"]: opts_key = f"options_{lc}" correct_key = f"correct_answer_{lc}" lc_options = quiz.get(opts_key, []) if len(lc_options) == 4: lc_after = _reorder(lc_options) quiz[opts_key] = lc_after quiz[correct_key] = _remap_correct( quiz.get(correct_key, ""), lc_options, lc_after ) return quiz # ══════════════════════════════════════════════════════════════ # MERGE – Final Mega-JSON (V15.0 updated) # ══════════════════════════════════════════════════════════════ def _merge_into_mega_json( self, master_en: Dict[str, Any], translations: Dict[str, Optional[Dict[str, Any]]], story_id: str, hour_slot: int, ) -> Dict[str, Any]: lang_codes = ["ar", "uk", "es", "de"] TTS_LANGS = ["en", "ar", "uk", "es", "de"] mega: Dict[str, Any] = { "story_id": story_id, "hour_slot": hour_slot, "pipeline": "V16.2-TrulyFree", "historical_era": master_en.get("historical_era", ""), "story_premise": master_en.get("story_premise", ""), "title_en": master_en.get("title_en", ""), "historical_context_en": master_en.get("historical_context_en", ""), "learning_goal_en": master_en.get("learning_goal_en", ""), # V13.0: On-This-Day anchor metadata "historical_anchor": master_en.get("historical_anchor", {}), } for lc in lang_codes: t = translations.get(lc) or {} mega[f"title_{lc}"] = t.get(f"title_{lc}", mega["title_en"]) mega[f"story_premise_{lc}"] = t.get(f"story_premise_{lc}", mega["story_premise"]) mega[f"historical_context_{lc}"] = t.get(f"historical_context_{lc}", mega["historical_context_en"]) mega[f"learning_goal_{lc}"] = t.get(f"learning_goal_{lc}", mega["learning_goal_en"]) # Trivia (with per-language fallback) en_trivia = master_en.get("historical_trivia", []) merged_trivia: List[Dict] = [] for ti, et in enumerate(en_trivia): entry: Dict[str, Any] = {"fact_en": et.get("fact_en", ""), "period": et.get("period", "")} for lc in lang_codes: t_tri = (translations.get(lc) or {}).get("historical_trivia", []) tt = t_tri[ti] if ti < len(t_tri) else {} entry[f"fact_{lc}"] = tt.get(f"fact_{lc}", entry["fact_en"]) merged_trivia.append(entry) mega["historical_trivia"] = merged_trivia # Per-language chapter maps trans_ch_map: Dict[str, Dict[int, Dict]] = {} for lc in lang_codes: t = translations.get(lc) or {} trans_ch_map[lc] = { ch["chapter_number"]: ch for ch in t.get("chapters", []) if isinstance(ch, dict) and "chapter_number" in ch } mega_chapters: List[Dict] = [] for en_ch in master_en.get("chapters", []): num = en_ch["chapter_number"] en_vocab = en_ch.get("vocabulary", []) en_quiz = en_ch.get("quiz", {}) texts: Dict[str, str] = {"en": en_ch.get("text_en", "")} for lc in lang_codes: tch = trans_ch_map.get(lc, {}).get(num, {}) texts[lc] = tch.get(f"text_{lc}", texts["en"]) # Vocabulary with translations vocab_list: List[Dict] = [] for vi, ev in enumerate(en_vocab): entry: Dict[str, Any] = { "word_en": ev.get("word", ""), "meaning_en": ev.get("meaning_en", ""), "example_en": ev.get("example", ""), } for lc in lang_codes: tch = trans_ch_map.get(lc, {}).get(num, {}) tvocab = tch.get("vocabulary", []) tv = tvocab[vi] if vi < len(tvocab) else {} entry[f"word_{lc}"] = tv.get(f"word_{lc}", entry["word_en"]) entry[f"meaning_{lc}"] = tv.get(f"meaning_{lc}", entry["meaning_en"]) entry[f"example_{lc}"] = tv.get(f"example_{lc}", entry["example_en"]) vocab_list.append(entry) # Quiz with translations quiz_entry: Dict[str, Any] = { "question_en": en_quiz.get("question_en", ""), "options_en": en_quiz.get("options_en", []), "correct_answer_en": en_quiz.get("correct_answer_en", ""), } for lc in lang_codes: tch = trans_ch_map.get(lc, {}).get(num, {}) tquiz = tch.get(f"quiz_{lc}", {}) quiz_entry[f"question_{lc}"] = tquiz.get("question", quiz_entry["question_en"]) quiz_entry[f"options_{lc}"] = tquiz.get("options", quiz_entry["options_en"]) quiz_entry[f"correct_answer_{lc}"] = tquiz.get("correct_answer", quiz_entry["correct_answer_en"]) # V15.0: shuffle quiz options BEFORE writing to mega-JSON. # This is the final, authoritative shuffle point. # correct_answer_* for every language is re-pointed after shuffle. quiz_entry = self._shuffle_quiz_options(quiz_entry) mega_chapters.append({ "chapter_number": num, "act_title": en_ch.get("act_title", ""), # V15.0: preserve 3-act label "cinematic_image_prompt": en_ch.get("cinematic_image_prompt", ""), "audio_environment": en_ch.get("audio_environment", {}), "image_url": "", "texts": texts, "audio_urls": {lang: "" for lang in TTS_LANGS}, "karaoke_urls": {lang: "" for lang in TTS_LANGS}, "vocabulary": vocab_list, "quiz": quiz_entry, }) mega["chapters"] = mega_chapters # FIX 4 (V12.0): Aggregate chapter quizzes at root level. mega["quiz"] = [ch["quiz"] for ch in mega_chapters] # V13.0 Req 6: Inject legacy interaction.choices adapter into every chapter # so the frontend _buildChapterCard() continues to work without modification. for ch in mega_chapters: ch["interaction"] = self._adapt_quiz_to_choices(ch.get("quiz", {}), lang_codes) return mega # ────────────────────────────────────────────────────────────────── # V13.0 Req 6 — Quiz Adapter: converts backend quiz format → # legacy frontend interaction.choices[] format # ────────────────────────────────────────────────────────────────── @staticmethod def _adapt_quiz_to_choices( quiz: Dict[str, Any], lang_codes: Optional[List[str]] = None, ) -> Dict[str, Any]: """ Convert the backend quiz structure: { "question_en": "...", "options_en": ["A. ...", ...], "correct_answer_en": "A. ...", "question_ar": "...", "options_ar": [...], "correct_answer_ar": "...", ... } into the legacy frontend `interaction` structure: { "question_en": "...", "question_ar": "...", ... "choices": [ {"text_en": "A. ...", "text_ar": "...", "is_correct": True/False}, ... ] } Both English and all translated lang variants are wired into each choice object so the frontend `c[`text_${tgt}`]` lookup works for every language the app supports. """ if not quiz: return {"choices": []} if lang_codes is None: lang_codes = ["ar", "uk", "es", "de"] all_langs = ["en"] + list(lang_codes) interaction: Dict[str, Any] = {} # Copy all question_ keys for every language for lang in all_langs: qkey = f"question_{lang}" if qkey in quiz: interaction[qkey] = quiz[qkey] # Build choices array from English options (authoritative) en_options = quiz.get("options_en", []) en_correct = quiz.get("correct_answer_en", "") choices: List[Dict[str, Any]] = [] for i, opt_en in enumerate(en_options): choice: Dict[str, Any] = { "text_en": opt_en, "is_correct": (opt_en == en_correct), "explanation_en": "", } # Wire translated option text for each lang for lc in lang_codes: lang_options = quiz.get(f"options_{lc}", []) choice[f"text_{lc}"] = lang_options[i] if i < len(lang_options) else opt_en choices.append(choice) interaction["choices"] = choices return interaction # ══════════════════════════════════════════════════════════════ # V10.0 — Pre-flight Sanity Check + Emergency Recovery # ══════════════════════════════════════════════════════════════ async def _pre_flight_check(self, mega: Dict[str, Any]) -> None: """ Validates the mega-JSON before it is returned to immersion.py. Checks: • historical_trivia is a non-empty list with non-empty fact_en strings • Every chapter has a non-empty quiz.question_en • Every chapter's quiz correct_answer_en is present in options_en (post-shuffle integrity check — V15.0) • Every chapter has non-empty vocabulary If anything is missing, triggers _emergency_recover_scholar. """ issues: List[str] = [] trivia = mega.get("historical_trivia", []) if not trivia or not any(t.get("fact_en") for t in trivia): issues.append("historical_trivia") for ch in mega.get("chapters", []): num = ch["chapter_number"] quiz = ch.get("quiz", {}) if not quiz or not quiz.get("question_en"): issues.append(f"chapter_{num}_quiz") else: # V15.0 post-shuffle integrity: correct_answer_en must be in options_en correct = quiz.get("correct_answer_en", "") options = quiz.get("options_en", []) if correct and options and correct not in options: _warn( f"[PreFlight] Chapter {num} correct_answer_en '{correct[:40]}' " f"not found in options_en — re-shuffling" ) fixed = self._shuffle_quiz_options(quiz) ch["quiz"] = fixed # Re-compute interaction choices with the fixed quiz lang_codes = ["ar", "uk", "es", "de"] ch["interaction"] = self._adapt_quiz_to_choices(fixed, lang_codes) vocab = ch.get("vocabulary", []) if not vocab or not any(v.get("word_en") for v in vocab): issues.append(f"chapter_{num}_vocabulary") if issues: _warn(f"[PreFlight] Missing or empty fields: {issues} — triggering emergency recovery") await self._emergency_recover_scholar(mega, missing_fields=issues) else: _ok("[PreFlight] All fields validated ✓") async def _emergency_recover_scholar( self, mega: Dict[str, Any], missing_fields: Optional[List[str]] = None, ) -> None: """ Single-pass emergency recovery using the validation Gemini key pool. Re-generates only the specific missing content (quiz, vocab, or trivia) rather than the full Scholar suite, to keep the call small and reliable. """ _critical("[Emergency] Scholar recovery fired — using validation Gemini key pool") missing = set(missing_fields or []) story_context = ( mega.get("historical_context_en", "") + " " + mega.get("story_premise", "") ).strip() # Recover missing trivia if "historical_trivia" in missing: try: trivia = await self._historian_trivia_content(story_context) # Keep existing non-empty items, replace only empty ones existing = mega.get("historical_trivia", [{"fact_en": "", "period": ""}] * 3) for i, (ex, new) in enumerate(zip(existing, trivia)): if not ex.get("fact_en") and new.get("fact_en"): existing[i] = new mega["historical_trivia"] = existing _ok("[Emergency] Trivia recovered ✓") except Exception as exc: _warn(f"[Emergency] Trivia recovery failed: {exc}") # Recover missing chapter quiz / vocabulary for ch in mega.get("chapters", []): num = ch["chapter_number"] needs_quiz = f"chapter_{num}_quiz" in missing needs_vocab = f"chapter_{num}_vocabulary" in missing if not (needs_quiz or needs_vocab): continue try: # Use validation keys — small, reliable call ch_text = ch.get("texts", {}).get("en", "")[:SCHOLAR_CHAPTER_MAX_CHARS] result = await self._scholar_chapter_content( chapter_num=num, chapter_text=ch_text, story_context=story_context, ) if needs_quiz and result.get("quiz", {}).get("question_en"): ch["quiz"] = result["quiz"] _ok(f"[Emergency] Chapter {num} quiz recovered ✓") if needs_vocab and any(v.get("word") for v in result.get("vocabulary", [])): # Rebuild vocab_list format for the mega-JSON raw_vocab = result["vocabulary"] lc_list = ["ar", "uk", "es", "de"] new_vocab: List[Dict] = [] for rv in raw_vocab: entry: Dict[str, Any] = { "word_en": rv.get("word", ""), "meaning_en": rv.get("meaning_en", ""), "example_en": rv.get("example", ""), } for lc in lc_list: entry[f"word_{lc}"] = rv.get("word", "") entry[f"meaning_{lc}"] = rv.get("meaning_en", "") entry[f"example_{lc}"] = rv.get("example", "") new_vocab.append(entry) ch["vocabulary"] = new_vocab _ok(f"[Emergency] Chapter {num} vocabulary recovered ✓") except Exception as exc: _warn(f"[Emergency] Chapter {num} recovery failed: {exc}") # ══════════════════════════════════════════════════════════════ # PUBLIC ENTRY POINT – V14.0 Agentic Pipeline # ══════════════════════════════════════════════════════════════ async def generate_immersion_story( self, hour: Optional[int] = None, lang_focus: Optional[str] = None, system_prompt: Optional[str] = None, user_prompt: Optional[str] = None, max_tokens: Optional[int] = None, ) -> Dict[str, Any]: """ V16.2 Five-Agent Documentary Pipeline: Step 0 — History Dice: Groq picks a real "On This Day" event. Falls back to _pick_category() if Groq unavailable. Step 1 — The Architect (Gemini/validation): On-This-Day outline seeded with the dice result, mandatory 3-Act structure. Step 2 — The Narrator × 3 (Groq/maestro, SEQUENTIAL): Act I (Setup), Act II (Details), Act III (Conclusion) → The Director × 3 (Groq): cinematic keywords + audio Step 3a — The Scholars (Gemini/enrichment, PARALLEL): vocab + quiz (randomised answer slot) + trivia Step 3b — Attach scholar content to chapters Step 4 — The Translators (Gemini/translation, PARALLEL): AR / ES / DE / UK (4 languages × 4 tasks = 16 concurrent) Number-formatting rule injected into every prompt. Step 5 — Merge into final Mega-JSON: • act_title preserved per chapter • _shuffle_quiz_options() applied (100% "always A" elimination) • interaction.choices.is_correct re-computed post-shuffle Step 6 — Pre-flight sanity check + emergency scholar recovery Note: The Auditor (Step 2b) has been removed in V16.2. """ if not self._groq_keys: raise HTTPException(503, "No Groq API keys configured. Set GROQ_KEY_1…GROQ_KEY_8.") now = datetime.datetime.utcnow() hour = hour if hour is not None else now.hour hour_slot = (hour // 6) * 6 date_iso = now.strftime("%Y-%m-%d") day_of_year = now.timetuple().tm_yday story_id = f"{date_iso}-{hour_slot:02d}-{uuid.uuid4().hex[:8]}" # dash-separated for URL/DB compatibility today_month_day = now.strftime("%B %-d") # ── Lang_focus safety — never raise 422 ─────────────────── # Accept None, empty string, or any invalid value gracefully. VALID_LANG_FOCUS = {"ar", "uk", "world"} if not lang_focus or lang_focus.strip().lower() not in VALID_LANG_FOCUS: if lang_focus: _warn( f"[Pipeline] Invalid lang_focus='{lang_focus}' received — " f"defaulting to 'world'" ) lang_focus = {0: "world", 6: "ar", 12: "uk", 18: "world"}.get(hour_slot, "world") else: lang_focus = lang_focus.strip().lower() # ── Step 0: History Dice ─────────────────────────────────── _info(f"🎲 Step 0 — History Dice for {today_month_day}…") dice_key = await self._acquire_groq_role("maestro", "HistoryDice") dice_event = await roll_historical_dice(dice_key, today_month_day) # Determine category — dice result overrides static picker if dice_event: category = dice_event.get("category", "Wars & Battles") cat_instruction = ( f"Focus this documentary on the following verified historical event " f"that occurred on {today_month_day}: " f"\"{dice_event.get('event', '')}\" ({dice_event.get('year', '')}). " f"Build the 3-act outline entirely around this specific event." ) _ok( f"Step 0 (Dice) → '{dice_event.get('event','?')}' " f"({dice_event.get('year','?')}) | cat={category}" ) else: _cat_idx, category, cat_instruction = _pick_category(hour_slot, day_of_year) _info(f"Step 0 (Dice fallback) → static category: {category}") _info( f"🚀 V16.2 TrulyFree Pipeline START | {story_id} | " f"slot={hour_slot:02d} | cat={category} | focus={lang_focus}" ) # ── Step 1: The Architect (Gemini/validation) ────────────── t1 = time.monotonic() outline = await self._generate_screenplay_outline( category=category, cat_instruction=cat_instruction, lang_focus=lang_focus, date_anchor=today_month_day, today_month_day=today_month_day, ) anchor = outline.get("historical_anchor", {}) _ok( f"Step 1 (Architect) done ({time.monotonic()-t1:.1f}s) | " f"\"{outline.get('title_en','?')}\" | " f"anchor={anchor.get('event','?')} ({anchor.get('year','?')})" ) # ── Character Integrity Firewall (V18.0) ──────────────────────── # Strip any character the Architect invented without a primary source. # Undocumented figures silently removed — Narrator uses institutional voice. raw_chars = outline.get("characters", []) # V18.5: GPT OSS 120B sometimes returns characters as strings — skip them verified_chars = [ c for c in raw_chars if isinstance(c, dict) and c.get("primary_source", "").strip() and c.get("primary_source", "").strip().lower() not in ("", "unknown", "n/a", "none") ] removed = len(raw_chars) - len(verified_chars) if removed: _warn( f"[CharFirewall] Removed {removed} undocumented character(s) " f"from outline — Narrator will use institutional voice for those roles." ) outline["characters"] = verified_chars # ── Step 2: The Narrator (Groq/maestro) + The Director (Groq) ── # STRICTLY SEQUENTIAL: Act I → Act II → Act III. # The Director runs immediately after each chapter. t2 = time.monotonic() chapters: List[Dict[str, Any]] = [] for chap_num in [1, 2, 3]: chapter = await self._write_chapter( chapter_num=chap_num, outline=outline, previous_chapters=chapters, character_descs=outline.get("characters", []), ) chapters.append(chapter) _ok( f"Step 2 (Narrator+Director) done ({time.monotonic()-t2:.1f}s) | " f"3 acts | " f"words=[{','.join(str(len(c['text_en'].split())) for c in chapters)}]" ) # ── Assemble master_en — Narrator output goes directly to Scholars ── master_en: Dict[str, Any] = { "title_en": outline["title_en"], "historical_era": outline["historical_era"], "story_premise": outline["story_premise"], "historical_context_en": outline["historical_context_en"], "learning_goal_en": outline["learning_goal_en"], "historical_anchor": outline.get("historical_anchor", {}), "chapters": chapters, } # ── Step 3a: The Scholars (Gemini, PARALLEL) ─────────────── t3 = time.monotonic() vocab_per_chapter, trivia_list, quiz_per_chapter = \ await self._generate_scholar_content_v10( chapters=chapters, historical_context=outline["historical_context_en"], story_premise=outline["story_premise"], ) # ── Step 3b: Attach scholar content ─────────────────────── for i, ch in enumerate(master_en["chapters"]): ch["vocabulary"] = vocab_per_chapter[i] ch["quiz"] = quiz_per_chapter[i] master_en["historical_trivia"] = trivia_list _ok( f"Step 3 (Scholars) done ({time.monotonic()-t3:.1f}s) — " f"4 Gemini tasks (parallel)" ) # ── Step 4: The Translators (Gemini, PARALLEL) ───────────── # 16 concurrent Gemini calls (4 languages × 4 tasks each). # Number-formatting rule injected into every translation prompt. t4 = time.monotonic() _info( "🌐 Step 4 — Per-chapter parallel Gemini translation " "(1 meta + 3 chapter tasks × 4 languages = 16 concurrent tasks)…" ) results = await asyncio.gather( self._translate_story_per_chapter_v11(master_en, "Arabic", "ar"), self._translate_story_per_chapter_v11(master_en, "Spanish", "es"), self._translate_story_per_chapter_v11(master_en, "German", "de"), self._translate_story_per_chapter_v11(master_en, "Ukrainian", "uk"), return_exceptions=True, ) translations: Dict[str, Optional[Dict]] = {} for lc, res in zip(["ar", "es", "de", "uk"], results): if isinstance(res, Exception): _warn(f"[Step4/{lc.upper()}] Exception: {res} — EN fallback") translations[lc] = None else: translations[lc] = res ok_langs = [lc for lc, v in translations.items() if v is not None] _ok(f"Step 4 done ({time.monotonic()-t4:.1f}s) | translated={ok_langs}") # ── Step 5: Merge into final Mega-JSON ──────────────────── mega = self._merge_into_mega_json(master_en, translations, story_id, hour_slot) mega["story_id"] = story_id mega["hour_slot"] = hour_slot mega["category"] = category mega["lang_focus"] = lang_focus mega["historical_anchor"] = outline.get("historical_anchor", {}) mega["pipeline"] = "V16.2-TrulyFree" if dice_event: mega["dice_event"] = dice_event # ── Step 6: Pre-flight Sanity Check ─────────────────────── await self._pre_flight_check(mega) _ok(f"🎯 V16.2 TrulyFree Pipeline COMPLETE | {story_id}") return mega # ────────────────────────────────────────────────────────────────── # Singleton # ────────────────────────────────────────────────────────────────── key_manager = SmartKeyManager() # ══════════════════════════════════════════════════════════════════ # Standalone FastAPI app (legacy / backup) # ══════════════════════════════════════════════════════════════════ app = FastAPI(title="Titan Gateway", version="16.2") @app.middleware("http") async def add_security_headers(request: Request, call_next): response = await call_next(request) response.headers["Cross-Origin-Opener-Policy"] = "same-origin" response.headers["Cross-Origin-Embedder-Policy"] = "require-corp" return response app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/health") async def health(): return { "status": "ok", "version": "16.2 — Truly Free", "v16_agents": { "0_dice": "Groq/maestro — roll_historical_dice(): 5 real On-This-Day events → random.choice()", "1_architect": f"Gemini/{GEMMA_MODEL_ID} (validation) — 3-Act outline seeded by dice result", "2_narrator": f"Groq/{IMMERSION_MAESTRO_MODEL} (maestro) — Iron Standard V18.0 narration (Act I→II→III)", "3_director": f"Groq/{IMMERSION_AUTHOR_MODEL} (generation) — keyword-only image prompt + audio env", "4_scholars": f"Gemini/{GEMMA_MODEL_ID} (enrichment) — vocab + quiz (A/B/C/D random) + trivia", "5_translators": f"Gemini/{GEMMA_MODEL_ID} (translation) — AR/ES/DE/UK + number rule (16 parallel)", }, "v162_changes": { "auditor_removed": "_audit_story_gemini() deleted — no more prompts.AUDITOR_* AttributeError", "pipeline_tag": "V16.2-TrulyFree", "flow": "Narrator → Scholars (direct, no intermediate Auditor step)", }, "v160_preserved": { "history_dice": "roll_historical_dice() — LLM picks 5 real events, random.choice() selects one", "lang_focus_safe": "Any invalid/empty lang_focus safely defaults to 'world' (no 422 errors)", "number_rule": "words(digits) format enforced in Narrator and all translation prompts", "director_prompt": "Keyword-only output, max 15 words, zero sentences, zero emotional language", }, "groq_keys": len(GROQ_API_KEYS), "gemini_keys": len(GEMINI_API_KEYS), "groq_pool": [GROQ_KEY_NAMES.get(k, f"…{k[-4:]}") for k in GROQ_API_KEYS], "gemini_pool": [GEMINI_KEY_NAMES.get(k, f"…{k[-4:]}") for k in GEMINI_API_KEYS], } @app.post("/api/v1/cinema/transcribe") async def transcribe(file: UploadFile = File(...)): try: audio_bytes = await file.read() result = await key_manager.transcribe_audio(audio_bytes, filename=file.filename) return JSONResponse(content=result) except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) @app.post("/api/v1/cinema/translate") async def translate(request: Request): try: body = await request.json() except Exception: raise HTTPException(400, "Invalid JSON") if isinstance(body, dict) and "segments" in body: segments, target_language = body["segments"], body.get("target_language", "Arabic") elif isinstance(body, list): segments, target_language = body, "Arabic" else: raise HTTPException(422, "Expected array or {segments, target_language}") if not isinstance(segments, list): raise HTTPException(422, "segments must be a list") for idx, seg in enumerate(segments): if not isinstance(seg, dict) or "id" not in seg or "text" not in seg: raise HTTPException(422, f"Segment {idx} missing id/text") result = await key_manager.translate_segments(segments, target_language) return JSONResponse(content={"translations": result}) @app.get("/") async def root(): return { "message": "Titan Gateway V16.2 — Truly Free Pipeline", "keys": f"{len(GROQ_API_KEYS)} Groq + {len(GEMINI_API_KEYS)} Gemini", "features": [ "V16.2: Auditor removed — no more AttributeError crashes", "V16.2: Pipeline: Narrator → Scholars (direct, no intermediate step)", "V16.2: Version tag: V16.2-TrulyFree", "V16.0: History Dice — LLM picks 5 real On-This-Day events, random.choice() selects one", "V16.0: words(numbers) rule enforced in narration + all translation prompts", "V16.0: Director outputs keyword-only image prompts (max 15 words)", "V16.0: lang_focus always safe — invalid input defaults to 'world'", "V15.0: 3-Act structure (Setup / Technical Detail / Legacy+Conclusion)", "V15.0: Quiz shuffler — 'always A' bias eliminated", "V15.0: German (de) TTS — de-DE-KillianNeural, rate=-25%", "Role-based key pools (generation/maestro/translation/enrichment/validation)", "Pre-flight sanity check + emergency scholar recovery", ], } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)