#!/usr/bin/env python3 """DaDaScribe upload helper for large files (up to 10 GB). Uploads a local audio/video file straight to DaDaScribe storage through the /v1/upload/* endpoints and, optionally, submits it for transcription. Library: from dadascribe_upload import upload_file, transcribe_file result = transcribe_file("YOUR_API_KEY", "talk.mp4", source_language="en") print(result["id"]) # batch id for /v1/status upload_token = upload_file("YOUR_API_KEY", "talk.mp4") # upload only Command line: export DADASCRIBE_API_KEY=YOUR_API_KEY python3 dadascribe_upload.py talk.mp4 --source-language en python3 dadascribe_upload.py talk.mp4 --source-language en --destination-language es,it --diarization "Ann,Bob" python3 dadascribe_upload.py talk.mp4 --upload-only # prints the upload_token Requires Python 3.8+ and no third-party packages. """ import argparse import json import mimetypes import os import sys import threading import time import urllib.error import urllib.request from concurrent.futures import ThreadPoolExecutor DEFAULT_API_BASE = "https://api.dadascribe.com/v1" URL_BATCH = 100 RETRIES = 3 class DadaScribeError(Exception): pass def _api_base(api_base): return (api_base or os.environ.get("DADASCRIBE_API_BASE") or DEFAULT_API_BASE).rstrip("/") def _post(api_base, api_key, path, payload, retry_server_errors=True): # /upload/complete and /transcribe are not safe to repeat after a server-side success, # so they are only retried on 429 (rejected before any work is done). data = json.dumps(payload).encode() attempt = 0 while True: request = urllib.request.Request( api_base + path, data=data, method="POST", headers={"Authorization": "Bearer " + api_key, "Content-Type": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=300) as response: return json.loads(response.read().decode()) except urllib.error.HTTPError as e: body = e.read().decode(errors="replace") try: message = json.loads(body).get("message", body) except ValueError: message = body retryable = e.code == 429 or (retry_server_errors and e.code >= 500) if retryable and attempt < RETRIES: time.sleep((5 if e.code == 429 else 1) * 2 ** attempt) attempt += 1 continue raise DadaScribeError("%s failed (HTTP %d): %s" % (path, e.code, message)) except (urllib.error.URLError, OSError) as e: if retry_server_errors and attempt < RETRIES: time.sleep(2 ** attempt) attempt += 1 continue raise DadaScribeError("%s failed: %s" % (path, getattr(e, "reason", e))) def _put(url, body, content_type, label): attempt = 0 while True: request = urllib.request.Request(url, data=body, method="PUT", headers={"Content-Type": content_type}) try: with urllib.request.urlopen(request, timeout=900) as response: return response.headers.get("ETag", "") except (urllib.error.URLError, OSError) as e: if attempt >= RETRIES: raise DadaScribeError("upload of %s failed: %s" % (label, e)) time.sleep(2 ** attempt) attempt += 1 def _upload_parts(api_base, api_key, path, size, init, concurrency, on_progress): part_size = int(init["part_size"]) count = int(init.get("part_count") or -(-size // part_size)) etags = {} sent = [0] lock = threading.Lock() def send(number, url): with open(path, "rb") as f: f.seek((number - 1) * part_size) chunk = f.read(part_size) etag = _put(url, chunk, "application/octet-stream", "part %d" % number) if on_progress: with lock: sent[0] += len(chunk) on_progress(sent[0], size) return number, etag with ThreadPoolExecutor(max_workers=concurrency) as pool: for first in range(1, count + 1, URL_BATCH): last = min(first + URL_BATCH - 1, count) urls = _post(api_base, api_key, "/upload/parts", { "token": init["token"], "upload_id": init["upload_id"], "part_from": first, "part_to": last, })["urls"] for number, etag in pool.map(lambda n: send(n, urls[str(n)]), range(first, last + 1)): etags[number] = etag return [{"partNumber": n, "etag": etags[n]} for n in sorted(etags)] def upload_file(api_key, path, api_base=None, concurrency=4, on_progress=None): """Upload a local file; returns the upload_token to pass to /v1/transcribe. on_progress(bytes_sent, total_bytes) is called as data is uploaded. """ api_base = _api_base(api_base) size = os.path.getsize(path) filename = os.path.basename(path) content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" init = _post(api_base, api_key, "/upload/init", {"filename": filename, "size": size, "content_type": content_type}) token = init["token"] try: if init["mode"] == "put": with open(path, "rb") as f: body = f.read() _put(init["url"], body, content_type, "file") if on_progress: on_progress(size, size) done = _post(api_base, api_key, "/upload/complete", {"token": token, "size": size}, retry_server_errors=False) else: parts = _upload_parts(api_base, api_key, path, size, init, concurrency, on_progress) done = _post(api_base, api_key, "/upload/complete", { "token": token, "upload_id": init["upload_id"], "parts": parts, "size": size, }, retry_server_errors=False) except BaseException: try: _post(api_base, api_key, "/upload/abort", {"token": token}) except Exception: pass raise return done["upload_token"] def transcribe_file(api_key, path, source_language, destination_language=None, diarization=None, api_base=None, concurrency=4, on_progress=None): """Upload a local file and submit it; returns the /v1/transcribe response ({"status", "id", "count"}).""" upload_token = upload_file(api_key, path, api_base=api_base, concurrency=concurrency, on_progress=on_progress) payload = {"upload_token": upload_token, "source-language": source_language} if destination_language: payload["destination-language"] = destination_language if diarization: payload["diarization"] = diarization return _post(_api_base(api_base), api_key, "/transcribe", payload, retry_server_errors=False) def main(argv=None): parser = argparse.ArgumentParser(description="Upload a file (up to 10 GB) to DaDaScribe and submit it for transcription.") parser.add_argument("file") parser.add_argument("--source-language", help="spoken language code, e.g. en (required unless --upload-only)") parser.add_argument("--destination-language", help="comma-separated translation codes, e.g. es,it") parser.add_argument("--diarization", help='comma-separated speaker names, e.g. "Ann,Bob"') parser.add_argument("--upload-only", action="store_true", help="only upload; print the upload_token") parser.add_argument("--api-key", default=os.environ.get("DADASCRIBE_API_KEY"), help="defaults to $DADASCRIBE_API_KEY") parser.add_argument("--api-base", help="defaults to $DADASCRIBE_API_BASE or " + DEFAULT_API_BASE) args = parser.parse_args(argv) if not args.api_key: parser.error("set DADASCRIBE_API_KEY or pass --api-key") if not args.upload_only and not args.source_language: parser.error("--source-language is required unless --upload-only is set") def progress(sent, total): if sys.stderr.isatty(): sys.stderr.write("\rUploaded %d%% (%d / %d MB)" % (sent * 100 // max(total, 1), sent >> 20, total >> 20)) sys.stderr.flush() try: if args.upload_only: print(upload_file(args.api_key, args.file, api_base=args.api_base, on_progress=progress)) else: result = transcribe_file(args.api_key, args.file, args.source_language, args.destination_language, args.diarization, api_base=args.api_base, on_progress=progress) print(json.dumps(result)) except (DadaScribeError, OSError) as e: print("\nerror: %s" % e, file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())