app/services/learn_lesson_builder.py

"""Turn one Learn Anything lesson into a playable, browser-ready lesson.

This is the piece that makes "Learn Anything" work economically. A lesson is
NOT rendered to an MP4 on a GPU. Instead we generate a lightweight lesson
"manifest": a continuous teaching script broken into narration beats, each
with a board (heading + lines) to show while it is spoken, plus per-beat
audio. The student's browser "plays" it like a class — audio + synced board —
so there is no video render cost at all.

Cost control: lessons are keyed by their content, source context, and owner.
Sensitive uploads therefore cannot select another student's cached manifest or
audio.

Voice: Deepgram Aura for English / English-medium ("Manglish") lessons. A
Malayalam-medium lesson routes to the local AI4Bharat provider instead
(Deepgram has no Malayalam voice) — see ``synthesize_beats``.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import re
import subprocess
import threading
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from app.core.config import BACKEND_DIR, PROJECT_ROOT, get_settings

logger = logging.getLogger(__name__)

# Writable cache for lesson manifests + audio.
# NEVER use PROJECT_ROOT/"public" — on HF Docker WORKDIR=/app, that resolves to
# /public which is not creatable (Permission denied → 500).
# Prefer: LEARN_LESSON_CACHE_DIR → /app/generated/... → backend/generated/...
# Tests may monkeypatch PUBLIC_ROOT.
PUBLIC_ROOT = BACKEND_DIR / "generated" / "learn-anything"

GROQ_URL = "https://api.groq.com/openai/v1/chat/completions"
# llama-4-scout was retired from Groq; 3.3-70b is the strongest current chat model.
GROQ_MODEL = "llama-3.3-70b-versatile"
DEEPGRAM_SPEAK_URL = "https://api.deepgram.com/v1/speak"

# ~10 minutes of natural speech is roughly 1200-1500 spoken words. The model
# tends to write short beats, so we ask for more of them to reach real length.
TARGET_BEATS = 16
MIN_BEATS = 12
MAX_BEATS = 20

# The lesson must remain usable even when a third-party provider or voice
# service is slow.  These are deliberately conservative defaults; deployments
# may lower them, but cannot raise them past the safe caps below.
DEFAULT_LESSON_LLM_TIMEOUT_SECONDS = 20.0
DEFAULT_LESSON_LLM_BUDGET_SECONDS = 35.0
DEFAULT_LESSON_AUDIO_BUDGET_SECONDS = 20.0
DEFAULT_LESSON_TTS_TIMEOUT_SECONDS = 15.0

# Prevent a user from opening the same uncached lesson repeatedly and fanning
# out parallel authoring jobs.
_lesson_build_locks: dict[str, threading.Lock] = {}
_lesson_build_locks_guard = threading.Lock()


class LessonBuildError(RuntimeError):
    pass


def _bounded_seconds(name: str, default: float, maximum: float) -> float:
    raw = _env(name)
    try:
        configured = float(raw) if raw else default
    except (TypeError, ValueError):
        configured = default
    return max(1.0, min(configured, maximum))


def _lesson_llm_timeout_seconds() -> float:
    return _bounded_seconds(
        "LEARN_LESSON_LLM_TIMEOUT_SECONDS",
        DEFAULT_LESSON_LLM_TIMEOUT_SECONDS,
        DEFAULT_LESSON_LLM_TIMEOUT_SECONDS,
    )


def _lesson_generation_budget_seconds() -> float:
    return _bounded_seconds(
        "LEARN_LESSON_GENERATION_BUDGET_SECONDS",
        DEFAULT_LESSON_LLM_BUDGET_SECONDS,
        DEFAULT_LESSON_LLM_BUDGET_SECONDS,
    )


def _lesson_sync_tts_enabled() -> bool:
    return _env("LEARN_LESSON_SYNC_TTS").lower() in {"1", "true", "yes", "on"}


@dataclass
class LessonBeat:
    kind: str  # hook | explain | example | analogy | checkpoint | recap
    narration: str
    board_heading: str
    board_lines: list[str] = field(default_factory=list)
    visual_hint: str = ""
    audio_src: str = ""
    start_second: float = 0.0
    duration_seconds: float = 0.0


def _clean_env_value(raw: str) -> str:
    return raw.strip().strip('"').strip("'").strip()


def _env(name: str) -> str:
    value = os.getenv(name)
    if value:
        return _clean_env_value(value)
    for filename in (".env.local", ".env", "backend/.env"):
        path = PROJECT_ROOT / filename
        if not path.exists():
            continue
        try:
            raw = path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            raw = path.read_text(encoding="utf-8", errors="replace")
        for line in raw.splitlines():
            if line.startswith(f"{name}="):
                return _clean_env_value(line.split("=", 1)[1])
    return ""


def lesson_hash(
    topic: str,
    lesson_title: str,
    level: str,
    medium: str,
    voice: str,
    *,
    context: str = "",
    user_id: str = "",
) -> str:
    payload = json.dumps(
        {
            "topic": topic.strip().lower(),
            "lesson": lesson_title.strip().lower(),
            "level": level.strip().lower(),
            "medium": medium.strip().lower(),
            "voice": voice.strip().lower(),
            "context": hashlib.sha256(context.encode("utf-8")).hexdigest(),
            "user": user_id,
            "schema": "learn-lesson-v2-private",
        },
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:20]


def _call_openai_compatible_json(
    *,
    url: str,
    api_key: str,
    model: str,
    prompt: str,
    max_tokens: int = 8000,
    provider_label: str,
    timeout: float | None = None,
) -> dict[str, Any]:
    """OpenAI-shaped chat completions that return a JSON object body."""
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.35,
        "max_tokens": max_tokens,
        "response_format": {"type": "json_object"},
    }
    request = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}",
            # Some CDN edges 403 urllib's default UA.
            "User-Agent": "DocDoe-LearnLesson/1.0",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(
            request,
            timeout=timeout if timeout is not None else _lesson_llm_timeout_seconds(),
        ) as response:
            result = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body = ""
        try:
            body = exc.read().decode("utf-8", errors="replace")[:400]
        except Exception:
            body = ""
        raise LessonBuildError(
            f"{provider_label} HTTP {exc.code}: {body or exc.reason}"
        ) from exc
    content = result["choices"][0]["message"]["content"]
    if isinstance(content, list):
        # Some providers return content parts; join text pieces.
        content = "".join(
            part.get("text", "") if isinstance(part, dict) else str(part)
            for part in content
        )
    return json.loads(content)


def _call_groq(
    prompt: str,
    api_key: str,
    max_tokens: int = 8000,
    timeout: float | None = None,
) -> dict[str, Any]:
    return _call_openai_compatible_json(
        url=GROQ_URL,
        api_key=api_key,
        model=GROQ_MODEL,
        prompt=prompt,
        max_tokens=max_tokens,
        provider_label="Groq",
        timeout=timeout,
    )


def _call_zai(
    prompt: str,
    api_key: str,
    max_tokens: int = 8000,
    timeout: float | None = None,
) -> dict[str, Any]:
    """Free Z.AI GLM-4.7-Flash for lesson authoring (OpenAI-compatible)."""
    base = (_env("ZAI_BASE_URL") or _env("Z_AI_BASE_URL") or "https://api.z.ai/api/paas/v4").rstrip("/")
    model = _env("ZAI_MODEL_TEXT") or _env("GLM_MODEL_TEXT") or "glm-4.7-flash"
    # Prefer the shared client when settings are loaded; only fall back for an
    # unavailable optional import.  Falling back after a timeout would make
    # one slow provider call twice as long and defeat the lesson SLA.
    try:
        from app.services.zai_client import chat_completions, extract_message_text
    except ImportError:
        return _call_openai_compatible_json(
            url=f"{base}/chat/completions",
            api_key=api_key,
            model=model,
            prompt=prompt,
            max_tokens=max_tokens,
            provider_label="Z.AI",
            timeout=timeout,
        )
    result = chat_completions(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
        temperature=0.35,
        response_format={"type": "json_object"},
        thinking_disabled=True,
        timeout=timeout,
    )
    content = extract_message_text(result)
    return json.loads(content)


def _script_llm_providers() -> list[dict[str, Any]]:
    """Lesson-script providers: free Z.AI Flash first, then Groq."""
    providers: list[dict[str, Any]] = []
    zai = _env("ZAI_API_KEY") or _env("Z_AI_API_KEY") or _env("GLM_API_KEY")
    if zai and len(zai) >= 16:
        providers.append(
            {
                "name": "zai",
                "key": zai,
                "call": lambda prompt, timeout=None, key=zai: _call_zai(
                    prompt, key, timeout=timeout or _lesson_llm_timeout_seconds()
                ),
            }
        )
    groq = _env("GROQ_API_KEY")
    if groq.startswith(("gsk-", "gsk_")) and len(groq) >= 20:
        providers.append(
            {
                "name": "groq",
                "key": groq,
                "call": lambda prompt, timeout=None, key=groq: _call_groq(
                    prompt, key, timeout=timeout or _lesson_llm_timeout_seconds()
                ),
            }
        )
    return providers


def _lesson_prompt(
    topic: str, lesson_title: str, level: str, medium: str, context: str
) -> str:
    medium_rule = (
        "Write the spoken narration in natural spoken Malayalam mixed with English technical terms "
        "(the way a Kerala tuition teacher actually talks: Malayalam sentences, English kept for "
        "technical vocabulary). Keep board_heading, board_lines, objectives, notes and flashcards in English."
        if medium.strip().lower() in {"malayalam", "manglish", "ml"}
        else "Write the spoken narration in clear, simple spoken Indian English."
    )
    return f"""You are one of the best teachers in the world making ONE ~10-minute lesson that a student experiences as a continuous, spoken class (not slides). Your goal: the student truly UNDERSTANDS, not just hears facts.

Lesson: "{lesson_title}"
Part of learning: "{topic}"
Learner level: {level or "beginner"}
{medium_rule}

HOW A GREAT LESSON IS BUILT (follow this arc across the beats)
1. HOOK — open with a real question, surprising fact, or everyday situation that makes the student curious about THIS lesson. No throat-clearing.
2. GROUND IT — connect to something the student already knows before introducing anything new.
3. EXPLAIN — teach the core idea. ALWAYS give the reason BEFORE the rule ("here's why, so the rule makes sense"). Build up, never dump.
4. WORKED EXAMPLE — walk through ONE concrete, specific example with real numbers/specifics, step by step, thinking out loud. This is the heart of the lesson — make it vivid and complete.
5. ANALOGY — at most one, and only if it genuinely makes the idea click.
6. MISCONCEPTION — name the exact mistake students usually make here and correct it directly ("A lot of students think X — but actually Y, because...").
7. CHECKPOINT — ask the student a question and give them a beat to think, then reveal and explain the answer. Make them do the thinking.
8. RECAP — warm, tight summary of what they can now do, and how it connects to the next thing.

TEACHING RULES
- Talk TO the student ("you"), warm, human, and encouraging. Sound like a person who loves this subject, not a textbook.
- Reason before rule, concrete before abstract, one idea per beat fully developed.
- CRITICAL LENGTH RULE: each beat's narration MUST be 90-140 words of what the teacher actually SAYS. Never write one- or two-sentence beats. A real teacher talks for 40-60 seconds per beat.
- The board is what appears on screen WHILE they speak — a short heading and 2-5 tight bullet fragments (not full sentences, not read aloud verbatim). The narration teaches; the board reinforces the keywords, the formula, or the example's steps.
- Total narration across all beats MUST be about 1300-1600 words — a full ~10 minute class, not a summary.
- Be accurate. If the lesson has a formula, definition, or process, state it precisely and correctly.

QUALITY OF SUPPORTING MATERIAL
- objectives: 3-5 crisp "By the end you can…" statements — the concrete skills this lesson delivers.
- notes: 6-10 revision notes a student writes in their notebook — self-contained, exam-ready, each a complete useful fact (include the key formula/definition/steps, not vague reminders).
- flashcards: 6-10 real question→answer pairs that test the hardest/most testable points (definitions, why-questions, one small applied problem). Answers must be correct and specific.

Return ONE JSON object, no markdown:
{{
  "lesson_title": "{lesson_title}",
  "summary": "one warm sentence describing what this class teaches",
  "objectives": ["By the end you can …", "By the end you can …"],
  "beats": [
    {{"kind": "hook|explain|example|analogy|checkpoint|recap", "narration": "what the teacher says (90-140 words)", "board_heading": "short title", "board_lines": ["tight fragment", "tight fragment"], "visual_hint": "optional: a simple diagram/idea to draw, or empty string"}}
  ],
  "notes": ["exam-ready revision note", "..."],
  "flashcards": [{{"front": "question", "back": "correct, specific answer"}}]
}}

Make {MIN_BEATS}-{MAX_BEATS} beats (aim for {TARGET_BEATS}), 3-5 objectives, 6-10 notes and 6-10 flashcards.
{f"Use this source material where relevant (stay faithful to it):{chr(10)}{context[:4000]}" if context.strip() else ""}
"""


def _continue_prompt(
    topic: str, lesson_title: str, level: str, medium: str, taught_headings: list[str]
) -> str:
    medium_rule = (
        "Continue in natural spoken Malayalam mixed with English technical terms; keep board text in English."
        if medium.strip().lower() in {"malayalam", "manglish", "ml"}
        else "Continue in clear, simple spoken Indian English."
    )
    already = "; ".join(taught_headings)
    return f"""You are continuing a live ~10-minute class on "{lesson_title}" (part of "{topic}", learner level {level or "beginner"}).
{medium_rule}

So far you have already taught these beats: {already}.
Now CONTINUE and FINISH the class with the remaining beats: a deeper worked example, a common mistake or misconception to avoid, a quick understanding check, and a warm recap. Do NOT repeat what was already taught.

Same rules: each beat's narration is what the teacher SAYS and MUST be 90-130 words. Board = short heading + 2-5 tight bullet lines.

Return ONE JSON object, no markdown:
{{
  "beats": [
    {{"kind": "example|checkpoint|recap|explain", "narration": "...", "board_heading": "...", "board_lines": ["..."], "visual_hint": ""}}
  ],
  "notes": ["6-10 concise revision notes covering the WHOLE lesson"],
  "flashcards": [{{"front": "question", "back": "answer"}}]
}}

Write 6-9 more beats and 6-10 flashcards.
"""


def _starter_reading_script(
    *, topic: str, lesson_title: str, level: str
) -> dict[str, Any]:
    """Return an honest, deterministic lesson for local/mock development.

    The direct Groq/Deepgram authoring pipeline must not run when the configured
    AI provider is ``mock``. This starter is deliberately labelled as reading
    mode by ``build_lesson``; it keeps roadmap study and resume flows usable
    without presenting generated audio as real.
    """
    normalized = " ".join([topic, lesson_title]).lower()
    if "python" in normalized:
        return {
            "lesson_title": lesson_title,
            "summary": "A beginner reading class on how Python instructions, values, decisions, repetition, and functions fit together.",
            "objectives": [
                "Explain what a Python program does",
                "Use variables and basic value types",
                "Recognise decisions, loops, and functions",
                "Trace a short program before running it",
            ],
            "beats": [
                {
                    "kind": "hook",
                    "narration": "A computer does not guess what you mean. It follows instructions in order. Python gives you a readable way to write those instructions. In this class, treat every line as a small command: store a value, make a decision, repeat an action, or reuse a group of instructions. That simple model is enough to begin reading real Python without memorising a long list of rules.",
                    "board_heading": "Code is a sequence of instructions",
                    "board_lines": [
                        "Read top to bottom",
                        "One clear action per line",
                        "Predict before you run",
                    ],
                    "visual_hint": "Draw three boxes labelled input, process, output.",
                },
                {
                    "kind": "explain",
                    "narration": "A variable is a name that refers to a value. For example, score = 5 gives the name score the integer value 5, while name = 'Asha' gives name a text value. Common beginner types are int for whole numbers, float for decimal numbers, str for text, and bool for True or False. The equals sign assigns a value here; it does not ask whether two values are equal.",
                    "board_heading": "Names and values",
                    "board_lines": [
                        "score = 5",
                        "name = 'Asha'",
                        "int, float, str, bool",
                    ],
                    "visual_hint": "Connect each variable name to its current value.",
                },
                {
                    "kind": "example",
                    "narration": "Trace this example: name = 'Asha', marks = 8, then print(name, marks). The first line stores text, the second stores a whole number, and print sends both values to the output. Change marks to 9 and only the printed number changes. This is a useful study habit: say what each line changes before you press Run. Tracing catches many mistakes faster than rereading the whole program.",
                    "board_heading": "Worked example",
                    "board_lines": [
                        "name = 'Asha'",
                        "marks = 8",
                        "print(name, marks)",
                        "Output: Asha 8",
                    ],
                    "visual_hint": "Use a two-column trace table: variable and value.",
                },
                {
                    "kind": "explain",
                    "narration": "Programs become useful when they can choose and repeat. An if statement runs a block only when its condition is True. A for loop repeats a block for each item in a sequence. Python uses indentation to show which lines belong inside that block, so spacing changes meaning. Read the condition first, then follow only the indented lines that should run.",
                    "board_heading": "Decide and repeat",
                    "board_lines": [
                        "if condition:",
                        "    run this block",
                        "for item in sequence:",
                        "    repeat this block",
                    ],
                    "visual_hint": "Draw a decision diamond leading to an indented block.",
                },
                {
                    "kind": "explain",
                    "narration": "A function gives a reusable name to a group of instructions. You define it with def, pass information through parameters, and use return when the function must send a result back. For example, def double(number): return number * 2 describes one job clearly. Calling double(4) produces 8. Functions reduce repetition and make each part of a program easier to test.",
                    "board_heading": "Reuse with functions",
                    "board_lines": [
                        "def double(number):",
                        "    return number * 2",
                        "double(4) -> 8",
                    ],
                    "visual_hint": "Show input 4 entering a function box and output 8 leaving it.",
                },
                {
                    "kind": "checkpoint",
                    "narration": "Pause and predict this without running it: total = 2, then for number in [1, 2, 3], total = total + number. The loop adds 1, then 2, then 3 to the starting value 2, so total becomes 8. If your answer differed, write the value after every pass. A trace table is the correction tool: it makes the changing state visible instead of asking you to hold every step in memory.",
                    "board_heading": "Checkpoint",
                    "board_lines": [
                        "Start: total = 2",
                        "+1 -> 3",
                        "+2 -> 5",
                        "+3 -> 8",
                    ],
                    "visual_hint": "Make one row for each loop pass.",
                },
                {
                    "kind": "recap",
                    "narration": "You now have a map for beginner Python. Values are stored behind variable names. If chooses a path, for repeats a block, and def creates a reusable function. Your next move is small: type the worked example, change one value, and predict the new output before running it. Learning programming comes from this short loop of predict, run, compare, and correct.",
                    "board_heading": "Your Python map",
                    "board_lines": [
                        "Variables store values",
                        "if chooses",
                        "for repeats",
                        "def reuses",
                        "Predict -> run -> correct",
                    ],
                    "visual_hint": "Keep this map beside your first practice program.",
                },
            ],
            "notes": [
                "Python executes instructions in a defined order.",
                "A variable name refers to a value; assignment uses =.",
                "Basic beginner types include int, float, str, and bool.",
                "An if statement runs its indented block when the condition is True.",
                "A for loop repeats its indented block for items in a sequence.",
                "A function is defined with def and can return a result.",
                "Trace changing variable values to debug a short program.",
            ],
            "flashcards": [
                {
                    "front": "What does = do in score = 5?",
                    "back": "It assigns the integer value 5 to the name score.",
                },
                {"front": "Which type stores text?", "back": "str"},
                {"front": "What controls a Python code block?", "back": "Indentation."},
                {
                    "front": "What does an if statement do?",
                    "back": "It runs a block when its condition is True.",
                },
                {
                    "front": "What does a for loop do?",
                    "back": "It repeats a block for each item in a sequence.",
                },
                {
                    "front": "Why use a function?",
                    "back": "To name and reuse a focused group of instructions.",
                },
            ],
        }

    clean_title = lesson_title.strip() or topic.strip() or "this topic"
    clean_level = level.strip() or "beginner"
    return {
        "lesson_title": clean_title,
        "summary": f"A {clean_level} starter reading class that turns {clean_title} into a definition, example, check, and next practice move.",
        "objectives": [
            f"State what {clean_title} means",
            "Identify the central terms",
            "Work through one concrete example",
            "Check your understanding without notes",
        ],
        "beats": [
            {
                "kind": "hook",
                "narration": f"Before collecting facts about {clean_title}, write one question you want this lesson to answer. That question gives the topic a purpose and makes it easier to notice which ideas matter.",
                "board_heading": "Start with one question",
                "board_lines": [
                    f"Topic: {clean_title}",
                    "What must I understand?",
                    "What can I explain after this?",
                ],
                "visual_hint": "Write your question at the top of the page.",
            },
            {
                "kind": "explain",
                "narration": f"Build a precise definition of {clean_title}: name the larger idea it belongs to, the feature that makes it distinct, and one boundary or condition. Keep this as a working definition and verify subject-specific facts against a trusted lesson or source.",
                "board_heading": "Build the definition",
                "board_lines": [
                    "Category",
                    "Distinct feature",
                    "Boundary or condition",
                ],
                "visual_hint": "Use a three-part definition box.",
            },
            {
                "kind": "example",
                "narration": f"Choose one concrete example of {clean_title}. Label which part of the definition appears in the example and which details are only background. A useful example should let you explain why it belongs, not merely name it.",
                "board_heading": "Test with an example",
                "board_lines": [
                    "Name the example",
                    "Match it to the definition",
                    "Explain why it fits",
                ],
                "visual_hint": "Draw arrows from example details to definition terms.",
            },
            {
                "kind": "checkpoint",
                "narration": f"Close your notes and explain {clean_title} in two sentences: one definition and one example with a reason. If you cannot connect the example to the definition, mark that exact missing link for revision instead of restarting the whole topic.",
                "board_heading": "Quick check",
                "board_lines": [
                    "Sentence 1: definition",
                    "Sentence 2: example + why",
                    "Mark the missing link",
                ],
                "visual_hint": "Answer aloud before reopening your notes.",
            },
            {
                "kind": "recap",
                "narration": f"Your next move for {clean_title} is now specific: verify the working definition, add one subject-correct example, then retry the two-sentence explanation tomorrow. That short retrieval step creates evidence of learning and gives the roadmap a real place to resume.",
                "board_heading": "Next move",
                "board_lines": ["Verify", "Add one example", "Recall tomorrow"],
                "visual_hint": "Schedule the two-sentence recall for tomorrow.",
            },
        ],
        "notes": [
            f"Working topic: {clean_title}.",
            "A strong definition gives category, distinct feature, and boundary.",
            "An example is useful only when you can explain why it fits.",
            "Mark the exact missing link instead of restarting everything.",
            "Retry the definition and example from memory the next day.",
        ],
        "flashcards": [
            {
                "front": f"What is your working definition of {clean_title}?",
                "back": "Give category, distinct feature, and boundary; verify subject facts against a trusted source.",
            },
            {
                "front": "What makes an example useful?",
                "back": "You can connect its details to the definition and explain why it fits.",
            },
            {
                "front": "What should you revise after a failed recall?",
                "back": "The exact missing link, not the entire topic.",
            },
        ],
    }


def _generate_with_retries(
    prompt: str,
    label: str,
    providers: list[dict[str, Any]] | None = None,
    *,
    deadline: float | None = None,
) -> dict[str, Any]:
    last_error: Exception | None = None
    chain = providers if providers is not None else _script_llm_providers()
    if not chain:
        raise LessonBuildError(
            "No lesson LLM key configured. Set ZAI_API_KEY (free GLM-4.7-Flash) "
            "or GROQ_API_KEY for Learn Anything authoring."
        )
    for provider in chain:
        if deadline is not None and time.monotonic() >= deadline:
            break
        name = str(provider.get("name") or "llm")
        call = provider["call"]
        # A second attempt used to multiply a slow provider chain into a
        # multi-minute request.  One bounded attempt per provider is enough;
        # the next provider or the authored reading class is the recovery path.
        for attempt in range(1, 2):
            if deadline is not None and time.monotonic() >= deadline:
                break
            try:
                remaining = None
                if deadline is not None:
                    remaining = max(
                        1.0,
                        min(_lesson_llm_timeout_seconds(), deadline - time.monotonic()),
                    )
                data = call(prompt, timeout=remaining) if remaining is not None else call(prompt)
                logger.info("Lesson %s authored via %s (attempt %s)", label, name, attempt)
                return data
            except Exception as exc:  # provider SDKs expose adapter-specific errors
                last_error = exc
                logger.warning(
                    "Lesson %s via %s attempt %s failed: %s",
                    label,
                    name,
                    attempt,
                    type(exc).__name__,
                )
    raise LessonBuildError(f"Lesson {label} generation failed: {last_error}")


def _narration_words(beats: list[dict[str, Any]]) -> int:
    return sum(len(str(beat.get("narration", "")).split()) for beat in beats)


def generate_lesson_script(
    *,
    topic: str,
    lesson_title: str,
    level: str = "",
    medium: str = "english",
    context: str = "",
) -> dict[str, Any]:
    providers = _script_llm_providers()
    if not providers:
        logger.warning(
            "ZAI_API_KEY/GROQ_API_KEY missing/invalid; serving starter reading class."
        )
        return _starter_reading_script(
            topic=topic, lesson_title=lesson_title, level=level
        )

    deadline = time.monotonic() + _lesson_generation_budget_seconds()
    try:
        data = _generate_with_retries(
            _lesson_prompt(topic, lesson_title, level, medium, context),
            "script",
            providers,
            deadline=deadline,
        )
    except LessonBuildError as exc:
        # Never leave 1000 students on a hard error when providers are out of quota:
        # fall back to a complete starter reading class and keep the product usable.
        logger.warning("Lesson script providers failed (%s); using starter class.", exc)
        return _starter_reading_script(
            topic=topic, lesson_title=lesson_title, level=level
        )

    beats = list(data.get("beats") or [])
    if not beats:
        return _starter_reading_script(
            topic=topic, lesson_title=lesson_title, level=level
        )

    # The model reliably writes ~4-5 minutes in one call, then stops. To reach a
    # real ~10-minute class, ask it to continue from what it already taught.
    # One continuation is enough; this stays a cheap 2-call generation.
    if _narration_words(beats) < 950 and time.monotonic() < deadline:
        try:
            taught = [
                str(beat.get("board_heading") or beat.get("kind")) for beat in beats
            ]
            more = _generate_with_retries(
                _continue_prompt(topic, lesson_title, level, medium, taught),
                "continuation",
                providers,
                deadline=deadline,
            )
            beats.extend(more.get("beats") or [])
            # Prefer the fuller notes/flashcards from whichever pass gave more.
            for field_name in ("notes", "flashcards"):
                if len(more.get(field_name) or []) > len(data.get(field_name) or []):
                    data[field_name] = more[field_name]
        except LessonBuildError as exc:
            logger.warning("Lesson continuation skipped: %s", exc)

    data["beats"] = beats
    return data


def _deepgram_synthesize(
    text: str,
    out_path: Path,
    api_key: str,
    model: str,
    *,
    timeout: float | None = None,
) -> None:
    url = f"{DEEPGRAM_SPEAK_URL}?model={model}&encoding=linear16&sample_rate=24000"
    request = urllib.request.Request(
        url,
        data=json.dumps({"text": text}).encode("utf-8"),
        headers={
            "Authorization": f"Token {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(
        request,
        timeout=timeout if timeout is not None else DEFAULT_LESSON_TTS_TIMEOUT_SECONDS,
    ) as response:
        audio = response.read()
    # Deepgram linear16 is headerless PCM; wrap it in a WAV container so the
    # browser (and ffprobe) can read it directly.
    out_path.parent.mkdir(parents=True, exist_ok=True)
    pcm_path = out_path.with_suffix(".pcm")
    pcm_path.write_bytes(audio)
    subprocess.run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "s16le",
            "-ar",
            "24000",
            "-ac",
            "1",
            "-i",
            str(pcm_path),
            str(out_path),
        ],
        check=True,
        capture_output=True,
    )
    pcm_path.unlink(missing_ok=True)


def _probe_duration(path: Path) -> float:
    completed = subprocess.run(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format=duration",
            "-of",
            "default=nw=1:nk=1",
            str(path),
        ],
        capture_output=True,
        check=True,
        text=True,
    )
    return round(float(completed.stdout.strip()), 3)


def synthesize_beats(
    beats: list[dict[str, Any]],
    out_dir: Path,
    medium: str,
    *,
    deadline: float | None = None,
) -> list[LessonBeat]:
    """Voice each beat. Deepgram for English; AI4Bharat for Malayalam medium."""
    use_malayalam = medium.strip().lower() in {"malayalam", "ml"}
    result: list[LessonBeat] = []
    cursor = 0.0

    ai4bharat_provider = None
    if use_malayalam:
        from app.schemas.video import VideoSceneAudioInput
        from app.services.tts_provider import AI4BharatIndicParlerTTSProvider

        ai4bharat_provider = (AI4BharatIndicParlerTTSProvider(), VideoSceneAudioInput)
    else:
        api_key = _env("DOCDOE_TTS_API_KEY")
        model = _env("DOCDOE_TTS_MODEL") or "aura-luna-en"
        if not api_key:
            raise LessonBuildError(
                "DOCDOE_TTS_API_KEY (Deepgram) missing; cannot voice English lesson."
            )

    for index, beat in enumerate(beats, start=1):
        if deadline is not None and time.monotonic() >= deadline:
            raise LessonBuildError("Lesson audio exceeded its bounded generation budget.")
        narration = str(beat.get("narration", "")).strip()
        if not narration:
            continue
        audio_path = out_dir / f"beat-{index:02d}.wav"
        if use_malayalam:
            provider, scene_cls = ai4bharat_provider
            scene = scene_cls(
                scene_id=index,
                type="concept",
                duration_seconds=20,
                voice_text=narration,
            )
            synth = provider.generate_scene_audio(
                scene=scene,
                output_file=audio_path,
                voice_mode="malayalam_soft",
                voice="Anjali",
                language="ml",
            )
            if synth.file_path != audio_path:
                Path(synth.file_path).replace(audio_path)
        else:
            remaining = None
            if deadline is not None:
                remaining = max(1.0, min(DEFAULT_LESSON_TTS_TIMEOUT_SECONDS, deadline - time.monotonic()))
            _deepgram_synthesize(
                narration,
                audio_path,
                api_key,
                model,
                timeout=remaining,
            )
        duration = _probe_duration(audio_path)
        result.append(
            LessonBeat(
                kind=str(beat.get("kind", "explain")),
                narration=narration,
                board_heading=str(beat.get("board_heading", "")),
                board_lines=[str(line) for line in (beat.get("board_lines") or [])],
                visual_hint=str(beat.get("visual_hint", "")),
                audio_src=(
                    f"/generated/learn-anything/{out_dir.parent.name}/{out_dir.name}/{audio_path.name}"
                ),
                start_second=round(cursor, 3),
                duration_seconds=duration,
            )
        )
        cursor += duration + 0.35  # small breath between beats
    return result


def _reading_beats(beats: list[dict[str, Any]]) -> list[LessonBeat]:
    """Turn authored beats into a fully navigable reading class."""
    result: list[LessonBeat] = []
    cursor = 0.0
    for beat in beats:
        narration = str(beat.get("narration", "")).strip()
        if not narration:
            continue
        duration = round(max(20.0, len(narration.split()) / 180 * 60), 3)
        result.append(
            LessonBeat(
                kind=str(beat.get("kind", "explain")),
                narration=narration,
                board_heading=str(beat.get("board_heading", "")),
                board_lines=[str(line) for line in (beat.get("board_lines") or [])],
                visual_hint=str(beat.get("visual_hint", "")),
                audio_src="",
                start_second=round(cursor, 3),
                duration_seconds=duration,
            )
        )
        cursor += duration
    return result


def _lesson_lock_for(key: str) -> threading.Lock:
    with _lesson_build_locks_guard:
        lock = _lesson_build_locks.get(key)
        if lock is None:
            lock = threading.Lock()
            _lesson_build_locks[key] = lock
        return lock


def _is_forbidden_cache_path(path: Path) -> bool:
    """Block the HF /public tree that resolves outside the writable container."""
    normalized = path.as_posix().replace("\\", "/")
    # Unix absolute
    if normalized == "/public" or normalized.startswith("/public/"):
        return True
    # Windows oddities if someone sets PUBLIC_ROOT = Path("/public/...")
    if normalized.lower().endswith(":/public") or "/public/generated" in normalized and normalized.startswith("/"):
        return True
    return False


def ensure_public_root() -> Path:
    """Ensure the lesson cache directory exists and is writable on HF + local."""
    candidates: list[Path] = []
    env = _env("LEARN_LESSON_CACHE_DIR")
    if env:
        candidates.append(Path(env))
    # Tests monkeypatch PUBLIC_ROOT to a temp path — prefer that when safe.
    if not _is_forbidden_cache_path(PUBLIC_ROOT):
        candidates.append(PUBLIC_ROOT)
    # HF Docker WORKDIR is /app (backend tree).
    candidates.append(Path("/app/generated/learn-anything"))
    # Monorepo / local backend package root.
    candidates.append(BACKEND_DIR / "generated" / "learn-anything")
    # Last-resort temp (always writable for reading-mode fallbacks).
    candidates.append(
        Path(os.getenv("TMPDIR") or os.getenv("TEMP") or "/tmp")
        / "docdoe-learn-lessons"
    )

    errors: list[str] = []
    seen: set[str] = set()
    for root in candidates:
        key = root.as_posix()
        if key in seen:
            continue
        seen.add(key)
        if _is_forbidden_cache_path(root):
            errors.append(f"skip forbidden path {root}")
            continue
        try:
            root.mkdir(parents=True, exist_ok=True)
            probe = root / ".write_probe"
            probe.write_text("ok", encoding="utf-8")
            probe.unlink(missing_ok=True)
            logger.info("Lesson cache using %s", root)
            return root
        except OSError as exc:
            errors.append(f"{root}: {exc}")
            continue

    raise LessonBuildError(
        "Lesson cache directory is not writable. Tried: " + "; ".join(errors)
    )


def build_lesson(
    *,
    topic: str,
    lesson_title: str,
    level: str = "",
    medium: str = "english",
    context: str = "",
    user_id: str | None = None,
    force: bool = False,
) -> dict[str, Any]:
    """Full pipeline: script -> audio -> playable manifest, cached privately.

    Every user's source context is isolated in its own cache namespace.
    Concurrent first opens for the same hash serialize on a per-hash lock so
    we do not fan out N identical LLM bills under load.
    """
    settings = get_settings()
    mock_mode = str(settings.ai_provider).strip().lower() == "mock"
    voice = (
        "reading-preview"
        if mock_mode
        else "ai4bharat-anjali"
        if medium.strip().lower() in {"malayalam", "ml"}
        else (_env("DOCDOE_TTS_MODEL") or "aura-luna-en")
    )
    owner = user_id or "unowned"
    key = lesson_hash(topic, lesson_title, level, medium, voice, context=context, user_id=owner)
    cache_root = ensure_public_root()
    out_dir = cache_root / owner / key
    manifest_path = out_dir / "lesson.json"

    if manifest_path.exists() and not force:
        return json.loads(manifest_path.read_text(encoding="utf-8"))

    lock = _lesson_lock_for(key)
    with lock:
        # Re-check inside the lock: another student may have finished while we waited.
        if manifest_path.exists() and not force:
            return json.loads(manifest_path.read_text(encoding="utf-8"))

        script = (
            _starter_reading_script(topic=topic, lesson_title=lesson_title, level=level)
            if mock_mode
            else generate_lesson_script(
                topic=topic,
                lesson_title=lesson_title,
                level=level,
                medium=medium,
                context=context,
            )
        )
        try:
            out_dir.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            # Do not write an untracked fallback that the authenticated media
            # router cannot authorize. `ensure_public_root` already tries the
            # configured cache, application cache, and an OS temp directory.
            raise LessonBuildError(
                f"Could not create private lesson cache folder ({out_dir}): {exc}"
            ) from exc
        delivery_mode = "reading" if mock_mode else "audio"
        delivery_notice = (
            "Local preview: this complete starter class is available in reading mode; no generated voice is being presented as real audio."
            if mock_mode
            else ""
        )
        if mock_mode:
            beats = _reading_beats(script["beats"])
        else:
            # Audio is intentionally opt-in for the synchronous request.  A
            # voice API call per beat made an uncached lesson wait for minutes,
            # even though the existing reading player was already complete and
            # useful.  Keep the fast path honest: a real provider-authored
            # script is still returned, with audio as a separate future
            # enhancement when the deployment explicitly enables it.
            if _lesson_sync_tts_enabled():
                try:
                    beats = synthesize_beats(
                        script["beats"],
                        out_dir,
                        medium,
                        deadline=time.monotonic() + _bounded_seconds(
                            "LEARN_LESSON_AUDIO_BUDGET_SECONDS",
                            DEFAULT_LESSON_AUDIO_BUDGET_SECONDS,
                            DEFAULT_LESSON_AUDIO_BUDGET_SECONDS,
                        ),
                    )
                except Exception as exc:
                    logger.warning(
                        "Lesson voice generation exceeded its budget; serving reading mode: %s",
                        type(exc).__name__,
                    )
                    beats = _reading_beats(script["beats"])
                    delivery_mode = "reading"
                    delivery_notice = "Audio is temporarily unavailable. The complete authored class is ready in reading mode, and your lesson position still saves."
            else:
                beats = _reading_beats(script["beats"])
                delivery_mode = "reading"
                delivery_notice = "This class is ready in reading mode so you can start immediately. Retry later if you want a voiced version."
        if not beats:
            raise LessonBuildError("No audible beats were produced for the lesson.")

        total_seconds = round(beats[-1].start_second + beats[-1].duration_seconds, 3)
        manifest = {
            "schema": "learn-lesson-v1",
            "lessonHash": key,
            "topic": topic,
            "lessonTitle": script.get("lesson_title", lesson_title),
            "level": level,
            "medium": medium,
            "voice": voice,
            "deliveryMode": delivery_mode,
            "deliveryNotice": delivery_notice,
            "isFallback": delivery_mode == "reading",
            "totalSeconds": total_seconds,
            "totalMinutes": round(total_seconds / 60, 2),
            "summary": str(script.get("summary", "")).strip(),
            "objectives": [
                str(item).strip()
                for item in (script.get("objectives") or [])
                if str(item).strip()
            ],
            "beats": [beat.__dict__ for beat in beats],
            "notes": [str(note) for note in (script.get("notes") or [])],
            "flashcards": [
                {"front": str(card.get("front", "")), "back": str(card.get("back", ""))}
                for card in (script.get("flashcards") or [])
                if card.get("front") and card.get("back")
            ],
        }
        # Atomic-ish write: write temp then replace so readers never see half JSON.
        tmp_path = manifest_path.with_suffix(".json.tmp")
        tmp_path.write_text(
            json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
        )
        tmp_path.replace(manifest_path)
        return manifest


def _slugify(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", value.strip().lower()).strip("-")


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(
        description="Build one Learn Anything lesson end-to-end."
    )
    parser.add_argument("--topic", required=True)
    parser.add_argument("--lesson", required=True)
    parser.add_argument("--level", default="beginner")
    parser.add_argument("--medium", default="english")
    parser.add_argument("--force", action="store_true")
    args = parser.parse_args()

    result = build_lesson(
        topic=args.topic,
        lesson_title=args.lesson,
        level=args.level,
        medium=args.medium,
        force=args.force,
    )
    print(
        json.dumps(
            {
                "lessonHash": result["lessonHash"],
                "lessonTitle": result["lessonTitle"],
                "totalMinutes": result["totalMinutes"],
                "beats": len(result["beats"]),
                "notes": len(result["notes"]),
                "flashcards": len(result["flashcards"]),
            },
            indent=2,
        )
    )