feat: support create subtitles (.srt)
This commit is contained in:
@@ -1,10 +1,12 @@
|
||||
# Transcribe
|
||||
|
||||
Requires Bash, FFmpeg, curl, jq, and standard coreutils.
|
||||
Requires Bash, FFmpeg, curl, jq, and standard coreutils; `--srt` also requires
|
||||
Python 3 and the accompanying `subtitles.py` file.
|
||||
|
||||
```sh
|
||||
./transcribe recording.m4a
|
||||
./transcribe video.mp4 > transcript.txt
|
||||
./transcribe --srt video.mp4 > subtitles.srt
|
||||
```
|
||||
|
||||
Accepts m4a, mp4, aac, flac, opus, mp3, ogg, wav, and other formats your
|
||||
@@ -13,6 +15,14 @@ at 64 kbps (requires FFmpeg's libmp3lame encoder),
|
||||
and sends it to OpenRouter's `microsoft/mai-transcribe-2` model. Temporary
|
||||
files are removed on exit. The transcript goes to stdout; errors to stderr.
|
||||
|
||||
`--srt` requests word timestamps using OpenRouter's `verbose_json` response.
|
||||
Words are grouped into cues of at most two lines (42 characters per line,
|
||||
except an unusually long unbroken word) and six seconds. Cues also break at
|
||||
sentence endings, pauses of at least 0.7 seconds, and speaker changes when
|
||||
speaker labels are available. Timing comes from the words, rounded to
|
||||
milliseconds. Subtitle output goes to stdout; missing or invalid word timestamps
|
||||
produce an error. Without `--srt`, output remains plain text.
|
||||
|
||||
One file per invocation; the whole recording is sent in one request, so
|
||||
the service's audio length and request size limits apply.
|
||||
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
"""Convert OpenRouter word timestamps into readable SRT cues."""
|
||||
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
import sys
|
||||
import textwrap
|
||||
|
||||
LINE_WIDTH = 42
|
||||
MAX_DURATION = 6.0
|
||||
PAUSE = 0.7
|
||||
|
||||
|
||||
def lines(text):
|
||||
return textwrap.wrap(
|
||||
text, width=LINE_WIDTH, break_long_words=False, break_on_hyphens=False
|
||||
)
|
||||
|
||||
|
||||
def timestamp(seconds):
|
||||
milliseconds = math.floor(seconds * 1000 + 0.5)
|
||||
seconds, milliseconds = divmod(milliseconds, 1000)
|
||||
minutes, seconds = divmod(seconds, 60)
|
||||
hours, minutes = divmod(minutes, 60)
|
||||
return f"{hours:02}:{minutes:02}:{seconds:02},{milliseconds:03}"
|
||||
|
||||
|
||||
def convert(response):
|
||||
if response.get("error"):
|
||||
raise ValueError(f"Transcription failed: {response['error']}")
|
||||
words = response.get("words")
|
||||
if not isinstance(words, list) or (not words and response.get("text", "").strip()):
|
||||
raise ValueError("SRT output requires word timestamps; the API returned none")
|
||||
|
||||
cues = []
|
||||
current = []
|
||||
previous_start = -1
|
||||
|
||||
def flush():
|
||||
if current:
|
||||
text = " ".join(word["word"] for word in current)
|
||||
start = current[0]["start"]
|
||||
end = min(max(word["end"] for word in current), start + MAX_DURATION)
|
||||
# Avoid zero-length cues after rounding very short word durations.
|
||||
end = max(end, math.floor(start * 1000 + 0.5) / 1000 + 0.001)
|
||||
cues.append((start, end, "\n".join(lines(text))))
|
||||
current.clear()
|
||||
|
||||
for item in words:
|
||||
if not isinstance(item, dict) or not isinstance(item.get("word"), str):
|
||||
raise ValueError("Invalid subtitle word: expected word text")
|
||||
start, end = item.get("start"), item.get("end")
|
||||
if (
|
||||
any(
|
||||
type(value) not in (int, float) or not math.isfinite(value)
|
||||
for value in (start, end)
|
||||
)
|
||||
or start < 0
|
||||
or end < start
|
||||
):
|
||||
raise ValueError("Invalid subtitle word: expected finite 0 <= start <= end")
|
||||
if start < previous_start:
|
||||
raise ValueError("Invalid subtitle words: timestamps are out of order")
|
||||
previous_start = start
|
||||
word = dict(item, word=" ".join(item["word"].split()))
|
||||
if not word["word"]:
|
||||
continue
|
||||
if len(lines(word["word"])) > 2:
|
||||
raise ValueError("Invalid subtitle word: text exceeds two lines")
|
||||
if current:
|
||||
candidate = " ".join(w["word"] for w in [*current, word])
|
||||
previous = current[-1]
|
||||
if (
|
||||
len(lines(candidate)) > 2
|
||||
or end - current[0]["start"] > MAX_DURATION
|
||||
or start - previous["end"] >= PAUSE
|
||||
or word.get("speaker") != previous.get("speaker")
|
||||
or (
|
||||
previous["end"] - current[0]["start"] >= 1.0
|
||||
and re.search(r'[.!?…]["\u201d\u2019\u0027)]*$', previous["word"])
|
||||
)
|
||||
):
|
||||
flush()
|
||||
current.append(word)
|
||||
flush()
|
||||
return "".join(
|
||||
f"{index}\n{timestamp(start)} --> {timestamp(end)}\n{text}\n\n"
|
||||
for index, (start, end, text) in enumerate(cues, 1)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
with open(sys.argv[1], encoding="utf-8") as source:
|
||||
result = convert(json.load(source))
|
||||
except (ValueError, OSError) as error:
|
||||
print(f"SRT conversion failed: {error}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
sys.stdout.write(result)
|
||||
@@ -0,0 +1,75 @@
|
||||
import unittest
|
||||
|
||||
from subtitles import convert, timestamp
|
||||
|
||||
|
||||
class SubtitleTests(unittest.TestCase):
|
||||
def test_minute_long_segment_becomes_short_cues(self):
|
||||
text = (
|
||||
"Okay, all right, brilliant. Oh dude, we've got it this time. "
|
||||
"You know what? I can feel it. Okay, we found the longer self "
|
||||
"tappers that are for these little SSD cooling fans. "
|
||||
) * 4
|
||||
tokens = text.split()
|
||||
words = [
|
||||
{"word": word, "start": 971.68 + i * 0.4, "end": 971.68 + i * 0.4 + 0.35}
|
||||
for i, word in enumerate(tokens)
|
||||
]
|
||||
result = convert(
|
||||
{
|
||||
"text": text,
|
||||
"words": words,
|
||||
"segments": [{"text": text, "start": 971.68, "end": 1037.42}],
|
||||
}
|
||||
)
|
||||
cues = result.strip().split("\n\n")
|
||||
self.assertGreater(len(cues), 10)
|
||||
recovered = []
|
||||
for index, cue in enumerate(cues, 1):
|
||||
number, timing, *body = cue.splitlines()
|
||||
self.assertEqual(int(number), index)
|
||||
self.assertLessEqual(len(body), 2)
|
||||
self.assertTrue(all(len(line) <= 42 for line in body))
|
||||
|
||||
def seconds(value):
|
||||
h, m, s = value.replace(",", ".").split(":")
|
||||
return int(h) * 3600 + int(m) * 60 + float(s)
|
||||
|
||||
start, end = map(seconds, timing.split(" --> "))
|
||||
self.assertLessEqual(end - start, 6.001)
|
||||
recovered.extend(" ".join(body).split())
|
||||
self.assertEqual(recovered, tokens)
|
||||
self.assertIn("00:16:11,680", result)
|
||||
|
||||
def test_pause_sentence_and_speaker_boundaries(self):
|
||||
result = convert(
|
||||
{
|
||||
"words": [
|
||||
{"word": "First sentence.", "start": 0, "end": 1.2, "speaker": 0},
|
||||
{"word": "Next", "start": 1.3, "end": 1.6, "speaker": 0},
|
||||
{"word": "pause", "start": 3, "end": 3.4, "speaker": 0},
|
||||
{"word": "speaker", "start": 3.5, "end": 4, "speaker": 1},
|
||||
]
|
||||
}
|
||||
)
|
||||
self.assertEqual(len(result.strip().split("\n\n")), 4)
|
||||
|
||||
def test_missing_and_invalid_timestamps(self):
|
||||
for response in (
|
||||
{"text": "speech"},
|
||||
{"text": "speech", "words": []},
|
||||
{"words": [{"word": "bad", "start": 2, "end": 1}]},
|
||||
{"words": [{"word": "bad", "start": float("nan"), "end": 1}]},
|
||||
{"error": {"message": "failed"}},
|
||||
):
|
||||
with self.assertRaises(ValueError):
|
||||
convert(response)
|
||||
|
||||
def test_empty_and_rounding(self):
|
||||
self.assertEqual(convert({"words": [], "text": ""}), "")
|
||||
self.assertEqual(timestamp(59.9996), "00:01:00,000")
|
||||
self.assertEqual(timestamp(3601.234), "01:00:01,234")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+23
-8
@@ -1,25 +1,34 @@
|
||||
#!/usr/bin/env bash
|
||||
|
||||
export OPENROUTER_API_KEY
|
||||
OPENROUTER_API_KEY="$(op read "op://API Keys/openrouter-mai-transcribe-2/credential")"
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
srt=false
|
||||
if [[ ${1:-} == --srt ]]; then
|
||||
srt=true
|
||||
shift
|
||||
fi
|
||||
if [[ $# == 1 && ($1 == -h || $1 == --help) ]]; then
|
||||
echo "Usage: ${0##*/} AUDIO_FILE"
|
||||
echo "Usage: ${0##*/} [--srt] AUDIO_FILE"
|
||||
echo 'Transcribe audio using OPENROUTER_API_KEY; print the text to stdout.'
|
||||
echo ' --srt Print SRT subtitles in short cues using word timestamps instead.'
|
||||
exit 0
|
||||
fi
|
||||
if [[ $# != 1 ]]; then
|
||||
echo "Usage: ${0##*/} AUDIO_FILE" >&2
|
||||
echo "Usage: ${0##*/} [--srt] AUDIO_FILE" >&2
|
||||
exit 1
|
||||
fi
|
||||
if [[ ! -f $1 || ! -r $1 ]]; then
|
||||
echo "Cannot read audio file: $1" >&2
|
||||
exit 1
|
||||
fi
|
||||
export OPENROUTER_API_KEY
|
||||
OPENROUTER_API_KEY="$(op read "op://API Keys/openrouter-mai-transcribe-2/credential")"
|
||||
: "${OPENROUTER_API_KEY:?Set OPENROUTER_API_KEY first}"
|
||||
for tool in ffmpeg curl jq base64 mktemp; do
|
||||
dependencies=(ffmpeg curl jq base64 mktemp)
|
||||
if "$srt"; then
|
||||
dependencies+=(python3)
|
||||
fi
|
||||
for tool in "${dependencies[@]}"; do
|
||||
command -v "$tool" >/dev/null || {
|
||||
echo "Missing dependency: $tool" >&2
|
||||
exit 1
|
||||
@@ -37,8 +46,10 @@ ffmpeg -nostdin -hide_banner -loglevel error -i "$(realpath -- "$1")" \
|
||||
|
||||
# Keep large audio payloads out of shell arguments (which have a size limit).
|
||||
base64 <"$tmp/audio.mp3" | tr -d '\n\r' >"$tmp/audio.b64"
|
||||
jq -n --rawfile data "$tmp/audio.b64" \
|
||||
'{model: "microsoft/mai-transcribe-2", input_audio: {data: $data, format: "mp3"}}' \
|
||||
jq -n --rawfile data "$tmp/audio.b64" --argjson srt "$srt" \
|
||||
'{model: "microsoft/mai-transcribe-2", input_audio: {data: $data, format: "mp3"}}
|
||||
+ (if $srt then {response_format: "verbose_json", timestamp_granularities: ["segment", "word"]}
|
||||
else {} end)' \
|
||||
>"$tmp/request.json"
|
||||
|
||||
curl --silent --show-error --fail-with-body \
|
||||
@@ -49,6 +60,10 @@ curl --silent --show-error --fail-with-body \
|
||||
cat "$tmp/response.json" >&2
|
||||
exit 1
|
||||
}
|
||||
if "$srt"; then
|
||||
python3 "$(dirname -- "$(realpath -- "$0")")/subtitles.py" "$tmp/response.json"
|
||||
else
|
||||
jq -r 'if (.text | type) == "string" then .text
|
||||
else error("Transcription failed: " + (.error.message // tojson)) end' \
|
||||
"$tmp/response.json"
|
||||
fi
|
||||
|
||||
Reference in New Issue
Block a user