Spaces:
Sleeping
Sleeping
| """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"] | |