Spaces:
Paused
Paused
File size: 3,787 Bytes
3d2098f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | """Small shared utilities: logging, JSON extraction, retries, ffprobe."""
from __future__ import annotations
import json
import logging
import re
import subprocess
import time
from typing import Any, Callable, Optional
log = logging.getLogger("reelstudio")
if not log.handlers:
_h = logging.StreamHandler()
_h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
log.addHandler(_h)
log.setLevel(logging.INFO)
# ---------------------------------------------------------------------------
# JSON extraction (LLMs love markdown fences)
# ---------------------------------------------------------------------------
_FENCE_RE = re.compile(r"```(?:json)?\s*(.*?)```", re.DOTALL)
def extract_json(text: str) -> Any:
"""Parse JSON out of an LLM response, stripping markdown fences.
Tries, in order: fenced block, the raw text, and the largest
{...} / [...] slice. Raises ValueError if nothing parses.
"""
candidates = []
m = _FENCE_RE.search(text)
if m:
candidates.append(m.group(1))
candidates.append(text.strip())
for opener, closer in (("{", "}"), ("[", "]")):
start, end = text.find(opener), text.rfind(closer)
if start != -1 and end > start:
candidates.append(text[start : end + 1])
for c in candidates:
try:
return json.loads(c)
except (json.JSONDecodeError, TypeError):
continue
raise ValueError(f"No parseable JSON in LLM response: {text[:200]!r}")
# ---------------------------------------------------------------------------
# Retry with exponential backoff
# ---------------------------------------------------------------------------
class HardAPIError(Exception):
"""Non-retryable API failure (400/401/403/404 class)."""
def retry_backoff(
fn: Callable[[], Any],
attempts: int = 4,
base_delay: float = 2.0,
retry_on: tuple = (Exception,),
label: str = "call",
) -> Any:
"""Run ``fn`` with exponential backoff. HardAPIError is never retried."""
last: Optional[Exception] = None
for i in range(attempts):
try:
return fn()
except HardAPIError:
raise
except retry_on as e: # noqa: PERF203
last = e
delay = base_delay * (2**i)
log.warning("%s failed (attempt %d/%d): %s — retrying in %.1fs", label, i + 1, attempts, e, delay)
if i < attempts - 1:
time.sleep(delay)
raise last # type: ignore[misc]
# ---------------------------------------------------------------------------
# FFmpeg helpers
# ---------------------------------------------------------------------------
def run_ffmpeg(args: list[str], label: str = "ffmpeg") -> None:
"""Run an ffmpeg command, raising with stderr tail on failure."""
cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", *args]
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
tail = (proc.stderr or "")[-2000:]
raise RuntimeError(f"{label} failed (exit {proc.returncode}): {tail}")
def ffprobe_duration(path: str) -> float:
"""True media duration in seconds via ffprobe."""
proc = subprocess.run(
[
"ffprobe", "-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
path,
],
capture_output=True,
text=True,
)
if proc.returncode != 0:
raise RuntimeError(f"ffprobe failed for {path}: {proc.stderr[-500:]}")
return float(proc.stdout.strip())
def slugify(text: str, max_len: int = 40) -> str:
s = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
return s[:max_len] or "untitled"
|