scripts/production_preflight.py

"""DocDoe production preflight checks.

Run before deploying to verify the production environment is safe:

    python backend/scripts/production_preflight.py

Exit codes:
    0 — no FAIL (WARN entries allowed)
    1 — at least one FAIL

The script never prints secret values. Sensitive env vars are masked
to first 4 + last 4 characters with the middle replaced by ``****``.
"""
from __future__ import annotations

import ast
import os
import shutil
import sys
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Callable, Iterable

# Ensure the script can be run from anywhere — fall back to repo-relative imports.
HERE = Path(__file__).resolve().parent
BACKEND_DIR = HERE.parent
REPO_ROOT = BACKEND_DIR.parent
if str(BACKEND_DIR) not in sys.path:
    sys.path.insert(0, str(BACKEND_DIR))


# ── Result types ──────────────────────────────────────────────────────────────

Status = str  # "PASS" | "WARN" | "FAIL"


@dataclass(frozen=True)
class Check:
    section: str
    name: str
    status: Status
    detail: str


def _pass(section: str, name: str, detail: str = "") -> Check:
    return Check(section, name, "PASS", detail)


def _warn(section: str, name: str, detail: str) -> Check:
    return Check(section, name, "WARN", detail)


def _fail(section: str, name: str, detail: str) -> Check:
    return Check(section, name, "FAIL", detail)


# ── Secret masking ────────────────────────────────────────────────────────────

def mask_secret(value: str | None) -> str:
    if not value:
        return "(missing)"
    if len(value) <= 8:
        return "****"
    return f"{value[:4]}****{value[-4:]}"


# ── Helpers to read env ───────────────────────────────────────────────────────

def env(name: str, default: str | None = None) -> str | None:
    raw = os.environ.get(name)
    if raw is None:
        return default
    stripped = raw.strip()
    return stripped if stripped else default


def env_bool(name: str, default: bool = False) -> bool:
    raw = env(name)
    if raw is None:
        return default
    return raw.lower() in {"1", "true", "yes", "on"}


# ── Section 1 — Environment ───────────────────────────────────────────────────

DEFAULT_JWT_SECRET = "change-this-local-dev-secret"
LOCAL_HOST_PREFIXES = ("http://localhost", "http://127.0.0.1", "http://0.0.0.0")


def check_environment(is_prod: bool) -> Iterable[Check]:
    sec = "Environment"
    yield _pass(sec, "ENVIRONMENT", f"set to '{env('ENVIRONMENT') or 'development'}'")

    db_url = env("DATABASE_URL")
    if not db_url:
        yield _fail(sec, "DATABASE_URL", "missing")
    elif is_prod and db_url.startswith("sqlite"):
        yield _fail(sec, "DATABASE_URL", "SQLite not allowed in production — use PostgreSQL")
    elif db_url.startswith("postgresql"):
        yield _pass(sec, "DATABASE_URL", "PostgreSQL configured")
    else:
        yield _warn(sec, "DATABASE_URL", f"non-PostgreSQL driver ({db_url.split(':', 1)[0]})")

    frontend = env("FRONTEND_BASE_URL")
    if is_prod and frontend:
        if frontend.startswith(LOCAL_HOST_PREFIXES):
            yield _fail(sec, "FRONTEND_BASE_URL", "points at localhost in production")
        elif not frontend.startswith("https://"):
            yield _warn(sec, "FRONTEND_BASE_URL", "should use https in production")
        else:
            yield _pass(sec, "FRONTEND_BASE_URL", frontend)
    elif frontend:
        yield _pass(sec, "FRONTEND_BASE_URL", frontend)
    else:
        yield _warn(sec, "FRONTEND_BASE_URL", "not set")

    cors = env("CORS_ORIGINS")
    if is_prod:
        if not cors:
            yield _fail(sec, "CORS_ORIGINS", "missing in production")
        else:
            origins = [o.strip() for o in cors.split(",") if o.strip()]
            bad = [o for o in origins if o == "*" or o.startswith(LOCAL_HOST_PREFIXES)]
            if bad:
                yield _fail(sec, "CORS_ORIGINS", f"unsafe entries: {bad}")
            else:
                yield _pass(sec, "CORS_ORIGINS", f"{len(origins)} origin(s) configured")
    else:
        yield _pass(sec, "CORS_ORIGINS", cors or "(default)")

    jwt = env("JWT_SECRET_KEY")
    if not jwt or jwt == DEFAULT_JWT_SECRET:
        if is_prod:
            yield _fail(sec, "JWT_SECRET_KEY", "missing or default value — generate with `openssl rand -hex 32`")
        else:
            yield _warn(sec, "JWT_SECRET_KEY", "using dev default")
    elif len(jwt) < 32:
        yield _fail(sec, "JWT_SECRET_KEY", f"too short ({len(jwt)} chars, need >=32)")
    else:
        yield _pass(sec, "JWT_SECRET_KEY", f"strong ({len(jwt)} chars, {mask_secret(jwt)})")

    auth_enabled = env_bool("AUTH_ENABLED")
    if is_prod and not auth_enabled:
        yield _fail(sec, "AUTH_ENABLED", "must be true in production")
    else:
        yield _pass(sec, "AUTH_ENABLED", str(auth_enabled))

    auth_provider = (env("AUTH_PROVIDER") or "jwt").lower()
    if auth_provider not in {"jwt", "supabase"}:
        yield _fail(sec, "AUTH_PROVIDER", "must be 'jwt' or 'supabase' for real users")
    else:
        yield _pass(sec, "AUTH_PROVIDER", auth_provider)
    if auth_provider == "supabase":
        supabase_url = env("SUPABASE_URL")
        supabase_jwt = env("SUPABASE_JWT_SECRET")
        service_key = env("SUPABASE_SERVICE_ROLE_KEY")
        if not supabase_url:
            yield _fail(sec, "SUPABASE_URL", "required when AUTH_PROVIDER=supabase")
        else:
            yield _pass(sec, "SUPABASE_URL", supabase_url)
        if not supabase_jwt:
            yield _fail(sec, "SUPABASE_JWT_SECRET", "required to verify student sessions")
        elif len(supabase_jwt) < 32:
            yield _fail(sec, "SUPABASE_JWT_SECRET", "too short (need >=32 characters)")
        else:
            yield _pass(sec, "SUPABASE_JWT_SECRET", mask_secret(supabase_jwt))
        if not service_key:
            yield _fail(
                sec,
                "SUPABASE_SERVICE_ROLE_KEY",
                "required for authenticated account deletion; backend only",
            )
        else:
            yield _pass(sec, "SUPABASE_SERVICE_ROLE_KEY", mask_secret(service_key))

    rate_limit = env_bool("RATE_LIMIT_ENABLED")
    if is_prod and not rate_limit:
        yield _warn(sec, "RATE_LIMIT_ENABLED", "should be true in production")
    else:
        yield _pass(sec, "RATE_LIMIT_ENABLED", str(rate_limit))


# ── Section 1b — Database recovery ──────────────────────────────────────

def check_database_recovery(is_prod: bool) -> Iterable[Check]:
    sec = "Database recovery"
    strategy = (env("DATABASE_BACKUP_STRATEGY") or "").lower()

    if not strategy:
        if is_prod:
            yield _fail(
                sec,
                "DATABASE_BACKUP_STRATEGY",
                "missing; use 'managed' for provider backups or 'pg_dump' for the bundled backup job",
            )
        else:
            yield _pass(sec, "DATABASE_BACKUP_STRATEGY", "not required in development")
        return
    if strategy not in {"managed", "pg_dump"}:
        yield _fail(sec, "DATABASE_BACKUP_STRATEGY", "must be 'managed' or 'pg_dump'")
        return
    yield _pass(sec, "DATABASE_BACKUP_STRATEGY", strategy)

    retention_raw = env("DATABASE_BACKUP_RETENTION_DAYS") or "0"
    try:
        retention_days = int(retention_raw)
    except ValueError:
        retention_days = 0
    if retention_days < 7:
        yield _fail(sec, "DATABASE_BACKUP_RETENTION_DAYS", "must be at least 7 days")
    else:
        yield _pass(sec, "DATABASE_BACKUP_RETENTION_DAYS", f"{retention_days} days")

    restore_tested_raw = env("DATABASE_RESTORE_TESTED_AT")
    if not restore_tested_raw:
        yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", "missing; record the latest successful restore drill date")
    else:
        try:
            restore_date = date.fromisoformat(restore_tested_raw)
            age_days = (date.today() - restore_date).days
            if age_days < -1:
                yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", "cannot be a future date")
            elif age_days > 90:
                yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", f"restore drill is stale ({age_days} days old)")
            else:
                yield _pass(sec, "DATABASE_RESTORE_TESTED_AT", f"{restore_tested_raw} ({max(age_days, 0)} days ago)")
        except ValueError:
            yield _fail(sec, "DATABASE_RESTORE_TESTED_AT", "must be an ISO date in YYYY-MM-DD format")

    if strategy == "pg_dump":
        backup_dir = env("DATABASE_BACKUP_DIR")
        if not backup_dir:
            yield _fail(sec, "DATABASE_BACKUP_DIR", "required when DATABASE_BACKUP_STRATEGY=pg_dump")
        else:
            yield _pass(sec, "DATABASE_BACKUP_DIR", "configured")
        for binary in ("pg_dump", "pg_restore"):
            if shutil.which(binary):
                yield _pass(sec, binary, "available")
            else:
                yield _fail(sec, binary, "not available on PATH")


# ── Section 2 — Storage ───────────────────────────────────────────────────────

def check_storage(is_prod: bool) -> Iterable[Check]:
    sec = "Storage"
    provider = (env("STORAGE_PROVIDER") or "local").lower()

    if provider not in {"local", "r2", "s3", "cloudinary"}:
        yield _fail(sec, "STORAGE_PROVIDER",
                    f"invalid value '{provider}' — must be local/r2/s3/cloudinary")
        return

    yield _pass(sec, "STORAGE_PROVIDER", provider)

    if provider == "cloudinary":
        cloud_name = env("CLOUDINARY_CLOUD_NAME")
        api_key = env("CLOUDINARY_API_KEY")
        api_secret = env("CLOUDINARY_API_SECRET")
        if not cloud_name:
            yield _fail(sec, "CLOUDINARY_CLOUD_NAME", "missing")
        else:
            yield _pass(sec, "CLOUDINARY_CLOUD_NAME", cloud_name)
        if not api_key:
            yield _fail(sec, "CLOUDINARY_API_KEY", "missing")
        else:
            yield _pass(sec, "CLOUDINARY_API_KEY", mask_secret(api_key))
        if not api_secret:
            yield _fail(sec, "CLOUDINARY_API_SECRET", "missing — required for uploads")
        else:
            yield _pass(sec, "CLOUDINARY_API_SECRET", mask_secret(api_secret))
        try:
            import cloudinary  # noqa: F401
            yield _pass(sec, "cloudinary package", "installed")
        except ImportError:
            yield _fail(sec, "cloudinary package",
                        "not installed — add 'cloudinary' to backend/requirements.txt")
        return

    if provider == "local":
        if is_prod:
            yield _warn(sec, "Local storage in production",
                        "generated videos lost on redeploy — see backend/STORAGE_SETUP.md")
        else:
            yield _pass(sec, "Local storage", "OK for development")
        return

    # R2 / S3 required vars
    bucket = env("STORAGE_BUCKET")
    access_key = env("STORAGE_ACCESS_KEY_ID")
    secret_key = env("STORAGE_SECRET_ACCESS_KEY")
    endpoint = env("STORAGE_ENDPOINT_URL")
    public_base = env("STORAGE_PUBLIC_BASE_URL")
    region = env("STORAGE_REGION")

    if not bucket:
        yield _fail(sec, "STORAGE_BUCKET", "missing")
    else:
        yield _pass(sec, "STORAGE_BUCKET", bucket)

    if not access_key:
        yield _fail(sec, "STORAGE_ACCESS_KEY_ID", "missing")
    else:
        yield _pass(sec, "STORAGE_ACCESS_KEY_ID", mask_secret(access_key))

    if not secret_key:
        yield _fail(sec, "STORAGE_SECRET_ACCESS_KEY", "missing")
    else:
        yield _pass(sec, "STORAGE_SECRET_ACCESS_KEY", mask_secret(secret_key))

    if provider == "r2":
        if not endpoint:
            yield _fail(sec, "STORAGE_ENDPOINT_URL", "required for R2")
        elif "r2.cloudflarestorage.com" not in endpoint and "cloudflare" not in endpoint:
            yield _warn(sec, "STORAGE_ENDPOINT_URL", "does not look like an R2 endpoint")
        else:
            yield _pass(sec, "STORAGE_ENDPOINT_URL", endpoint)
    elif provider == "s3":
        if not region:
            yield _warn(sec, "STORAGE_REGION", "not set — boto3 will use AWS default")

    if is_prod and public_base:
        if public_base.startswith(LOCAL_HOST_PREFIXES):
            yield _fail(sec, "STORAGE_PUBLIC_BASE_URL", "points at localhost in production")
        elif not public_base.startswith("https://"):
            yield _warn(sec, "STORAGE_PUBLIC_BASE_URL", "should use https in production")
        else:
            yield _pass(sec, "STORAGE_PUBLIC_BASE_URL", public_base)
    elif public_base:
        yield _pass(sec, "STORAGE_PUBLIC_BASE_URL", public_base)
    else:
        yield _warn(sec, "STORAGE_PUBLIC_BASE_URL",
                    "not set — public URLs may fall back to provider default")

    # Optional: try to instantiate the boto3 provider without uploading
    if bucket and access_key and secret_key:
        try:
            import boto3  # noqa: F401
            yield _pass(sec, "boto3", "installed")
        except ImportError:
            yield _fail(sec, "boto3", "not installed — run `pip install boto3`")


# ── Section 3 — Video render dependencies ────────────────────────────────────

def _has_binary(name: str) -> bool:
    return shutil.which(name) is not None


def check_video_deps() -> Iterable[Check]:
    sec = "Video render deps"

    for binary in ("ffmpeg", "ffprobe"):
        if _has_binary(binary):
            yield _pass(sec, binary, f"available at {shutil.which(binary)}")
        else:
            yield _fail(sec, binary,
                        f"{binary} not found in PATH — video render/validation will fail")

    for binary in ("node", "npm"):
        if _has_binary(binary):
            yield _pass(sec, binary, f"available at {shutil.which(binary)}")
        else:
            yield _fail(sec, binary, f"{binary} not found in PATH — render CLI will fail")

    package_json = REPO_ROOT / "package.json"
    if not package_json.exists():
        yield _fail(sec, "package.json", "not found at repo root")
        return

    try:
        import json as _json
        data = _json.loads(package_json.read_text(encoding="utf-8"))
    except Exception as exc:
        yield _fail(sec, "package.json", f"could not parse: {exc.__class__.__name__}")
        return

    scripts = data.get("scripts", {})
    if "render:from-json" in scripts:
        yield _pass(sec, "render:from-json script", "defined in package.json")
    else:
        yield _fail(sec, "render:from-json script",
                    "missing from package.json — backend invokes this for render")

    deps = {**data.get("dependencies", {}), **data.get("devDependencies", {})}
    if "remotion" in deps or "@remotion/cli" in deps:
        version = deps.get("remotion") or deps.get("@remotion/cli", "unknown")
        yield _pass(sec, "remotion package", f"declared ({version})")
    else:
        yield _fail(sec, "remotion package", "not declared in package.json")


# ── Section 4 — AI / TTS config ──────────────────────────────────────────────

def check_ai_config(is_prod: bool) -> Iterable[Check]:
    sec = "AI / TTS"

    ai_provider = (env("AI_PROVIDER") or "").lower()
    if not ai_provider:
        yield _fail(sec, "AI_PROVIDER", "not set")
    else:
        yield _pass(sec, "AI_PROVIDER", ai_provider)

    if is_prod and env_bool("AI_FALLBACK_TO_MOCK"):
        yield _fail(sec, "AI_FALLBACK_TO_MOCK",
                    "must be false in production — mock responses must not reach students")
    else:
        yield _pass(sec, "AI_FALLBACK_TO_MOCK", str(env_bool("AI_FALLBACK_TO_MOCK")))

    tts_provider = (env("TTS_PROVIDER") or "").lower()
    if tts_provider:
        yield _pass(sec, "TTS_PROVIDER", tts_provider)
    else:
        yield _warn(sec, "TTS_PROVIDER", "not set — falling back to default")

    video_tts_provider = (env("VIDEO_TTS_PROVIDER") or "ai4bharat").lower()
    if video_tts_provider:
        yield _pass(sec, "VIDEO_TTS_PROVIDER", video_tts_provider)
    else:
        yield _warn(sec, "VIDEO_TTS_PROVIDER", "not set - study video voice defaults to ai4bharat")

    if ai_provider == "openai":
        key = env("OPENAI_API_KEY")
        if not key:
            yield _fail(sec, "OPENAI_API_KEY", "required when AI_PROVIDER=openai")
        else:
            yield _pass(sec, "OPENAI_API_KEY", mask_secret(key))

    needs_sarvam = ai_provider == "sarvam" or tts_provider == "sarvam" or video_tts_provider == "sarvam"
    sarvam_key = env("SARVAM_API_KEY")
    if needs_sarvam:
        if not sarvam_key:
            yield _fail(sec, "SARVAM_API_KEY",
                        "required when AI_PROVIDER or TTS_PROVIDER is sarvam")
        else:
            yield _pass(sec, "SARVAM_API_KEY", mask_secret(sarvam_key))

    ai4bharat_names = {"ai4bharat", "indic_parler", "indic_parler_tts"}
    needs_ai4bharat = tts_provider in ai4bharat_names or video_tts_provider in ai4bharat_names
    if needs_ai4bharat:
        hf_key = env("HUGGINGFACE_API_KEY") or env("HF_TOKEN")
        if not hf_key:
            yield _fail(
                sec,
                "HUGGINGFACE_API_KEY",
                "required for the initial gated AI4Bharat model download",
            )
        else:
            yield _pass(sec, "HUGGINGFACE_API_KEY", mask_secret(hf_key))

    if ai_provider == "openrouter":
        key = env("OPENROUTER_API_KEY")
        if not key:
            yield _fail(sec, "OPENROUTER_API_KEY", "required when AI_PROVIDER=openrouter")
        else:
            yield _pass(sec, "OPENROUTER_API_KEY", mask_secret(key))

    if ai_provider == "gemini":
        key = env("GEMINI_API_KEY")
        if not key:
            yield _fail(sec, "GEMINI_API_KEY", "required when AI_PROVIDER=gemini")
        else:
            yield _pass(sec, "GEMINI_API_KEY", mask_secret(key))

    if ai_provider == "auto":
        configured = [
            name
            for name, key in (
                ("openai", env("OPENAI_API_KEY")),
                ("sarvam", sarvam_key),
                ("openrouter", env("OPENROUTER_API_KEY")),
                ("gemini", env("GEMINI_API_KEY")),
            )
            if key
        ]
        if configured:
            yield _pass(sec, "AI_PROVIDER_AUTO_KEYS", ",".join(configured))
        else:
            yield _fail(sec, "AI_PROVIDER_AUTO_KEYS", "AI_PROVIDER=auto requires at least one real AI key")

    # Expanded TTS provider coverage
    tts_providers = {
        "TTS_PROVIDER": env("TTS_PROVIDER"),
        "VIDEO_TTS_PROVIDER": env("VIDEO_TTS_PROVIDER") or "ai4bharat",
    }
    for k, v in tts_providers.items():
        if v:
            yield _pass(sec, f"{k}_set", v)
        else:
            yield _warn(sec, f"{k}_set", "falls back to default (ai4bharat)")

    # More providers sanity (edge, sarvam, mock, openai-tts etc recognized in code)
    known_tts = {
        "ai4bharat",
        "indic_parler",
        "indic_parler_tts",
        "edge",
        "sarvam",
        "mock",
        "openai",
        "gcp",
        "azure",
        "kokoro",
        "hybrid",
    }
    for k, v in tts_providers.items():
        if v and v.lower() not in known_tts:
            yield _warn(sec, f"{k}_unknown", f"'{v}' not in common known list; ensure provider impl exists")


# ── Section 5b — Billing / Stripe (if wired) ─────────────────────────────────
def check_billing_stripe(is_prod: bool) -> Iterable[Check]:
    """Fail closed when Stripe is partially configured.

    A fully absent Stripe configuration is a valid free-only deployment. Once
    any Stripe setting is supplied, Checkout, signed webhook activation, and
    both student-facing recurring Prices must all be ready together.
    """

    sec = "Billing / Stripe"
    stripe_key = env("STRIPE_SECRET_KEY")
    webhook_secret = env("STRIPE_WEBHOOK_SECRET")
    price_ids = env("STRIPE_PRICE_IDS")
    any_billing_config = bool(stripe_key or webhook_secret or price_ids)

    if not any_billing_config:
        detail = "not configured; paid checkout is safely disabled"
        if is_prod:
            yield _warn(sec, "Stripe billing", detail)
        else:
            yield _pass(sec, "Stripe billing", detail)
        return

    if not stripe_key:
        yield _fail(sec, "STRIPE_SECRET_KEY", "missing while other Stripe settings are present")
    elif is_prod and stripe_key.startswith(("sk_test_", "rk_test_")):
        yield _fail(sec, "STRIPE_SECRET_KEY", "test-mode key configured in production")
    elif is_prod and stripe_key.startswith("sk_live_"):
        yield _warn(
            sec,
            "STRIPE_SECRET_KEY",
            f"{mask_secret(stripe_key)}; prefer a restricted rk_live_ key",
        )
    else:
        yield _pass(sec, "STRIPE_SECRET_KEY", mask_secret(stripe_key))

    if not webhook_secret:
        yield _fail(
            sec,
            "STRIPE_WEBHOOK_SECRET",
            "missing; checkout remains disabled because entitlements cannot be activated safely",
        )
    elif not webhook_secret.startswith("whsec_"):
        yield _fail(sec, "STRIPE_WEBHOOK_SECRET", "must be a Stripe endpoint signing secret")
    else:
        yield _pass(sec, "STRIPE_WEBHOOK_SECRET", mask_secret(webhook_secret))

    mappings: dict[str, str] = {}
    for pair in (price_ids or "").split(","):
        if "=" not in pair:
            continue
        key, value = pair.split("=", 1)
        mappings[key.strip().lower()] = value.strip()
    required_plans = {"popular_299", "premium_599"}
    missing_plans = sorted(
        plan
        for plan in required_plans
        if not mappings.get(plan, "").startswith("price_")
    )
    if missing_plans:
        yield _fail(
            sec,
            "STRIPE_PRICE_IDS",
            f"missing valid recurring Price mapping(s): {', '.join(missing_plans)}",
        )
    else:
        yield _pass(sec, "STRIPE_PRICE_IDS", "Popular and Premium Prices configured")


# ── Section 5c — Diagram logic coverage (generalized video) — static for env robustness ──────────────────
def check_diagram_logic() -> Iterable[Check]:
    sec = "Diagram logic"
    vg_path = BACKEND_DIR / "app" / "routes" / "video_generator.py"
    if not vg_path.exists():
        yield _fail(sec, "diagram module import", "video_generator.py source not found")
        return
    try:
        with open(vg_path, encoding="utf-8") as f:
            tree = ast.parse(f.read(), filename=str(vg_path))

        # Collect defined function names (static, no import/runtime deps needed)
        func_names = {
            node.name
            for node in ast.walk(tree)
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
        }

        # Find _DIAGRAM_KEYWORDS assign and approximate length (static tuple/list)
        kw_len = 0
        for node in ast.walk(tree):
            if isinstance(node, ast.Assign):
                for t in node.targets:
                    if isinstance(t, ast.Name) and t.id == "_DIAGRAM_KEYWORDS":
                        if isinstance(node.value, (ast.Tuple, ast.List)):
                            kw_len = len(node.value.elts)
                        elif isinstance(node.value, ast.Constant) and isinstance(node.value.value, (tuple, list)):
                            kw_len = len(node.value.value)
                        break

        has_needs = "_needs_diagram" in func_names
        has_ensure = "_ensure_diagram_scene" in func_names
        has_generic = "_generic_diagram_scene" in func_names
        has_keywords = kw_len > 5

        if has_needs and has_ensure and has_generic and has_keywords:
            yield _pass(sec, "_ensure_diagram_scene generalized", "present + _needs + _generic + keywords (static)")
            # Behavioral smoke skipped (would require full runtime env + deps); presence + keyword count is sufficient for preflight
            yield _pass(sec, "diagram keyword trigger", f"generalized keywords present (len={kw_len})")
        else:
            missing = []
            if not has_needs:
                missing.append("_needs_diagram")
            if not has_ensure:
                missing.append("_ensure_diagram_scene")
            if not has_generic:
                missing.append("_generic_diagram_scene")
            if not has_keywords:
                missing.append("_DIAGRAM_KEYWORDS (len<=5)")
            yield _fail(sec, "_ensure_diagram_scene generalized", f"missing or incomplete: {', '.join(missing)}")
    except SyntaxError as exc:
        yield _fail(sec, "diagram module import", f"ast parse error in video_generator.py: {exc}")
    except Exception as exc:  # noqa: BLE001
        yield _fail(sec, "diagram module import", f"could not statically inspect video_generator.py: {exc.__class__.__name__}")

# ── Section 5d — New 10kx features (spaced rep + streaming Ask) — static presence ──────────────────
def check_new_features() -> Iterable[Check]:
    sec = "New features (10kx)"
    # Spaced repetition
    fc_path = BACKEND_DIR / "app" / "routes" / "flashcards.py"
    try:
        fc_src = fc_path.read_text(encoding="utf-8") if fc_path.exists() else ""
        has_sm2 = "_sm2_update" in fc_src or "SM-2" in fc_src
        has_due = "/due" in fc_src or "get_due_cards" in fc_src
        has_review = "/review" in fc_src or "record_flashcard_review" in fc_src
        if has_sm2 and has_due and has_review:
            yield _pass(sec, "spaced repetition (SM-2)", "due/review endpoints + scheduler helpers present")
        else:
            yield _warn(sec, "spaced repetition (SM-2)", "helpers/endpoints may be incomplete")
    except Exception:
        yield _warn(sec, "spaced repetition (SM-2)", "could not inspect flashcards.py")

    # Streaming Ask / StudyCast
    ai_path = BACKEND_DIR / "app" / "services" / "ai_provider.py"
    ask_path = BACKEND_DIR / "app" / "routes" / "ask.py"
    try:
        ai_src = ai_path.read_text(encoding="utf-8") if ai_path.exists() else ""
        ask_src = ask_path.read_text(encoding="utf-8") if ask_path.exists() else ""
        has_gen_stream = "def generate_streaming" in ai_src or "generate_streaming" in ai_src
        has_ask_stream = "/ask/stream" in ask_src or "ask_stream" in ask_src
        if has_gen_stream and has_ask_stream:
            yield _pass(sec, "streaming Ask/StudyCast", "generate_streaming + /ask/stream present")
        else:
            yield _warn(sec, "streaming Ask/StudyCast", "streaming support may be partial")
    except Exception:
        yield _warn(sec, "streaming Ask/StudyCast", "could not inspect ai_provider/ask sources")


# ── Section 5 — URL safety ───────────────────────────────────────────────────

def check_url_safety(is_prod: bool) -> Iterable[Check]:
    if not is_prod:
        yield _pass("URL safety", "skipped", "not in production mode")
        return

    sec = "URL safety"
    candidates = {
        "FRONTEND_BASE_URL": env("FRONTEND_BASE_URL"),
        "STORAGE_PUBLIC_BASE_URL": env("STORAGE_PUBLIC_BASE_URL"),
        "GOOGLE_OAUTH_REDIRECT_URI": env("GOOGLE_OAUTH_REDIRECT_URI"),
    }
    for name, value in candidates.items():
        if not value:
            continue
        if value.startswith(LOCAL_HOST_PREFIXES):
            yield _fail(sec, name, f"localhost URL in production: {value}")
        elif value.startswith("http://"):
            yield _warn(sec, name, f"not https: {value}")
        else:
            yield _pass(sec, name, value)


# ── Section 6 — Deployment target ─────────────────────────────────────────────

def check_deploy_target(is_prod: bool) -> Iterable[Check]:
    sec = "Deployment target"
    target = (env("DEPLOY_TARGET") or "").lower()

    if not target:
        if is_prod:
            yield _warn(sec, "DEPLOY_TARGET",
                        "not set — recommend setting to huggingface/railway/flyio/docker")
        else:
            yield _pass(sec, "DEPLOY_TARGET", "not set (dev)")
        return

    if target not in {"huggingface", "railway", "flyio", "docker", "local"}:
        yield _warn(sec, "DEPLOY_TARGET", f"unrecognized value '{target}'")
        return

    yield _pass(sec, "DEPLOY_TARGET", target)

    if target == "huggingface":
        # HF Spaces sets PORT=7860 and expects 0.0.0.0 bind.
        port = env("PORT") or "7860"
        if port != "7860":
            yield _warn(sec, "PORT",
                        f"HF Spaces expects 7860; got {port} (HF auto-sets via $PORT)")
        else:
            yield _pass(sec, "PORT", "7860 (HF Spaces default)")

        # Frontend on Vercel — backend CORS should include vercel.app or custom domain.
        cors = env("CORS_ORIGINS") or ""
        if cors and not any(
            "vercel.app" in o or ".docdoe.ai" in o or "docdoe.ai" in o
            for o in [origin.strip() for origin in cors.split(",")]
        ):
            yield _warn(sec, "Vercel origin in CORS",
                        "no vercel.app or docdoe.ai origin in CORS_ORIGINS — frontend may not reach backend")

        frontend = env("FRONTEND_BASE_URL") or ""
        if frontend and "vercel.app" not in frontend and "docdoe.ai" not in frontend:
            yield _warn(sec, "Vercel frontend URL",
                        f"FRONTEND_BASE_URL '{frontend}' does not look like Vercel/docdoe.ai")


# ── Runner ────────────────────────────────────────────────────────────────────

def collect_all_checks() -> list[Check]:
    is_prod = (env("ENVIRONMENT") or "development").lower() == "production"
    sections: list[Callable[[], Iterable[Check]]] = [
        lambda: check_environment(is_prod),
        lambda: check_database_recovery(is_prod),
        lambda: check_storage(is_prod),
        lambda: check_video_deps(),
        lambda: check_ai_config(is_prod),
        lambda: check_billing_stripe(is_prod),
        lambda: check_diagram_logic(),
        lambda: check_new_features(),
        lambda: check_url_safety(is_prod),
        lambda: check_deploy_target(is_prod),
    ]
    out: list[Check] = []
    for fn in sections:
        out.extend(fn())
    return out


def render_table(checks: list[Check]) -> str:
    lines = []
    current_section = None
    width_status = 6
    width_name = max((len(c.name) for c in checks), default=24)
    width_name = min(max(width_name, 24), 48)
    for c in checks:
        if c.section != current_section:
            lines.append("")
            lines.append(f"-- {c.section} ".ljust(80, "-"))
            current_section = c.section
        icon = {"PASS": "[+]", "WARN": "[!]", "FAIL": "[x]"}[c.status]
        lines.append(
            f"  {icon} {c.status:<{width_status}} {c.name:<{width_name}} {c.detail}"
        )
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    # Allow load from .env file if present (dev convenience).
    # Skip in test mode (PREFLIGHT_SKIP_ENV_FILE=1) so tests aren't polluted.
    env_file = REPO_ROOT / "backend" / ".env"
    if env_file.exists() and not os.environ.get("PREFLIGHT_SKIP_ENV_FILE"):
        try:
            for raw_line in env_file.read_text(encoding="utf-8").splitlines():
                line = raw_line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, _, value = line.partition("=")
                key = key.strip()
                value = value.strip().strip('"').strip("'")
                os.environ.setdefault(key, value)
        except Exception:
            pass  # don't fail preflight on .env parse errors

    checks = collect_all_checks()
    print("DocDoe Production Preflight")
    print("=" * 80)
    print(render_table(checks))
    print()

    fail_count = sum(1 for c in checks if c.status == "FAIL")
    warn_count = sum(1 for c in checks if c.status == "WARN")
    pass_count = sum(1 for c in checks if c.status == "PASS")
    print(f"Summary: {pass_count} PASS · {warn_count} WARN · {fail_count} FAIL")

    if fail_count:
        print("\nFAILED — fix the items above before deploying.")
        return 1
    if warn_count:
        print("\nReady to deploy with WARNINGS. Review them above.")
    else:
        print("\nAll checks passed.")
    return 0


if __name__ == "__main__":
    sys.exit(main())