diff options
Diffstat (limited to 'working/meeting-transcription-service/src')
4 files changed, 901 insertions, 0 deletions
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:])) |