File size: 1,693 Bytes
bbe6bab | 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 | """
SadTalker client — sends portrait + audio to the MI300X server, gets MP4 back.
Server (droplet port 8003):
POST /generate multipart: audio file + profile_id form field → MP4 bytes
GET /health → {"status":"ok"}
"""
import os
import tempfile
import httpx
TALKING_HEAD_URL = os.getenv("TALKING_HEAD_URL", "")
TIMEOUT = float(os.getenv("TALKING_HEAD_TIMEOUT", "120"))
def generate_talking_video(audio_path: str, profile_id: str) -> str | None:
"""
Animate the profile portrait to match the given audio.
Returns path to a temp MP4 file, or None if unavailable/failed.
"""
if not TALKING_HEAD_URL or not audio_path:
return None
if not os.path.exists(audio_path):
return None
try:
with open(audio_path, "rb") as f:
audio_bytes = f.read()
suffix = os.path.splitext(audio_path)[1] or ".mp3"
mime = "audio/mpeg" if suffix == ".mp3" else "audio/wav"
r = httpx.post(
f"{TALKING_HEAD_URL.rstrip('/')}/generate",
files={"audio": (f"audio{suffix}", audio_bytes, mime)},
data={"profile_id": profile_id},
timeout=TIMEOUT,
)
r.raise_for_status()
tmp = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
tmp.write(r.content)
tmp.close()
return tmp.name
except Exception as e:
print(f"[TalkingHead] {e}")
return None
def is_available() -> bool:
if not TALKING_HEAD_URL:
return False
try:
r = httpx.get(f"{TALKING_HEAD_URL.rstrip('/')}/health", timeout=5.0)
return r.status_code == 200
except Exception:
return False
|