model_runtime.py

"""HTTP client for storyteler Modal backend."""
from __future__ import annotations
import os
import httpx

API_URL = os.environ.get(
    "STORYTELER_API_URL",
    "https://rafalbogusdxc--storyteler-api.modal.run",
).rstrip("/")

TIMEOUT_S = 900


class BackendError(RuntimeError):
    pass


def _post(base_url: str, path: str, payload: dict, timeout: float = TIMEOUT_S) -> dict:
    url = f"{base_url}{path}"
    try:
        resp = httpx.post(url, json=payload, timeout=timeout, follow_redirects=True)
        resp.raise_for_status()
        return resp.json()
    except httpx.ConnectError as e:
        raise BackendError(
            f"Cannot reach backend at {base_url} — is the Modal app deployed? ({e})"
        ) from e
    except httpx.ReadTimeout as e:
        raise BackendError(
            "Backend timed out — likely a cold start. Try again in ~1 minute."
        ) from e
    except httpx.HTTPStatusError as e:
        raise BackendError(
            f"Backend error {e.response.status_code}: {e.response.text[:300]}"
        ) from e


def health() -> dict:
    try:
        resp = httpx.get(f"{API_URL}/health", timeout=10, follow_redirects=True)
        resp.raise_for_status()
        return resp.json()
    except Exception:
        return {"status": "unreachable", "url": API_URL}


def enhance_prompt(raw_text: str, parent_context: str | None = None) -> str:
    data = _post(API_URL, "/enhance", {
        "raw_text": raw_text,
        "parent_context": parent_context,
    })
    return data["enhanced"]


def suggest_continuations(parent_description: str) -> list[str]:
    data = _post(API_URL, "/suggest", {
        "parent_description": parent_description,
    })
    return data["suggestions"]


def generate_images(prompt: str, ref_image_b64: str | None = None,
                    seeds: list[int] | None = None) -> list[str]:
    payload = {"prompt": prompt}
    if ref_image_b64:
        payload["ref_image_b64"] = ref_image_b64
    if seeds:
        payload["seeds"] = seeds
    data = _post(API_URL, "/generate-images", payload)
    return data["images_b64"]


def enhance_video_prompts(scene_description: str, n: int = 4) -> list[str]:
    data = _post(API_URL, "/enhance-video", {
        "scene_description": scene_description,
        "n": n,
    })
    return data["enhanced"]


def draft_narration(story_so_far: str, scene_description: str) -> str:
    data = _post(API_URL, "/draft-narration", {
        "story_so_far": story_so_far,
        "scene_description": scene_description,
    })
    return data["text"]


def generate_narration(text: str, voice: str = "af_heart") -> str:
    data = _post(API_URL, "/generate-narration", {
        "text": text,
        "voice": voice,
    })
    return data["audio_b64"]


def generate_video(image_b64: str, prompt: str) -> str:
    payload = {
        "image_b64": image_b64,
        "prompt": prompt,
    }
    data = _post(API_URL, "/generate-video", payload)
    return data["video_b64"]


def image_from_frame(frame_b64: str, prompt: str) -> list[str]:
    data = _post(API_URL, "/image-from-frame", {
        "frame_b64": frame_b64,
        "prompt": prompt,
    })
    return data["images_b64"]