#!/usr/bin/env python3 """Run one video transcript-and-summary provider on the benchmark video. run.py --video PATH --out OUT_DIR --provider NAME Providers: twelvelabs, echosaw (API-driven), screenapp, otter (UI-only services; pass --import FILE with the JSON their web app returns for the uploaded recording, see the docstrings below). Each writes OUT_DIR/.json with: transcript plain-text transcript as returned by the service summary the service's own summary of the meeting wall_seconds per-phase timings (upload, index/process, transcript, summary) and total version model / API version string reported or documented by the service run_at UTC timestamp Credentials are read from the environment and never written to disk: TWELVELABS_API_KEY twelvelabs ECHOSAW_API_KEY, ECHOSAW_API_BASE echosaw (gamma /v1 REST API) """ import argparse import datetime as dt import json import os import time SUMMARY_PROMPT = ( "Summarize this meeting recording in 5-8 sentences. Cover the purpose of the " "meeting, who the participants are (by role, not name), the main topics " "discussed, and any decisions or action items." ) def now(): return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") # ---------------------------------------------------------------- Twelve Labs def run_twelvelabs(video): from twelvelabs import TwelveLabs from twelvelabs.types import VideoContext_AssetId client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"]) timings = {} t = time.time() with open(video, "rb") as f: asset = client.assets.create(method="direct", file=f, filename=os.path.basename(video)) while client.assets.retrieve(asset.id).status not in ("ready", "failed"): time.sleep(3) timings["upload"] = round(time.time() - t, 1) t = time.time() index = client.indexes.create( index_name=f"curated-lists-bench-{int(t)}", models=[{"model_name": "marengo3.0", "model_options": ["visual", "audio"]}], ) indexed = client.indexes.indexed_assets.create(index.id, asset_id=asset.id) while True: st = client.indexes.indexed_assets.retrieve(index.id, indexed.id) if st.status in ("ready", "failed"): break time.sleep(5) timings["index"] = round(time.time() - t, 1) video_id = st.id t = time.time() v = client.indexes.videos.retrieve(index.id, video_id, transcription=True) segs = v.transcription or [] transcript = " ".join(s.value for s in segs if getattr(s, "value", None)) timings["transcript"] = round(time.time() - t, 1) t = time.time() a = client.analyze( model_name="pegasus1.5", video=VideoContext_AssetId(type="asset_id", asset_id=asset.id), prompt=SUMMARY_PROMPT, temperature=0.2, ) summary = a.data if isinstance(a.data, str) else json.dumps(a.data) timings["summary"] = round(time.time() - t, 1) return { "transcript": transcript, "summary": summary, "wall_seconds": {**timings, "total": round(sum(timings.values()), 1)}, "version": "pegasus1.5 analyze + marengo3.0 index (visual+audio) transcription, twelvelabs SDK 1.3.4", "notes": {"transcript_segments": len(segs)}, } # ---------------------------------------------------------------- Echosaw def run_echosaw(video): """Echosaw public API: POST /v1/analyze/url with a fetchable URL of the video (ECHOSAW_VIDEO_URL; the file itself is not uploaded from this machine), poll /v1/analysis/status/{mediaId}, then read /v1/analysis/results/{mediaId}. Transcript + summary are produced by the single analysis pass, so "index" is the full pipeline time and transcript/summary phases are only the fetch.""" import requests base = os.environ["ECHOSAW_API_BASE"].rstrip("/") h = {"X-Api-Key": os.environ["ECHOSAW_API_KEY"]} timings = {} t = time.time() r = requests.post(f"{base}/v1/analyze/url", headers=h, json={ "url": os.environ["ECHOSAW_VIDEO_URL"], "mediaType": "video", "originalFilename": os.path.basename(video), }) r.raise_for_status() media_id = r.json()["mediaId"] timings["upload"] = round(time.time() - t, 1) t = time.time() while True: s = requests.get(f"{base}/v1/analysis/status/{media_id}", headers=h).json() if s["status"].lower() in ("complete", "failed"): break time.sleep(10) timings["index"] = round(time.time() - t, 1) t = time.time() rep = requests.get(f"{base}/v1/analysis/results/{media_id}", headers=h).json() timings["transcript"] = round(time.time() - t, 1) timings["summary"] = 0.0 summ = rep["summary"] summary = f"{summ['title']}\n\n{summ['description']}\n\n{rep['semantics']['summary']}" return { "transcript": rep["transcript"]["text"], "summary": summary, "wall_seconds": {**timings, "total": round(sum(timings.values()), 1)}, "version": "Echosaw gamma pipeline, /v1/analyze/url", "notes": { "transcript_segments": len(rep["transcript"]["segments"]), "billed_minutes": rep["billing"]["billedMinutes"], "rate_per_minute_usd": rep["billing"]["ratePerMinute"], "timeline_entries": len(rep["timeline"]), "key_phrases": rep["semantics"]["keyPhrases"], }, } # ---------------------------------------------------------------- ScreenApp (UI) def run_screenapp(video, imported): """ScreenApp has no public API on the free plan. Upload the video in the web app (Media Files > New > Upload File), then save the JSON the app fetches for the recording (/app/api/files//meta, /transcript, /summary) as {"meta": ..., "transcript": ..., "summary": ...} and pass it with --import. Turnaround is upload-started (noted by hand) to summary.generatedAt.""" d = imported tr, su, meta = d["transcript"], d["summary"], d["meta"] started = dt.datetime.fromisoformat(d["upload_started_at"]) done = dt.datetime.fromisoformat(su["generatedAt"].replace("Z", "+00:00")) total = round((done - started).total_seconds(), 1) summary = su["overview"] + "\n\n" + "\n".join( f"{c['title']}: {c['notes']}" for c in su["chapters"] ) return { "transcript": tr["text"], "summary": summary, "wall_seconds": {"upload": None, "index": None, "transcript": None, "summary": None, "total": total}, "version": ( f"ScreenApp web app, free plan; transcript provider '{tr['provider']}', " f"summary provider '{su['provider']}' (as reported by the app)" ), "notes": { "transcript_segments": len(tr["segments"]), "speakers": len(meta.get("speakerNames") or {}), "chapters": len(su["chapters"]), }, } # ---------------------------------------------------------------- Otter (UI) def run_otter(video, imported): """Otter has no self-serve API on the Basic plan. Import the video in the web app (Import > Browse files), then save the JSON the app fetches for the conversation (/forward/api/v1/speech and /abstract_summary) as {"transcript": joined text, "segments": n, "created_at": ..., "modified_time": ..., "abstract_summary": ..., "outline": ..., "summary": keywords} and pass --import. Turnaround is created_at (upload finished) to modified_time (processing done).""" d = imported total = round(d["modified_time"] - d["created_at"], 1) summary = d["abstract_summary"]["short_summary"] return { "transcript": d["transcript"], "summary": summary, "wall_seconds": {"upload": None, "index": None, "transcript": None, "summary": None, "total": total}, "version": "Otter web app, Basic plan, English (US) import", "notes": { "transcript_segments": d["segments"], "outline_sections": len(d["outline"]), "keywords": d["summary"], }, } PROVIDERS = { "twelvelabs": lambda v, _i: run_twelvelabs(v), "echosaw": lambda v, _i: run_echosaw(v), "screenapp": run_screenapp, "otter": run_otter, } def main(): ap = argparse.ArgumentParser() ap.add_argument("--video", required=True) ap.add_argument("--out", required=True) ap.add_argument("--provider", required=True, choices=sorted(PROVIDERS)) ap.add_argument("--import", dest="imported", help="JSON captured from a UI-only service") a = ap.parse_args() os.makedirs(a.out, exist_ok=True) imported = None if a.imported: with open(a.imported) as f: imported = json.load(f) result = PROVIDERS[a.provider](a.video, imported) result["run_at"] = now() result["video"] = os.path.basename(a.video) with open(os.path.join(a.out, f"{a.provider}.json"), "w") as f: json.dump(result, f, indent=1) print(json.dumps(result["wall_seconds"]), len(result["transcript"].split()), "words") if __name__ == "__main__": main()