diff options
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() == [] |