#!/usr/bin/env python3 """Tiny stdlib-only fal.ai client used to build the opus55sites asset library. Usage: fal.py image --prompt "..." --out path.png [--size 1920x1088|landscape_16_9] [--quality high|medium|low] [--ref ref1.jpg --ref ref2.jpg] # refs switch to openai/gpt-image-2/edit (style reference) [--transparent] fal.py video --image still.png --prompt "..." --out clip.mp4 [--model veo|kling] [--duration 6] fal.py optimize-image in.png out.webp [--width 2000] [--q 82] fal.py optimize-video in.mp4 out-base [--width 1280] # writes out-base.mp4 (+ poster out-base.jpg) Reads FAL_API_KEY from env or .env. Every call is appended to tools/fal-log.jsonl. """ import argparse, base64, json, mimetypes, os, subprocess, sys, time, urllib.request, urllib.error HERE = os.path.dirname(os.path.abspath(__file__)) LOG = os.path.join(HERE, "fal-log.jsonl") def key(): k = os.environ.get("FAL_API_KEY") if k: return k for line in open(".env"): if line.startswith("FAL_API_KEY="): return line.split("=", 1)[1].strip().strip('"') sys.exit("FAL_API_KEY not found") def req(url, data=None, method=None): h = {"Authorization": f"Key {key()}", "Content-Type": "application/json"} body = json.dumps(data).encode() if data is not None else None r = urllib.request.Request(url, data=body, headers=h, method=method or ("POST" if body else "GET")) try: with urllib.request.urlopen(r, timeout=120) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: raise RuntimeError(f"{e.code} {e.read()[:600]!r}") def run(model, payload, timeout=900): sub = req(f"https://queue.fal.run/{model}", payload) status_url, resp_url = sub["status_url"], sub["response_url"] t0 = time.time() while True: s = req(status_url) if s.get("status") == "COMPLETED": break if time.time() - t0 > timeout: raise RuntimeError(f"timeout waiting on {model}") time.sleep(4) return req(resp_url) def data_uri(path): if path.startswith("http"): return path mt = mimetypes.guess_type(path)[0] or "image/png" return f"data:{mt};base64," + base64.b64encode(open(path, "rb").read()).decode() def download(url, out): os.makedirs(os.path.dirname(os.path.abspath(out)), exist_ok=True) with urllib.request.urlopen(url, timeout=300) as r, open(out, "wb") as f: f.write(r.read()) def log(entry): entry["ts"] = time.strftime("%Y-%m-%dT%H:%M:%S") with open(LOG, "a") as f: f.write(json.dumps(entry) + "\n") def size_arg(s): if "x" in s: w, h = s.lower().split("x") return {"width": int(w), "height": int(h)} return s def cmd_image(a): payload = {"prompt": a.prompt, "quality": a.quality, "output_format": "png", "num_images": 1, "image_size": size_arg(a.size)} if a.transparent: payload["background"] = "transparent" model = "openai/gpt-image-2" if a.ref: model = "openai/gpt-image-2/edit" payload["image_urls"] = [data_uri(r) for r in a.ref] res = run(model, payload) url = res["images"][0]["url"] download(url, a.out) log({"kind": "image", "model": model, "out": a.out, "url": url, "prompt": a.prompt, "refs": a.ref or []}) print(json.dumps({"out": a.out, "url": url})) def cmd_video(a): img = data_uri(a.image) if a.model == "veo": model = "fal-ai/veo3.1/fast/image-to-video" payload = {"prompt": a.prompt, "image_url": img, "duration": f"{a.duration}s", "resolution": "1080p", "generate_audio": False, "aspect_ratio": a.aspect} else: model = "fal-ai/kling-video/v2.5-turbo/pro/image-to-video" payload = {"prompt": a.prompt, "image_url": img, "duration": "10" if a.duration > 5 else "5"} res = run(model, payload, timeout=1500) url = res["video"]["url"] download(url, a.out) log({"kind": "video", "model": model, "out": a.out, "url": url, "prompt": a.prompt, "image": a.image, "duration": a.duration}) print(json.dumps({"out": a.out, "url": url})) def cmd_opt_image(a): subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", a.inp, "-vf", f"scale='min({a.width},iw)':-2", "-c:v", "libwebp", "-quality", str(a.q), a.out], check=True) print(a.out, os.path.getsize(a.out) // 1024, "KB") def cmd_opt_video(a): base = a.out subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", a.inp, "-an", "-vf", f"scale='min({a.width},iw)':-2,format=yuv420p", "-c:v", "libx264", "-preset", "slow", "-crf", str(a.crf), "-movflags", "+faststart", base + ".mp4"], check=True) subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", base + ".mp4", "-frames:v", "1", "-q:v", "4", base + ".jpg"], check=True) print(base + ".mp4", os.path.getsize(base + ".mp4") // 1024, "KB") def main(): p = argparse.ArgumentParser() sp = p.add_subparsers(dest="cmd", required=True) i = sp.add_parser("image") i.add_argument("--prompt", required=True); i.add_argument("--out", required=True) i.add_argument("--size", default="landscape_16_9"); i.add_argument("--quality", default="high") i.add_argument("--ref", action="append"); i.add_argument("--transparent", action="store_true") v = sp.add_parser("video") v.add_argument("--image", required=True); v.add_argument("--prompt", required=True) v.add_argument("--out", required=True); v.add_argument("--model", default="veo", choices=["veo", "kling"]) v.add_argument("--duration", type=int, default=6); v.add_argument("--aspect", default="16:9") oi = sp.add_parser("optimize-image") oi.add_argument("inp"); oi.add_argument("out"); oi.add_argument("--width", type=int, default=2000) oi.add_argument("--q", type=int, default=82) ov = sp.add_parser("optimize-video") ov.add_argument("inp"); ov.add_argument("out"); ov.add_argument("--width", type=int, default=1280) ov.add_argument("--crf", type=int, default=26) a = p.parse_args() {"image": cmd_image, "video": cmd_video, "optimize-image": cmd_opt_image, "optimize-video": cmd_opt_video}[a.cmd](a) if __name__ == "__main__": main()