1#!/usr/bin/env python3
2"""OmniHuman 1.5 — generate lipsync video from image + audio via fal.ai."""
3import argparse
4import json
5import os
6from pathlib import Path
7from urllib.request import urlopen
8 
9 
10def load_fal_key() -> str:
11    for name in ("FAL_KEY", "FAL_AI_KEY", "FAL_API_KEY"):
12        value = os.getenv(name)
13        if value:
14            return value.strip()
15    raise SystemExit("Missing fal.ai key. Set FAL_KEY or FAL_AI_KEY.")
16 
17 
18def main() -> None:
19    parser = argparse.ArgumentParser(description="OmniHuman 1.5 lipsync: image + audio → video")
20    parser.add_argument("--image", required=True, help="Character image path.")
21    parser.add_argument("--audio", required=True, help="Audio file path (mp3/wav).")
22    parser.add_argument("--output", required=True, help="Output mp4 path.")
23    parser.add_argument("--json-out", help="Optional path to save raw fal response JSON.")
24    args = parser.parse_args()
25 
26    os.environ["FAL_KEY"] = load_fal_key()
27    import fal_client
28 
29    image_path = Path(args.image).expanduser().resolve()
30    audio_path = Path(args.audio).expanduser().resolve()
31    output_path = Path(args.output).expanduser().resolve()
32    output_path.parent.mkdir(parents=True, exist_ok=True)
33 
34    print(f"Uploading image: {image_path}")
35    image_url = fal_client.upload_file(str(image_path))
36    print(f"Uploading audio: {audio_path}")
37    audio_url = fal_client.upload_file(str(audio_path))
38 
39    model = "fal-ai/bytedance/omnihuman/v1.5"
40    print(f"Submitting to {model}...")
41 
42    def _print_logs(update):
43        status = getattr(update, "status", None)
44        if status:
45            print(f"STATUS: {status}")
46 
47    result = fal_client.subscribe(
48        model,
49        arguments={"image_url": image_url, "audio_url": audio_url},
50        with_logs=True,
51        on_queue_update=_print_logs,
52    )
53 
54    if args.json_out:
55        json_out = Path(args.json_out).expanduser().resolve()
56        json_out.parent.mkdir(parents=True, exist_ok=True)
57        json_out.write_text(json.dumps(result, ensure_ascii=False, indent=2))
58 
59    video = result.get("video") if isinstance(result, dict) else None
60    if not isinstance(video, dict) or not video.get("url"):
61        raise SystemExit(f"Unexpected result: {json.dumps(result, ensure_ascii=False)[:2000]}")
62 
63    print(f"Downloading: {video['url']}")
64    with urlopen(video["url"], timeout=300) as resp:
65        output_path.write_bytes(resp.read())
66    print(f"Saved: {output_path}")
67    print(json.dumps({"ok": True, "output": str(output_path), "duration": result.get("duration")}))
68 
69 
70if __name__ == "__main__":
71    main()
72