yt_subs.py - YouTube subtitle generator (SRT/TXT/JSON) via Microsoft MAI-Transcribe 2 on OpenRouter. Single-file uv script, PEP 723 inline deps, 60 languages, auto language ID, word-level timings, optional diarization.

srt_tools.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Split, validate and merge SRT files for the subtitling pipeline.

  srt_tools.py split  tr_raw.srt blocks/ --cues 120
  srt_tools.py check  blocks/en_001.srt
  srt_tools.py merge  blocks/ en_final.srt --prefix en_

`check` enforces the Netflix Turkish spec used by the leyla-ile-mecnun skill:
42 chars/line, 2 lines, 17 CPS, 0.833-7 s. Exit code 1 if anything fails, so
an agent can gate on it.
"""
import argparse, os, re, sys

MAX_LINE, MAX_LINES, MAX_CPS, MIN_DUR, MAX_DUR = 42, 2, 17.0, 0.833, 7.0


def parse(path):
    cues, block = [], []
    for raw in open(path, encoding='utf-8').read().split('\n\n'):
        block = [l for l in raw.strip().split('\n') if l.strip()]
        if len(block) < 2 or '-->' not in block[1]:
            continue
        a, b = [x.strip() for x in block[1].split('-->')]
        cues.append({'n': int(block[0]), 'start': a, 'end': b,
                     'lines': block[2:]})
    return cues


def sec(t):
    h, m, rest = t.split(':')
    s, ms = rest.split(',')
    return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000


def render(cues):
    out = []
    for i, c in enumerate(cues, 1):
        out.append('%d\n%s --> %s\n%s\n' % (i, c['start'], c['end'],
                                            '\n'.join(c['lines'])))
    return '\n'.join(out)


def cmd_split(a):
    cues = parse(a.srt)
    os.makedirs(a.outdir, exist_ok=True)
    blocks, cur = [], []
    for i, c in enumerate(cues):
        cur.append(c)
        if len(cur) < a.cues:
            continue
        # past the target, break at the next real pause so a block ends on a
        # scene boundary rather than mid-exchange
        nxt = cues[i + 1] if i + 1 < len(cues) else None
        if nxt is None or sec(nxt['start']) - sec(c['end']) >= a.gap:
            blocks.append(cur)
            cur = []
    if cur:
        blocks.append(cur)
    for i, b in enumerate(blocks, 1):
        p = os.path.join(a.outdir, '%s%03d.srt' % (a.prefix, i))
        open(p, 'w', encoding='utf-8').write(render(b))
        print('%s  %d cues  %s -> %s' % (p, len(b), b[0]['start'], b[-1]['end']))
    print('%d blocks' % len(blocks))


def cmd_check(a):
    bad = 0
    for path in a.srt:
        for c in parse(path):
            d = sec(c['end']) - sec(c['start'])
            text = ' '.join(c['lines'])
            n = len(text)
            errs = []
            if d < MIN_DUR:
                errs.append('short %.2fs' % d)
            if d > MAX_DUR:
                errs.append('long %.2fs' % d)
            if d > 0 and n / d > MAX_CPS:
                errs.append('%.1f cps' % (n / d))
            if len(c['lines']) > MAX_LINES:
                errs.append('%d lines' % len(c['lines']))
            wide = [l for l in c['lines'] if len(l) > MAX_LINE]
            if wide:
                errs.append('%d chars' % max(len(l) for l in wide))
            if re.search(r'\[S\d+\]', text):
                errs.append('speaker label')
            if errs:
                bad += 1
                print('%s cue %d: %s' % (os.path.basename(path), c['n'],
                                         ', '.join(errs)))
    print('FAIL: %d cue(s)' % bad if bad else 'OK')
    return 1 if bad else 0


def cmd_merge(a):
    files = sorted(f for f in os.listdir(a.blockdir)
                   if f.startswith(a.prefix) and f.endswith('.srt'))
    if not files:
        sys.exit('no %s*.srt in %s' % (a.prefix, a.blockdir))
    cues = []
    for f in files:
        cues += parse(os.path.join(a.blockdir, f))
    open(a.out, 'w', encoding='utf-8').write(render(cues))
    print('%d files, %d cues -> %s' % (len(files), len(cues), a.out))


p = argparse.ArgumentParser(description=__doc__,
                            formatter_class=argparse.RawDescriptionHelpFormatter)
sub = p.add_subparsers(dest='cmd', required=True)

s = sub.add_parser('split', help='split an srt into blocks on scene gaps')
s.add_argument('srt'); s.add_argument('outdir')
s.add_argument('--cues', type=int, default=120, help='target cues per block')
s.add_argument('--gap', type=float, default=1.5, help='pause that ends a block')
s.add_argument('--prefix', default='tr_')
s.set_defaults(fn=cmd_split)

s = sub.add_parser('check', help='validate against the subtitle spec')
s.add_argument('srt', nargs='+')
s.set_defaults(fn=cmd_check)

s = sub.add_parser('merge', help='reassemble blocks into one srt')
s.add_argument('blockdir'); s.add_argument('out')
s.add_argument('--prefix', default='en_')
s.set_defaults(fn=cmd_merge)

a = p.parse_args()
sys.exit(a.fn(a) or 0)

yt_subs.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["yt-dlp"]
# ///
"""YouTube subtitle generator - SRT / TXT / JSON via MAI-Transcribe 2 on OpenRouter.

Transcribes a YouTube video (or any local audio/video file) into subtitles with
word-accurate timing. Handles 60 languages with automatic language detection,
and optional speaker labels.

-------------------------------------------------------------------------------
QUICK START
-------------------------------------------------------------------------------
  1. Install uv (https://docs.astral.sh/uv/) and ffmpeg:
         brew install uv ffmpeg          # macOS
  2. Get an OpenRouter API key (https://openrouter.ai/keys):
         export OPENROUTER_API_KEY=sk-or-...
  3. Run it:
         uv run yt_subs.py "https://youtu.be/VIDEO_ID" -o out

uv reads the inline dependency block above and installs what it needs on first
run - there is nothing else to set up. You can also mark the file executable
(chmod +x yt_subs.py) and run ./yt_subs.py directly.

-------------------------------------------------------------------------------
EXAMPLES
-------------------------------------------------------------------------------
  # A YouTube video, auto-detecting the language
  uv run yt_subs.py "https://youtu.be/VIDEO_ID" -o out

  # A local file, telling it the language (a strong hint - only use it when sure)
  uv run yt_subs.py podcast.m4a -o out --lang tr --name episode_12

  # Label who is speaking (see the caveat under LIMITS)
  uv run yt_subs.py interview.mp3 -o out --diarize

  # Shorter cues for small screens
  uv run yt_subs.py talk.mp4 -o out --max-chars 42 --max-secs 4

-------------------------------------------------------------------------------
OPTIONS
-------------------------------------------------------------------------------
  source            YouTube URL, or a local audio/video file
  -o, --out DIR     output directory (default: subs)
  --name STEM       output filename stem (default: subtitles)
  --lang CODE       ISO-639-1 hint such as tr, en, de. Omit to auto-detect.
  --diarize         separate speakers. Improves cue splitting; ids stay out
                    of the srt unless --label-speakers is given
  --label-speakers  prefix cues with [S0] etc. Off by default - professional
                    subtitles carry no speaker labels. Review aid only.
  --video           also keep the source video as mp4. Downloads once and
                    extracts the audio from it, so the source is fetched a
                    single time. Ignored for local files.
  --height N        max video height with --video (default: 1080)
  --max-chars N     max characters per cue (default: 84)
  --max-secs N      max seconds per cue (default: 6)
  --gap N           pause in seconds that forces a new cue (default: 0.7)
  --chunk-seconds N audio seconds per API request (default: 1600)

-------------------------------------------------------------------------------
OUTPUT
-------------------------------------------------------------------------------
  <name>.srt   standard subtitles - drop into any video player or upload to
               YouTube directly
  <name>.txt   plain text, one cue per line, for reading or search indexing
  <name>.json  every cue with start/end/text/speaker, plus metadata (detected
               language, duration, word count, cost) - use this to build your
               own formats (VTT, CSV, a transcript page)

  A work/ subdirectory holds the downloaded audio, the chunks, and one cached
  JSON response per chunk. Keep it and reruns are nearly free; delete it to
  start clean.

-------------------------------------------------------------------------------
COST AND SPEED
-------------------------------------------------------------------------------
  About $0.10 per hour of audio. A 30-minute video costs roughly $0.05 and
  finishes in well under a minute; transcription runs far faster than realtime,
  so download time usually dominates.

-------------------------------------------------------------------------------
LIMITS AND GOTCHAS (found by testing, not guessed)
-------------------------------------------------------------------------------
  * Long audio is rejected by the provider, so the script always splits the
    audio into chunks. ~31 minutes in one request works; 40 minutes fails
    consistently. The 1600s default sits safely below that line.

  * Speaker IDs from --diarize restart at 0 in every chunk. Within one chunk
    they are reliable; across chunks they are NOT the same person. For audio
    under ~26 minutes there is only one chunk, so labels hold for the whole
    file. Longer videos need voice-matching across chunks, which this script
    does not attempt.

  * The API returns intermittent 500/503/429 errors even on requests that are
    perfectly valid - the identical chunk often succeeds on retry. The script
    retries with exponential backoff, and caches each chunk as it lands, so a
    failed run resumes instead of paying twice.

  * Cue timing comes from word-level timestamps rather than the API segment
    boundaries, which can stretch back over leading silence and produce cues
    that appear seconds early.

  * --lang is a strong hint, not a filter. Leaving it off is usually better;
    set it only when auto-detection actually gets the language wrong.

-------------------------------------------------------------------------------
TROUBLESHOOTING
-------------------------------------------------------------------------------
  "set OPENROUTER_API_KEY"     export the key in the shell you are running in
  "ffmpeg not found on PATH"   brew install ffmpeg (or your package manager)
  "no speech found"            the audio is silent, or music-only
  A 400 on one chunk           usually a corrupt or empty chunk; delete the
                               work/ directory and rerun
  Download fails               update yt-dlp: uv cache clean, then rerun

Requires: Python 3.11+, ffmpeg and ffprobe on PATH. yt-dlp is installed
automatically by uv and is only used for URLs.
"""
import argparse, base64, json, os, re, shutil, subprocess, sys, time, urllib.error, urllib.request

API = 'https://openrouter.ai/api/v1/audio/transcriptions'
MODEL = 'microsoft/mai-transcribe-2'
# Provider rejects long audio: 31 min observed OK, 40 min failed repeatedly.
CHUNK_SECONDS = 1600
# ffmpeg's segmenter can leave a sub-second remainder; the API rejects those.
MIN_CHUNK_SECONDS = 1.0
SENT_END = ('.', '?', '!', '…', ':')


def run(cmd, **kw):
    return subprocess.run(cmd, check=True, capture_output=True, text=True, **kw)


def duration(path):
    return float(run(['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
                      '-of', 'csv=p=0', path]).stdout.strip())


def fetch_video(source, work, height=1080):
    """Download the best mp4 up to `height` and return its path."""
    dest = os.path.join(work, 'video.mp4')
    if os.path.exists(dest):
        print('  cached video')
        return dest
    import yt_dlp
    # Resolution first, H.264 only as a tie-break. YouTube often carries a
    # height in AV1/VP9 but not in H.264, and hard-preferring the codec would
    # silently downgrade the picture.
    fmt = ('bestvideo[height<=%d]+bestaudio/best[height<=%d]' % (height, height))
    opts = {'format': fmt, 'merge_output_format': 'mp4',
            'format_sort': ['res', 'vcodec:h264', 'acodec:aac'],
            'outtmpl': os.path.join(work, 'video.%(ext)s'),
            'quiet': True, 'no_warnings': True, 'noprogress': True}
    with yt_dlp.YoutubeDL(opts) as ydl:
        ydl.download([source])
    if not os.path.exists(dest):
        sys.exit('download produced no mp4 in %s' % work)
    return dest


def audio_from(path, work):
    """Extract a 16 kHz mono mp3 next to `path` for transcription."""
    dest = os.path.join(work, 'source.mp3')
    if os.path.exists(dest):
        return dest
    run(['ffmpeg', '-y', '-i', path, '-vn', '-ac', '1', '-ar', '16000',
         '-b:a', '64k', dest])
    return dest


def fetch_audio(source, work, keep_video=False, height=1080):
    """Return a local audio path for a URL or an existing file.

    With keep_video, the mp4 is downloaded once and the audio extracted from
    it, so the episode is fetched a single time.
    """
    if not re.match(r'https?://', source):
        if not os.path.exists(source):
            sys.exit('no such file: %s' % source)
        return source
    if keep_video:
        return audio_from(fetch_video(source, work, height), work)
    dest = os.path.join(work, 'source.mp3')
    if os.path.exists(dest):
        print('  cached download')
        return dest
    import yt_dlp
    opts = {'format': 'bestaudio/best',
            'outtmpl': os.path.join(work, 'source.%(ext)s'),
            'postprocessors': [{'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3'}],
            'quiet': True, 'no_warnings': True, 'noprogress': True}
    with yt_dlp.YoutubeDL(opts) as ydl:
        ydl.download([source])
    if not os.path.exists(dest):
        sys.exit('download produced no mp3 in %s' % work)
    return dest


def split(audio, work, chunk_seconds=CHUNK_SECONDS):
    """Normalise to 16 kHz mono and cut into provider-sized chunks."""
    pat = os.path.join(work, 'chunk_%03d.mp3')
    existing = sorted(f for f in os.listdir(work) if re.match(r'chunk_\d+\.mp3$', f))
    if not existing:
        run(['ffmpeg', '-v', 'error', '-y', '-i', audio, '-f', 'segment',
             '-segment_time', str(chunk_seconds), '-ac', '1', '-ar', '16000',
             '-b:a', '32k', pat])
        existing = sorted(f for f in os.listdir(work) if re.match(r'chunk_\d+\.mp3$', f))
    return [os.path.join(work, f) for f in existing]


def transcribe(path, key, lang, diarize, retries=5):
    """One chunk -> API response. Retries: 429/500/503 are transient here."""
    cache = path[:-4] + '.json'
    if os.path.exists(cache) and os.path.getsize(cache) > 0:
        return json.load(open(cache))
    body = {
        'model': MODEL,
        'input_audio': {'data': base64.b64encode(open(path, 'rb').read()).decode(),
                        'format': 'mp3'},
        'response_format': 'verbose_json',
        'timestamp_granularities': ['word', 'segment'],
    }
    if lang:
        body['language'] = lang
    if diarize:
        # Only reachable via the JSON body; the multipart form drops unknown fields.
        body['provider'] = {'options': {'azure': {'diarization': {'enabled': True}}}}
    for attempt in range(retries):
        req = urllib.request.Request(API, data=json.dumps(body).encode(),
                                     headers={'Authorization': 'Bearer ' + key,
                                              'Content-Type': 'application/json'})
        try:
            d = json.load(urllib.request.urlopen(req, timeout=1800))
            json.dump(d, open(cache, 'w'), ensure_ascii=False)
            return d
        except urllib.error.HTTPError as e:
            if e.code in (429, 500, 502, 503) and attempt < retries - 1:
                wait = 2 ** attempt
                print('    HTTP %d, retrying in %ds' % (e.code, wait))
                time.sleep(wait)
                continue
            sys.exit('%s failed: HTTP %s %s' % (path, e.code, e.read().decode()[:200]))
    sys.exit('%s failed after %d attempts' % (path, retries))


def collect(chunks, key, lang, diarize):
    """Transcribe every chunk and merge onto one global timeline."""
    words, cost, langs = [], 0.0, []
    offset = 0.0
    for i, c in enumerate(chunks, 1):
        span = duration(c)
        if span < MIN_CHUNK_SECONDS:
            print('  chunk %d/%d: %.2fs remainder, skipped' % (i, len(chunks), span))
            offset += span
            continue
        print('  chunk %d/%d' % (i, len(chunks)))
        d = transcribe(c, key, lang, diarize)
        cost += (d.get('usage') or {}).get('cost', 0.0)
        if d.get('language'):
            langs.append(d['language'])
        for w in d.get('words') or []:
            words.append({'word': w['word'], 'start': w['start'] + offset,
                          'end': w['end'] + offset, 'speaker': w.get('speaker')})
        offset += span
    return words, cost, (langs[0] if langs else None), offset


def build_cues(words, max_chars, max_secs, gap):
    """Group words into readable cues. Word timings are the reliable ones;
    segment boundaries from the API can absorb leading silence."""
    cues, cur = [], []
    for w in words:
        if cur:
            prev = cur[-1]
            text_len = sum(len(x['word']) + 1 for x in cur)
            speaker_change = w.get('speaker') != prev.get('speaker')
            if (w['start'] - prev['end'] > gap
                    or text_len + len(w['word']) > max_chars
                    or w['end'] - cur[0]['start'] > max_secs
                    or speaker_change
                    or (prev['word'].endswith(SENT_END) and text_len > max_chars * 0.4)):
                cues.append(cur)
                cur = []
        cur.append(w)
    if cur:
        cues.append(cur)
    return [{'start': c[0]['start'], 'end': c[-1]['end'],
             'text': ' '.join(x['word'] for x in c).strip(),
             'speaker': c[0].get('speaker')} for c in cues if c]


def ts(t):
    h, r = divmod(t, 3600)
    m, s = divmod(r, 60)
    return '%02d:%02d:%02d,%03d' % (h, m, int(s), round(s % 1 * 1000))


def write_outputs(cues, out, stem, meta, label_speakers):
    def tag(c):
        return '[S%s] ' % c['speaker'] if label_speakers and c['speaker'] is not None else ''
    srt = os.path.join(out, stem + '.srt')
    with open(srt, 'w') as f:
        for i, c in enumerate(cues, 1):
            f.write('%d\n%s --> %s\n%s%s\n\n' % (i, ts(c['start']), ts(c['end']), tag(c), c['text']))
    txt = os.path.join(out, stem + '.txt')
    with open(txt, 'w') as f:
        for c in cues:
            f.write('%s%s\n' % (tag(c), c['text']))
    js = os.path.join(out, stem + '.json')
    json.dump({**meta, 'cues': cues}, open(js, 'w'), ensure_ascii=False, indent=1)
    return srt, txt, js


def main():
    p = argparse.ArgumentParser(description='Generate subtitles from a YouTube URL or audio file.')
    p.add_argument('source', help='YouTube URL or local audio/video file')
    p.add_argument('-o', '--out', default='subs', help='output directory (default: subs)')
    p.add_argument('--lang', help='ISO-639-1 hint, e.g. tr. Omit to auto-detect.')
    p.add_argument('--video', action='store_true',
                   help='also keep the source video as mp4. Downloads once and '
                        'extracts the audio from it, so the episode is fetched '
                        'a single time. Ignored for local files.')
    p.add_argument('--height', type=int, default=1080,
                   help='max video height with --video (default: 1080)')
    p.add_argument('--diarize', action='store_true',
                   help='ask the provider to separate speakers. Improves cue '
                        'splitting and segment timing; IDs stay out of the srt '
                        'unless --label-speakers is given')
    p.add_argument('--label-speakers', action='store_true',
                   help='prefix each cue with its speaker id, e.g. [S0]. Off by '
                        'default: professional interlingual subtitles carry no '
                        'speaker labels. Useful for review, not for delivery. '
                        '(IDs restart per chunk, so they are not consistent '
                        'across a long video.)')
    p.add_argument('--max-chars', type=int, default=84, help='max characters per cue')
    p.add_argument('--max-secs', type=float, default=6.0, help='max seconds per cue')
    p.add_argument('--gap', type=float, default=0.7, help='pause that forces a new cue')
    p.add_argument('--name', default='subtitles', help='output filename stem')
    p.add_argument('--chunk-seconds', type=int, default=CHUNK_SECONDS,
                   help='audio seconds per API request (default %d; provider '
                        'rejects long audio)' % CHUNK_SECONDS)
    a = p.parse_args()

    key = os.environ.get('OPENROUTER_API_KEY')
    if not key:
        sys.exit('set OPENROUTER_API_KEY')
    for tool in ('ffmpeg', 'ffprobe'):
        if not shutil.which(tool):
            sys.exit('%s not found on PATH' % tool)

    os.makedirs(a.out, exist_ok=True)
    work = os.path.join(a.out, 'work')
    os.makedirs(work, exist_ok=True)

    print('fetching audio')
    audio = fetch_audio(a.source, work, a.video, a.height)
    if a.video:
        src = os.path.join(work, 'video.mp4')
        if os.path.exists(src):
            keep = os.path.join(a.out, a.name + '.mp4')
            if not os.path.exists(keep):
                shutil.copy2(src, keep)
            print('  video: %s' % keep)
    print('splitting (%d s chunks)' % a.chunk_seconds)
    chunks = split(audio, work, a.chunk_seconds)
    print('transcribing %d chunk(s)' % len(chunks))
    words, cost, lang, total = collect(chunks, key, a.lang, a.diarize)
    if not words:
        sys.exit('no speech found')
    cues = build_cues(words, a.max_chars, a.max_secs, a.gap)
    meta = {'source': a.source, 'model': MODEL, 'language': lang,
            'audio_seconds': round(total, 1), 'words': len(words),
            'cue_count': len(cues), 'cost_usd': round(cost, 4)}
    srt, txt, js = write_outputs(cues, a.out, a.name, meta, a.label_speakers)
    print('\n%d cues, %d words, %.1f min, language=%s, cost $%.4f'
          % (len(cues), len(words), total / 60, lang, cost))
    for f in (srt, txt, js):
        print('  %s' % f)


if __name__ == '__main__':
    main()