modelbeast/server/operators/ffprobe/run.py
2026-07-12 21:05:22 +10:00

37 lines
1.3 KiB
Python

import argparse
import json
import subprocess
from pathlib import Path
ap = argparse.ArgumentParser()
ap.add_argument("--input", required=True)
ap.add_argument("--outdir", required=True)
ap.add_argument("--params", default="{}")
args = ap.parse_args()
out = subprocess.run(
["ffprobe", "-v", "error", "-print_format", "json",
"-show_format", "-show_streams", args.input],
capture_output=True, text=True, check=True)
probe = json.loads(out.stdout)
summary = {"streams": []}
fmt = probe.get("format", {})
summary["container"] = fmt.get("format_long_name")
summary["duration_s"] = float(fmt.get("duration", 0) or 0)
summary["size_bytes"] = int(fmt.get("size", 0) or 0)
for s in probe.get("streams", []):
entry = {"type": s.get("codec_type"), "codec": s.get("codec_name")}
if s.get("codec_type") == "video":
entry["resolution"] = f"{s.get('width')}x{s.get('height')}"
entry["fps"] = s.get("avg_frame_rate")
entry["frames"] = s.get("nb_frames")
summary["streams"].append(entry)
print(json.dumps(summary, indent=2))
outdir = Path(args.outdir)
name = Path(args.input).stem + ".probe.json"
(outdir / name).write_text(json.dumps({"summary": summary, "raw": probe}, indent=2))
(outdir / "result.json").write_text(json.dumps(
{"outputs": [{"path": name, "meta": {"summary": summary}}]}))