"""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"]