Build an OpusClip clone: the whole pipeline in Python

Times are wall clock from our runs. Workers AI bills $0.011 per 1,000 neurons, and the first 10,000 neurons a day are free.

1. 1

   Install, download and pull the audio

   You need Python, ffmpeg, MediaPipe for faces and a Cloudflare account for Workers AI. Whisper only needs mono 16 kHz audio, which keeps the upload small.

   python3 -m venv venv && . venv/bin/activate
   pip install -q mediapipe==1.2.2
   curl -sO https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_full_range/float16/1/blaze_face_full_range.tflite
   curl -sO https://api.clipsdk.com/f/lib/demo/rocket-ranch-podcast.mp4
   ffmpeg -v error -i rocket-ranch-podcast.mp4 -vn -ac 1 -ar 16000 -b:a 32k audio.mp3
   export CF_ACCOUNT_ID=your-account-id CF_API_TOKEN=your-workers-ai-token
   

2. 2

   Call Workers AI from Python

   Both models run on Cloudflare's REST API, so one small helper covers the transcript and the ranking call.

   import json
   import os
   import urllib.request
   
   ACCOUNT = os.environ["CF_ACCOUNT_ID"]
   TOKEN = os.environ["CF_API_TOKEN"]
   
   def run(model, payload):
       request = urllib.request.Request(
           f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT}/ai/run/{model}",
           data=json.dumps(payload).encode(),
           headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"},
       )
       with urllib.request.urlopen(request, timeout=300) as response:
           return json.load(response)["result"]
   

3. 3

   Transcribe with word timestamps

   Whisper large v3 turbo returns segments with a start and end time per word. The 10-minute episode came back as a full word-level transcript in under a minute.

   import base64
   import json
   from workers_ai import run
   
   audio = base64.b64encode(open("audio.mp3", "rb").read()).decode()
   result = run("@cf/openai/whisper-large-v3-turbo", {"audio": audio, "language": "en"})
   
   words = [
       {"word": w["word"].strip(), "start": w["start"], "end": w["end"]}
       for segment in result["segments"]
       for w in segment["words"]
   ]
   json.dump(words, open("words.json", "w"))
   
   print(f'{result["transcription_info"]["duration"]:.1f} s of audio, {len(words)} words')
   print(words[:4])
   
   # $ time python3 transcribe.py
   # 570.3 s of audio, 1697 words
   # [{'word': 'It', 'start': 0, 'end': 0.48}, {'word': 'is', 'start': 0.48, 'end': 0.64}, {'word': 'an', 'start': 0.64, 'end': 0.76}, {'word': 'exciting', 'start': 0.76, 'end': 1.32}]
   #
   # real	0m55.102s
   # user	0m0.118s
   # sys	0m0.029s
   

4. 4

   Number the transcript lines

   The model should answer with line numbers, not timestamps it might invent. A line ends at a full stop, a question mark or a pause over 0.7 s.

   import json
   
   words = json.load(open("words.json"))
   
   # One numbered line per sentence, or per stretch of speech that ends in a pause
   lines, current = [], []
   for i, word in enumerate(words):
       current.append(word)
       next_start = words[i + 1]["start"] if i + 1 < len(words) else None
       if word["word"][-1] in ".?!" or next_start is None or next_start - word["end"] > 0.7:
           lines.append({
               "id": len(lines),
               "start": current[0]["start"],
               "end": word["end"],
               "text": " ".join(w["word"] for w in current),
           })
           current = []
   
   json.dump(lines, open("lines.json", "w"))
   print(len(lines), "lines")
   for line in lines[60:63]:
       print(f'[{line["id"]}] {line["start"]:.1f}-{line["end"]:.1f}: {line["text"]}')
   
   # $ python3 lines.py
   # 158 lines
   # [60] 236.7-237.9: Now the science begins.
   # [61] 237.9-246.9: The scary part of getting off the earth, traveling, you know, 300 million miles and landing on the surface is over, now the science can begin.
   # [62] 247.9-260.0: You know, one of the things I think of when they do the EDL, when it lands, is that I think, I don't know if it was during Curiosity or one of the recent missions where, I believe there was a choreographed dance move.
   

5. 5

   Let the model pick the moments

   Gemma 4 reads the numbered transcript and returns 3 clips as JSON. The line numbers turn back into exact start and end times. Check what it returns before you cut: our LLM moments guide shows the checks.

   import json
   from workers_ai import run
   
   lines = json.load(open("lines.json"))
   transcript = "\n".join(f'[{l["id"]}] {l["start"]:.1f}-{l["end"]:.1f}: {l["text"]}' for l in lines)
   
   prompt = f"""Pick the 3 best 20 to 60 second clips from this podcast for TikTok, Reels and Shorts.
   Each clip needs a hook in its first line, one complete thought, and a payoff at the end.
   Clips are runs of whole lines and must not overlap. Lines are [id] start-end seconds: text.
   Score each clip 0 to 100 for how well it works on its own.
   Return JSON: {{"clips": [{{"first_line": 0, "last_line": 0, "title": "", "score": 0}}]}}
   
   {transcript}"""
   
   result = run("@cf/google/gemma-4-26b-a4b-it", {
       "messages": [{"role": "user", "content": prompt}],
       "response_format": {"type": "json_object"},
   })
   clips = json.loads(result["choices"][0]["message"]["content"])["clips"]
   
   by_id = {l["id"]: l for l in lines}
   for clip in clips:
       clip["start"], clip["end"] = by_id[clip["first_line"]]["start"], by_id[clip["last_line"]]["end"]
       print(f'{clip["start"]:.1f}-{clip["end"]:.1f} s, score {clip["score"]}: {clip["title"]}')
   json.dump(clips, open("clips.json", "w"), indent=2)
   print(result["usage"])
   
   # $ time python3 pick.py
   # 79.6-123.6 s, score 95: The Seven Minutes of Terror
   # 187.8-211.9 s, score 88: The Joy of a Successful Landing
   # 426.0-462.1 s, score 92: Searching for Biosignatures
   # {'prompt_tokens': 5112, 'completion_tokens': 4189, 'total_tokens': 9301, 'prompt_tokens_details': {'cached_tokens': 0}, 'neurons': 160.71817016601562}
   #
   # real	1m25.179s
   # user	0m0.075s
   # sys	0m0.030s
   

6. 6

   Reframe, caption and render

   The top pick goes to ffmpeg in one pass: a 9:16 crop centred on the biggest face, captions 3 words at a time from the word timestamps, and a 1080x1920 MP4.

   import json
   import subprocess
   import cv2
   import mediapipe as mp
   from mediapipe.tasks.python import BaseOptions, vision
   
   SRC = "rocket-ranch-podcast.mp4"
   clip = json.load(open("clips.json"))[0]
   start, end = clip["start"], clip["end"]
   
   # Reframe: centre a 9:16 crop on the median position of the biggest face
   video = cv2.VideoCapture(SRC)
   W, H = int(video.get(cv2.CAP_PROP_FRAME_WIDTH)), int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
   crop_w = round(H * 9 / 16 / 2) * 2
   detector = vision.FaceDetector.create_from_options(vision.FaceDetectorOptions(
       base_options=BaseOptions(model_asset_path="blaze_face_full_range.tflite")))
   xs = []
   for i in range(10):
       video.set(cv2.CAP_PROP_POS_MSEC, (start + (end - start) * (i + 0.5) / 10) * 1000)
       ok, frame = video.read()
       found = detector.detect(mp.Image(image_format=mp.ImageFormat.SRGB, data=cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))).detections
       if found:
           box = max(found, key=lambda d: d.bounding_box.width).bounding_box
           xs.append(box.origin_x + box.width / 2)
   centre = sorted(xs)[len(xs) // 2] if xs else W / 2
   left = int(min(max(centre - crop_w / 2, 0), W - crop_w))
   
   # Captions: 3 words at a time, mid-frame, timed from the word timestamps
   def ts(seconds):
       return f"{int(seconds // 3600)}:{int(seconds % 3600 // 60):02d}:{seconds % 60:05.2f}"
   
   words = [w for w in json.load(open("words.json")) if start <= w["start"] < end]
   events = []
   for i in range(0, len(words), 3):
       group = words[i:i + 3]
       text = " ".join(w["word"] for w in group).upper()
       events.append(f"Dialogue: 0,{ts(group[0]['start'] - start)},{ts(group[-1]['end'] - start)},Caption,,0,0,0,,{text}")
   open("captions.ass", "w").write("\n".join([
       "[Script Info]", "PlayResX: 1080", "PlayResY: 1920", "",
       "[V4+ Styles]",
       "Format: Name, Fontname, Fontsize, PrimaryColour, OutlineColour, Bold, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV",
       "Style: Caption,DejaVu Sans,80,&H00FFFFFF,&H00000000,1,1,6,0,5,130,130,0", "",
       "[Events]", "Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text",
       *events,
   ]))
   
   # Render: cut, crop, scale to 1080x1920 and burn the captions in one pass
   subprocess.run(["ffmpeg", "-v", "error", "-y", "-ss", f"{start}", "-t", f"{end - start}", "-i", SRC,
                   "-vf", f"crop={crop_w}:{H}:{left}:0,scale=1080:1920,ass=captions.ass",
                   "-c:v", "libx264", "-crf", "20", "-c:a", "aac", "clip.mp4"], check=True)
   print(f"crop x={left} from {len(xs)}/10 face samples, {len(events)} captions, clip.mp4 is {end - start:.1f} s")
   
   # $ time python3 render.py
   # crop x=476 from 5/10 face samples, 50 captions, clip.mp4 is 44.0 s
   #
   # real	1m49.951s
   # user	3m10.295s
   # sys	0m11.257s