# ============================================================ # 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 → "V22.0-CompoundResearcher" # [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 = "moonshotai/kimi-k2-instruct" # Narrator — Kimi K2, superior long-form prose, minimal padding # V22.0 — Compound Researcher replaces gpt-oss-120b for the Historian step COMPOUND_RESEARCHER_MODEL: str = "groq/compound-beta" # Historian — autonomous web search agent # 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]]: """ V24.0 — History Dice with deduplication. Loads the last 120 used historical topics and injects them into the prompt so the LLM avoids repeating them. Also filters the returned list client-side for a double layer of dedup protection. Returns a dict {"event": "...", "year": "...", "category": "..."} on success, or None if the call fails (caller falls back to _pick_category). """ # Load recent topics for dedup (same mechanism as non-historical types) _recent = await _load_and_cache_topics("historical") _excl_block = "" if _recent: _excl_list = ", ".join(f'"{t}"' for t in _recent[-30:]) _excl_block = ( f"\nDO NOT suggest any of these recently used topics: {_excl_list}\n" "Choose completely different events — strict prohibition on repetition.\n" ) prompt = prompts.DICE_PROMPT_TEMPLATE.format( today_month_day=today_month_day, exclusion_block=_excl_block, ) try: data = await api_clients.call_groq_json( api_key=api_key, prompt=prompt, caller="HistoryDice", max_tokens=800, temperature=0.5, # slightly higher → more variety ) 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 # V24.0 — Client-side dedup filter (second layer) _recent_lower = [r.lower() for r in _recent] fresh = [e for e in events if e.get("event", "").lower() not in _recent_lower] pool = fresh[:5] if fresh else events[:5] chosen = random.choice(pool) _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 # ── Topic deduplication: last 120 topics per type (stored on HF dataset) ──── # V23.1: moved from /tmp (wiped on restart) to HuggingFace dataset # so the used-topics list survives Space restarts and redeploys. _DEDUP_REPO_PATH = "dedup/used_topics_{story_type}.json" async def _hf_load_topics(story_type: str) -> List[str]: """Download used-topics list from HF dataset. Returns [] on any failure.""" try: import httpx as _httpx hf_token = HF_TOKEN dataset_id = os.environ.get("HF_DATASET_ID", "").strip() if not hf_token or not dataset_id: return [] repo_path = _DEDUP_REPO_PATH.format(story_type=story_type) url = f"https://huggingface.co/datasets/{dataset_id}/resolve/main/{repo_path}" async with _httpx.AsyncClient(timeout=10.0) as hx: r = await hx.get(url, headers={"Authorization": f"Bearer {hf_token}"}) if r.status_code == 200: return r.json() return [] except Exception: return [] async def _hf_save_topics(story_type: str, topics: List[str]) -> None: """Upload used-topics list to HF dataset. Silently ignores failures.""" try: from huggingface_hub import HfApi as _HfApi import io as _io hf_token = HF_TOKEN dataset_id = os.environ.get("HF_DATASET_ID", "").strip() if not hf_token or not dataset_id: return repo_path = _DEDUP_REPO_PATH.format(story_type=story_type) api = _HfApi(token=hf_token) data = json.dumps(topics, ensure_ascii=False, indent=2).encode() api.upload_file( path_or_fileobj=_io.BytesIO(data), path_in_repo=repo_path, repo_id=dataset_id, repo_type="dataset", ) except Exception: pass def _get_recent_topics(story_type: str) -> List[str]: """Return list of recently used topic titles (local /tmp cache only).""" try: path = f"/tmp/titan_used_topics_{story_type}.json" with open(path) as f: return json.load(f) except Exception: return [] def _save_topic_local(story_type: str, topic: str, topics: List[str]) -> None: """Write topics list to local /tmp cache.""" try: path = f"/tmp/titan_used_topics_{story_type}.json" with open(path, "w") as f: json.dump(topics, f, ensure_ascii=False) except Exception: pass async def _load_and_cache_topics(story_type: str) -> List[str]: """ Load topics from /tmp cache first (fast). If empty, fetch from HF dataset and warm the /tmp cache. """ local = _get_recent_topics(story_type) if local: return local # /tmp was wiped (restart) — reload from HF dataset remote = await _hf_load_topics(story_type) if remote: _save_topic_local(story_type, "", remote) return remote async def _save_topic(story_type: str, topic: str) -> None: """ Add topic to used list (keep last 120). Saves to both /tmp (fast) and HF dataset (persistent). """ recent = await _load_and_cache_topics(story_type) norm = topic.lower().strip() if norm not in [r.lower() for r in recent]: recent.append(topic) recent = recent[-120:] _save_topic_local(story_type, topic, recent) await _hf_save_topics(story_type, recent) async def roll_dice_for_type( api_key: str, story_type: str, today_month_day: str, ) -> Optional[Dict[str, str]]: """ Roll the dice for any story type. - historical: uses existing roll_historical_dice() - origin/mystery/science/explore: asks LLM to pick a topic for today Returns {"event": "...", "year": "...", "category": "..."} or None. """ if story_type == "historical": return await roll_historical_dice(api_key, today_month_day) # Build exclusion note from recent topics _recent = await _load_and_cache_topics(story_type) _excl = "" if _recent: _excl_list = ", ".join(f'"{t}"' for t in _recent[-20:]) _excl = f"\n\nDO NOT suggest any of these recently used topics: {_excl_list}\nChoose completely different subjects." TYPE_DICE_PROMPTS = { "origin": f"""Today is {today_month_day}. Generate 5 fascinating origin stories — inventions, discoveries, or practices — that are connected to this date OR are simply among the most surprising and little-known origin stories in history. For each, give: a short descriptive title (the thing being explained), and the approximate year of origin. Return ONLY a JSON array, no markdown: [{{"event": "The invention of X", "year": "YYYY", "category": "Origin Story"}}] Focus on variety: technology, food, language, medicine, daily habits. Be specific.{_excl}""", "mystery": f"""Today is {today_month_day}. Generate 5 compelling true mysteries or crimes — unsolved cases, historical puzzles, or famous crimes — connected to this date OR among the most gripping documented mysteries in history. For each, give: a short descriptive title and the approximate year. Return ONLY a JSON array, no markdown: [{{"event": "The disappearance of X", "year": "YYYY", "category": "True Mystery"}}] Focus on variety: crimes, disappearances, historical puzzles, unsolved deaths.{_excl}""", "science": f"""Today is {today_month_day}. Generate 5 fascinating science topics — natural phenomena, biological mechanisms, physics concepts, or chemistry processes — that are connected to this date OR are among the most counterintuitive and surprising things science has discovered. For each, give: a short descriptive title and the approximate year of key discovery. Return ONLY a JSON array, no markdown: [{{"event": "Why X happens", "year": "YYYY", "category": "Science"}}] Focus on variety: biology, physics, chemistry, astronomy, earth science.{_excl}""", "explore": f"""Today is {today_month_day}. Generate 5 extraordinary places, creatures, or phenomena that EXIST RIGHT NOW and that most people have never heard of — hidden corners of Earth, deep-sea wonders, space discoveries, vanished civilizations, strange animals, extreme environments, or little-known living cultures. For each, give: a short descriptive title and the approximate year of discovery/documentation. Return ONLY a JSON array, no markdown: [{{"event": "The bioluminescent bay of X", "year": "YYYY", "category": "Exploration"}}] Focus on variety: deep ocean, space, remote geography, biology, archaeology, anthropology.{_excl}""", } prompt = TYPE_DICE_PROMPTS.get(story_type) if not prompt: return None try: raw = await api_clients.call_groq_chat_completion( api_key=api_key, system_prompt="You are a creative researcher. Return only valid JSON arrays.", user_prompt=prompt, max_tokens=700, temperature=0.95, model="llama-3.3-70b-versatile", ) import json as _json text = raw.strip().lstrip("```json").lstrip("```").rstrip("```").strip() events = _json.loads(text) if not isinstance(events, list) or not events: return None # Filter out any recently used topics _recent_lower = [r.lower() for r in _recent] fresh = [e for e in events if e.get("event","").lower() not in _recent_lower] chosen = random.choice(fresh[:5] if fresh else events[:5]) _ok( f"[Dice/{story_type}] → '{chosen.get('event','?')}' " f"({chosen.get('year','?')})" ) return chosen except Exception as exc: _warn(f"[Dice/{story_type}] Failed: {exc}") 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 _run_historian( self, dice_event: dict, category: str, lang_focus: str, today_month_day: str, story_type: str = "historical", ) -> str: """ V22.0 — Step 0.5: The Compound Researcher (groq/compound-beta). UPGRADE from V19.0: - Model changed: openai/gpt-oss-120b → groq/compound-beta - compound-beta has LIVE web search built-in (autonomous agent) - Uses COMPOUND_RESEARCHER_SYSTEM_PROMPTS per story_type - Returns Markdown dossier WITH real Wikimedia image URLs + ESL vocab - Fallback: if compound fails 3 times, falls back to gpt-oss-120b The dossier is passed to: 1. The Architect (llama-4-scout) for JSON structuring 2. prompts.parse_compound_image_urls() to extract real_image_urls """ focus_map = { "ar": "Focus on events from Arab, Islamic, Berber, or broader Middle Eastern history.", "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"]) # V22.0: Use compound-specific system prompts _sys = prompts.COMPOUND_RESEARCHER_SYSTEM_PROMPTS.get( story_type, prompts.COMPOUND_RESEARCHER_SYSTEM_PROMPTS["historical"] ) # Build user prompt — same structure as before _usr_tpl = prompts.HISTORIAN_PROMPT_TEMPLATES.get(story_type) or prompts.HISTORIAN_PROMPT_TEMPLATE _fmt_args = dict( event=dice_event.get("event", ""), year=dice_event.get("year", ""), category=category, lang_focus=lang_focus, today_month_day=today_month_day, ) if story_type == "historical": _fmt_args["focus_instruction"] = focus_instruction user_prompt = _usr_tpl.format(**_fmt_args) # ── Attempt 1-3: groq/compound-beta (live web search) ───── last_error = None for _att in range(1, 4): _key = await self._acquire_groq_role("maestro", "CompoundResearcher") try: _info( f"[CompoundResearcher/{story_type}] attempt {_att}/3 | " f"model=groq/compound-beta | key=\u2026{_key[-4:]}" ) dossier = await api_clients.call_compound_researcher( api_key=_key, system_prompt=_sys, user_prompt=user_prompt, caller=f"CompoundResearcher/{story_type}", ) if len(dossier) < 200: raise ValueError(f"Dossier too short: {len(dossier)} chars") _ok( f"[CompoundResearcher] \u2705 Dossier ready — {len(dossier)} chars | " f"type={story_type}" ) return dossier except GroqRateLimitError: await self._mark_groq_rl(_key, "CompoundResearcher") last_error = "RateLimit" except Exception as _exc: _es = str(_exc) if "500" in _es or "502" in _es or "503" in _es: await self._mark_groq_server_error(_key, "CompoundResearcher") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * _att) last_error = _es _warn(f"[CompoundResearcher] attempt {_att} error: {_es[:100]}") # ── Fallback: gpt-oss-120b (no web search) ──────────────── _warn( "[CompoundResearcher] compound-beta failed 3 times — " "falling back to openai/gpt-oss-120b (no web search)" ) _sys_fallback = prompts.HISTORIAN_SYSTEM_PROMPTS.get(story_type) or prompts.HISTORIAN_SYSTEM_PROMPT for _att in range(1, 3): _key = await self._acquire_groq_role("maestro", "Historian/fallback") try: _info( f"[Historian/fallback] attempt {_att}/2 | " f"model=openai/gpt-oss-120b | key=\u2026{_key[-4:]}" ) response = await api_clients.call_groq_chat_completion( api_key=_key, system_prompt=_sys_fallback, user_prompt=user_prompt, max_tokens=4000, temperature=0.3, model="openai/gpt-oss-120b", ) dossier = response.strip() if len(dossier) < 200: raise ValueError(f"Fallback dossier too short: {len(dossier)} chars") _ok(f"[Historian/fallback] \u2705 Fallback dossier ready — {len(dossier)} chars") return dossier except GroqRateLimitError: await self._mark_groq_rl(_key, "Historian/fallback") last_error = "RateLimit" except Exception as _exc: _es = str(_exc) if "500" in _es or "502" in _es or "503" in _es: await self._mark_groq_server_error(_key, "Historian/fallback") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * _att) last_error = _es _warn(f"[Historian/fallback] attempt {_att} error: {_es[:100]}") raise HTTPException(502, f"Historian (compound + fallback) failed: {last_error}") async def _generate_screenplay_outline( self, category: str, cat_instruction: str, lang_focus: str, date_anchor: str, today_month_day: str, historian_notes: str = "", ) -> Dict[str, Any]: """ V19.0 — Architect now uses llama-4-scout (JSON formatting only). Reads the Historian dossier and structures it into JSON. llama-4-scout TPM=30K vs gpt-oss-120b TPM=8K → fewer 400 errors. """ focus_map = { "ar": "Focus on events from Arab, Islamic, Berber, or broader Middle Eastern history.", "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"]) # V20.0: inject story_type instruction into cat_instruction _st = getattr(self, "_current_story_type", "historical") TYPE_ARCHITECT_INSTRUCTIONS = { "explore": ( "\n\nSTORY TYPE — EXPLORER / DISCOVERY:" "\nThis is a documentary about a real place, creature, phenomenon, or culture" "\nthat exists RIGHT NOW or was recently discovered." "\nAct I: Arrival — put the reader there. Coordinates, sensory details, scale." "\nAct II: The extraordinary — what science has found, what lives there, what happens." "\nAct III: The mystery and significance — what is still unknown, why it matters." "\nNever use past tense for things that still exist — write as if the camera is rolling NOW." "\nEvery fact must be documentable — expedition report, scientific paper, satellite data." ), "mystery": ( "\n\nSTORY TYPE — TRUE MYSTERY / CRIME:" "\nThis is a forensic investigation documentary, NOT a history lecture." "\nThe 3 acts must follow: Setup (the crime/disappearance) → " "Investigation (evidence, suspects, theories) → Unresolved (what remains unknown)." "\nEvery chapter must feel like a detective case file — facts first, questions last." "\nThe final act MUST end with what is STILL unsolved and WHY." ), "science": ( "\n\nSTORY TYPE — SCIENCE EXPLAINER:" "\nThis is a science documentary, NOT a history of science." "\nAct I: The phenomenon and why it puzzled scientists." "\nAct II: The mechanism — HOW it actually works at a physical/biological level." "\nAct III: Real-world implications and what is still unknown." "\nEvery chapter must explain the MECHANISM, not just list discoveries." ), "origin": ( "\n\nSTORY TYPE — ORIGIN STORY:" "\nThis is a 'how it began' documentary." "\nAct I: The world BEFORE — the need/problem/accident that started everything." "\nAct II: The moment of invention/discovery and how it spread." "\nAct III: How it transformed the world and what it looks like today." "\nFocus on human stories and surprising facts, not dry chronology." ), } _type_instr = TYPE_ARCHITECT_INSTRUCTIONS.get(_st, "") _full_cat = cat_instruction + _type_instr prompt = prompts.ARCHITECT_PROMPT_TEMPLATE.format( today_month_day=today_month_day, category=category, cat_instruction=_full_cat, focus_instruction=focus_instruction, historian_notes=historian_notes or "(No dossier — use your own verified knowledge)", ) _REQUIRED = {"title_en", "historical_era", "chapters"} last_schema_error = None for _att in range(1, 7): _key = await self._acquire_groq_role("generation", "Architect") try: _info( f"[Architect] Groq attempt {_att}/6 | " f"role=generation | model=meta-llama/llama-4-scout-17b-16e-instruct | key=\u2026{_key[-4:]}" ) _data = await api_clients.call_groq_json( api_key=_key, prompt=prompt, caller="Architect", max_tokens=8000, temperature=0.2, model="meta-llama/llama-4-scout-17b-16e-instruct", ) _missing = _REQUIRED - _data.keys() if _missing: raise ValueError(f"Missing root keys: {_missing}") _chs = _data.get("chapters", []) if not isinstance(_chs, list) or not _chs: raise ValueError("chapters is empty or not a list") _ACT_DEFAULTS = { 0: {"chapter_number": 1, "act_title": "Act I -- The Setup / The Shock"}, 1: {"chapter_number": 2, "act_title": "Act II -- Mechanics & Consequence"}, 2: {"chapter_number": 3, "act_title": "Act III -- The Verified Aftermath"}, } _repaired = False for _ci, _ch in enumerate(_chs): if not isinstance(_ch, dict): _warn(f"[Architect] chapters[{_ci}] is {type(_ch).__name__} — repairing") _defaults = _ACT_DEFAULTS.get(_ci, {"chapter_number": _ci+1, "act_title": f"Chapter {_ci+1}"}) _str_content = str(_ch) if isinstance(_ch, str) else "" import re as _re3 _pseudo_aps = [ s.strip() for s in _re3.split(r"[.!?]\s+", _str_content) if len(s.strip()) > 20 ][:8] _chs[_ci] = { "chapter_number": _defaults["chapter_number"], "act_title": _defaults["act_title"], "summary": _str_content[:300], "action_points": _pseudo_aps, "plot_points": [], "location": _chs[0].get("location","") if isinstance(_chs[0],dict) else "", "ambiance": _chs[0].get("ambiance","") if isinstance(_chs[0],dict) else "", "sensory_cue": "", } _repaired = True if _repaired: _data["chapters"] = _chs # story_bgm validation + inference (always runs) _VALID_BGM = { "bgm_epic.mp3","bgm_suspense.mp3","bgm_inspiring.mp3", "bgm_melancholic.mp3","bgm_classical.mp3","bgm_ancient.mp3", } if _data.get("story_bgm") not in _VALID_BGM: _ctx = ( _data.get("historical_era","") + " " + _data.get("story_premise","") + " " + historian_notes[:500] ).lower() if any(w in _ctx for w in ["war","battle","military","conquest","siege","revolt","revolution","invasion","crusade"]): _bgm = "bgm_epic.mp3" elif any(w in _ctx for w in ["assassination","conspiracy","espionage","intrigue","plot","betrayal","coup"]): _bgm = "bgm_suspense.mp3" elif any(w in _ctx for w in ["discovery","exploration","founding","breakthrough","invention","reform","renaissance"]): _bgm = "bgm_inspiring.mp3" elif any(w in _ctx for w in ["collapse","defeat","disaster","fall","decline","tragedy","famine","plague"]): _bgm = "bgm_melancholic.mp3" elif any(w in _ctx for w in [" bc"," bce","ancient","babylon","persia","egypt","greece","rome","mesopotamia"]): _bgm = "bgm_ancient.mp3" else: _bgm = "bgm_classical.mp3" _data["story_bgm"] = _bgm _warn(f"[Architect] story_bgm inferred \u2192 {_bgm}") else: _ok(f"[Architect] story_bgm from LLM \u2192 {_data['story_bgm']}") if _repaired: _warn("[Architect] Repaired malformed chapters — pipeline continues") return _data except GroqRateLimitError: await self._mark_groq_rl(_key, "Architect") last_schema_error = "RateLimit" except Exception as _exc: _es = str(_exc) if "500" in _es or "502" in _es or "503" in _es: await self._mark_groq_server_error(_key, "Architect") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * _att) last_schema_error = _es _warn(f"[Architect] attempt {_att} error: {_es[:120]} — retrying") raise HTTPException(502, f"Architect failed after 6 attempts: {last_schema_error}") 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." ) # V20.0: per-type Narrator persona _nov_override = prompts.NOVELIST_SYSTEM_PROMPTS.get( getattr(self, "_current_story_type", "historical") ) system_prompt = _nov_override or 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) # V21.0 — Normalize paragraph breaks: # Ensure \n\n separates paragraphs (Groq sometimes returns \n only). # Replace 3+ newlines with double newline, but keep double newlines intact. content = _re.sub(r'\n{3,}', '\n\n', 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" ) # V18.9 — Repetition guard: detect looping sentences import re as _re2 _sents = [ s.strip() for s in _re2.split(r"[.!?،؟]\s+", content) if len(s.strip()) > 40 ] if _sents: from collections import Counter as _Counter _counts = _Counter(_sents) _max_repeat = _counts.most_common(1)[0][1] if _max_repeat >= 3: raise ValueError( f"[Novelist/Ch{chapter_num}] Repetition loop detected " f"(sentence repeated {_max_repeat}x) — 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 Extracts cinematic_image_prompt only (audio_environment removed in V16.3). 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], "pexels_search_query": ch_outline.get("location", ""), } 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)") # Derive audio_environment from outline ambiance (Architect always sets this) _ambiance = ch_outline.get("ambiance", "monastery") return { "chapter_number": chapter_num, "act_title": act_title, "text_en": text_en, "cinematic_image_prompt": metadata.get("cinematic_image_prompt", ""), "pexels_search_query": metadata.get("pexels_search_query", ""), "audio_environment": { "primary_ambience": _ambiance, "intensity": "medium", "dynamic_cues": ch_outline.get("sensory_cue", ""), "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]: """ V21.0 — Fixed re-lettering bug. Previous version's _reorder() preserved original A/B/C/D letter prefixes after shuffling, producing malformed lists like [C. opt, B. opt, D. opt, A. opt]. Fix: strip all letter prefixes BEFORE shuffling, reorder by permutation, then re-apply A/B/C/D sequentially. Correct answer tracked by bare content. English canonical permutation applied to all other language variants identically. """ import re as _re quiz = dict(quiz) letters = ["A", "B", "C", "D"] def _strip_prefix(opt: str) -> str: return _re.sub(r"^[A-Da-d][.\)]\s*", "", opt).strip() def _apply_perm(options: List[str], perm: List[int]) -> List[str]: """Strip prefixes, reorder, re-letter A/B/C/D in sequence.""" stripped = [_strip_prefix(o) for o in options] reordered = [stripped[i] for i in perm] return [f"{letters[pos]}. {reordered[pos]}" for pos in range(4)] def _find_correct(correct_bare: str, new_opts: List[str]) -> str: for opt in new_opts: if _strip_prefix(opt) == correct_bare: return opt return new_opts[0] if new_opts else correct_bare en_options = quiz.get("options_en", []) if len(en_options) != 4: return quiz perm = list(range(4)) random.shuffle(perm) en_correct_bare = _strip_prefix(quiz.get("correct_answer_en", "")) en_new = _apply_perm(en_options, perm) quiz["options_en"] = en_new quiz["correct_answer_en"] = _find_correct(en_correct_bare, en_new) 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_correct_bare = _strip_prefix(quiz.get(correct_key, "")) lc_new = _apply_perm(lc_options, perm) quiz[opts_key] = lc_new quiz[correct_key] = _find_correct(lc_correct_bare, lc_new) 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, compound_img_urls: list = None, # V22.0: real Wikimedia URLs from compound-beta ) -> 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": "V22.0-CompoundResearcher", "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 } _img_urls = compound_img_urls or [] # V22.0 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) # V22.0: attach real_image_url from compound dossier (index = chapter_num - 1) _real_img = _img_urls[num - 1] if (num - 1) < len(_img_urls) else "" mega_chapters.append({ "chapter_number": num, "act_title": en_ch.get("act_title", ""), "cinematic_image_prompt": en_ch.get("cinematic_image_prompt", ""), "pexels_search_query": en_ch.get("pexels_search_query", ""), "audio_environment": en_ch.get("audio_environment", {}), "real_image_url": _real_img, # V22.0: real Wikimedia URL from compound "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, story_type_override: Optional[str] = None, custom_topic: Optional[str] = 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 # V23.0: one story per hour — no slot snapping 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: story_type from override (scheduler) or default ---------- # The scheduler now sends story_type directly — no rotation formula needed VALID_TYPES = {"historical","origin","mystery","science","explore"} story_type = story_type_override if story_type_override in VALID_TYPES else "historical" self._current_story_type = story_type _info(f"[Step0] story_type={story_type} | {today_month_day}") dice_key = await self._acquire_groq_role("maestro", "HistoryDice") if custom_topic: dice_event = {"event": custom_topic, "year": "", "category": story_type} _ok(f"[Step0] custom_topic -> {custom_topic}") else: dice_event = await roll_dice_for_type(dice_key, story_type, today_month_day) if dice_event: category = dice_event.get("category", story_type) _event_str = dice_event.get("event", "") _year_str = dice_event.get("year", "") _type_label = prompts.STORY_TYPES[story_type]["label_en"] cat_instruction = ( f"Story type: {_type_label}. " f"Topic: \"{_event_str}\" ({_year_str}). " f"Build the entire 3-act outline around this topic." ) _ok(f"[Step0/Dice/{story_type}] {_event_str} ({_year_str})") # Save to dedup list if _event_str: await _save_topic(story_type, _event_str) else: _cat_idx, category, cat_instruction = _pick_category(hour_slot, day_of_year) _info(f"[Step0/fallback] static category: {category}") _info(f"Step 0 (Dice fallback) → static category: {category}") _info( f"🚀 V22.0 CompoundResearcher+Architect Pipeline START | {story_id} | " f"slot={hour_slot:02d} | cat={category} | focus={lang_focus}" ) # ── Step 0.5: The Compound Researcher (groq/compound-beta) ── # V22.0: compound-beta searches the web autonomously. # Dossier contains: verified facts + ESL vocab + 3 Wikimedia URLs. t05 = time.monotonic() historian_notes = "" compound_img_urls: list = [] # V22.0: real Wikimedia URLs from compound if dice_event: try: historian_notes = await self._run_historian( dice_event=dice_event, category=category, lang_focus=lang_focus, today_month_day=today_month_day, story_type=story_type, ) # V23.1: extract WIKI_QUERY strings from compound dossier # (compound now returns search queries, not fake URLs) compound_img_urls = prompts.parse_compound_image_urls(historian_notes) _ok( f"Step 0.5 (CompoundResearcher) done ({time.monotonic()-t05:.1f}s) | " f"{len(historian_notes)} chars | " f"{len(compound_img_urls)} wiki queries found" ) except Exception as _hist_exc: _warn(f"Step 0.5 (CompoundResearcher) failed: {_hist_exc} — Architect will use own knowledge") historian_notes = "" compound_img_urls = [] # ── Step 1: The Architect (llama-4-scout) ────────────────── # JSON formatting only — reads Historian dossier and structures it. # llama-4-scout TPM=30K → fewer 400 errors than gpt-oss-120b. 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, historian_notes=historian_notes, ) 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 # ── V18.7: Chapters integrity guard ──────────────────────────── _outline_chapters = outline.get("chapters", []) if not _outline_chapters or not isinstance(_outline_chapters[0], dict): _warn("[ChaptersGuard] outline['chapters'] malformed — re-running Architect") 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, historian_notes=historian_notes, ) _retry_chapters = outline.get("chapters", []) if not _retry_chapters or not isinstance(_retry_chapters[0], dict): raise ValueError("[ChaptersGuard] Architect returned malformed chapters after retry") # ── 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, compound_img_urls=compound_img_urls, # V22.0: real Wikimedia URLs ) 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"] = "V22.0-CompoundResearcher" # V20.0: story type metadata _st = getattr(self, "_current_story_type", "historical") mega["story_type"] = _st mega["story_type_label"] = prompts.STORY_TYPES.get(_st, {}).get("label_en", "") mega["story_type_emoji"] = prompts.STORY_TYPES.get(_st, {}).get("emoji", "") mega["story_type_color"] = prompts.STORY_TYPES.get(_st, {}).get("color", "coral") # story_bgm from outline mega["story_bgm"] = outline.get("story_bgm", "bgm_classical.mp3") 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 # ══════════════════════════════════════════════════════════════ # TITAN MENTOR — Chunking & Load Balancing Pipeline # Route: POST /api/v1/immersion/generate-lesson # # Three-agent sequential pipeline: # Agent 1 — Architect (Gemini/validation): splits raw text → 3-5 chapters JSON # Agent 2 — Lecturer (Gemini/enrichment): audio script ~800 words per chapter # Agent 3 — Translator (Gemini/translation): translates script to target_lang # # Load balancing: each agent uses a different Gemini role key pool, # distributing load across all available keys. # All story pipeline code above is UNTOUCHED. # ══════════════════════════════════════════════════════════════ async def generate_mentor_lesson( self, text: str, subject_name: str, subject_type: str, target_lang: str, ) -> Dict[str, Any]: """ Titan Mentor three-agent pipeline. Step 1 — Architect (Gemini/validation keys): Splits raw text into 3-5 logical chapters → JSON array. Image placeholders [IMAGE_001] preserved in source_text. Step 2 — For each chapter (SEQUENTIAL, different Gemini key pools): a. Lecturer (Gemini/enrichment) → audio script ~800 words b. Translator(Gemini/translation) → translated script Step 3 — Returns structured JSON ready for immersion.py. Returns: { "subject_name", "subject_type", "target_lang", "total_chapters", "chapters": [ { "chapter_number", "title", "source_text", "script_en", "script_translated", "target_lang", "audio_url_en": "", # filled by immersion.py "audio_url_translated": "" # filled by immersion.py }, ... ] } """ import re as _re_mentor # ── Count [IMAGE_NNN] placeholders in raw text ───────────── image_count = len(_re_mentor.findall(r'\[IMAGE_\d{3}\]', text)) _info( f"[MentorArchitect] Starting | subject={subject_name} | " f"type={subject_type} | lang={target_lang} | images={image_count}" ) # ══════════════════════════════════════════════════════════ # Step 1 — Architect: split text into chapters # Uses Gemini/validation key pool (GOOGLE_KEY_4-5) # ══════════════════════════════════════════════════════════ architect_prompt = prompts.MENTOR_ARCHITECT_PROMPT.format( subject_name=subject_name, image_count=image_count, raw_text=text, ) chapters_data: Dict[str, Any] = {} last_err: Optional[str] = None for _att in range(1, 5): _key = await self._acquire_gemini_role("validation", "MentorArchitect") try: _info(f"[MentorArchitect] attempt {_att}/4 | key=…{_key[-4:]}") chapters_data = await api_clients.call_gemini_json( api_key=_key, prompt=architect_prompt, caller="MentorArchitect", max_tokens=6000, temperature=0.2, ) _raw_chs = chapters_data.get("chapters", []) if not _raw_chs or not isinstance(_raw_chs, list): raise ValueError( f"Architect returned empty or invalid chapters: {str(chapters_data)[:120]}" ) _ok( f"[MentorArchitect] ✅ {len(_raw_chs)} chapters extracted | " f"subject={subject_name}" ) break except Exception as _exc: _es = str(_exc) if "429" in _es or "rate" in _es.lower() or "quota" in _es.lower(): await self._mark_gemini_rl(_key, "MentorArchitect") elif "500" in _es or "502" in _es or "503" in _es or "server_error" in _es.lower(): await self._mark_gemini_server_error(_key, "MentorArchitect") else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * _att) last_err = _es _warn(f"[MentorArchitect] attempt {_att} error: {_es[:120]}") else: raise HTTPException( 502, f"MentorArchitect failed after 4 attempts: {last_err}" ) raw_chapters: List[Dict[str, Any]] = chapters_data.get("chapters", []) # ══════════════════════════════════════════════════════════ # Step 2 — Loop: Lecturer + Translator per chapter # Sequential (each chapter's script feeds the translator) # ══════════════════════════════════════════════════════════ result_chapters: List[Dict[str, Any]] = [] for ch in raw_chapters: ch_num = ch.get("chapter_number", len(result_chapters) + 1) ch_title = ch.get("title", f"Chapter {ch_num}") ch_source = ch.get("source_text", "") _info(f"[MentorPipeline] ── Ch{ch_num}: '{ch_title}' ──") # ── 2a: Lecturer — write audio script ───────────────── # Uses Gemini/enrichment key pool (GOOGLE_KEY_1-3) lecturer_prompt = prompts.MENTOR_LECTURER_PROMPT.format( subject_type=subject_type, subject_name=subject_name, chapter_number=ch_num, chapter_title=ch_title, source_text=ch_source, ) chapter_script = "" last_err = None for _att in range(1, 5): _key = await self._acquire_gemini_role( "enrichment", f"MentorLecturer/Ch{ch_num}" ) try: _info( f"[MentorLecturer/Ch{ch_num}] attempt {_att}/4 | " f"key=…{_key[-4:]}" ) raw_script = await api_clients.call_gemini_text( api_key=_key, prompt=lecturer_prompt, caller=f"MentorLecturer/Ch{ch_num}", max_tokens=2000, temperature=0.4, ) _word_count = len(raw_script.split()) if _word_count < 100: raise ValueError( f"Lecturer output too short: {_word_count} words" ) chapter_script = raw_script.strip() _ok( f"[MentorLecturer/Ch{ch_num}] ✅ {_word_count} words" ) break except Exception as _exc: _es = str(_exc) if "429" in _es or "rate" in _es.lower() or "quota" in _es.lower(): await self._mark_gemini_rl( _key, f"MentorLecturer/Ch{ch_num}" ) elif "500" in _es or "502" in _es or "503" in _es or "server_error" in _es.lower(): await self._mark_gemini_server_error( _key, f"MentorLecturer/Ch{ch_num}" ) else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * _att) last_err = _es _warn( f"[MentorLecturer/Ch{ch_num}] attempt {_att} error: {_es[:120]}" ) else: # Non-fatal: fall back to raw source text so pipeline continues _warn( f"[MentorLecturer/Ch{ch_num}] All attempts failed — " f"using source_text as fallback script" ) chapter_script = ch_source # ── 2b: Translator — translate audio script ──────────── # Uses Gemini/translation key pool (GOOGLE_KEY_1-5) # Skipped when target_lang is English or empty translated_script = "" _skip_translation = target_lang.strip().lower() in ( "", "en", "english", "anglais" ) if not _skip_translation and chapter_script: translator_prompt = prompts.MENTOR_TRANSLATOR_PROMPT.format( target_lang=target_lang, subject_name=subject_name, chapter_number=ch_num, chapter_script=chapter_script, ) last_err = None for _att in range(1, 5): _key = await self._acquire_gemini_role( "translation", f"MentorTranslator/Ch{ch_num}" ) try: _info( f"[MentorTranslator/Ch{ch_num}] attempt {_att}/4 | " f"lang={target_lang} | key=…{_key[-4:]}" ) raw_trans = await api_clients.call_gemini_text( api_key=_key, prompt=translator_prompt, caller=f"MentorTranslator/Ch{ch_num}", max_tokens=2500, temperature=0.15, ) _trans_words = len(raw_trans.split()) if _trans_words < 50: raise ValueError( f"Translation too short: {_trans_words} words" ) translated_script = raw_trans.strip() _ok( f"[MentorTranslator/Ch{ch_num}] ✅ {_trans_words} words" ) break except Exception as _exc: _es = str(_exc) if "429" in _es or "rate" in _es.lower() or "quota" in _es.lower(): await self._mark_gemini_rl( _key, f"MentorTranslator/Ch{ch_num}" ) elif "500" in _es or "502" in _es or "503" in _es or "server_error" in _es.lower(): await self._mark_gemini_server_error( _key, f"MentorTranslator/Ch{ch_num}" ) else: await asyncio.sleep(RETRY_BACKOFF_SECONDS * _att) last_err = _es _warn( f"[MentorTranslator/Ch{ch_num}] attempt {_att} error: {_es[:120]}" ) else: _warn( f"[MentorTranslator/Ch{ch_num}] All attempts failed — " f"translated_script will be empty" ) result_chapters.append({ "chapter_number": ch_num, "title": ch_title, "source_text": ch_source, "script_en": chapter_script, "script_translated": translated_script, "target_lang": target_lang, "audio_url_en": "", # filled by immersion.py after TTS "audio_url_translated": "", # filled by immersion.py after TTS }) _ok(f"[MentorPipeline] Ch{ch_num} complete ✅") # ══════════════════════════════════════════════════════════ # Step 3 — Return final structured payload # ══════════════════════════════════════════════════════════ _ok( f"[MentorPipeline] 🎓 All {len(result_chapters)} chapters done | " f"subject={subject_name}" ) return { "subject_name": subject_name, "subject_type": subject_type, "target_lang": target_lang, "total_chapters": len(result_chapters), "chapters": result_chapters, } # ────────────────────────────────────────────────────────────────── # 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": "V22.0-CompoundResearcher", "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)