diff options
| author | Craig Jennings <c@cjennings.net> | 2026-09-25 13:05:42 -0400 |
|---|---|---|
| committer | Craig Jennings <c@cjennings.net> | 2026-09-25 13:05:42 -0400 |
| commit | d447db5ac9063ec12afb8f5bc7b3a0b05c75bad4 (patch) | |
| tree | 1c0231a2528f779a83741937e0da6fea0c409da9 /working/meeting-transcription-service | |
| parent | adab9abae11792eaa17711a63a74efa68885a0be (diff) | |
| download | archsetup-main.tar.gz archsetup-main.zip | |
This is a self-hosted transcription service: whisper.cpp for the words, pyannote for the speaker labels. It has been running on ratio since 2026-09-17, with velox as the offline fallback. A systemd path unit watches a filesystem queue and starts a oneshot worker per job. There is no network listener. ssh is the transport, systemd is the daemon, and the filesystem is the queue.
It lands in working/ rather than its final home because two decisions come first. I haven't picked where the code lives in this repo. The Hugging Face token the diarization model needs on its first download also has to be handled, since anyone can read this repo.
Neither blocks the service, which already runs. Both block the install path this repo owes it.
The accompanying note lists what each machine needs. The torch venv is 1.3 GB and the whisper model is a separate download, so the note describes both rather than carrying them here.
Diffstat (limited to 'working/meeting-transcription-service')
13 files changed, 2191 insertions, 0 deletions
diff --git a/working/meeting-transcription-service/2026-09-19-work-handoff.org b/working/meeting-transcription-service/2026-09-19-work-handoff.org new file mode 100644 index 0000000..d4da1d9 --- /dev/null +++ b/working/meeting-transcription-service/2026-09-19-work-handoff.org @@ -0,0 +1,44 @@ +#+TITLE: Handoff from work: the meeting transcription service needs an install home +#+AUTHOR: Craig Jennings + +* What this is +A self-hosted meeting transcription service (whisper.cpp plus pyannote speaker diarization) that runs on +ratio, with velox as the offline fallback. It has been running on both machines since 2026-09-17. The code +arrives as meeting-transcription-service.tar.gz alongside this note. I picked archsetup as its home because +it is machine setup: a worker, two systemd user units, a Python venv and a model file. The client script and +its Emacs backend entry are handled separately. + +* How it works +- A job queue under ~/.local/state/meeting-transcribe/ with incoming/, work/, done/ and failed/. A systemd + user path unit (meeting-transcribe.path) starts a oneshot worker (meeting-transcribe.service) when a job + lands in incoming/. +- The worker (src/transcribe-worker) converts audio to 16 kHz mono, runs whisper-cli at word level, runs + pyannote (speaker-diarization-community-1) through src/diarize.py with the job's speaker count, and + merges the two by timestamp with src/merge_transcript.py. One job at a time, behind a lock. +- A finished job leaves done/<id>.txt. A failed job leaves failed/<id>.log. Nothing half-written reaches done/. +- No network listener. Tailscale ssh is the transport and the login, systemd is the daemon, the + filesystem is the queue. + +* What the install has to provide, per machine +- ~/.local/share/pyannote-diarize/.venv :: Python 3.12, torch (CPU build), pyannote.audio 4.0.7. About + 1.3 GB. Built with uv. +- ~/.local/share/whisper-models/ggml-large-v3-turbo-q5_0.bin :: the whisper model; ratio's and velox's + copies have the same checksum. whisper-cpp itself must be installed (it was already on velox). +- The src/ scripts placed where the units expect them, and the two user units enabled. +- Linger on, so the path unit runs without a login session. +- One-time, online, by hand: a Hugging Face token and acceptance of the pyannote model terms, so the + diarization model can be cached. After that neither machine needs Hugging Face. The token is a + credential: it must not be written into this repo, which is publicly cloneable. + +* State today +- Installed by hand on ratio and velox; both verified. No NVIDIA GPU and no ROCm on ratio, so it is CPU only. +- make check in the bundle runs pytest (129 tests; whisper, pyannote, ssh and scp are faked at the process + boundary), pyright and shellcheck. All green on 2026-09-19. +- Known rough edge: when two local runs overlap, the second finds the lock held, the worker returns + silently, and the client says "finished without producing a transcript". Rerunning fixes it, but the + message should say the lock was held. + +* What I'm asking archsetup for +Take ownership of the service side: the worker, diarize.py, merge_transcript.py, the two units, and an +install step that builds the venv, fetches the model and enables the units on both daily drivers. Keep +anything work-specific out of the repo; the code bundle has none (scanned before sending). diff --git a/working/meeting-transcription-service/Makefile b/working/meeting-transcription-service/Makefile new file mode 100644 index 0000000..2ed7dad --- /dev/null +++ b/working/meeting-transcription-service/Makefile @@ -0,0 +1,15 @@ +# Checks for the transcription service. transcribe-worker has no .py suffix, so +# pyright's directory scan skips it; it is named explicitly here. +.PHONY: check test types lint + +check: test types lint + +test: + python3 -m pytest tests -q + +types: + pyright + pyright src/transcribe-worker + +lint: + shellcheck src/ratio-transcribe diff --git a/working/meeting-transcription-service/pyrightconfig.json b/working/meeting-transcription-service/pyrightconfig.json new file mode 100644 index 0000000..8228120 --- /dev/null +++ b/working/meeting-transcription-service/pyrightconfig.json @@ -0,0 +1,4 @@ +{ + "extraPaths": ["src"], + "include": ["src", "tests"] +} diff --git a/working/meeting-transcription-service/src/diarize.py b/working/meeting-transcription-service/src/diarize.py new file mode 100644 index 0000000..a6ae85e --- /dev/null +++ b/working/meeting-transcription-service/src/diarize.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Run pyannote speaker diarization on one audio file and write the turns as JSON. + +Usage: diarize.py AUDIO OUT_JSON [--speakers N | --min-speakers N --max-speakers N] + +Output is a list of {"start", "end", "speaker"} in seconds, which is what +merge_transcript.py reads. HF_TOKEN is only needed the first time, to download +the gated model; after that the cached copy loads offline. + +pyannote and torch are imported inside run(), so the pure helpers here can be +tested without the multi-gigabyte environment. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from collections.abc import Iterable +from pathlib import Path +from typing import Any + +MODEL = "pyannote/speaker-diarization-community-1" + + +def turns_from_tracks(tracks: Iterable[tuple[Any, Any, str]]) -> list[dict[str, Any]]: + """Convert pyannote (segment, track, label) triples into sorted turn dicts. + + Times are rounded to milliseconds. Segments with no length are dropped. + """ + turns = [ + {"start": round(float(seg.start), 3), "end": round(float(seg.end), 3), "speaker": str(label)} + for seg, _track, label in tracks + if float(seg.end) > float(seg.start) + ] + return sorted(turns, key=lambda t: (t["start"], t["end"])) + + +def _positive_int(value: str) -> int: + number = int(value) + if number < 1: + raise argparse.ArgumentTypeError("must be 1 or more") + return number + + +def parse_args(argv: list[str]) -> argparse.Namespace: + """Parse the command line. Exits with usage on bad or contradictory counts.""" + parser = argparse.ArgumentParser(description="Speaker diarization with pyannote.") + parser.add_argument("audio") + parser.add_argument("out") + parser.add_argument("--speakers", type=_positive_int, help="exact number of speakers") + parser.add_argument("--min-speakers", type=_positive_int) + parser.add_argument("--max-speakers", type=_positive_int) + args = parser.parse_args(argv) + if args.speakers is not None and (args.min_speakers or args.max_speakers): + parser.error("--speakers cannot be combined with --min-speakers/--max-speakers") + if args.min_speakers and args.max_speakers and args.min_speakers > args.max_speakers: + parser.error("--min-speakers cannot exceed --max-speakers") + return args + + +def pipeline_kwargs(args: argparse.Namespace) -> dict[str, int]: + """Only the speaker-count options that were actually given.""" + options = { + "num_speakers": args.speakers, + "min_speakers": args.min_speakers, + "max_speakers": args.max_speakers, + } + return {name: value for name, value in options.items() if value is not None} + + +def run(args: argparse.Namespace) -> list[dict[str, Any]]: + """Load the pipeline, diarize the audio, and return the turns.""" + # Heavy, and only installed in the service venv, so imported here on purpose. + from pyannote.audio import Pipeline # pyright: ignore[reportMissingImports] + + pipeline = Pipeline.from_pretrained(MODEL, token=os.environ.get("HF_TOKEN") or None) + if pipeline is None: + raise RuntimeError(f"could not load {MODEL}: check HF_TOKEN and that its terms are accepted") + output = pipeline(args.audio, **pipeline_kwargs(args)) + # The exclusive variant never overlaps two speakers, which is what a + # word-by-word merge wants. Older pipelines return the annotation itself. + annotation = getattr(output, "exclusive_speaker_diarization", None) + if annotation is None: + annotation = getattr(output, "speaker_diarization", output) + return turns_from_tracks(annotation.itertracks(yield_label=True)) + + +def main(argv: list[str]) -> int: + """CLI entry point. Writes OUT_JSON atomically; non-zero on any failure.""" + args = parse_args(argv) + if not Path(args.audio).is_file(): + print(f"Error: audio file not found: {args.audio}", file=sys.stderr) + return 1 + try: + turns = run(args) + except Exception as err: # noqa: BLE001 - report any model failure and exit non-zero + print(f"Error: diarization failed: {err}", file=sys.stderr) + return 1 + if not turns: + print("Error: diarization found no speech", file=sys.stderr) + return 1 + out = Path(args.out) + partial = out.with_name(out.name + ".partial") + partial.write_text(json.dumps(turns), encoding="utf-8") + partial.replace(out) + speakers = len({t["speaker"] for t in turns}) + print(f"{len(turns)} turns, {speakers} speakers -> {out}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/working/meeting-transcription-service/src/merge_transcript.py b/working/meeting-transcription-service/src/merge_transcript.py new file mode 100644 index 0000000..13aec75 --- /dev/null +++ b/working/meeting-transcription-service/src/merge_transcript.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +"""Merge whisper word timings with speaker turns into transcript lines. + +Input is two JSON files: whisper-cli's ``-oj`` output (run at word level) and +the diarizer's list of speaker turns. Output is one line per stretch of speech, +``HH:MM:SS Speaker A: text``, the same shape the hosted services produced. + +Standard library only, so it runs under any Python 3.10+ without the venv. +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from pathlib import Path + +DEFAULT_MAX_GAP_S = 3.0 +DEFAULT_SPEECH_MARGIN_S = 2.0 +DEFAULT_MAX_LOOP_S = 30.0 + + +@dataclass(frozen=True) +class Unit: + """A piece of transcribed text with its start and end in seconds.""" + + start: float + end: float + text: str + + +@dataclass(frozen=True) +class Turn: + """A stretch of audio the diarizer attributes to one speaker.""" + + start: float + end: float + speaker: str + + +def _timestamp(seconds: float) -> str: + """Render seconds as HH:MM:SS, floored.""" + whole = int(seconds) + return f"{whole // 3600:02d}:{whole % 3600 // 60:02d}:{whole % 60:02d}" + + +def _speaker_name(index: int) -> str: + """Name speakers A-Z in order of first speech, then by number.""" + return chr(ord("A") + index) if index < 26 else str(index) + + +def _speaker_for(unit: Unit, turns: list[Turn]) -> str: + """Pick the turn a unit belongs to. + + The turn overlapping most of the unit wins. A unit overlapping nothing (a + zero-length word, or one whisper timed into a silence) goes to the turn + nearest its midpoint, because whisper's word timings drift by a few hundred + milliseconds and dropping the word would be worse than a near guess. + """ + best = max(turns, key=lambda t: min(unit.end, t.end) - max(unit.start, t.start)) + if min(unit.end, best.end) - max(unit.start, best.start) > 0: + return best.speaker + mid = (unit.start + unit.end) / 2 + + def distance(turn: Turn) -> float: + if turn.start <= mid <= turn.end: + return 0.0 + return min(abs(mid - turn.start), abs(mid - turn.end)) + + return min(turns, key=distance).speaker + + +def merge(units: list[Unit], turns: list[Turn], max_gap_s: float = DEFAULT_MAX_GAP_S) -> list[str]: + """Return transcript lines for ``units`` labelled by ``turns``. + + Consecutive units from one speaker share a line. The line breaks when the + speaker changes, or when the speaker pauses longer than ``max_gap_s``, so a + long monologue still carries usable timestamps. + + Raises: + ValueError: if there are no turns, no spoken words, or ``max_gap_s`` is negative. + """ + if max_gap_s < 0: + raise ValueError("max_gap_s must not be negative") + spoken = sorted((u for u in units if u.text.strip()), key=lambda u: (u.start, u.end)) + if not spoken: + raise ValueError("no speech: the transcription holds no words") + if not turns: + raise ValueError("no speaker turns: the diarization is empty") + ordered_turns = sorted(turns, key=lambda t: (t.start, t.end)) + + names: dict[str, str] = {} + lines: list[tuple[float, str, list[str]]] = [] + previous_end = 0.0 + for unit in spoken: + speaker = _speaker_for(unit, ordered_turns) + name = names.setdefault(speaker, _speaker_name(len(names))) + if lines and lines[-1][1] == name and unit.start - previous_end <= max_gap_s: + lines[-1][2].append(unit.text) + else: + lines.append((unit.start, name, [unit.text])) + previous_end = max(previous_end, unit.end) + + return [ + f"{_timestamp(start)} Speaker {name}: {' '.join(''.join(parts).split())}" + for start, name, parts in lines + ] + + +def drop_outside_speech( + units: list[Unit], turns: list[Turn], margin_s: float = DEFAULT_SPEECH_MARGIN_S +) -> tuple[list[Unit], int]: + """Return the units that belong to speech, and how many were dropped. + + Whisper invents words ("Thank you.") when it is handed silence. The diarizer + marks where people actually spoke, so a unit that touches no turn and sits + more than ``margin_s`` from the nearest one is treated as invented. The margin + protects real words the diarizer clipped off the edge of a turn. + + Raises: + ValueError: if there are no turns, or ``margin_s`` is negative. + """ + if margin_s < 0: + raise ValueError("margin_s must not be negative") + if not turns: + raise ValueError("no speaker turns: the diarization is empty") + + def gap(unit: Unit) -> float: + # Seconds between the unit and its nearest turn; zero when they touch. + # Rounded to the millisecond, the resolution of whisper's offsets. + nearest = min(max(turn.start - unit.end, unit.start - turn.end, 0.0) for turn in turns) + return round(nearest, 3) + + kept = [unit for unit in units if gap(unit) <= margin_s] + return kept, len(units) - len(kept) + + +def _unit(item: dict) -> Unit: + """A Unit from a whisper segment or token dict (offsets in milliseconds). + + whisper-cli clamps a token's start to its segment's start without moving the + end, so some tokens arrive ending before they begin. Those become zero-length + at their start; left alone they corrupt both the overlap and the pause maths. + """ + start = item["offsets"]["from"] / 1000 + end = item["offsets"]["to"] / 1000 + return Unit(start, max(start, end), item["text"]) + + +def find_repetition( + text: str, min_words: int = 3, max_words: int = 12, min_repeats: int = 4 +) -> tuple[str, int] | None: + """Find a phrase repeated back to back, whisper's hallucination signature. + + Returns the phrase and its repeat count, or None. Four consecutive repeats of + a phrase of three or more words is the line: people say a thing two or three + times, and single-word runs ("yeah yeah yeah") are ordinary speech. + """ + words = text.split() + keys = [w.lower().strip(".,!?;:\"'") for w in words] + for size in range(min_words, max_words + 1): + for i in range(len(keys) - size * min_repeats + 1): + phrase = keys[i : i + size] + if len(set(phrase)) < 2: + continue + count = 1 + while keys[i + count * size : i + (count + 1) * size] == phrase: + count += 1 + if count >= min_repeats: + return " ".join(words[i : i + size]), count + return None + + +def collapse_repetitions( + units: list[Unit], + min_words: int = 3, + max_words: int = 12, + min_repeats: int = 4, + max_loop_s: float = DEFAULT_MAX_LOOP_S, +) -> tuple[list[Unit], list[tuple[str, int]]]: + """Collapse whisper's repetition loops, keeping one copy of the phrase. + + Whisper sometimes gets stuck and emits the same phrase over and over. A short + loop (up to ``max_loop_s`` of audio) costs a few seconds of speech, so it is + collapsed to a single occurrence and reported. A longer one means real speech + was lost for a stretch, and that is raised instead, so the job fails rather + than hand back a transcript with a hole in it. + + Returns the surviving units and a list of (phrase, repeat count) for every + loop collapsed. Units are matched word by word, so a phrase spread over word + units and a phrase sitting in one segment unit are both found. + + Raises: + ValueError: if a loop lasts longer than ``max_loop_s``, or the limit is negative. + """ + if max_loop_s < 0: + raise ValueError("max_loop_s must not be negative") + units = list(units) + collapsed: list[tuple[str, int]] = [] + + def find() -> tuple[int, int, int] | None: + # (first word index, phrase size, repeat count) of the earliest loop, or None + words = [(w, ui) for ui, u in enumerate(units) for w in u.text.split()] + keys = [w.lower().strip(".,!?;:\"'") for w, _ in words] + best: tuple[int, int, int] | None = None + for size in range(min_words, max_words + 1): + for i in range(len(keys) - size * min_repeats + 1): + if best is not None and i >= best[0]: + break + phrase = keys[i : i + size] + if len(set(phrase)) < 2: + continue + count = 1 + while keys[i + count * size : i + (count + 1) * size] == phrase: + count += 1 + if count >= min_repeats: + best = (i, size, count) + break + return best + + while True: + hit = find() + if hit is None: + return units, collapsed + i, size, count = hit + words = [(w, ui) for ui, u in enumerate(units) for w in u.text.split()] + phrase_text = " ".join(w for w, _ in words[i : i + size]) + doomed = set(range(i + size, i + count * size)) # word indexes of the repeats + touched = {ui for wi, (_, ui) in enumerate(words) if wi in doomed or i <= wi < i + size} + span_start = min(units[ui].start for ui in touched) + span_end = max(units[ui].end for ui in touched) + duration = round(span_end - span_start, 3) + if duration > max_loop_s: + raise ValueError( + f"whisper looped: {phrase_text!r} repeats {count} times over {duration:.0f} s; " + "rerun whisper with -mc 0" + ) + # Rebuild every touched unit from the words it keeps. A unit that held only + # repeats disappears; one that also held the first copy or later speech keeps + # those words. Every pass removes (count - 1) * size words, so this ends. + last_kept = max(ui for wi, (_, ui) in enumerate(words) if i <= wi < i + size) + rebuilt: list[Unit] = [] + for ui, unit in enumerate(units): + if ui not in touched: + rebuilt.append(unit) + continue + keep = [w for wi, (w, wui) in enumerate(words) if wui == ui and wi not in doomed] + if not keep: + continue + # The kept copy takes over the time the loop occupied, so the merge does + # not read the removed stretch as a pause and break the line there. + end = max(unit.end, span_end) if ui == last_kept else unit.end + rebuilt.append(Unit(unit.start, end, " " + " ".join(keep))) + units = rebuilt + collapsed.append((phrase_text, count)) + + +def load_whisper_json(path: str | Path) -> list[Unit]: + """Read whisper-cli JSON output. Offsets there are in milliseconds. + + With ``-ojf`` each segment carries its tokens and their offsets; those become + word-level units, which is what lets a speaker change land mid-segment. Plain + ``-oj`` output, or a segment with no tokens, falls back to the segment itself. + + Raises: + ValueError: if the file is not whisper's JSON shape. + """ + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + units: list[Unit] = [] + for item in data["transcription"]: + tokens = item.get("tokens") or [] + if not tokens: + units.append(_unit(item)) + continue + for token in tokens: + text = token["text"] + if not text or text.startswith("[_"): # [_BEG_], [_TT_123], [_EOT_] + continue + unit = _unit(token) + if units and not text.startswith(" "): + # A sub-word piece or punctuation: it belongs to the word before it. + previous = units[-1] + units[-1] = Unit(previous.start, max(previous.end, unit.end), previous.text + text) + else: + units.append(unit) + return units + except OSError as err: + raise ValueError(f"{path}: cannot read whisper output ({err.strerror})") from err + except (json.JSONDecodeError, KeyError, TypeError) as err: + raise ValueError(f"{path}: not whisper-cli JSON output ({err!r})") from err + + +def load_turns_json(path: str | Path) -> list[Turn]: + """Read the diarizer's turns: a list of {start, end, speaker}, in seconds. + + Raises: + ValueError: if the file is not that shape. + """ + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(data, list): + raise TypeError("expected a list of turns") + return [Turn(float(t["start"]), float(t["end"]), str(t["speaker"])) for t in data] + except OSError as err: + raise ValueError(f"{path}: cannot read speaker turns ({err.strerror})") from err + except (json.JSONDecodeError, KeyError, TypeError, ValueError) as err: + raise ValueError(f"{path}: not a speaker-turns file ({err!r})") from err + + +def main(argv: list[str]) -> int: + """CLI: ``merge_transcript.py WHISPER_JSON TURNS_JSON`` prints the transcript.""" + if len(argv) != 2: + print("usage: merge_transcript.py WHISPER_JSON TURNS_JSON", file=sys.stderr) + return 2 + try: + turns = load_turns_json(argv[1]) + units, _dropped = drop_outside_speech(load_whisper_json(argv[0]), turns) + units, _collapsed = collapse_repetitions(units) + lines = merge(units, turns) + except ValueError as err: + print(f"Error: {err}", file=sys.stderr) + return 1 + print("\n".join(lines)) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/working/meeting-transcription-service/src/ratio-transcribe b/working/meeting-transcription-service/src/ratio-transcribe new file mode 100755 index 0000000..8db59b6 --- /dev/null +++ b/working/meeting-transcription-service/src/ratio-transcribe @@ -0,0 +1,203 @@ +#!/usr/bin/env bash +# ratio-transcribe - Transcribe audio on my own transcription host, with speaker labels +# Usage: ratio-transcribe <audio-file> [language] +# +# Same contract as assemblyai-transcribe: the transcript goes to stdout, one line +# per speaker turn ("HH:MM:SS Speaker A: text"); progress and errors go to stderr; +# any failure exits non-zero with nothing on stdout. +# +# The work happens on a host that runs the meeting-transcribe queue (whisper-cpp +# plus pyannote). This script copies the audio over ssh, drops a job into the +# queue, waits, and prints the result. The job id is a hash of the audio and its +# options, so if the connection drops or the laptop sleeps, running the same +# command again just collects the finished transcript. If the host can't be +# reached at all, the same queue and worker run on this machine instead. +# +# Optional environment: +# SPEAKERS exact number of speakers, when you know it +# MIN_SPEAKERS, MAX_SPEAKERS a range instead +# TRANSCRIBE_HOST ssh name of the host (default: ratio) +# TRANSCRIBE_TIMEOUT seconds to wait for the job (default: 3600) +# TRANSCRIBE_POLL seconds between checks (default: 10) +# TRANSCRIBE_LOCAL=1 skip the host and run here +# TRANSCRIBE_WORKER path to the local worker + +set -euo pipefail + +AUDIO="${1:-}" +LANG_CODE="${2:-en}" +HOST="${TRANSCRIBE_HOST:-ratio}" +TIMEOUT="${TRANSCRIBE_TIMEOUT:-3600}" +POLL="${TRANSCRIBE_POLL:-10}" +WORKER="${TRANSCRIBE_WORKER:-$HOME/.local/share/pyannote-diarize/src/transcribe-worker}" +STATE=".local/state/meeting-transcribe" # relative to the home directory, on either machine + +if [[ -z "$AUDIO" ]]; then + echo "Usage: ratio-transcribe <audio-file> [language]" >&2 + echo "Example: SPEAKERS=3 ratio-transcribe meeting.m4a en" >&2 + exit 1 +fi + +if [[ ! -f "$AUDIO" ]]; then + echo "Error: Audio file not found: $AUDIO" >&2 + exit 1 +fi +# scp reads "name:with:colons" as host:path; an absolute path removes the ambiguity. +AUDIO="$(realpath -- "$AUDIO")" + +# Everything below ends up in a job file and on command lines, so check it first. +if [[ ! "$LANG_CODE" =~ ^[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*$ ]]; then + echo "Error: Invalid language code: $LANG_CODE" >&2 + exit 1 +fi + +for name in SPEAKERS MIN_SPEAKERS MAX_SPEAKERS; do + value="${!name:-}" + if [[ -n "$value" && ! "$value" =~ ^[1-9][0-9]*$ ]]; then + echo "Error: $name must be a positive whole number of speakers, got: $value" >&2 + exit 1 + fi +done +if [[ -n "${SPEAKERS:-}" && ( -n "${MIN_SPEAKERS:-}" || -n "${MAX_SPEAKERS:-}" ) ]]; then + echo "Error: give an exact SPEAKERS count or a MIN/MAX speaker range, not both" >&2 + exit 1 +fi +if [[ -n "${MIN_SPEAKERS:-}" && -n "${MAX_SPEAKERS:-}" ]] && (( MIN_SPEAKERS > MAX_SPEAKERS )); then + echo "Error: MIN_SPEAKERS cannot exceed MAX_SPEAKERS (speaker range)" >&2 + exit 1 +fi + +for tool in jq sha256sum; do + if ! command -v "$tool" &> /dev/null; then + echo "Error: $tool command not found" >&2 + exit 1 + fi +done + +EXT="${AUDIO##*.}" +[[ "$EXT" =~ ^[A-Za-z0-9]{1,5}$ ]] || EXT="bin" +EXT="${EXT,,}" + +if [[ -n "${SPEAKERS:-}" ]]; then + COUNT_TAG="s${SPEAKERS}" +elif [[ -n "${MIN_SPEAKERS:-}${MAX_SPEAKERS:-}" ]]; then + COUNT_TAG="r${MIN_SPEAKERS:-x}-${MAX_SPEAKERS:-x}" +else + COUNT_TAG="auto" +fi +JOB_ID="$(sha256sum "$AUDIO" | cut -c1-16)-${LANG_CODE,,}-${COUNT_TAG}" + +JOB_JSON=$(jq -cn \ + --arg language "$LANG_CODE" \ + --arg name "$(basename "$AUDIO")" \ + --arg speakers "${SPEAKERS:-}" --arg min "${MIN_SPEAKERS:-}" --arg max "${MAX_SPEAKERS:-}" \ + '{language: $language} + + (if $speakers != "" then {speakers: ($speakers | tonumber)} else {} end) + + (if $min != "" then {min_speakers: ($min | tonumber)} else {} end) + + (if $max != "" then {max_speakers: ($max | tonumber)} else {} end) + + {original_name: $name}') + +# ssh reads stdin unless told not to, which would swallow the input of any loop +# this script is called from. Only the job-file upload needs stdin. +remote() { ssh -n -o BatchMode=yes -o ConnectTimeout=8 "$HOST" "$@"; } +remote_with_stdin() { ssh -o BatchMode=yes -o ConnectTimeout=8 "$HOST" "$@"; } + +# One word for where the job stands on the host: done, failed, queued or new. +remote_status() { + remote "cd $STATE 2>/dev/null || { echo new; exit 0; } + if [ -e done/$JOB_ID.txt ]; then echo done + elif [ -e failed/$JOB_ID.log ]; then echo failed + elif [ -d incoming/$JOB_ID ] || [ -d work/$JOB_ID ]; then echo queued + else echo new; fi" +} + +print_transcript() { # $1 = the transcript text + if [[ -z "${1//[[:space:]]/}" ]]; then + echo "Error: the transcript came back empty" >&2 + exit 1 + fi + echo "Transcription complete! (${SECONDS}s total)" >&2 + printf '%s\n' "$1" +} + +run_remote() { + local status + status=$(remote_status) + + if [[ "$status" == "failed" ]]; then + echo "An earlier attempt at this job failed; trying again..." >&2 + remote "rm -f $STATE/failed/$JOB_ID.log" + status="new" + fi + + if [[ "$status" == "new" ]]; then + echo "Uploading audio file to $HOST..." >&2 + # Copy into uploading/, then rename into incoming/. The queue only ever sees + # a complete job. + remote "mkdir -p $STATE/incoming $STATE/uploading/$JOB_ID" + scp -q -o BatchMode=yes "$AUDIO" "$HOST:$STATE/uploading/$JOB_ID/audio.$EXT" < /dev/null + printf '%s' "$JOB_JSON" | remote_with_stdin "cat > $STATE/uploading/$JOB_ID/job.json" + remote "mv $STATE/uploading/$JOB_ID $STATE/incoming/$JOB_ID" + echo "Job $JOB_ID queued. Waiting for completion..." >&2 + elif [[ "$status" == "queued" ]]; then + echo "Job $JOB_ID is already queued on $HOST. Waiting for completion..." >&2 + fi + + while true; do + # A dropped connection is not a failed job; keep asking until the timeout. + status=$(remote_status 2> /dev/null) || status="unreachable" + case "$status" in + done) + print_transcript "$(remote "cat $STATE/done/$JOB_ID.txt")" + return 0 + ;; + failed) + echo "Error: transcription failed on $HOST" >&2 + remote "cat $STATE/failed/$JOB_ID.log" >&2 || true + exit 1 + ;; + esac + if (( SECONDS >= TIMEOUT )); then + echo "Error: no result after ${TIMEOUT}s. The job is still with $HOST;" >&2 + echo "run the same command again to collect the transcript." >&2 + exit 1 + fi + sleep "$POLL" + [[ "$status" == "unreachable" ]] || echo "Processing... (${SECONDS}s elapsed)" >&2 + done +} + +run_local() { + if [[ ! -x "$WORKER" ]]; then + echo "Error: $HOST is unreachable and there is no local worker at $WORKER" >&2 + exit 1 + fi + local state="$HOME/$STATE" + if [[ ! -s "$state/done/$JOB_ID.txt" ]]; then + echo "Running the transcription locally (this machine is slower; expect a wait)..." >&2 + rm -f "$state/failed/$JOB_ID.log" + rm -rf "$state/uploading/$JOB_ID" + mkdir -p "$state/incoming" "$state/uploading/$JOB_ID" + cp "$AUDIO" "$state/uploading/$JOB_ID/audio.$EXT" + printf '%s' "$JOB_JSON" > "$state/uploading/$JOB_ID/job.json" + [[ -d "$state/incoming/$JOB_ID" ]] || mv "$state/uploading/$JOB_ID" "$state/incoming/$JOB_ID" + HF_HUB_OFFLINE=1 "$WORKER" >&2 < /dev/null + fi + if [[ -e "$state/failed/$JOB_ID.log" ]]; then + echo "Error: local transcription failed" >&2 + cat "$state/failed/$JOB_ID.log" >&2 + exit 1 + fi + if [[ ! -e "$state/done/$JOB_ID.txt" ]]; then + echo "Error: the local worker finished without producing a transcript" >&2 + exit 1 + fi + print_transcript "$(< "$state/done/$JOB_ID.txt")" +} + +if [[ -z "${TRANSCRIBE_LOCAL:-}" ]] && remote true 2> /dev/null; then + run_remote +else + [[ -n "${TRANSCRIBE_LOCAL:-}" ]] || echo "$HOST is unreachable." >&2 + run_local +fi diff --git a/working/meeting-transcription-service/src/transcribe-worker b/working/meeting-transcription-service/src/transcribe-worker new file mode 100755 index 0000000..12e64b0 --- /dev/null +++ b/working/meeting-transcription-service/src/transcribe-worker @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +"""Drain the meeting-transcription queue, one job at a time. + +Layout under the state directory (default ~/.local/state/meeting-transcribe): + + incoming/<id>/ audio.<ext> and an optional job.json, dropped by the client + work/<id>/ the job being processed + done/<id>.txt the transcript, plus done/<id>.json with run metadata + failed/<id>.log what went wrong, by stage + +The client uploads into uploading/<id>/ and renames the folder into incoming/ when +the copy is complete, so the queue only ever lists whole jobs. A folder whose name +starts with a dot is skipped as well, as a second line of defence. Nothing is +written into done/ except by rename, so a reader never sees half a transcript. + +Standard library only. whisper-cli and ffmpeg come from PATH; the diarizer runs in +its own virtualenv and is called as a subprocess. +""" + +from __future__ import annotations + +import argparse +import fcntl +import json +import os +import re +import shutil +import subprocess +import sys +import time +from dataclasses import dataclass, field +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import merge_transcript # noqa: E402 + +MAX_ATTEMPTS = 2 +LANGUAGE_RE = re.compile(r"^[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*$") +SHARE = Path.home() / ".local/share" + + +@dataclass +class Config: + """Where things live and how the tools are called.""" + + home: Path = Path.home() / ".local/state/meeting-transcribe" + whisper_model: Path = SHARE / "whisper-models/ggml-large-v3-turbo-q5_0.bin" + diarize_cmd: list[str] = field( + default_factory=lambda: [ + str(SHARE / "pyannote-diarize/.venv/bin/python"), + str(Path(__file__).resolve().parent / "diarize.py"), + ] + ) + threads: int = max(1, (os.cpu_count() or 4) // 2) + + +class JobError(Exception): + """A job failed at a named stage.""" + + def __init__(self, stage: str, detail: str) -> None: + super().__init__(f"{stage}: {detail}") + self.stage = stage + self.detail = detail + + +def _run(stage: str, argv: list[str]) -> None: + """Run one tool; raise JobError carrying the tail of its stderr on failure.""" + try: + result = subprocess.run(argv, capture_output=True, text=True, check=False, stdin=subprocess.DEVNULL) + except OSError as err: + raise JobError(stage, f"cannot run {argv[0]}: {err.strerror}") from err + if result.returncode != 0: + tail = "\n".join(result.stderr.strip().splitlines()[-15:]) + raise JobError(stage, f"exit {result.returncode}\n{tail}") + + +def _read_job(folder: Path) -> dict: + """Load and validate job.json. A missing file means all defaults.""" + path = folder / "job.json" + if not path.exists(): + return {} + try: + job = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(job, dict): + raise ValueError("expected an object") + except (OSError, ValueError) as err: + raise JobError("job", f"job.json is unreadable: {err}") from err + language = job.get("language", "en") + if not isinstance(language, str) or not LANGUAGE_RE.match(language): + raise JobError("job", f"job.json: bad language {language!r}") + for key in ("speakers", "min_speakers", "max_speakers"): + value = job.get(key) + if value is not None and (isinstance(value, bool) or not isinstance(value, int) or value < 1): + raise JobError("job", f"job.json: {key} must be a positive whole number, got {value!r}") + return job + + +def _diarize_options(job: dict) -> list[str]: + options: list[str] = [] + if job.get("speakers") is not None: + return ["--speakers", str(job["speakers"])] + if job.get("min_speakers") is not None: + options += ["--min-speakers", str(job["min_speakers"])] + if job.get("max_speakers") is not None: + options += ["--max-speakers", str(job["max_speakers"])] + return options + + +def process(folder: Path, config: Config) -> tuple[list[str], dict]: + """Run one job folder through the pipeline. Returns transcript lines and metadata.""" + job = _read_job(folder) + audio = next((p for p in sorted(folder.iterdir()) if p.name.startswith("audio.")), None) + if audio is None: + raise JobError("job", "no audio file in the job folder") + + wav = folder / "speech.wav" + _run("ffmpeg", ["ffmpeg", "-v", "error", "-y", "-i", str(audio), "-ar", "16000", "-ac", "1", str(wav)]) + + # -ojf keeps normal segments (better text) and adds token offsets for the merge. + # -mc 0 stops whisper feeding its own output back in, which is what sends it + # into repetition loops on long meetings. + prefix = folder / "words" + _run("whisper", [ + "whisper-cli", "-m", str(config.whisper_model), "-f", str(wav), + "-l", job.get("language", "en"), "-ojf", "-mc", "0", + "-t", str(config.threads), "-of", str(prefix), + ]) + + turns = folder / "turns.json" + _run("diarize", [*config.diarize_cmd, str(wav), str(turns), *_diarize_options(job)]) + + try: + loaded_turns = merge_transcript.load_turns_json(turns) + # Whisper invents words in silence. Drop them before the loop guard, so a + # quiet meeting isn't mistaken for a repetition loop. + units, dropped = merge_transcript.drop_outside_speech( + merge_transcript.load_whisper_json(prefix.with_suffix(".json")), loaded_turns + ) + # A short stutter is collapsed and recorded; a long loop still fails the job. + units, collapsed = merge_transcript.collapse_repetitions(units) + lines = merge_transcript.merge(units, loaded_turns) + except ValueError as err: + raise JobError("merge", str(err)) from err + + meta = { + "original_name": job.get("original_name"), + "language": job.get("language", "en"), + "speakers_requested": job.get("speakers"), + "speakers_found": len({t.speaker for t in loaded_turns}), + "lines": len(lines), + "words": sum(len(line.split()) - 3 for line in lines), + "dropped_outside_speech": dropped, + "loops_collapsed": [[phrase, count] for phrase, count in collapsed], + } + return lines, meta + + +def _write_atomic(path: Path, text: str) -> None: + partial = path.with_name(path.name + ".partial") + partial.write_text(text, encoding="utf-8") + partial.replace(path) + + +def _fail(failed_dir: Path, job_id: str, stage: str, detail: str) -> None: + """Record a failed job. Falls back to stderr if even the log cannot be written.""" + try: + _write_atomic(failed_dir / f"{job_id}.log", f"stage: {stage}\n{detail}\n") + except OSError as err: + print(f"{job_id}: {stage}: {detail} (and the failure log could not be written: {err})", file=sys.stderr) + + +def _bump_attempts(folder: Path) -> int: + """Count this run against the job, tolerating a missing or broken job.json.""" + path = folder / "job.json" + try: + job = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {} + if not isinstance(job, dict): + return 1 + except (OSError, ValueError): + return 1 # _read_job reports the real problem + try: + attempts = int(job.get("attempts") or 0) + 1 + except (TypeError, ValueError): + attempts = 1 # a malformed counter counts as a first try + job["attempts"] = attempts + path.write_text(json.dumps(job), encoding="utf-8") + return attempts + + +def drain(config: Config) -> dict[str, int]: + """Process every waiting job. Returns counts of done and failed jobs.""" + dirs = {name: config.home / name for name in ("incoming", "work", "done", "failed")} + for folder in dirs.values(): + folder.mkdir(parents=True, exist_ok=True) + + counts = {"done": 0, "failed": 0} + with open(config.home / "lock", "w", encoding="utf-8") as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + return counts # another worker is draining; it will reach these jobs + + while True: + # Jobs stranded in work/ by a crash go first, then arrivals, oldest first. + waiting = sorted( + (p for d in (dirs["work"], dirs["incoming"]) for p in d.iterdir() + if p.is_dir() and not p.name.startswith(".")), + key=lambda p: (p.parent.name != "work", p.stat().st_mtime), + ) + if not waiting: + return counts + source = waiting[0] + job_id = source.name + done_txt = dirs["done"] / f"{job_id}.txt" + if done_txt.exists(): + shutil.rmtree(source) + continue + + folder = dirs["work"] / job_id + if source != folder: + source.rename(folder) + started = time.monotonic() + try: + if _bump_attempts(folder) > MAX_ATTEMPTS: + raise JobError("worker", f"gave up after {MAX_ATTEMPTS} attempts; the job kept crashing") + lines, meta = process(folder, config) + meta["seconds"] = round(time.monotonic() - started, 1) + _write_atomic(dirs["done"] / f"{job_id}.json", json.dumps(meta)) + _write_atomic(done_txt, "\n".join(lines) + "\n") + (dirs["failed"] / f"{job_id}.log").unlink(missing_ok=True) + counts["done"] += 1 + except JobError as err: + _fail(dirs["failed"], job_id, err.stage, err.detail) + counts["failed"] += 1 + except Exception as err: # noqa: BLE001 - one bad job must never stop the queue + _fail(dirs["failed"], job_id, "worker", f"{type(err).__name__}: {err}") + counts["failed"] += 1 + finally: + shutil.rmtree(folder, ignore_errors=True) + + +def main(argv: list[str]) -> int: + """CLI: drain the queue once and exit. systemd's path unit calls this.""" + parser = argparse.ArgumentParser(description="Drain the meeting-transcription queue.") + parser.add_argument("--home", type=Path, help="state directory (default ~/.local/state/meeting-transcribe)") + args = parser.parse_args(argv) + config = Config(home=args.home) if args.home else Config() + counts = drain(config) + print(f"done={counts['done']} failed={counts['failed']}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/working/meeting-transcription-service/systemd/meeting-transcribe.path b/working/meeting-transcription-service/systemd/meeting-transcribe.path new file mode 100644 index 0000000..821b292 --- /dev/null +++ b/working/meeting-transcription-service/systemd/meeting-transcribe.path @@ -0,0 +1,12 @@ +[Unit] +Description=Watch the meeting-transcription queue for new jobs + +[Path] +# incoming/ only ever holds complete jobs: the client uploads into uploading/ and +# renames the finished folder across. So "not empty" always means real work, and the +# worker emptying the folder is what lets this unit go quiet again. +DirectoryNotEmpty=%h/.local/state/meeting-transcribe/incoming +MakeDirectory=yes + +[Install] +WantedBy=default.target diff --git a/working/meeting-transcription-service/systemd/meeting-transcribe.service b/working/meeting-transcription-service/systemd/meeting-transcribe.service new file mode 100644 index 0000000..92f5ef4 --- /dev/null +++ b/working/meeting-transcription-service/systemd/meeting-transcribe.service @@ -0,0 +1,9 @@ +[Unit] +Description=Transcribe queued meeting recordings (whisper + pyannote) + +[Service] +Type=oneshot +ExecStart=%h/.local/share/pyannote-diarize/src/transcribe-worker +# The pyannote model is cached after its first download; never reach for the network. +Environment=HF_HUB_OFFLINE=1 +Nice=5 diff --git a/working/meeting-transcription-service/tests/test_diarize.py b/working/meeting-transcription-service/tests/test_diarize.py new file mode 100644 index 0000000..ef009b8 --- /dev/null +++ b/working/meeting-transcription-service/tests/test_diarize.py @@ -0,0 +1,74 @@ +"""Tests for diarize's pure parts. The pyannote pipeline itself is not loaded here.""" + +import sys +from collections import namedtuple +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +import diarize # noqa: E402 + +Segment = namedtuple("Segment", "start end") + + +class TestTurnsFromTracks: + def test_diarize_turns_from_tracks_converts_and_sorts(self): + """Normal: (segment, track, label) triples become sorted turn dicts.""" + tracks = [(Segment(5.0, 9.25), "_", "SPEAKER_01"), (Segment(0.5, 4.0), "_", "SPEAKER_00")] + assert diarize.turns_from_tracks(tracks) == [ + {"start": 0.5, "end": 4.0, "speaker": "SPEAKER_00"}, + {"start": 5.0, "end": 9.25, "speaker": "SPEAKER_01"}, + ] + + def test_diarize_turns_from_tracks_rounds_to_milliseconds(self): + """Boundary: float noise from the model is rounded away.""" + tracks = [(Segment(0.03096875, 1.9980000000000002), "_", "SPEAKER_00")] + assert diarize.turns_from_tracks(tracks) == [ + {"start": 0.031, "end": 1.998, "speaker": "SPEAKER_00"} + ] + + def test_diarize_turns_from_tracks_drops_empty_segments(self): + """Boundary: zero or negative length segments carry no speech.""" + tracks = [(Segment(2.0, 2.0), "_", "SPEAKER_00"), (Segment(3.0, 2.5), "_", "SPEAKER_00")] + assert diarize.turns_from_tracks(tracks) == [] + + def test_diarize_turns_from_tracks_empty_input_is_empty_list(self): + """Boundary: no tracks.""" + assert diarize.turns_from_tracks([]) == [] + + +class TestParseArgs: + def test_diarize_parse_args_speakers_sets_exact_count(self): + """Normal: --speakers pins the count.""" + args = diarize.parse_args(["a.wav", "out.json", "--speakers", "3"]) + assert (args.audio, args.out, args.speakers) == ("a.wav", "out.json", 3) + + def test_diarize_parse_args_defaults_let_the_model_estimate(self): + """Normal: no count given.""" + args = diarize.parse_args(["a.wav", "out.json"]) + assert args.speakers is None and args.min_speakers is None and args.max_speakers is None + + @pytest.mark.parametrize("argv", [ + ["a.wav", "out.json", "--speakers", "0"], + ["a.wav", "out.json", "--speakers", "-2"], + ["a.wav", "out.json", "--speakers", "three"], + ["a.wav", "out.json", "--speakers", "3", "--max-speakers", "5"], + ["a.wav", "out.json", "--min-speakers", "4", "--max-speakers", "2"], + ["a.wav"], + ]) + def test_diarize_parse_args_rejects_bad_counts(self, argv): + """Error: non-positive, non-numeric, contradictory or missing arguments.""" + with pytest.raises(SystemExit): + diarize.parse_args(argv) + + +class TestPipelineKwargs: + def test_diarize_pipeline_kwargs_only_passes_what_was_given(self): + """Normal: unset options are not forwarded to the model.""" + args = diarize.parse_args(["a.wav", "o.json", "--min-speakers", "2", "--max-speakers", "4"]) + assert diarize.pipeline_kwargs(args) == {"min_speakers": 2, "max_speakers": 4} + args = diarize.parse_args(["a.wav", "o.json", "--speakers", "3"]) + assert diarize.pipeline_kwargs(args) == {"num_speakers": 3} + assert diarize.pipeline_kwargs(diarize.parse_args(["a.wav", "o.json"])) == {} diff --git a/working/meeting-transcription-service/tests/test_merge_transcript.py b/working/meeting-transcription-service/tests/test_merge_transcript.py new file mode 100644 index 0000000..a584043 --- /dev/null +++ b/working/meeting-transcription-service/tests/test_merge_transcript.py @@ -0,0 +1,510 @@ +"""Tests for merge_transcript: whisper words + speaker turns -> transcript lines.""" + +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +import merge_transcript as mt # noqa: E402 +from merge_transcript import Turn, Unit # noqa: E402 + + +def words(start, *tokens, step=0.5): + """Evenly spaced word units beginning at ``start`` seconds.""" + return [ + Unit(start + i * step, start + (i + 1) * step, f" {token}") + for i, token in enumerate(tokens) + ] + + +class TestMergeNormal: + def test_merge_transcript_merge_two_speakers_yields_one_line_per_turn(self): + """Normal: words fall into the turn that contains them.""" + units = words(0.0, "Good", "morning.") + words(6.0, "Sounds", "good.") + turns = [Turn(0.0, 5.0, "SPEAKER_00"), Turn(5.5, 9.0, "SPEAKER_01")] + assert mt.merge(units, turns) == [ + "00:00:00 Speaker A: Good morning.", + "00:00:06 Speaker B: Sounds good.", + ] + + def test_merge_transcript_merge_letters_follow_order_of_first_speech(self): + """Normal: Speaker A is whoever talks first, whatever pyannote called them.""" + units = words(0.0, "First.") + words(4.0, "Second.") + words(8.0, "Again.") + turns = [ + Turn(0.0, 3.0, "SPEAKER_02"), + Turn(3.5, 7.0, "SPEAKER_00"), + Turn(7.5, 10.0, "SPEAKER_02"), + ] + assert mt.merge(units, turns) == [ + "00:00:00 Speaker A: First.", + "00:00:04 Speaker B: Second.", + "00:00:08 Speaker A: Again.", + ] + + def test_merge_transcript_merge_same_speaker_across_turns_stays_one_line(self): + """Normal: pyannote splits a speaker's run into turns; the line does not.""" + units = words(0.0, "One", "two") + words(1.2, "three.") + turns = [Turn(0.0, 1.0, "SPEAKER_00"), Turn(1.1, 2.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["00:00:00 Speaker A: One two three."] + + def test_merge_transcript_merge_long_pause_starts_a_new_line(self): + """Normal: a pause past max_gap_s breaks the line so timestamps stay useful.""" + units = words(0.0, "Before.") + words(10.0, "After.") + turns = [Turn(0.0, 12.0, "SPEAKER_00")] + assert mt.merge(units, turns, max_gap_s=3.0) == [ + "00:00:00 Speaker A: Before.", + "00:00:10 Speaker A: After.", + ] + + def test_merge_transcript_merge_word_in_a_gap_goes_to_nearest_turn(self): + """Normal: whisper's timing drifts; a word between turns joins the closer one.""" + units = [Unit(4.6, 4.8, " late"), Unit(5.2, 5.4, " early")] + turns = [Turn(0.0, 4.5, "SPEAKER_00"), Turn(5.5, 9.0, "SPEAKER_01")] + assert mt.merge(units, turns) == [ + "00:00:04 Speaker A: late", + "00:00:05 Speaker B: early", + ] + + def test_merge_transcript_merge_overlapping_turns_pick_the_larger_overlap(self): + """Normal: with overlapped speech, the word goes where most of it sits.""" + # The anchor word fixes SPEAKER_00 as A, so the contested word's label + # actually shows which turn won: 0.2 s of overlap with A, 0.9 s with B. + units = [Unit(1.0, 1.5, " anchor"), Unit(4.0, 5.0, " contested")] + turns = [Turn(0.0, 4.2, "SPEAKER_00"), Turn(4.1, 9.0, "SPEAKER_01")] + assert mt.merge(units, turns) == [ + "00:00:01 Speaker A: anchor", + "00:00:04 Speaker B: contested", + ] + + +class TestMergeBoundary: + def test_merge_transcript_merge_timestamp_past_an_hour_is_floored(self): + """Boundary: 3725.9 s renders as 01:02:05.""" + units = [Unit(3725.9, 3726.4, " Late.")] + turns = [Turn(3700.0, 3800.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["01:02:05 Speaker A: Late."] + + def test_merge_transcript_merge_zero_length_and_blank_units_are_handled(self): + """Boundary: whisper emits empty units and zero-length words.""" + units = [Unit(0.0, 0.0, ""), Unit(2.18, 2.18, " we"), Unit(2.18, 2.54, " have")] + turns = [Turn(0.0, 5.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["00:00:02 Speaker A: we have"] + + def test_merge_transcript_merge_twenty_seventh_speaker_gets_a_number(self): + """Boundary: letters run out after Z.""" + units, turns = [], [] + for i in range(27): + units += [Unit(i * 10.0, i * 10.0 + 1, f" s{i}")] + turns += [Turn(i * 10.0, i * 10.0 + 5, f"SPEAKER_{i:02d}")] + lines = mt.merge(units, turns) + assert lines[25] == "00:04:10 Speaker Z: s25" + assert lines[26] == "00:04:20 Speaker 26: s26" + + def test_merge_transcript_merge_unicode_and_inner_spacing_survive(self): + """Boundary: non-ASCII text; whitespace collapses to single spaces.""" + units = [Unit(0.0, 0.5, " Բարև,"), Unit(0.5, 1.0, " Երևան"), Unit(1.0, 1.5, " — café.")] + turns = [Turn(0.0, 2.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["00:00:00 Speaker A: Բարև, Երևան — café."] + + def test_merge_transcript_merge_unsorted_input_is_sorted_first(self): + """Boundary: neither list has to arrive in time order.""" + units = words(6.0, "Second.") + words(0.0, "First.") + turns = [Turn(5.0, 9.0, "SPEAKER_01"), Turn(0.0, 4.0, "SPEAKER_00")] + assert mt.merge(units, turns) == [ + "00:00:00 Speaker A: First.", + "00:00:06 Speaker B: Second.", + ] + + +class TestMergeError: + def test_merge_transcript_merge_no_turns_raises(self): + """Error: words with no diarization cannot be labelled.""" + with pytest.raises(ValueError, match="no speaker turns"): + mt.merge(words(0.0, "Hello."), []) + + def test_merge_transcript_merge_no_words_raises(self): + """Error: an empty transcription is a failure, not an empty transcript.""" + with pytest.raises(ValueError, match="no speech"): + mt.merge([Unit(0.0, 0.0, " ")], [Turn(0.0, 5.0, "SPEAKER_00")]) + + def test_merge_transcript_merge_negative_gap_setting_raises(self): + """Error: max_gap_s must not be negative.""" + with pytest.raises(ValueError, match="max_gap_s"): + mt.merge(words(0.0, "Hi."), [Turn(0.0, 1.0, "SPEAKER_00")], max_gap_s=-1) + + +class TestLoaders: + def test_merge_transcript_load_whisper_json_reads_millisecond_offsets(self, tmp_path): + """Normal: whisper-cli -oj stores offsets in milliseconds.""" + path = tmp_path / "w.json" + path.write_text(json.dumps({"transcription": [ + {"offsets": {"from": 0, "to": 190}, "text": " if"}, + {"offsets": {"from": 190, "to": 590}, "text": " there's"}, + ]})) + assert mt.load_whisper_json(path) == [Unit(0.0, 0.19, " if"), Unit(0.19, 0.59, " there's")] + + def test_merge_transcript_load_turns_json_reads_seconds(self, tmp_path): + """Normal: the diarizer writes start/end in seconds.""" + path = tmp_path / "t.json" + path.write_text(json.dumps([{"start": 0.5, "end": 4.25, "speaker": "SPEAKER_00"}])) + assert mt.load_turns_json(path) == [Turn(0.5, 4.25, "SPEAKER_00")] + + @pytest.mark.parametrize("content", ["not json", "{}", '{"transcription": [{"text": "x"}]}']) + def test_merge_transcript_load_whisper_json_malformed_raises(self, tmp_path, content): + """Error: a truncated or foreign file is rejected with its path named.""" + path = tmp_path / "w.json" + path.write_text(content) + with pytest.raises(ValueError, match="w.json"): + mt.load_whisper_json(path) + + @pytest.mark.parametrize("content", ["not json", "{}", '[{"start": 1}]']) + def test_merge_transcript_load_turns_json_malformed_raises(self, tmp_path, content): + """Error: same for the turns file.""" + path = tmp_path / "t.json" + path.write_text(content) + with pytest.raises(ValueError, match="t.json"): + mt.load_turns_json(path) + + +class TestLoadersMissingFile: + def test_merge_transcript_load_whisper_json_missing_file_raises_value_error(self, tmp_path): + """Error: a missing whisper file is a clean ValueError naming the path.""" + with pytest.raises(ValueError, match="gone.json"): + mt.load_whisper_json(tmp_path / "gone.json") + + def test_merge_transcript_load_turns_json_missing_file_raises_value_error(self, tmp_path): + """Error: a missing turns file (the diarizer failed upstream) is a clean ValueError.""" + with pytest.raises(ValueError, match="gone.json"): + mt.load_turns_json(tmp_path / "gone.json") + + def test_merge_transcript_main_missing_turns_file_exits_one_without_traceback(self, tmp_path, capsys): + """Error: the CLI reports it in one line and prints nothing on stdout.""" + w = tmp_path / "w.json" + w.write_text('{"transcription": [{"offsets": {"from": 0, "to": 500}, "text": " Hi."}]}') + assert mt.main([str(w), str(tmp_path / "gone.json")]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "gone.json" in captured.err and "Traceback" not in captured.err + + +class TestCli: + def _files(self, tmp_path, transcription, turns): + w = tmp_path / "w.json" + t = tmp_path / "t.json" + w.write_text(json.dumps({"transcription": transcription})) + t.write_text(json.dumps(turns)) + return str(w), str(t) + + def test_merge_transcript_main_prints_transcript_and_returns_zero(self, tmp_path, capsys): + """Normal: the CLI writes the lines to stdout.""" + w, t = self._files( + tmp_path, + [{"offsets": {"from": 0, "to": 500}, "text": " Hello."}], + [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}], + ) + assert mt.main([w, t]) == 0 + assert capsys.readouterr().out == "00:00:00 Speaker A: Hello.\n" + + def test_merge_transcript_main_failure_prints_nothing_on_stdout(self, tmp_path, capsys): + """Error: a failed merge exits 1 with the reason on stderr only.""" + w, t = self._files(tmp_path, [], [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]) + assert mt.main([w, t]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "no speech" in captured.err + + def test_merge_transcript_main_wrong_argument_count_returns_two(self, capsys): + """Error: usage.""" + assert mt.main([]) == 2 + assert "usage" in capsys.readouterr().err.lower() + + +def tok(start_ms, end_ms, text): + return {"text": text, "offsets": {"from": start_ms, "to": end_ms}} + + +class TestLoadWhisperTokens: + """whisper-cli -ojf keeps normal segments and adds per-token offsets inside each.""" + + def _write(self, tmp_path, segments): + path = tmp_path / "full.json" + path.write_text(json.dumps({"transcription": segments})) + return path + + def test_merge_transcript_load_whisper_json_prefers_tokens_over_segments(self, tmp_path): + """Normal: with tokens present, units are words, not whole segments.""" + path = self._write(tmp_path, [{ + "offsets": {"from": 0, "to": 2000}, "text": " we basically", + "tokens": [tok(0, 0, "[_BEG_]"), tok(10, 400, " we"), tok(500, 1900, " basically"), tok(2000, 2000, "[_TT_100]")], + }]) + assert mt.load_whisper_json(path) == [Unit(0.01, 0.4, " we"), Unit(0.5, 1.9, " basically")] + + def test_merge_transcript_load_whisper_json_joins_subword_and_punctuation_tokens(self, tmp_path): + """Normal: a token with no leading space continues the previous word.""" + path = self._write(tmp_path, [{ + "offsets": {"from": 0, "to": 3000}, "text": " Saturday, yes.", + "tokens": [tok(0, 300, " Sat"), tok(300, 700, "urday"), tok(700, 750, ","), tok(900, 1300, " yes"), tok(1300, 1350, ".")], + }]) + assert mt.load_whisper_json(path) == [Unit(0.0, 0.75, " Saturday,"), Unit(0.9, 1.35, " yes.")] + + def test_merge_transcript_load_whisper_json_segment_without_tokens_falls_back(self, tmp_path): + """Boundary: plain -oj output, or a segment whose token list is empty.""" + path = self._write(tmp_path, [ + {"offsets": {"from": 0, "to": 1000}, "text": " First.", "tokens": []}, + {"offsets": {"from": 1000, "to": 2000}, "text": " Second."}, + ]) + assert mt.load_whisper_json(path) == [Unit(0.0, 1.0, " First."), Unit(1.0, 2.0, " Second.")] + + def test_merge_transcript_load_whisper_json_leading_continuation_token_stands_alone(self, tmp_path): + """Boundary: the very first token has nothing to attach to.""" + path = self._write(tmp_path, [{ + "offsets": {"from": 0, "to": 500}, "text": "ing on", + "tokens": [tok(0, 200, "ing"), tok(200, 500, " on")], + }]) + assert mt.load_whisper_json(path) == [Unit(0.0, 0.2, "ing"), Unit(0.2, 0.5, " on")] + + def test_merge_transcript_load_whisper_json_token_without_offsets_raises(self, tmp_path): + """Error: a malformed token is rejected, naming the file.""" + path = self._write(tmp_path, [{"offsets": {"from": 0, "to": 1}, "text": " x", "tokens": [{"text": " x"}]}]) + with pytest.raises(ValueError, match="full.json"): + mt.load_whisper_json(path) + + +class TestFindRepetition: + def test_merge_transcript_find_repetition_clean_text_returns_none(self): + """Normal: ordinary speech, even with stock phrases scattered about.""" + text = " ".join(f"point {i} and i don't know if that works for us" for i in range(8)) + assert mt.find_repetition(text) is None + + def test_merge_transcript_find_repetition_back_to_back_phrase_is_reported(self): + """Normal: whisper's hallucination loop, the same phrase again and again.""" + text = "okay so " + "We don't know where we're going to be. " * 6 + "anyway moving on" + hit = mt.find_repetition(text) + assert hit is not None + phrase, count = hit + assert count >= 6 + assert "where we're going to be" in phrase.lower() + + def test_merge_transcript_find_repetition_three_repeats_is_tolerated(self): + """Boundary: people do say a thing three times; four in a row is the line.""" + assert mt.find_repetition("go back to this area " * 3) is None + assert mt.find_repetition("go back to this area " * 4) is not None + + def test_merge_transcript_find_repetition_short_fillers_are_not_loops(self): + """Boundary: 'yeah yeah yeah yeah yeah' is speech, not a loop.""" + assert mt.find_repetition("yeah " * 9 + "no no no no no") is None + + def test_merge_transcript_find_repetition_empty_text_returns_none(self): + """Boundary: nothing to scan.""" + assert mt.find_repetition("") is None + + def test_merge_transcript_main_loop_in_transcript_fails(self, tmp_path, capsys): + """Error: a looping transcription exits 1 with nothing on stdout.""" + words_ = ("we don't know where we're going to be " * 5).split() + # 40 words at 3 s each: two minutes of loop, far past the 30 s a short stutter gets + segs = [{"offsets": {"from": i * 3_000, "to": i * 3_000 + 2_500}, "text": f" {w}"} for i, w in enumerate(words_)] + w = tmp_path / "w.json" + w.write_text(json.dumps({"transcription": segs})) + t = tmp_path / "t.json" + t.write_text(json.dumps([{"start": 0.0, "end": 130.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(w), str(t)]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "repeat" in captured.err.lower() + + +class TestInvertedTokenTimes: + """whisper-cli sometimes clamps a token's start to its segment, leaving end < start.""" + + def test_merge_transcript_load_whisper_json_inverted_token_becomes_zero_length(self, tmp_path): + """Boundary: an end before the start is treated as a zero-length word at the start.""" + path = tmp_path / "inv.json" + path.write_text(json.dumps({"transcription": [{ + "offsets": {"from": 13120, "to": 16160}, "text": " which you", + "tokens": [tok(13120, 9580, " which"), tok(13120, 10140, " you")], + }]})) + assert mt.load_whisper_json(path) == [Unit(13.12, 13.12, " which"), Unit(13.12, 13.12, " you")] + + def test_merge_transcript_merge_inverted_tokens_do_not_split_a_sentence(self, tmp_path): + """Normal: the real case, after a pause longer than max_gap_s.""" + path = tmp_path / "inv.json" + path.write_text(json.dumps({"transcription": [ + {"offsets": {"from": 5360, "to": 8640}, "text": " blue pixels", + "tokens": [tok(5360, 7000, " blue"), tok(7000, 8640, " pixels")]}, + {"offsets": {"from": 13120, "to": 16160}, "text": " which you know", + "tokens": [tok(13120, 9580, " which"), tok(13120, 10140, " you"), tok(13120, 10890, " know")]}, + ]})) + turns = [Turn(0.0, 8.8, "SPEAKER_00"), Turn(13.3, 16.0, "SPEAKER_00")] + assert mt.merge(mt.load_whisper_json(path), turns) == [ + "00:00:05 Speaker A: blue pixels", + "00:00:13 Speaker A: which you know", + ] + + +class TestDropOutsideSpeech: + """Whisper invents words in silence; the diarizer knows where the speech is.""" + + def test_merge_transcript_drop_outside_speech_removes_words_far_from_any_turn(self): + """Normal: a hallucinated run in a silent stretch goes; real words stay.""" + real = words(0.0, "Good", "morning.") + invented = words(60.0, "Thank", "you.", "Thank", "you.", step=5.0) + kept, dropped = mt.drop_outside_speech(real + invented, [Turn(0.0, 3.0, "SPEAKER_00")]) + assert kept == real + assert dropped == 4 + + def test_merge_transcript_drop_outside_speech_keeps_words_between_close_turns(self): + """Normal: a word in a short gap between two turns is speech the diarizer clipped.""" + units = words(0.0, "One.") + words(3.2, "and") + words(4.0, "two.") + turns = [Turn(0.0, 3.0, "SPEAKER_00"), Turn(4.0, 6.0, "SPEAKER_01")] + assert mt.drop_outside_speech(units, turns) == (units, 0) + + def test_merge_transcript_drop_outside_speech_word_exactly_at_the_margin_is_kept(self): + """Boundary: the margin is inclusive.""" + unit = Unit(5.0, 5.5, " edge") + assert mt.drop_outside_speech([unit], [Turn(0.0, 3.0, "S")], margin_s=2.0) == ([unit], 0) + + def test_merge_transcript_drop_outside_speech_word_just_past_the_margin_is_dropped(self): + """Boundary: one millisecond further and it goes.""" + unit = Unit(5.001, 5.5, " edge") + assert mt.drop_outside_speech([unit], [Turn(0.0, 3.0, "S")], margin_s=2.0) == ([], 1) + + def test_merge_transcript_drop_outside_speech_zero_margin_needs_contact_with_a_turn(self): + """Boundary: margin 0 keeps a word touching a turn and drops one that is not.""" + touching, apart = Unit(3.0, 3.4, " touch"), Unit(3.5, 3.9, " apart") + kept, dropped = mt.drop_outside_speech([touching, apart], [Turn(0.0, 3.0, "S")], margin_s=0.0) + assert (kept, dropped) == ([touching], 1) + + def test_merge_transcript_drop_outside_speech_before_the_first_turn_counts_too(self): + """Boundary: silence at the start of a recording.""" + early = Unit(1.0, 1.5, " Thanks.") + assert mt.drop_outside_speech([early], [Turn(30.0, 40.0, "S")]) == ([], 1) + + def test_merge_transcript_drop_outside_speech_no_units_is_a_no_op(self): + """Boundary: nothing in, nothing out.""" + assert mt.drop_outside_speech([], [Turn(0.0, 3.0, "S")]) == ([], 0) + + def test_merge_transcript_drop_outside_speech_no_turns_raises(self): + """Error: without turns there is no way to tell speech from silence.""" + with pytest.raises(ValueError, match="no speaker turns"): + mt.drop_outside_speech(words(0.0, "Hello."), []) + + def test_merge_transcript_drop_outside_speech_negative_margin_raises(self): + """Error: a negative margin is a caller bug.""" + with pytest.raises(ValueError, match="margin"): + mt.drop_outside_speech(words(0.0, "Hello."), [Turn(0.0, 3.0, "S")], margin_s=-1.0) + + def test_merge_transcript_main_silence_hallucinations_do_not_trip_the_loop_guard(self, tmp_path, capsys): + """Normal: a run of invented thank-yous in silence is dropped, not reported as a loop.""" + transcription = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}] + [ + {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."} + for i in range(12) + ] + whisper = tmp_path / "w.json" + whisper.write_text(json.dumps({"transcription": transcription})) + turns = tmp_path / "t.json" + turns.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(whisper), str(turns)]) == 0 + assert capsys.readouterr().out == "00:00:00 Speaker A: Good morning.\n" + + def test_merge_transcript_main_a_loop_inside_speech_still_fails(self, tmp_path, capsys): + """Error: a real repetition loop happens while someone is talking, and is still caught.""" + transcription = [ + {"offsets": {"from": i * 1_000, "to": i * 1_000 + 900}, "text": " where we're going to be"} + for i in range(60) # a full minute of the same phrase + ] + whisper = tmp_path / "w.json" + whisper.write_text(json.dumps({"transcription": transcription})) + turns = tmp_path / "t.json" + turns.write_text(json.dumps([{"start": 0.0, "end": 70.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(whisper), str(turns)]) == 1 + assert "looped" in capsys.readouterr().err + + +class TestCollapseRepetitions: + """A short stutter is collapsed to one occurrence; a long loop still fails the job.""" + + def test_merge_transcript_collapse_repetitions_short_loop_keeps_one_copy(self): + """Normal: a six-word phrase said four times in ten seconds becomes one phrase.""" + units = words(0.0, "So", "anyway,") + words(1.0, *("fair, it's not going to be".split() * 4), step=0.4) + words(12.0, "done.") + kept, collapsed = mt.collapse_repetitions(units) + assert " ".join(u.text.strip() for u in kept) == "So anyway, fair, it's not going to be done." + assert collapsed == [("fair, it's not going to be", 4)] + + def test_merge_transcript_collapse_repetitions_clean_units_are_untouched(self): + """Normal: ordinary speech passes through with nothing collapsed.""" + units = words(0.0, "We", "have", "detection", "today,", "and", "we", "have", "a", "plan.") + assert mt.collapse_repetitions(units) == (units, []) + + def test_merge_transcript_collapse_repetitions_two_loops_both_collapse(self): + """Normal: separate stutters are each collapsed and each reported.""" + units = (words(0.0, *("go back to this area".split() * 4), step=0.3) + + words(10.0, "then") + + words(11.0, *("where we're going to be".split() * 5), step=0.3)) + kept, collapsed = mt.collapse_repetitions(units) + assert " ".join(u.text.strip() for u in kept) == "go back to this area then where we're going to be" + assert collapsed == [("go back to this area", 4), ("where we're going to be", 5)] + + def test_merge_transcript_collapse_repetitions_three_repeats_are_speech(self): + """Boundary: three repeats is emphasis, not a loop, and stays.""" + units = words(0.0, *("this is the thing".split() * 3)) + assert mt.collapse_repetitions(units) == (units, []) + + def test_merge_transcript_collapse_repetitions_single_word_runs_stay(self): + """Boundary: "yeah yeah yeah yeah" is ordinary speech.""" + units = words(0.0, *(["yeah"] * 8)) + assert mt.collapse_repetitions(units) == (units, []) + + def test_merge_transcript_collapse_repetitions_loop_at_the_limit_is_collapsed(self): + """Boundary: a loop lasting exactly max_loop_s is still a short one.""" + units = words(0.0, *("we do not know where".split() * 4), step=1.5) # 20 words, 30.0 s + kept, collapsed = mt.collapse_repetitions(units, max_loop_s=30.0) + assert collapsed == [("we do not know where", 4)] and len(kept) == 5 + + def test_merge_transcript_collapse_repetitions_loop_past_the_limit_raises(self): + """Error: a loop longer than max_loop_s means real speech was lost, so the job fails.""" + units = words(0.0, *("we do not know where".split() * 4), step=1.6) # 32.0 s + with pytest.raises(ValueError, match="looped"): + mt.collapse_repetitions(units, max_loop_s=30.0) + + def test_merge_transcript_collapse_repetitions_segment_units_collapse_too(self): + """Boundary: whisper's segment fallback puts a whole phrase in one unit.""" + units = [Unit(i * 1.0, i * 1.0 + 0.9, " where we're going to be") for i in range(6)] + kept, collapsed = mt.collapse_repetitions(units) + assert len(kept) == 1 and collapsed == [("where we're going to be", 6)] + + def test_merge_transcript_collapse_repetitions_negative_limit_raises(self): + """Error: a negative limit is a caller bug.""" + with pytest.raises(ValueError, match="max_loop_s"): + mt.collapse_repetitions(words(0.0, "hi"), max_loop_s=-1.0) + + def test_merge_transcript_main_short_loop_is_collapsed_not_fatal(self, tmp_path, capsys): + """Normal: the CLI prints the collapsed transcript and exits 0.""" + words_ = "we have detection today " .split() + ("fair, it's not going to be " * 4).split() + "easy.".split() + segs = [{"offsets": {"from": i * 300, "to": i * 300 + 250}, "text": f" {w}"} for i, w in enumerate(words_)] + w = tmp_path / "w.json"; w.write_text(json.dumps({"transcription": segs})) + t = tmp_path / "t.json"; t.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(w), str(t)]) == 0 + assert capsys.readouterr().out == "00:00:00 Speaker A: we have detection today fair, it's not going to be easy.\n" + + +class TestCollapseRepetitionsInsideUnits: + """The repeat can live inside one unit's text, which is what whisper's segment fallback emits.""" + + def test_merge_transcript_collapse_repetitions_loop_inside_one_unit_is_collapsed(self): + """Error case turned regression: a single unit holding the phrase four times must not hang.""" + unit = Unit(0.0, 8.0, " " + " ".join(["we do not know where"] * 4)) + kept, collapsed = mt.collapse_repetitions(unit and [unit]) + assert [u.text for u in kept] == [" we do not know where"] + assert collapsed == [("we do not know where", 4)] + + def test_merge_transcript_collapse_repetitions_partial_unit_keeps_its_other_words(self): + """Boundary: a unit holding the last repeat and real words after it keeps the real words.""" + units = words(0.0, *("go back to this area".split() * 3), step=0.5) + [ + Unit(7.5, 9.0, " go back to this area and then we stopped.") + ] + kept, collapsed = mt.collapse_repetitions(units) + assert " ".join(u.text.strip() for u in kept) == "go back to this area and then we stopped." + assert collapsed == [("go back to this area", 4)] diff --git a/working/meeting-transcription-service/tests/test_ratio_transcribe.py b/working/meeting-transcription-service/tests/test_ratio_transcribe.py new file mode 100644 index 0000000..1a10b3e --- /dev/null +++ b/working/meeting-transcription-service/tests/test_ratio_transcribe.py @@ -0,0 +1,263 @@ +"""Tests for ratio-transcribe, the client. + +ssh and scp are replaced by fakes that act on a temp directory standing in for the +remote home, so the client's real logic (job ids, upload-then-rename, polling, +collecting, the local fallback) runs against a filesystem it can't tell from the host. +""" + +import json +import os +import subprocess +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parent.parent / "src" / "ratio-transcribe" +STATE = ".local/state/meeting-transcribe" + +FAKE_SSH = r"""#!/usr/bin/env bash +# ssh [-o k=v]... host command... -> run the command with HOME at the fake remote +[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255 +no_stdin="" +while [[ "$1" == -* ]]; do [[ "$1" == "-n" ]] && no_stdin=1; [[ "$1" == "-o" ]] && shift; shift; done +shift # host +printf '%s\n' "$*" >> "$FAKE_SSH_LOG" +if [[ -n "$no_stdin" ]]; then + ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" < /dev/null ) + rc=$? +else + ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" ) + rc=$? + cat > /dev/null # like the real ssh, drain whatever stdin the command left behind +fi +if [[ "$*" == *"incoming/"* && "$*" == mv* && -n "${FAKE_ON_SUBMIT:-}" ]]; then + ( cd "$FAKE_REMOTE_HOME" && bash -c "$FAKE_ON_SUBMIT" ) +fi +exit $rc +""" + +FAKE_SCP = r"""#!/usr/bin/env bash +[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255 +while [[ "$1" == -* ]]; do [[ "$1" == "-o" ]] && shift; shift; done +printf '%s -> %s\n' "$1" "$2" >> "$FAKE_SCP_LOG" +cp "$1" "$FAKE_REMOTE_HOME/${2#*:}" +""" + +# What the worker would do, compressed: finish or fail whatever sits in incoming/. +WORKER_OK = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d"); + printf '00:00:00 Speaker A: Hello from the host.\\n' > done/$id.txt; rm -rf "$d"; done""" +WORKER_FAIL = f"""cd {STATE}; mkdir -p failed; for d in incoming/*/; do id=$(basename "$d"); + printf 'stage: whisper\\nexit 3\\n' > failed/$id.log; rm -rf "$d"; done""" +WORKER_EMPTY = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d"); + : > done/$id.txt; rm -rf "$d"; done""" + + +class Rig: + def __init__(self, tmp_path): + self.tmp = tmp_path + self.remote = tmp_path / "remote-home" + self.remote.mkdir() + self.local_home = tmp_path / "local-home" + self.local_home.mkdir() + self.bin = tmp_path / "bin" + self.bin.mkdir() + for name, body in (("ssh", FAKE_SSH), ("scp", FAKE_SCP)): + path = self.bin / name + path.write_text(body) + path.chmod(0o755) + self.ssh_log = tmp_path / "ssh.log" + self.scp_log = tmp_path / "scp.log" + self.audio = tmp_path / "meeting.m4a" + self.audio.write_bytes(b"pretend audio") + + def run(self, args=None, on_submit=WORKER_OK, **env_extra): + env = { + "PATH": f"{self.bin}:{os.environ['PATH']}", + "HOME": str(self.local_home), + "FAKE_REMOTE_HOME": str(self.remote), + "FAKE_SSH_LOG": str(self.ssh_log), + "FAKE_SCP_LOG": str(self.scp_log), + "TRANSCRIBE_HOST": "testhost", + "TRANSCRIBE_POLL": "0", + "TRANSCRIBE_TIMEOUT": "5", + } + if on_submit: + env["FAKE_ON_SUBMIT"] = on_submit + env.update({k: str(v) for k, v in env_extra.items()}) + if args is None: + args = [str(self.audio)] + return subprocess.run([str(SCRIPT), *args], env=env, capture_output=True, text=True, timeout=60) + + def state(self, *parts): + return self.remote.joinpath(STATE, *parts) + + def uploads(self): + return self.scp_log.read_text().splitlines() if self.scp_log.exists() else [] + + +@pytest.fixture +def rig(tmp_path): + return Rig(tmp_path) + + +class TestClientNormal: + def test_ratio_transcribe_new_recording_is_uploaded_and_transcript_printed(self, rig): + """Normal: upload, wait, print. stdout is the transcript and nothing else.""" + result = rig.run() + assert result.returncode == 0, result.stderr + assert result.stdout == "00:00:00 Speaker A: Hello from the host.\n" + assert len(rig.uploads()) == 1 + + def test_ratio_transcribe_job_file_carries_language_speakers_and_name(self, rig): + """Normal: the options reach the host inside job.json.""" + rig.run(args=[str(rig.audio), "es"], on_submit=None, SPEAKERS=3, TRANSCRIBE_TIMEOUT=0) + jobs = list(rig.state("incoming").glob("*/job.json")) + assert len(jobs) == 1 + assert json.loads(jobs[0].read_text()) == { + "language": "es", "speakers": 3, "original_name": "meeting.m4a", + } + assert (jobs[0].parent / "audio.m4a").read_bytes() == b"pretend audio" + + def test_ratio_transcribe_finished_job_is_collected_without_uploading_again(self, rig): + """Normal: rerunning after a dropped connection costs one ssh round trip.""" + first = rig.run() + rig.scp_log.unlink() + second = rig.run(on_submit=None) + assert second.returncode == 0 + assert second.stdout == first.stdout + assert rig.uploads() == [] + + def test_ratio_transcribe_job_id_depends_on_audio_and_options(self, rig): + """Normal: same file with a different speaker count is a different job.""" + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0, SPEAKERS=3) + assert len(list(rig.state("incoming").iterdir())) == 2 + + def test_ratio_transcribe_earlier_failure_is_retried(self, rig): + """Normal: a stale failure log does not block a fresh attempt.""" + assert rig.run(on_submit=WORKER_FAIL).returncode == 1 + result = rig.run(on_submit=WORKER_OK) + assert result.returncode == 0 + assert "Hello from the host" in result.stdout + assert list(rig.state("failed").glob("*.log")) == [] + + +class TestClientBoundary: + def test_ratio_transcribe_job_still_queued_is_not_uploaded_twice(self, rig): + """Boundary: a second run while the first job waits just joins the wait.""" + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + assert len(rig.uploads()) == 1 + + def test_ratio_transcribe_upload_lands_in_incoming_only_by_rename(self, rig): + """Boundary: the audio is copied into uploading/, never straight into incoming/.""" + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + assert "/uploading/" in rig.uploads()[0] + assert list(rig.state("uploading").iterdir()) == [] + + def test_ratio_transcribe_awkward_filename_survives(self, rig): + """Boundary: spaces and quotes in the recording's name.""" + odd = rig.tmp / "someone's \"weekly\" sync.m4a" + odd.write_bytes(b"x") + result = rig.run(args=[str(odd)], on_submit=None, TRANSCRIBE_TIMEOUT=0) + job = json.loads(next(rig.state("incoming").glob("*/job.json")).read_text()) + assert job["original_name"] == "someone's \"weekly\" sync.m4a", result.stderr + + def test_ratio_transcribe_colon_in_a_relative_filename_still_uploads(self, rig): + """Boundary: scp reads "standup 14:30.m4a" as host:path; the client hands it an absolute path.""" + (rig.tmp / "standup 14:30.m4a").write_bytes(b"x") + env = { + "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home), + "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log), + "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK, + "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5", + } + result = subprocess.run( + [str(SCRIPT), "standup 14:30.m4a"], cwd=rig.tmp, env=env, capture_output=True, text=True, timeout=60, + ) + assert result.returncode == 0, result.stderr + assert rig.uploads()[0].startswith("/") + + def test_ratio_transcribe_host_unreachable_falls_back_to_local_worker(self, rig): + """Boundary: offline, the same queue and worker run on this machine.""" + worker = rig.tmp / "fake-worker" + worker.write_text( + "#!/usr/bin/env bash\n" + f"cd \"$HOME/{STATE}\" && mkdir -p done && for d in incoming/*/; do id=$(basename \"$d\");\n" + "printf '00:00:00 Speaker A: Local fallback.\\n' > done/$id.txt; rm -rf \"$d\"; done\n" + ) + worker.chmod(0o755) + result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(worker)) + assert result.returncode == 0, result.stderr + assert result.stdout == "00:00:00 Speaker A: Local fallback.\n" + assert "local" in result.stderr.lower() + + +class TestClientStdin: + def test_ratio_transcribe_leaves_the_callers_stdin_alone(self, rig): + """Boundary: in a `while read` loop the client must not eat the loop's input.""" + env = { + "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home), + "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log), + "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK, + "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5", + } + result = subprocess.run( + ["bash", "-c", '"$0" "$1" > /dev/null 2>&1; cat', str(SCRIPT), str(rig.audio)], + env=env, input="next line of the caller's loop\n", capture_output=True, text=True, timeout=60, + ) + assert result.stdout == "next line of the caller's loop\n" + + +class TestClientError: + def test_ratio_transcribe_no_arguments_prints_usage(self, rig): + """Error: usage.""" + result = rig.run(args=[]) + assert result.returncode == 1 and "Usage: ratio-transcribe" in result.stderr and result.stdout == "" + + def test_ratio_transcribe_missing_file_fails_before_any_ssh(self, rig): + """Error: no such recording.""" + result = rig.run(args=[str(rig.tmp / "nope.m4a")]) + assert result.returncode == 1 and "not found" in result.stderr + assert not rig.ssh_log.exists() + + @pytest.mark.parametrize("env", [{"SPEAKERS": "0"}, {"SPEAKERS": "three"}, {"MIN_SPEAKERS": "-1"}, + {"MIN_SPEAKERS": "4", "MAX_SPEAKERS": "2"}, {"SPEAKERS": "3", "MAX_SPEAKERS": "5"}]) + def test_ratio_transcribe_bad_speaker_settings_are_rejected(self, rig, env): + """Error: non-positive, non-numeric or contradictory counts.""" + result = rig.run(**env) + assert result.returncode == 1 and "speaker" in result.stderr.lower() + assert not rig.ssh_log.exists() + + def test_ratio_transcribe_bad_language_is_rejected(self, rig): + """Error: the language ends up in a job file and a command line.""" + result = rig.run(args=[str(rig.audio), "en; rm -rf /"]) + assert result.returncode == 1 and "language" in result.stderr.lower() + assert not rig.ssh_log.exists() + + def test_ratio_transcribe_failed_job_reports_the_log_and_prints_nothing(self, rig): + """Error: the host's failure log comes back on stderr.""" + result = rig.run(on_submit=WORKER_FAIL) + assert result.returncode == 1 + assert result.stdout == "" + assert "stage: whisper" in result.stderr + + def test_ratio_transcribe_timeout_says_the_job_is_still_running(self, rig): + """Error: giving up waiting is not the job failing.""" + result = rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + assert result.returncode == 1 + assert result.stdout == "" + assert "again" in result.stderr.lower() + + def test_ratio_transcribe_empty_transcript_is_a_failure(self, rig): + """Error: a zero-byte result is never passed off as a transcript.""" + result = rig.run(on_submit=WORKER_EMPTY, TRANSCRIBE_TIMEOUT=1) + assert result.returncode == 1 + assert result.stdout == "" + + def test_ratio_transcribe_offline_without_local_worker_names_what_is_missing(self, rig): + """Error: no host and nothing installed locally.""" + result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(rig.tmp / "absent")) + assert result.returncode == 1 + assert "absent" in result.stderr diff --git a/working/meeting-transcription-service/tests/test_transcribe_worker.py b/working/meeting-transcription-service/tests/test_transcribe_worker.py new file mode 100644 index 0000000..bfffd31 --- /dev/null +++ b/working/meeting-transcription-service/tests/test_transcribe_worker.py @@ -0,0 +1,359 @@ +"""Tests for transcribe-worker: the queue drain on the transcription host. + +ffmpeg, whisper-cli and the diarizer are replaced by small fake executables (the +process boundary). The queue handling, the merge and the loop guard run for real. +""" + +import importlib.machinery +import importlib.util +import json +import os +import stat +import sys +from pathlib import Path + +import pytest + +SRC = Path(__file__).resolve().parent.parent / "src" +sys.path.insert(0, str(SRC)) + + +def _load_worker(): + loader = importlib.machinery.SourceFileLoader("transcribe_worker", str(SRC / "transcribe-worker")) + spec = importlib.util.spec_from_loader("transcribe_worker", loader) + assert spec is not None + module = importlib.util.module_from_spec(spec) + sys.modules["transcribe_worker"] = module # dataclasses looks the module up by name + loader.exec_module(module) + return module + + +worker = _load_worker() + +FAKE_FFMPEG = """#!/usr/bin/env bash +# last argument is the output; the one after -i is the input +while [[ $# -gt 1 ]]; do [[ "$1" == "-i" ]] && in="$2"; shift; done +cp "$in" "$1" +exit "${FAKE_FFMPEG_EXIT:-0}" +""" + +FAKE_WHISPER = """#!/usr/bin/env bash +printf '%s\\n' "$*" >> "$FAKE_LOG" +while [[ $# -gt 0 ]]; do [[ "$1" == "-of" ]] && prefix="$2"; shift; done +[[ "${FAKE_WHISPER_EXIT:-0}" == "0" ]] && cp "$FAKE_WHISPER_JSON" "$prefix.json" +exit "${FAKE_WHISPER_EXIT:-0}" +""" + +FAKE_DIARIZE = """#!/usr/bin/env bash +printf 'diarize %s\\n' "$*" >> "$FAKE_LOG" +[[ "${FAKE_DIARIZE_EXIT:-0}" == "0" ]] && cp "$FAKE_TURNS_JSON" "$2" +echo "fake diarizer says hello" >&2 +exit "${FAKE_DIARIZE_EXIT:-0}" +""" + + +def whisper_json(*words, step_ms=400): + return {"transcription": [ + {"offsets": {"from": i * step_ms, "to": i * step_ms + 300}, "text": f" {w}"} + for i, w in enumerate(words) + ]} + + +class Rig: + def __init__(self, tmp_path, monkeypatch): + self.home = tmp_path / "state" + self.bin = tmp_path / "bin" + self.bin.mkdir() + for name, body in (("ffmpeg", FAKE_FFMPEG), ("whisper-cli", FAKE_WHISPER), ("fake-diarize", FAKE_DIARIZE)): + path = self.bin / name + path.write_text(body) + path.chmod(path.stat().st_mode | stat.S_IXUSR) + self.log = tmp_path / "calls.log" + self.whisper_json = tmp_path / "whisper.json" + self.turns_json = tmp_path / "turns.json" + self.set_whisper(whisper_json("Good", "morning.")) + self.turns_json.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}])) + monkeypatch.setenv("PATH", f"{self.bin}:{os.environ['PATH']}") + monkeypatch.setenv("FAKE_LOG", str(self.log)) + monkeypatch.setenv("FAKE_WHISPER_JSON", str(self.whisper_json)) + monkeypatch.setenv("FAKE_TURNS_JSON", str(self.turns_json)) + self.monkeypatch = monkeypatch + self.config = worker.Config( + home=self.home, + whisper_model=tmp_path / "model.bin", + diarize_cmd=[str(self.bin / "fake-diarize")], + threads=2, + ) + + def set_whisper(self, data): + self.whisper_json.write_text(json.dumps(data)) + + def submit(self, job_id, job=None, audio=b"audio", where="incoming"): + folder = self.home / where / job_id + folder.mkdir(parents=True) + (folder / "audio.m4a").write_bytes(audio) + if job is not False: + (folder / "job.json").write_text(json.dumps(job if job is not None else {"language": "en"})) + return folder + + def calls(self): + return self.log.read_text().splitlines() if self.log.exists() else [] + + +@pytest.fixture +def rig(tmp_path, monkeypatch): + return Rig(tmp_path, monkeypatch) + + +class TestDrainNormal: + def test_transcribe_worker_drain_one_job_writes_transcript_to_done(self, rig): + """Normal: a job in incoming/ ends as done/<id>.txt and leaves nothing behind.""" + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n" + assert not (rig.home / "incoming" / "job1").exists() + assert not (rig.home / "work" / "job1").exists() + assert not (rig.home / "failed" / "job1.log").exists() + + def test_transcribe_worker_drain_writes_metadata_beside_the_transcript(self, rig): + """Normal: done/<id>.json records what ran, for the client and for debugging.""" + rig.submit("job1", {"language": "en", "speakers": 3, "original_name": "standup.mkv"}) + worker.drain(rig.config) + meta = json.loads((rig.home / "done" / "job1.json").read_text()) + assert meta["original_name"] == "standup.mkv" + assert meta["speakers_found"] == 1 + assert meta["lines"] == 1 + assert meta["seconds"] >= 0 + + def test_transcribe_worker_drain_passes_language_and_speaker_count_through(self, rig): + """Normal: job options reach whisper and the diarizer.""" + rig.submit("job1", {"language": "es", "speakers": 3}) + worker.drain(rig.config) + whisper_call = next(c for c in rig.calls() if not c.startswith("diarize")) + diarize_call = next(c for c in rig.calls() if c.startswith("diarize")) + assert "-l es" in whisper_call + assert "-mc 0" in whisper_call and "-ojf" in whisper_call + assert diarize_call.endswith("--speakers 3") + + def test_transcribe_worker_drain_min_and_max_speakers_are_forwarded(self, rig): + """Normal: a range instead of an exact count.""" + rig.submit("job1", {"min_speakers": 2, "max_speakers": 5}) + worker.drain(rig.config) + diarize_call = next(c for c in rig.calls() if c.startswith("diarize")) + assert "--min-speakers 2" in diarize_call and "--max-speakers 5" in diarize_call + + def test_transcribe_worker_drain_processes_every_job_oldest_first(self, rig): + """Normal: the queue drains completely, in arrival order.""" + first = rig.submit("b-first") + rig.submit("a-second") + os.utime(first, (1, 1)) + assert worker.drain(rig.config) == {"done": 2, "failed": 0} + diarize_calls = [c for c in rig.calls() if c.startswith("diarize")] + assert "b-first" in diarize_calls[0] and "a-second" in diarize_calls[1] + + +class TestDrainBoundary: + def test_transcribe_worker_drain_empty_queue_is_a_no_op(self, rig): + """Boundary: nothing to do, and the folders get created.""" + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "incoming").is_dir() and (rig.home / "done").is_dir() + + def test_transcribe_worker_drain_ignores_uploads_still_in_flight(self, rig): + """Boundary: a dot-prefixed folder is left alone (second line of defence behind uploading/).""" + rig.submit(".tmp-job9") + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "incoming" / ".tmp-job9").exists() + + def test_transcribe_worker_drain_job_already_done_is_not_rerun(self, rig): + """Boundary: resubmitting finished work costs nothing.""" + (rig.home / "done").mkdir(parents=True) + (rig.home / "done" / "job1.txt").write_text("already here\n") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "already here\n" + assert not (rig.home / "incoming" / "job1").exists() + assert rig.calls() == [] + + def test_transcribe_worker_drain_missing_job_file_uses_defaults(self, rig): + """Boundary: audio with no job.json still transcribes, in English, count estimated.""" + rig.submit("job1", job=False) + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + + def test_transcribe_worker_drain_retries_a_job_left_in_work_by_a_crash(self, rig): + """Boundary: a job stranded in work/ is picked up again.""" + rig.submit("job1", where="work") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + + def test_transcribe_worker_drain_gives_up_on_a_job_that_keeps_crashing(self, rig): + """Boundary: the second stranding is a failure, not a loop.""" + rig.submit("job1", {"language": "en", "attempts": 2}, where="work") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "attempt" in (rig.home / "failed" / "job1.log").read_text() + + +class TestDrainError: + def test_transcribe_worker_drain_whisper_failure_is_logged_with_its_stage(self, rig): + """Error: whisper exits non-zero.""" + rig.monkeypatch.setenv("FAKE_WHISPER_EXIT", "3") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + log = (rig.home / "failed" / "job1.log").read_text() + assert "whisper" in log + assert not (rig.home / "done" / "job1.txt").exists() + assert not (rig.home / "work" / "job1").exists() + + def test_transcribe_worker_drain_diarizer_failure_keeps_its_stderr(self, rig): + """Error: the diarizer fails; its own words land in the log.""" + rig.monkeypatch.setenv("FAKE_DIARIZE_EXIT", "1") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + log = (rig.home / "failed" / "job1.log").read_text() + assert "diarize" in log and "fake diarizer says hello" in log + + def test_transcribe_worker_drain_looping_transcription_fails(self, rig): + """Error: whisper's repetition loop is a failed job, never a transcript.""" + rig.set_whisper(whisper_json(*("we do not know where we are going".split() * 5), step_ms=1000)) # 40 s + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "looped" in (rig.home / "failed" / "job1.log").read_text() + assert not (rig.home / "done" / "job1.txt").exists() + + def test_transcribe_worker_drain_short_loop_is_collapsed_and_recorded(self, rig): + """Normal: a stutter of a few seconds is collapsed to one copy and noted in the metadata.""" + rig.set_whisper(whisper_json("Okay,", *("fair, it's not going to be".split() * 4), "easy.")) + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Okay, fair, it's not going to be easy.\n" + meta = json.loads((rig.home / "done" / "job1.json").read_text()) + assert meta["loops_collapsed"] == [["fair, it's not going to be", 4]] + + def test_transcribe_worker_drain_malformed_job_file_fails_that_job_only(self, rig): + """Error: one bad job does not stop the queue.""" + bad = rig.submit("bad") + (bad / "job.json").write_text("{not json") + os.utime(bad, (1, 1)) + rig.submit("good") + assert worker.drain(rig.config) == {"done": 1, "failed": 1} + assert (rig.home / "done" / "good.txt").exists() + assert "job.json" in (rig.home / "failed" / "bad.log").read_text() + + def test_transcribe_worker_drain_job_without_audio_fails(self, rig): + """Error: a job folder holding no audio file.""" + folder = rig.submit("job1") + (folder / "audio.m4a").unlink() + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "audio" in (rig.home / "failed" / "job1.log").read_text() + + @pytest.mark.parametrize("bad", [{"speakers": 0}, {"speakers": "three"}, {"language": "en; rm -rf"}]) + def test_transcribe_worker_drain_rejects_bad_option_values(self, rig, bad): + """Error: options are validated before they reach a command line.""" + rig.submit("job1", bad) + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert rig.calls() == [] + + +class TestDrainSilenceHallucinations: + def _silent_stretch(self, rig): + spoken = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}] + invented = [ + {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."} + for i in range(12) + ] + rig.set_whisper({"transcription": spoken + invented}) + rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])) + + def test_transcribe_worker_drain_words_invented_in_silence_are_dropped_not_failed(self, rig): + """Normal: a quiet meeting transcribes; the invented run never reaches the transcript.""" + self._silent_stretch(rig) + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n" + + def test_transcribe_worker_drain_metadata_counts_the_dropped_words(self, rig): + """Normal: the count is on record, so a transcript that lost a lot is visible.""" + self._silent_stretch(rig) + rig.submit("job1") + worker.drain(rig.config) + assert json.loads((rig.home / "done" / "job1.json").read_text())["dropped_outside_speech"] == 12 + + def test_transcribe_worker_drain_recording_with_no_speech_at_all_fails_cleanly(self, rig): + """Error: everything whisper produced sits in silence.""" + rig.set_whisper({"transcription": [{"offsets": {"from": 60_000, "to": 61_000}, "text": " Thank you."}]}) + rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])) + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "no speech" in (rig.home / "failed" / "job1.log").read_text() + + +class TestRunStdin: + def test_transcribe_worker_run_gives_tools_no_stdin(self, tmp_path): + """Boundary: ffmpeg reads stdin when it can, which would eat a calling loop's input.""" + out = tmp_path / "stdin-target" + read_end, write_end = os.pipe() + saved = os.dup(0) + os.dup2(read_end, 0) + try: + worker._run("probe", ["bash", "-c", f"readlink /proc/self/fd/0 > {out}"]) + finally: + os.dup2(saved, 0) + for fd in (saved, read_end, write_end): + os.close(fd) + assert out.read_text().strip() == "/dev/null" + + +class TestDrainResilience: + def test_transcribe_worker_drain_null_attempts_counts_as_a_first_try(self, rig): + """Error: a job.json with "attempts": null must not take the worker down; it is a first try.""" + odd = rig.submit("odd", {"language": "en", "attempts": None}) + os.utime(odd, (1, 1)) + rig.submit("good") + assert worker.drain(rig.config) == {"done": 2, "failed": 0} + assert (rig.home / "done" / "odd.txt").exists() + assert (rig.home / "done" / "good.txt").exists() + + def test_transcribe_worker_drain_unexpected_error_in_one_job_is_logged_and_the_queue_goes_on(self, rig): + """Error: an exception the pipeline never anticipated lands in failed/ with its type, not on the run.""" + first = rig.submit("first") + os.utime(first, (1, 1)) + rig.submit("second") + + def explode(job: dict) -> list[str]: + raise RuntimeError("boom") + + rig.monkeypatch.setattr(worker, "_diarize_options", explode) + assert worker.drain(rig.config) == {"done": 0, "failed": 2} # both reached, neither crashed the run + for job_id in ("first", "second"): + log = (rig.home / "failed" / f"{job_id}.log").read_text() + assert "RuntimeError" in log and "boom" in log + assert not (rig.home / "work" / job_id).exists() + + def test_transcribe_worker_drain_unwritable_done_dir_is_a_failed_job_not_a_crash(self, rig): + """Error: an OSError while writing the transcript lands in failed/, and the run survives.""" + if os.geteuid() == 0: + pytest.skip("root ignores directory permissions") + (rig.home / "done").mkdir(parents=True) + (rig.home / "done").chmod(0o500) + try: + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "job1" in "".join(p.name for p in (rig.home / "failed").iterdir()) + finally: + (rig.home / "done").chmod(0o700) + + def test_transcribe_worker_drain_ffmpeg_failure_is_logged_with_its_stage(self, rig): + """Error: ffmpeg exits non-zero.""" + rig.monkeypatch.setenv("FAKE_FFMPEG_EXIT", "2") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "ffmpeg" in (rig.home / "failed" / "job1.log").read_text() + + def test_transcribe_worker_drain_returns_at_once_when_another_worker_holds_the_lock(self, rig): + """Boundary: a second drain does not touch the queue while the first holds the lock.""" + import fcntl + rig.home.mkdir(parents=True, exist_ok=True) + rig.submit("job1") + with open(rig.home / "lock", "w", encoding="utf-8") as held: + fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB) + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "incoming" / "job1").exists() + assert rig.calls() == [] |