diff --git a/.importlinter b/.importlinter index 9a4fb79b..d63a189d 100644 --- a/.importlinter +++ b/.importlinter @@ -28,6 +28,8 @@ source_modules = aai_cli.streaming aai_cli.telemetry aai_cli.theme + aai_cli.transcribe_batch + aai_cli.transcribe_exec aai_cli.transcribe_render aai_cli.ws aai_cli.youtube diff --git a/README.md b/README.md index b3269d06..41bd2191 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ Your key is written to a git-ignored `.env` (never sent to the browser). Use `-- | --- | --- | | `assembly login` / `logout` / `whoami` | Manage the stored API key. | | `assembly doctor` | Check your environment (API key, network, ffmpeg, microphone, agent tooling). | -| `assembly transcribe ` | Transcribe a file, URL, or YouTube/podcast page URL (`--sample`, `--llm`, `--show-code`). | +| `assembly transcribe ` | Transcribe a file, URL, or YouTube/podcast page URL (`--sample`, `--llm`, `--show-code`) — or a directory/glob/stdin list as a resumable batch. | | `assembly transcripts list` / `get ` | Browse and fetch past transcripts. | | `assembly stream [file]` | Real-time transcription from a file or the microphone. | | `assembly agent` | *Run* a live two-way voice conversation (to **build** a voice agent app, use `assembly init voice-agent`). | @@ -225,6 +225,19 @@ done A Ctrl-C in a pipe hits both sides; to stop just the producer and let the consumer finish, signal the producer (`timeout -s INT 30s assembly stream …`) or end on a natural pause (`assembly stream --inactivity-timeout 5`). +## Batch transcription + +Point `assembly transcribe` at a directory or glob — or pipe a list of paths/URLs with `--from-stdin` — and it transcribes everything concurrently behind a live progress table: + +```sh +assembly transcribe ./recordings # every audio file under the directory +assembly transcribe "calls/*.mp3" # glob (quote it so your shell doesn't expand it) +find . -name "*.wav" | assembly transcribe --from-stdin # composes with find/ls/yt-dlp +assembly transcribe ./recordings --concurrency 8 # default is 4 at a time +``` + +Each source gets a `.aai.json` sidecar with the full transcript payload. The sidecar is also the resume marker: a re-run skips any source whose sidecar records a completed transcription of the same bytes (hash-checked for local files), so retrying a partly-failed batch only pays for what's missing. `--force` re-transcribes everything. Under `--json`, batch mode emits one NDJSON record per source as it finishes (`{source, status, id, sidecar}`, or `{source, status, error}` on failure); the exit code is `1` if any source failed. + ## API Key & Security `assembly` resolves your key in order: the `ASSEMBLYAI_API_KEY` environment variable, then the OS keyring (written only by `assembly login`). Two things worth knowing: diff --git a/aai_cli/commands/transcribe.py b/aai_cli/commands/transcribe.py index 43501f38..c1832c8b 100644 --- a/aai_cli/commands/transcribe.py +++ b/aai_cli/commands/transcribe.py @@ -15,6 +15,7 @@ llm, options, output, + transcribe_batch, transcribe_exec, ) @@ -32,6 +33,8 @@ epilog=examples_epilog( [ ("Transcribe a local file", "assembly transcribe call.mp3"), + ("Batch-transcribe a folder", "assembly transcribe ./recordings"), + ("Batch-transcribe a glob", 'assembly transcribe "calls/*.mp3"'), ("Try it with the hosted sample", "assembly transcribe --sample"), ("Transcribe a YouTube video", "assembly transcribe https://youtu.be/dtp6b76pMak"), ("Transcribe a podcast page", 'assembly transcribe "https://podcasts.apple.com/…"'), @@ -44,8 +47,14 @@ ) def transcribe( ctx: typer.Context, - source: str | None = typer.Argument(None, help="Audio file, URL, or YouTube/podcast URL."), + source: str | None = typer.Argument( + None, help="Audio file, URL, YouTube/podcast URL, or a directory/glob (batch mode)." + ), sample: bool = typer.Option(False, "--sample", help="Use the hosted wildfires.mp3 sample."), + # batch mode + from_stdin: bool = options.batch_from_stdin_option(), + concurrency: int = options.batch_concurrency_option(), + force: bool = options.batch_force_option(), # model & language speech_model: aai.SpeechModel | None = typer.Option( None, @@ -334,13 +343,17 @@ def transcribe( help="Print the equivalent Python SDK code and exit (does not transcribe).", ), ) -> None: - """Transcribe an audio file, URL, or YouTube/podcast link. + """Transcribe an audio file, URL, or YouTube/podcast link — or a whole batch. Quickest start: assembly transcribe call.mp3 (or --sample for the hosted demo). Save with --out FILE, or pipe one field with -o text. YouTube and podcast-page URLs (any page yt-dlp can extract) are downloaded first, then transcribed. + Batch mode: pass a directory or glob (or pipe a list with --from-stdin) to + transcribe many sources concurrently. Each source gets a .aai.json sidecar + with the full result, and a re-run skips sources already transcribed. + Curated flags cover common features; --config KEY=VALUE and --config-file reach every other field. Analysis (summary, chapters, ...) renders in human mode. """ @@ -403,6 +416,22 @@ def body(state: AppState, json_mode: bool) -> None: transcribe_exec.validate_speakers_expected(merged) + sources = transcribe_batch.expand_sources(source, from_stdin=from_stdin, sample=sample) + if sources is not None: + transcribe_batch.reject_single_source_flags( + out=out, output_field=output_field, llm_prompt=llm_prompt, show_code=show_code + ) + transcribe_batch.run_batch( + config.resolve_api_key(profile=state.profile), + sources, + transcription_config=config_builder.construct_transcription_config(merged), + concurrency=concurrency, + force=force, + json_mode=json_mode, + quiet=state.quiet, + ) + return + if show_code: # Print-only: build the equivalent script and exit without transcribing or # authenticating (raw stdout, so `--show-code > script.py` runs). No diff --git a/aai_cli/help_panels.py b/aai_cli/help_panels.py index c9a746e9..6eee4499 100644 --- a/aai_cli/help_panels.py +++ b/aai_cli/help_panels.py @@ -35,6 +35,7 @@ OPT_TRANSLATION = "Translation" OPT_ADVANCED = "Advanced" OPT_LLM = "LLM Transform" +OPT_BATCH = "Batch" # many-source mode: --from-stdin, --concurrency, --force # stream-specific panels (real-time concerns that file transcription has no equivalent for) OPT_CAPTURE = "Audio Capture" OPT_TURNS = "Turn Detection" diff --git a/aai_cli/options.py b/aai_cli/options.py index af5d0f92..e1c16bfc 100644 --- a/aai_cli/options.py +++ b/aai_cli/options.py @@ -8,8 +8,51 @@ import typer +from aai_cli import help_panels + +DEFAULT_BATCH_CONCURRENCY = 4 + def json_option(help_text: str = "Output raw JSON.") -> bool: """The standard ``--json``/``-j`` flag; pass ``help_text`` where the shape differs.""" flag: bool = typer.Option(False, "--json", "-j", help=help_text) return flag + + +# Batch-mode flags for `transcribe` (see transcribe_batch.py). Defined here because +# this module owns the FBT003 carve-out for Typer's boolean positional defaults. + + +def batch_from_stdin_option() -> bool: + """The ``--from-stdin`` flag: batch mode fed one path/URL per stdin line.""" + flag: bool = typer.Option( + False, + "--from-stdin", + help="Batch mode: read audio paths/URLs from stdin, one per line " + "(composes with find/ls/yt-dlp output).", + rich_help_panel=help_panels.OPT_BATCH, + ) + return flag + + +def batch_concurrency_option() -> int: + """The ``--concurrency`` option: how many sources transcribe at once in batch mode.""" + value: int = typer.Option( + DEFAULT_BATCH_CONCURRENCY, + "--concurrency", + min=1, + help="How many sources to transcribe at once in batch mode.", + rich_help_panel=help_panels.OPT_BATCH, + ) + return value + + +def batch_force_option() -> bool: + """The ``--force`` flag: re-transcribe even when a completed sidecar exists.""" + flag: bool = typer.Option( + False, + "--force", + help="Batch mode: re-transcribe sources whose sidecar already records a completed run.", + rich_help_panel=help_panels.OPT_BATCH, + ) + return flag diff --git a/aai_cli/transcribe_batch.py b/aai_cli/transcribe_batch.py new file mode 100644 index 00000000..358f4085 --- /dev/null +++ b/aai_cli/transcribe_batch.py @@ -0,0 +1,348 @@ +"""Batch transcription: directories, globs, and stdin lists with sidecar resume. + +``assembly transcribe`` switches to batch mode when the source is a directory or a +glob pattern, or when ``--from-stdin`` supplies one path/URL per line. Sources run +concurrently behind a live progress table; each finished source gets a +``.aai.json`` sidecar holding the full transcript. The sidecar doubles as +the resume marker — a re-run skips any source whose sidecar records a completed +transcription of the same bytes — so retrying a partly-failed batch only pays for +what's missing (``--force`` re-transcribes everything). +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import json +import re +from collections.abc import Generator +from concurrent.futures import ThreadPoolExecutor, as_completed +from contextlib import contextmanager +from pathlib import Path +from typing import TYPE_CHECKING + +from rich.live import Live +from rich.markup import escape + +from aai_cli import client, jsonshape, output, stdio, theme, transcribe_exec +from aai_cli.errors import CLIError, NotAuthenticated, UsageError + +if TYPE_CHECKING: + import assemblyai as aai + from rich.table import Table + +SIDECAR_SUFFIX = ".aai.json" + +# What a directory scan picks up (an explicit glob or stdin list is taken as-is). +AUDIO_EXTENSIONS = frozenset( + { + ".3gp", + ".aac", + ".aif", + ".aiff", + ".amr", + ".flac", + ".m4a", + ".m4b", + ".mka", + ".mkv", + ".mov", + ".mp2", + ".mp3", + ".mp4", + ".mpga", + ".oga", + ".ogg", + ".opus", + ".wav", + ".webm", + ".wma", + } +) + +_URL_PREFIXES = ("http://", "https://") +_GLOB_CHARS = frozenset("*?[") + + +def expand_sources(source: str | None, *, from_stdin: bool, sample: bool) -> list[str] | None: + """The batch source list, or ``None`` when this is a single-source invocation. + + Batch mode triggers on ``--from-stdin``, a directory (scanned recursively for + audio files), or a glob pattern that names no existing file. A plain file, URL, + ``-`` (audio piped on stdin), or ``--sample`` stays on the single-source path. + """ + if from_stdin: + return _stdin_sources(source, sample=sample) + if source is None or sample or source == "-" or source.startswith(_URL_PREFIXES): + return None + path = Path(source) + if path.is_dir(): + return _directory_sources(path) + if not path.exists() and _GLOB_CHARS.intersection(source): + return _glob_sources(source) + return None + + +def _stdin_sources(source: str | None, *, sample: bool) -> list[str]: + if source is not None or sample: + raise UsageError( + "--from-stdin reads sources from stdin; don't also pass a source or --sample." + ) + lines = list(dict.fromkeys(stdio.iter_piped_stdin_lines())) # dedupe, keep order + if not lines: + raise UsageError( + "No sources received on stdin.", + suggestion="Pipe one path or URL per line, e.g. " + "find . -name '*.mp3' | assembly transcribe --from-stdin.", + ) + return lines + + +def _directory_sources(path: Path) -> list[str]: + files = sorted( + str(p) for p in path.rglob("*") if p.is_file() and p.suffix.lower() in AUDIO_EXTENSIONS + ) + if not files: + raise UsageError( + f"No audio files found under {path}.", + suggestion="Recognized extensions: " + ", ".join(sorted(AUDIO_EXTENSIONS)) + ".", + ) + return files + + +def _glob_sources(pattern: str) -> list[str]: + # pathlib globs are always relative, so peel an absolute pattern's anchor off + # and glob from there ("" anchors at the working directory; Path("") is "."). + anchor = Path(pattern).anchor + matches = sorted( + str(p) + for p in Path(anchor).glob(pattern.removeprefix(anchor)) + if p.is_file() and not str(p).endswith(SIDECAR_SUFFIX) + ) + if not matches: + raise UsageError(f"No files match {pattern}.") + return matches + + +def reject_single_source_flags( + *, + out: Path | None, + output_field: object | None, + llm_prompt: list[str] | None, + show_code: bool, +) -> None: + """Batch mode writes one sidecar per source; the single-result flags don't apply.""" + if show_code: + raise UsageError( + "--show-code generates code for a single source, not a batch.", + suggestion="Pass one file or URL with --show-code.", + ) + if out is not None or output_field is not None or llm_prompt: + raise UsageError( + "--out, -o/--output, and --llm apply to a single source, not a batch.", + suggestion=f"Each source gets a '{SIDECAR_SUFFIX}' sidecar with the full result.", + ) + + +def sidecar_path(source: str) -> Path: + """Where ``source``'s sidecar lives: ``.aai.json`` next to a local file, or + a slug + URL-hash name in the working directory for a URL.""" + if source.startswith(_URL_PREFIXES): + digest = hashlib.sha256(source.encode()).hexdigest()[:8] + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", source.partition("://")[2]).strip("-.")[:64] + return Path(f"{slug}-{digest}{SIDECAR_SUFFIX}") + return Path(source + SIDECAR_SUFFIX) + + +def _source_digest(source: str) -> str | None: + """SHA-256 of a local file's bytes; ``None`` for URLs (and paths that aren't files).""" + if source.startswith(_URL_PREFIXES) or not Path(source).is_file(): + return None + with Path(source).open("rb") as f: + return hashlib.file_digest(f, "sha256").hexdigest() + + +def resumable_record(sidecar: Path, *, digest: str | None) -> dict[str, object] | None: + """The sidecar's record when it marks a completed transcription of the same bytes. + + ``None`` (transcribe again) when the sidecar is missing or corrupt, the run + didn't complete, or a local file's hash no longer matches the recorded one. + """ + try: + record = jsonshape.as_mapping(json.loads(sidecar.read_text())) + except (OSError, ValueError): + return None + if record is None or record.get("status") != "completed": + return None + if digest is not None and record.get("source_sha256") != digest: + return None + return record + + +def _write_sidecar( + sidecar: Path, *, source: str, transcript: aai.Transcript, digest: str | None +) -> None: + record: dict[str, object] = { + "source": source, + "id": transcript.id, + "status": client.status_str(transcript), + "transcript": client.transcript_json_payload(transcript), + } + if digest is not None: + record["source_sha256"] = digest + sidecar.write_text(json.dumps(record, indent=2, default=str) + "\n") + + +@dataclasses.dataclass +class _Item: + source: str + status: str = "queued" # queued → processing → completed | skipped | failed + transcript_id: str = "" + detail: str = "" # sidecar path when completed/skipped; the error message when failed + + def record(self) -> dict[str, str]: + """The NDJSON record emitted for this source under ``--json``.""" + rec = {"source": self.source, "status": self.status} + if self.transcript_id: + rec["id"] = self.transcript_id + if self.status == "failed": + rec["error"] = self.detail + elif self.detail: + rec["sidecar"] = self.detail + return rec + + +def _transcribe_one( + api_key: str, item: _Item, *, transcription_config: aai.TranscriptionConfig, force: bool +) -> None: + """Worker body: resume from the sidecar, or transcribe and write one. + + A per-source failure is recorded on the item and the batch carries on — except + NotAuthenticated, which re-raises so ``_drain`` aborts the batch (one rejected + key fails every source identically, and auto-login should trigger once). + """ + try: + sidecar = sidecar_path(item.source) + digest = _source_digest(item.source) + if not force and (record := resumable_record(sidecar, digest=digest)) is not None: + item.transcript_id = str(record.get("id") or "") + item.status, item.detail = "skipped", str(sidecar) + return + item.status = "processing" + transcript = transcribe_exec.run_transcription( + api_key, item.source, sample=False, transcription_config=transcription_config + ) + _write_sidecar(sidecar, source=item.source, transcript=transcript, digest=digest) + item.transcript_id = transcript.id or "" + item.status, item.detail = "completed", str(sidecar) + except CLIError as err: + item.status, item.detail = "failed", err.message + if isinstance(err, NotAuthenticated): + raise + + +def _render_table(items: list[_Item]) -> Table: + table = output.data_table("Source", "Status", "Transcript", "Result") + for item in items: + table.add_row( + escape(item.source), + theme.status_text(item.status), + item.transcript_id, + escape(item.detail), + ) + return table + + +@contextmanager +def _progress_table(items: list[_Item], *, json_mode: bool) -> Generator[None]: + """Render the batch as a live-updating table (human mode). + + Rich renders nothing while running on a non-interactive console and prints the + final frame once on stop, so piped/agent runs still get the result table. JSON + mode skips Rich entirely — NDJSON per source is the output. + """ + if json_mode: + yield + return + with Live( + get_renderable=lambda: _render_table(items), + console=output.console, + refresh_per_second=4, # pragma: no mutate (cosmetic refresh cadence) + ): + yield + + +def _drain( + api_key: str, + items: list[_Item], + *, + transcription_config: aai.TranscriptionConfig, + concurrency: int, + force: bool, + json_mode: bool, +) -> None: + """Run the workers, emitting one NDJSON record per finished source under ``--json``. + + The first exception that escapes a worker (NotAuthenticated, or a bug) drops the + not-yet-started sources and re-raises once the in-flight ones drain. + """ + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = { + pool.submit( + _transcribe_one, + api_key, + item, + transcription_config=transcription_config, + force=force, + ): item + for item in items + } + for future in as_completed(futures): + if (exc := future.exception()) is not None: + pool.shutdown(cancel_futures=True) + raise exc + if json_mode: + output.emit_ndjson(futures[future].record()) + + +def _summarize(items: list[_Item], *, json_mode: bool, quiet: bool) -> None: + failed = sum(1 for item in items if item.status == "failed") + if failed: + raise CLIError( + f"{failed} of {len(items)} sources failed.", + error_type="batch_failed", + suggestion="Re-run the same command to retry only the failures — " + "completed sources are skipped via their sidecars.", + ) + completed = sum(1 for item in items if item.status == "completed") + skipped = len(items) - completed + if not json_mode and not quiet: + output.error_console.print(output.success(f"Transcribed {completed}, skipped {skipped}.")) + + +def run_batch( + api_key: str, + sources: list[str], + *, + transcription_config: aai.TranscriptionConfig, + concurrency: int, + force: bool, + json_mode: bool, + quiet: bool, +) -> None: + """Transcribe ``sources`` concurrently, writing one sidecar per source. + + Raises CLIError (exit 1) when any source failed so scripts can trust the exit + code; a re-run resumes from the sidecars and retries only the failures. + """ + items = [_Item(source) for source in sources] + with _progress_table(items, json_mode=json_mode): + _drain( + api_key, + items, + transcription_config=transcription_config, + concurrency=concurrency, + force=force, + json_mode=json_mode, + ) + _summarize(items, json_mode=json_mode, quiet=quiet) diff --git a/tests/__snapshots__/test_cli_output_snapshots.ambr b/tests/__snapshots__/test_cli_output_snapshots.ambr index 63bd59b8..60dd1e57 100644 --- a/tests/__snapshots__/test_cli_output_snapshots.ambr +++ b/tests/__snapshots__/test_cli_output_snapshots.ambr @@ -933,7 +933,7 @@ Usage: assembly transcribe [OPTIONS] [SOURCE] - Transcribe an audio file, URL, or YouTube/podcast link. + Transcribe an audio file, URL, or YouTube/podcast link — or a whole batch. Quickest start: assembly transcribe call.mp3 (or --sample for the hosted demo). @@ -941,12 +941,17 @@ Save with --out FILE, or pipe one field with -o text. YouTube and podcast-page URLs (any page yt-dlp can extract) are downloaded first, then transcribed. + Batch mode: pass a directory or glob (or pipe a list with --from-stdin) to + transcribe many sources concurrently. Each source gets a .aai.json sidecar + with the full result, and a re-run skips sources already transcribed. + Curated flags cover common features; --config KEY=VALUE and --config-file reach every other field. Analysis (summary, chapters, ...) renders in human mode. ╭─ Arguments ──────────────────────────────────────────────────────────────────╮ - │ source [SOURCE] Audio file, URL, or YouTube/podcast URL. │ + │ source [SOURCE] Audio file, URL, YouTube/podcast URL, or a │ + │ directory/glob (batch mode). │ ╰──────────────────────────────────────────────────────────────────────────────╯ ╭─ Options ────────────────────────────────────────────────────────────────────╮ │ --sample Use the hosted │ @@ -968,6 +973,18 @@ │ transcribe). │ │ --help Show this message and exit. │ ╰──────────────────────────────────────────────────────────────────────────────╯ + ╭─ Batch ──────────────────────────────────────────────────────────────────────╮ + │ --from-stdin Batch mode: read audio paths/URLs │ + │ from stdin, one per line │ + │ (composes with find/ls/yt-dlp │ + │ output). │ + │ --concurrency INTEGER RANGE [x>=1] How many sources to transcribe at │ + │ once in batch mode. │ + │ [default: 4] │ + │ --force Batch mode: re-transcribe sources │ + │ whose sidecar already records a │ + │ completed run. │ + ╰──────────────────────────────────────────────────────────────────────────────╯ ╭─ Model & Language ───────────────────────────────────────────────────────────╮ │ --speech-model [best|nano|slam-1|univ Speech model. │ │ ersal] │ @@ -1069,6 +1086,10 @@ Examples Transcribe a local file $ assembly transcribe call.mp3 + Batch-transcribe a folder + $ assembly transcribe ./recordings + Batch-transcribe a glob + $ assembly transcribe "calls/*.mp3" Try it with the hosted sample $ assembly transcribe --sample Transcribe a YouTube video diff --git a/tests/test_source_validation.py b/tests/test_source_validation.py index 294a9ac9..aac8277e 100644 --- a/tests/test_source_validation.py +++ b/tests/test_source_validation.py @@ -67,15 +67,15 @@ def test_resolve_audio_source_rejects_directory(tmp_path): def test_transcribe_directory_source_fails_before_credentials(mocker, tmp_path): - # No key configured: a directory must read as "not a file", never trigger a login - # (or an upload attempt). + # No key configured: a directory is batch mode, and an empty one must read as + # "no audio files", never trigger a login (or an upload attempt). tx = mocker.patch("aai_cli.commands.transcribe.client.transcribe", autospec=True) result = runner.invoke(app, ["transcribe", str(tmp_path)]) assert result.exit_code == 2 # Rich may wrap the long tmp path mid-token (even inside a word), so compare with # all whitespace removed — matching the pattern in test_share.py. packed = "".join(result.output.split()) - assert "".join(f"Not a file: {tmp_path}".split()) in packed + assert "".join(f"No audio files found under {tmp_path}".split()) in packed assert "starting browser login" not in result.output tx.assert_not_called() diff --git a/tests/test_transcribe_batch.py b/tests/test_transcribe_batch.py new file mode 100644 index 00000000..69705071 --- /dev/null +++ b/tests/test_transcribe_batch.py @@ -0,0 +1,387 @@ +"""Batch-mode runs for `assembly transcribe`: sidecar resume, concurrency, failures, +and the per-source NDJSON / progress-table output. + +Source selection (glob/directory/stdin expansion, rejected flags) lives in +test_transcribe_batch_sources.py. +""" + +import hashlib +import json + +import pytest +from typer.testing import CliRunner + +from aai_cli import config, transcribe_batch +from aai_cli.errors import auth_failure +from aai_cli.main import app + +runner = CliRunner() + +_TRANSCRIBE = "aai_cli.commands.transcribe.client.transcribe" + + +@pytest.fixture(autouse=True) +def workdir(tmp_path, monkeypatch): + # Batch sources and sidecars are resolved relative to the working directory; + # isolate each test in its own tmp cwd. + monkeypatch.chdir(tmp_path) + + +def _auth(): + config.set_api_key("default", "sk_live") + + +def _fake_transcript(mocker, source="x"): + t = mocker.MagicMock() + t.id = f"t_{source}" + t.text = f"text of {source}" + t.status = "completed" + t.json_response = {"id": t.id, "text": t.text, "status": "completed"} + return t + + +def _patch_transcribe(mocker, monkeypatch): + """Patch client.transcribe with a fake that records the audio args it saw.""" + seen = [] + + def fake(api_key, audio, *, config): + seen.append(audio) + return _fake_transcript(mocker, audio) + + monkeypatch.setattr(_TRANSCRIBE, fake) + return seen + + +def _ndjson(result): + return [json.loads(line) for line in result.output.splitlines() if line.startswith("{")] + + +def test_sidecar_file_is_two_space_indented_with_trailing_newline(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + _patch_transcribe(mocker, monkeypatch) + runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + text = (tmp_path / "a.mp3.aai.json").read_text() + assert text.startswith('{\n "source"') # indent=2 + assert text.endswith("}\n") + + +# --- sidecar resume ------------------------------------------------------------- + + +def _completed_sidecar(tmp_path, name, data, transcript_id="t_old"): + record = { + "source": name, + "id": transcript_id, + "status": "completed", + "transcript": {"id": transcript_id}, + "source_sha256": hashlib.sha256(data).hexdigest(), + } + (tmp_path / f"{name}.aai.json").write_text(json.dumps(record)) + + +def test_rerun_skips_sources_with_completed_sidecars(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "b.mp3").write_bytes(b"bbb") + _completed_sidecar(tmp_path, "a.mp3", b"aaa") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert seen == ["b.mp3"] # only the unfinished source pays + records = {r["source"]: r for r in _ndjson(result)} + assert records["a.mp3"] == { + "source": "a.mp3", + "status": "skipped", + "id": "t_old", + "sidecar": "a.mp3.aai.json", + } + + +def test_changed_file_bytes_invalidate_the_sidecar(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"new-bytes") + _completed_sidecar(tmp_path, "a.mp3", b"old-bytes") # hash no longer matches + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert seen == ["a.mp3"] + + +def test_force_retranscribes_despite_completed_sidecar(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + _completed_sidecar(tmp_path, "a.mp3", b"aaa") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--force", "--json"]) + assert result.exit_code == 0 + assert seen == ["a.mp3"] + + +def test_corrupt_sidecar_retranscribes(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "a.mp3.aai.json").write_text("not json{") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert seen == ["a.mp3"] + + +def test_non_completed_sidecar_retranscribes(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "a.mp3.aai.json").write_text(json.dumps({"status": "error"})) + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert seen == ["a.mp3"] + + +def test_non_dict_sidecar_retranscribes(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "a.mp3.aai.json").write_text(json.dumps(["completed"])) + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert seen == ["a.mp3"] + + +def test_url_source_resumes_on_sidecar_alone(tmp_path, mocker, monkeypatch): + # URLs have no local bytes to hash: a completed sidecar is the whole resume check. + _auth() + url = "https://example.com/ep.mp3" + sidecar = transcribe_batch.sidecar_path(url) + sidecar.write_text(json.dumps({"status": "completed", "id": "t_old"})) + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "--from-stdin", "--json"], input=url + "\n") + assert result.exit_code == 0 + assert seen == [] + assert _ndjson(result) == [ + {"source": url, "status": "skipped", "id": "t_old", "sidecar": str(sidecar)} + ] + + +def test_url_sidecar_omits_source_hash(tmp_path, mocker, monkeypatch): + _auth() + url = "https://example.com/ep.mp3" + _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "--from-stdin", "--json"], input=url + "\n") + assert result.exit_code == 0 + record = json.loads(transcribe_batch.sidecar_path(url).read_text()) + assert record["source"] == url + assert "source_sha256" not in record + + +def test_url_sidecar_name_is_a_slug_plus_url_hash(): + url = "https://example.com/shows/ep.mp3?token=a/b" + digest = hashlib.sha256(url.encode()).hexdigest()[:8] + assert ( + str(transcribe_batch.sidecar_path(url)) + == f"example.com-shows-ep.mp3-token-a-b-{digest}.aai.json" + ) + + +def test_url_sidecar_slug_truncates_to_64_chars(): + url = "https://example.com/" + "x" * 100 + name = transcribe_batch.sidecar_path(url).name + slug = name.removesuffix(".aai.json").rsplit("-", 1)[0] + assert slug == ("example.com/" + "x" * 100).replace("/", "-")[:64] + assert len(slug) == 64 + + +# --- failures, exit codes, auth -------------------------------------------------- + + +def test_partial_failure_exits_1_and_completes_the_rest(tmp_path, mocker, monkeypatch): + from aai_cli.errors import APIError + + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + (tmp_path / "b.mp3").write_bytes(b"b") + + def fake(api_key, audio, *, config): + if audio == "a.mp3": + raise APIError("upload exploded") + return _fake_transcript(mocker, audio) + + monkeypatch.setattr(_TRANSCRIBE, fake) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 1 + records = {r["source"]: r for r in _ndjson(result) if "source" in r} + assert records["a.mp3"] == {"source": "a.mp3", "status": "failed", "error": "upload exploded"} + assert records["b.mp3"]["status"] == "completed" + assert (tmp_path / "b.mp3.aai.json").exists() + assert not (tmp_path / "a.mp3.aai.json").exists() + assert "1 of 2 sources failed." in result.output + assert "Re-run the same command" in result.output # resume is the retry path + + +def test_rejected_key_aborts_the_batch_with_exit_4(tmp_path, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + + def fake(api_key, audio, *, config): + raise auth_failure() + + monkeypatch.setattr(_TRANSCRIBE, fake) + monkeypatch.setattr("aai_cli.context._interactive_session", lambda: False) + result = runner.invoke(app, ["transcribe", "*.mp3"]) + assert result.exit_code == 4 + assert "rejected" in result.output + + +def test_auth_failure_drops_not_yet_started_sources(tmp_path, monkeypatch): + # The abort path must shut the pool down with cancel_futures=True — that's what + # keeps one rejected key from burning an upload per queued source. Asserted on + # the shutdown call because which queued futures actually get dropped is a race. + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + seen = {} + real = transcribe_batch.ThreadPoolExecutor + + class Capture(real): + def shutdown(self, wait=True, *, cancel_futures=False): + seen.setdefault("cancel_futures", cancel_futures) # first call wins + super().shutdown(wait=wait, cancel_futures=cancel_futures) + + monkeypatch.setattr(transcribe_batch, "ThreadPoolExecutor", Capture) + + def fake(api_key, audio, *, config): + raise auth_failure() + + monkeypatch.setattr(_TRANSCRIBE, fake) + monkeypatch.setattr("aai_cli.context._interactive_session", lambda: False) + result = runner.invoke(app, ["transcribe", "*.mp3"]) + assert result.exit_code == 4 + assert seen["cancel_futures"] is True + + +# --- output rendering ------------------------------------------------------------- + + +def test_human_mode_prints_result_table_and_summary(tmp_path, mocker, monkeypatch): + # 2 transcribed + 1 skipped: asymmetric on purpose, so a summary that counted + # the *non*-completed sources would read "Transcribed 1, skipped 2" and fail. + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "b.mp3").write_bytes(b"bbb") + (tmp_path / "c.mp3").write_bytes(b"ccc") + _completed_sidecar(tmp_path, "c.mp3", b"ccc") + _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3"]) + assert result.exit_code == 0 + assert "Source" in result.output and "Status" in result.output + assert "completed" in result.output and "skipped" in result.output + assert "a.mp3.aai.json" in result.output + assert "Transcribed 2, skipped 1." in result.output + assert "{" not in result.output # human mode emits no NDJSON + + +def test_json_mode_emits_no_table_or_summary(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert "Source" not in result.output # no table + assert "Transcribed" not in result.output # no human summary + + +def test_quiet_suppresses_the_summary_but_not_the_table(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["--quiet", "transcribe", "*.mp3"]) + assert result.exit_code == 0 + assert "Source" in result.output # the table is the data + assert "Transcribed" not in result.output + + +# --- concurrency ------------------------------------------------------------------ + + +def _capture_pool_size(monkeypatch): + seen = {} + real = transcribe_batch.ThreadPoolExecutor + + class Capture(real): + def __init__(self, max_workers=None, **kwargs): + seen["max_workers"] = max_workers + super().__init__(max_workers=max_workers, **kwargs) + + monkeypatch.setattr(transcribe_batch, "ThreadPoolExecutor", Capture) + return seen + + +def test_default_concurrency_is_four(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + _patch_transcribe(mocker, monkeypatch) + seen = _capture_pool_size(monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert seen["max_workers"] == 4 + + +def test_concurrency_flag_sets_pool_size(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + _patch_transcribe(mocker, monkeypatch) + seen = _capture_pool_size(monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--concurrency", "1", "--json"]) + assert result.exit_code == 0 + assert seen["max_workers"] == 1 + + +def test_concurrency_below_one_is_rejected(tmp_path): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + result = runner.invoke(app, ["transcribe", "*.mp3", "--concurrency", "0"]) + assert result.exit_code == 2 + + +def test_batch_runs_sources_concurrently(tmp_path, mocker, monkeypatch): + # With concurrency 2 and two sources, both workers must be in flight at once. + import threading + + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + (tmp_path / "b.mp3").write_bytes(b"b") + barrier = threading.Barrier(2, timeout=10) + + def fake(api_key, audio, *, config): + barrier.wait() # deadlocks (and times out) unless both run concurrently + return _fake_transcript(mocker, audio) + + monkeypatch.setattr(_TRANSCRIBE, fake) + result = runner.invoke(app, ["transcribe", "*.mp3", "--concurrency", "2", "--json"]) + assert result.exit_code == 0 + assert len(_ndjson(result)) == 2 + + +# --- unit edges -------------------------------------------------------------------- + + +def test_resumable_record_requires_matching_hash(): + # Direct unit check of the three local-file outcomes: match, mismatch, missing. + import pathlib + + sidecar = pathlib.Path("side.aai.json") + sidecar.write_text(json.dumps({"status": "completed", "source_sha256": "abc"})) + assert transcribe_batch.resumable_record(sidecar, digest="abc") is not None + assert transcribe_batch.resumable_record(sidecar, digest="other") is None + assert transcribe_batch.resumable_record(pathlib.Path("missing.json"), digest="abc") is None + + +def test_item_record_minimal_shape_before_any_work(): + # A not-yet-finished item serializes to just {source, status}: no empty id/sidecar keys. + assert transcribe_batch._Item("x.mp3").record() == {"source": "x.mp3", "status": "queued"} + + +def test_source_digest_is_sha256_of_file_bytes(tmp_path): + (tmp_path / "a.mp3").write_bytes(b"payload") + assert transcribe_batch._source_digest("a.mp3") == hashlib.sha256(b"payload").hexdigest() + assert transcribe_batch._source_digest("https://example.com/a.mp3") is None + assert transcribe_batch._source_digest("missing.mp3") is None diff --git a/tests/test_transcribe_batch_sources.py b/tests/test_transcribe_batch_sources.py new file mode 100644 index 00000000..d3c1f0fc --- /dev/null +++ b/tests/test_transcribe_batch_sources.py @@ -0,0 +1,206 @@ +"""Batch-mode source selection for `assembly transcribe`: glob/directory/stdin +expansion and the single-source flags batch mode rejects. + +The batch *run* (sidecar resume, concurrency, failures, output) lives in +test_transcribe_batch.py. +""" + +import json + +import pytest +from typer.testing import CliRunner + +from aai_cli import config, transcribe_batch +from aai_cli.errors import UsageError +from aai_cli.main import app + +runner = CliRunner() + +_TRANSCRIBE = "aai_cli.commands.transcribe.client.transcribe" + + +@pytest.fixture(autouse=True) +def workdir(tmp_path, monkeypatch): + # Batch sources and sidecars are resolved relative to the working directory; + # isolate each test in its own tmp cwd. + monkeypatch.chdir(tmp_path) + + +def _auth(): + config.set_api_key("default", "sk_live") + + +def _patch_transcribe(mocker, monkeypatch): + """Patch client.transcribe with a fake that records the audio args it saw.""" + seen = [] + + def fake(api_key, audio, *, config): + seen.append(audio) + t = mocker.MagicMock() + t.id = f"t_{audio}" + t.text = f"text of {audio}" + t.status = "completed" + t.json_response = {"id": t.id, "text": t.text, "status": "completed"} + return t + + monkeypatch.setattr(_TRANSCRIBE, fake) + return seen + + +def test_glob_skips_sidecar_files(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "stale.aai.json").write_text("{}") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*", "--json"]) + assert result.exit_code == 0 + assert seen == ["a.mp3"] # the stray sidecar is never treated as audio + + +def test_absolute_glob_pattern_matches_from_its_anchor(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", str(tmp_path / "*.mp3"), "--json"]) + assert result.exit_code == 0 + assert seen == [str(tmp_path / "a.mp3")] + + +def test_glob_without_matches_exits_2(mocker, monkeypatch): + _auth() + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "missing-*.mp3"]) + assert result.exit_code == 2 + assert "No files match" in result.output + assert seen == [] + + +def test_existing_file_with_glob_chars_stays_single_source(tmp_path, mocker, monkeypatch): + # A real file whose name contains [ ] must not be re-interpreted as a pattern. + _auth() + (tmp_path / "take[1].mp3").write_bytes(b"aaa") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "take[1].mp3", "-o", "id"]) + assert result.exit_code == 0 + assert seen == ["take[1].mp3"] + assert result.output.strip() == "t_take[1].mp3" # single-source output, no sidecar table + + +def test_directory_scan_is_recursive_and_audio_only(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "calls").mkdir() + (tmp_path / "calls" / "sub").mkdir() + (tmp_path / "calls" / "a.mp3").write_bytes(b"a") + (tmp_path / "calls" / "sub" / "b.WAV").write_bytes(b"b") # extension match is case-insensitive + (tmp_path / "calls" / "notes.txt").write_text("not audio") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "calls", "--json"]) + assert result.exit_code == 0 + assert sorted(seen) == ["calls/a.mp3", "calls/sub/b.WAV"] + + +def test_directory_without_audio_exits_2(tmp_path): + _auth() + (tmp_path / "empty").mkdir() + (tmp_path / "empty" / "notes.txt").write_text("x") + result = runner.invoke(app, ["transcribe", "empty"]) + assert result.exit_code == 2 + assert "No audio files found" in result.output + assert ".mp3" in result.output # the suggestion lists recognized extensions + + +def test_from_stdin_reads_deduped_lines(tmp_path, mocker, monkeypatch): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + (tmp_path / "b.mp3").write_bytes(b"b") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke( + app, ["transcribe", "--from-stdin", "--json"], input="a.mp3\n\na.mp3\nb.mp3\n" + ) + assert result.exit_code == 0 + assert sorted(seen) == ["a.mp3", "b.mp3"] # blank line dropped, duplicate collapsed + + +def test_stdin_source_list_dedupes_preserving_order(monkeypatch): + import io + + monkeypatch.setattr("sys.stdin", io.StringIO("b.mp3\na.mp3\nb.mp3\n")) + assert transcribe_batch.expand_sources(None, from_stdin=True, sample=False) == [ + "b.mp3", + "a.mp3", + ] + + +def test_from_stdin_with_empty_stdin_exits_2(): + _auth() + result = runner.invoke(app, ["transcribe", "--from-stdin"], input="") + assert result.exit_code == 2 + assert "No sources received on stdin" in result.output + + +def test_from_stdin_rejects_source_argument(): + _auth() + result = runner.invoke(app, ["transcribe", "a.mp3", "--from-stdin"], input="b.mp3\n") + assert result.exit_code == 2 + assert "--from-stdin reads sources from stdin" in result.output + + +def test_from_stdin_rejects_sample(): + _auth() + result = runner.invoke(app, ["transcribe", "--sample", "--from-stdin"], input="b.mp3\n") + assert result.exit_code == 2 + assert "--from-stdin reads sources from stdin" in result.output + + +@pytest.mark.parametrize("source", ["-", "https://example.com/a.mp3", None]) +def test_non_batch_sources_return_none(source): + assert transcribe_batch.expand_sources(source, from_stdin=False, sample=False) is None + + +def test_sample_returns_none_even_without_source(): + assert transcribe_batch.expand_sources(None, from_stdin=False, sample=True) is None + + +def test_expand_sources_directory_error_message_names_the_path(tmp_path): + (tmp_path / "calls").mkdir() + with pytest.raises(UsageError, match="No audio files found under calls"): + transcribe_batch.expand_sources("calls", from_stdin=False, sample=False) + + +@pytest.mark.parametrize( + "extra", + [["--out", "x.txt"], ["-o", "text"], ["--llm", "summarize"], ["--show-code"]], +) +def test_batch_rejects_single_source_flags(tmp_path, extra): + _auth() + (tmp_path / "a.mp3").write_bytes(b"a") + result = runner.invoke(app, ["transcribe", "*.mp3", *extra]) + assert result.exit_code == 2 + assert "single source" in result.output + + +def test_glob_batch_writes_per_source_sidecars(tmp_path, mocker, monkeypatch): + import hashlib + + _auth() + (tmp_path / "a.mp3").write_bytes(b"aaa") + (tmp_path / "b.mp3").write_bytes(b"bbb") + seen = _patch_transcribe(mocker, monkeypatch) + result = runner.invoke(app, ["transcribe", "*.mp3", "--json"]) + assert result.exit_code == 0 + assert sorted(seen) == ["a.mp3", "b.mp3"] + records = {r["source"]: r for r in map(json.loads, result.output.splitlines())} + assert records["a.mp3"] == { + "source": "a.mp3", + "status": "completed", + "id": "t_a.mp3", + "sidecar": "a.mp3.aai.json", + } + sidecar = json.loads((tmp_path / "a.mp3.aai.json").read_text()) + assert sidecar["status"] == "completed" + assert sidecar["transcript"] == { + "id": "t_a.mp3", + "text": "text of a.mp3", + "status": "completed", + } + assert sidecar["source_sha256"] == hashlib.sha256(b"aaa").hexdigest()