"""Google Veo & Runway – Text-to-Video für Jami-Promos."""

from __future__ import annotations

import json
import os
import time
from pathlib import Path
from typing import Any

import requests

ROOT = Path(__file__).resolve().parents[1]
SECRETS = ROOT / "data" / "jami-secrets.local.json"

RUNWAY_BASE = "https://api.dev.runwayml.com/v1"
RUNWAY_VERSION = "2024-11-06"


def _load_secrets() -> dict:
    if not SECRETS.is_file():
        return {}
    try:
        return json.loads(SECRETS.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}


def gemini_api_key() -> str:
    key = os.environ.get("GEMINI_API_KEY", "").strip()
    if key:
        return key
    key = os.environ.get("GOOGLE_API_KEY", "").strip()
    if key:
        return key
    data = _load_secrets()
    return str(
        data.get("gemini_api_key")
        or data.get("google_api_key")
        or data.get("GEMINI_API_KEY")
        or ""
    ).strip()


def runway_api_key() -> str:
    key = os.environ.get("RUNWAYML_API_SECRET", "").strip()
    if key:
        return key
    key = os.environ.get("RUNWAY_API_KEY", "").strip()
    if key:
        return key
    data = _load_secrets()
    return str(data.get("runway_api_key") or data.get("RUNWAYML_API_SECRET") or "").strip()


def download_video(url: str, dest: Path, *, timeout: int = 300) -> Path:
    dest.parent.mkdir(parents=True, exist_ok=True)
    with requests.get(url, stream=True, timeout=timeout) as resp:
        resp.raise_for_status()
        with dest.open("wb") as f:
            for chunk in resp.iter_content(65536):
                if chunk:
                    f.write(chunk)
    return dest


def _runway_headers() -> dict[str, str]:
    key = runway_api_key()
    if not key:
        raise RuntimeError(
            "Runway API-Key fehlt. Setze RUNWAYML_API_SECRET oder "
            'runway_api_key in data/jami-secrets.local.json · Key: https://dev.runwayml.com'
        )
    return {
        "Authorization": f"Bearer {key}",
        "Content-Type": "application/json",
        "X-Runway-Version": RUNWAY_VERSION,
    }


def runway_wait_task(task_id: str, *, timeout_sec: float = 900, poll_sec: float = 5.0) -> dict:
    deadline = time.time() + timeout_sec
    last: dict[str, Any] = {}
    while time.time() < deadline:
        r = requests.get(
            f"{RUNWAY_BASE}/tasks/{task_id}",
            headers=_runway_headers(),
            timeout=60,
        )
        r.raise_for_status()
        last = r.json()
        status = str(last.get("status") or "").upper()
        if status in ("SUCCEEDED", "COMPLETED", "COMPLETE"):
            return last
        if status in ("FAILED", "CANCELLED", "CANCELED"):
            raise RuntimeError(str(last.get("failure") or last.get("failureCode") or "Runway Task fehlgeschlagen"))
        time.sleep(poll_sec)
    raise TimeoutError(f"Runway Timeout nach {int(timeout_sec)}s (Status: {last.get('status')})")


def runway_text_to_video(
    prompt: str,
    dest: Path,
    *,
    model: str = "gen4.5",
    ratio: str = "720:1280",
    duration: int = 8,
    timeout_sec: float = 900,
) -> Path:
    """Runway text-to-video (gen4.5, veo3.1, seedance2, …)."""
    body = {
        "model": model,
        "promptText": prompt.strip()[:4000],
        "ratio": ratio,
        "duration": max(2, min(15, int(duration))),
    }
    r = requests.post(
        f"{RUNWAY_BASE}/text_to_video",
        headers=_runway_headers(),
        json=body,
        timeout=60,
    )
    if r.status_code >= 400:
        raise RuntimeError(f"Runway API {r.status_code}: {r.text[:400]}")
    data = r.json()
    task_id = str(data.get("id") or "")
    if not task_id:
        raise RuntimeError("Runway: keine Task-ID erhalten")

    task = runway_wait_task(task_id, timeout_sec=timeout_sec)
    outputs = task.get("output") or []
    if not outputs:
        raise RuntimeError("Runway: kein Video in der Antwort")
    url = str(outputs[0])
    return download_video(url, dest)


def generate_veo_video(
    prompt: str,
    dest: Path,
    *,
    model: str = "veo-3.1-generate-preview",
    duration: int = 8,
    aspect_ratio: str = "9:16",
    resolution: str = "1080p",
    timeout_sec: float = 1200,
    poll_sec: float = 15.0,
) -> Path:
    """Google Veo über Gemini API (google-genai SDK)."""
    key = gemini_api_key()
    if not key:
        raise RuntimeError(
            "Gemini/Google API-Key fehlt für Veo. Setze GEMINI_API_KEY oder "
            'gemini_api_key in data/jami-secrets.local.json · https://aistudio.google.com/apikey'
        )
    try:
        from google import genai
        from google.genai import types
    except ImportError as e:
        raise RuntimeError("pip install google-genai  # Veo SDK fehlt") from e

    client = genai.Client(api_key=key)
    config_kwargs: dict[str, Any] = {
        "number_of_videos": 1,
        "duration_seconds": max(4, min(8, int(duration))),
        "aspect_ratio": aspect_ratio,
        "enhance_prompt": True,
    }
    if resolution:
        config_kwargs["resolution"] = resolution

    operation = client.models.generate_videos(
        model=model,
        prompt=prompt.strip()[:4000],
        config=types.GenerateVideosConfig(**config_kwargs),
    )

    deadline = time.time() + timeout_sec
    while not operation.done:
        if time.time() > deadline:
            raise TimeoutError(f"Veo Timeout nach {int(timeout_sec)}s")
        time.sleep(poll_sec)
        operation = client.operations.get(operation)

    if not operation.result or not operation.result.generated_videos:
        raise RuntimeError("Veo: keine Videos in der Antwort")

    generated = operation.result.generated_videos[0]
    dest.parent.mkdir(parents=True, exist_ok=True)
    client.files.download(file=generated.video)
    generated.video.save(str(dest))
    return dest


def generate_ai_video(
    prompt: str,
    dest: Path,
    *,
    provider: str = "runway",
    model: str | None = None,
    ratio: str = "720:1280",
    duration: int = 8,
    aspect_ratio: str = "9:16",
    resolution: str = "1080p",
    timeout_sec: float = 900,
) -> tuple[Path, str]:
    """Einheitlicher Einstieg – provider: runway | veo."""
    prov = (provider or "runway").strip().lower()
    if prov == "veo":
        m = model or "veo-3.1-generate-preview"
        path = generate_veo_video(
            prompt,
            dest,
            model=m,
            duration=duration,
            aspect_ratio=aspect_ratio,
            resolution=resolution,
            timeout_sec=max(timeout_sec, 1200),
        )
        return path, f"veo/{m}"
    if prov == "runway":
        m = model or "gen4.5"
        path = runway_text_to_video(
            prompt,
            dest,
            model=m,
            ratio=ratio,
            duration=duration,
            timeout_sec=timeout_sec,
        )
        return path, f"runway/{m}"
    raise ValueError(f"Unbekannter Provider: {provider} (runway | veo)")
