Compare commits

..

No commits in common. 'e7c0ee774008bb02ae4a0e7010bae455d95a02e9' and '22a215c02a85d2808de00fff2687265abcaad303' have entirely different histories.

  1. 13
      .gitignore
  2. 37
      README.md
  3. 189
      docs/superpowers/plans/2026-08-30-audio-preprocessing-chunking.md
  4. 199
      docs/superpowers/plans/2026-08-30-openai-transcription.md
  5. 168
      docs/superpowers/plans/2026-08-30-transcription-desktop-milestone-1.md
  6. 70
      docs/superpowers/specs/2026-08-30-audio-preprocessing-chunking-design.md
  7. 125
      docs/superpowers/specs/2026-08-30-openai-transcription-design.md
  8. 27
      pyproject.toml
  9. 6
      requirements.txt
  10. 4
      src/voice_transcriptor/__init__.py
  11. 3
      src/voice_transcriptor/__main__.py
  12. 40
      src/voice_transcriptor/app.py
  13. 20
      src/voice_transcriptor/formatting.py
  14. 52
      src/voice_transcriptor/models.py
  15. 2
      src/voice_transcriptor/services/__init__.py
  16. 54
      src/voice_transcriptor/services/chunking.py
  17. 39
      src/voice_transcriptor/services/credentials.py
  18. 199
      src/voice_transcriptor/services/job_manifest.py
  19. 108
      src/voice_transcriptor/services/media_probe.py
  20. 164
      src/voice_transcriptor/services/preprocessing.py
  21. 123
      src/voice_transcriptor/services/settings.py
  22. 183
      src/voice_transcriptor/services/transcription.py
  23. 1
      src/voice_transcriptor/ui/__init__.py
  24. 163
      src/voice_transcriptor/ui/main_window.py
  25. 203
      src/voice_transcriptor/ui/settings_dialog.py
  26. 50
      tests/test_chunking.py
  27. 88
      tests/test_credentials.py
  28. 16
      tests/test_formatting.py
  29. 113
      tests/test_job_manifest.py
  30. 124
      tests/test_media_probe.py
  31. 100
      tests/test_preprocessing.py
  32. 144
      tests/test_settings.py
  33. 93
      tests/test_transcription.py
  34. 26
      tests/ui/test_app.py
  35. 91
      tests/ui/test_main_window.py
  36. 191
      tests/ui/test_settings_dialog.py

13
.gitignore vendored

@ -1,13 +0,0 @@
__pycache__/
*.py[cod]
.pytest_cache/
.venv/
venv/
dist/
build/
*.spec
.coverage
.idea/
.vscode/
.superpowers/
.worktrees/

37
README.md

@ -1,37 +0,0 @@
# Voice Transcriptor
Windows desktop application for resumable OpenAI transcription of long audio and video recordings.
## Setup and run
Install Python 3.12 or newer plus FFmpeg/FFprobe, ensure both executables are on `PATH`, then install dependencies and launch:
```powershell
python -m pip install -r requirements.txt
python -m voice_transcriptor
```
The app accepts `.m4a`, `.mp3`, `.wav`, `.mp4`, `.mov`, `.webm`, and `.mkv`. It probes the source with FFprobe and streams it through FFmpeg into mono 24 kHz AAC-LC `.m4a` chunks at 64 kbps. Video audio is selected directly, with no large intermediate extraction file.
Chunking is duration-based, not tied to a presumed universal upload-size limit. Defaults are 15-minute chunks with 15 seconds of overlap. Change duration, overlap, and temporary-file retention under **Settings → Advanced**.
Each transcription job is stored under `<output directory>/voice-transcriptor-jobs/<job-id>/`. Its atomic JSON manifest retains the exact source start/end time of every chunk plus `pending`, `processing`, `completed`, or `failed` status and raw transcript text. Successful chunks are saved immediately. Restarting the same source with the same effective settings resumes an incomplete job and never retranscribes completed chunks.
The default model is `gpt-transcribe`. Settings provides an editable selector seeded with `gpt-transcribe`, `gpt-4o-transcribe`, and `gpt-4o-mini-transcribe`, so future compatible transcription model IDs can be entered. The current official API documentation may reject `gpt-transcribe` for accounts where that identifier is unavailable; the application reports that as a permanent API error without silently changing models.
Language defaults to Brazilian Portuguese. The app sends ISO language `pt` and a prompt that preserves the spoken language without translation, Brazilian spelling and punctuation, numbers, names, technical terminology, and acronyms. Add names and specialized vocabulary in **Settings → Context / Vocabulary**.
Every preprocessed chunk is submitted independently to the OpenAI Audio Transcriptions API. Raw chunk transcripts are concatenated in source order; overlap deduplication is intentionally not implemented yet. Rate limits, timeouts, connection errors, and server errors use finite exponential backoff. Permanent API failures are not retried indefinitely. Active preprocessing, retry waits, and transcription can be cancelled without blocking the GUI.
## API key
Store the API key as a Windows Credential Manager generic credential whose target name is exactly `OPENAI_API_KEY`. The application retrieves that existing target through the Windows keyring backend. Saving a replacement key from Settings updates the same target. API keys and authorization headers are never written to settings, job manifests, transcripts, or logs.
## Tests
```powershell
python -m pytest -v
python -m compileall -q src tests
```
Tests use fake API endpoints and do not make live OpenAI requests.

189
docs/superpowers/plans/2026-08-30-audio-preprocessing-chunking.md

@ -1,189 +0,0 @@
# Audio Preprocessing and Chunking Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build a cancellable, incremental FFmpeg pipeline that creates duration-based overlapping speech-audio chunks, exact timestamp manifests, and GUI-visible progress.
**Architecture:** Pure `Decimal` chunk calculations feed a service that probes once and launches one streaming FFmpeg process per chunk. Typed callbacks and cancellation stay UI-independent; a Qt worker adapts them into signals, while settings persistence supplies duration, overlap, and retention policy.
**Tech Stack:** Python 3.12+, `decimal`, `dataclasses`, `subprocess`, FFmpeg/FFprobe, PySide6, pytest, pytest-qt.
**Spec:** `docs/superpowers/specs/2026-08-30-audio-preprocessing-chunking-design.md`
## Global Constraints
- Accept only `m4a`, `mp3`, `wav`, `mp4`, `mov`, `webm`, and `mkv` through the preprocessing boundary.
- Use FFprobe before FFmpeg and require a positive finite duration plus an audio stream.
- Default chunk duration is 900 seconds and overlap is 15 seconds; chunking is duration-based, not based on a presumed universal upload-size limit.
- Encode direct-from-source mono 24 kHz AAC-LC at 64 kbps in `.m4a` without a large intermediate file.
- Never load the recording into Python memory; one streaming FFmpeg child may run at a time.
- Manifest timestamps use exact decimal seconds serialized as strings.
- Job artifacts live in one verified job-specific temporary directory; remove it unless retention is enabled.
- Do not call any transcription API.
---
## File Map
- `src/voice_transcriptor/models.py`: shared immutable settings/media/chunk/progress/result values.
- `src/voice_transcriptor/services/chunking.py`: pure exact chunk calculations and validation.
- `src/voice_transcriptor/services/media_probe.py`: source stream metadata from FFprobe.
- `src/voice_transcriptor/services/preprocessing.py`: process lifecycle, manifest, cancellation, progress, and cleanup.
- `src/voice_transcriptor/services/settings.py`: backward-compatible advanced settings persistence.
- `src/voice_transcriptor/ui/settings_dialog.py`: collapsible advanced preprocessing controls.
- `src/voice_transcriptor/ui/main_window.py`: selection, metadata, worker, progress, and cancellation UI.
- `src/voice_transcriptor/app.py`, `src/voice_transcriptor/__main__.py`: concrete wiring and entry points.
- `tests/test_chunking.py`: exact calculation contracts.
- `tests/test_preprocessing.py`: FFmpeg/manifest/lifecycle service contracts.
- `tests/test_media_probe.py`, `tests/test_settings.py`: extended existing contracts.
- `tests/ui/test_settings_dialog.py`, `tests/ui/test_main_window.py`, `tests/ui/test_app.py`: Qt behavior and startup.
### Task 1: Exact chunk model and calculations
**Files:**
- Modify: `src/voice_transcriptor/models.py`
- Create: `src/voice_transcriptor/services/chunking.py`
- Create: `tests/test_chunking.py`
**Interfaces:**
- Produces: `ChunkBoundary(index: int, start_seconds: Decimal, end_seconds: Decimal)` with derived `duration_seconds`.
- Produces: `calculate_chunk_boundaries(total_duration, chunk_duration=Decimal("900"), overlap=Decimal("15")) -> tuple[ChunkBoundary, ...]`.
- Produces: `total_chunk_duration(boundaries) -> Decimal` and `source_timestamp(boundary, local_seconds) -> Decimal`.
- [ ] **Step 1: Write failing literal boundary tests.** Include `600 -> [(0, 600)]`, `900 -> [(0, 900)]`, `1800 -> [(0, 900), (885, 1785), (1770, 1800)]`, `901.25 -> [(0, 900), (885, 901.25)]`, and zero-overlap `1800 -> [(0, 900), (900, 1800)]`.
- [ ] **Step 2: Write failing aggregate/offset/validation tests.** Assert the 1800-second default boundaries total `1830`, chunk two local `12.5` maps to source `897.5`, multi-hour `28800` ends exactly at `28800`, and invalid non-positive duration/chunk or overlap outside `[0, chunk)` raises `ValueError`.
- [ ] **Step 3: Run `python -m pytest tests/test_chunking.py -v` and confirm failure because the module/interfaces do not exist.**
- [ ] **Step 4: Implement exact calculation minimally.** Convert inputs with `Decimal(str(value))`, reject non-finite values, advance by `chunk - overlap`, and stop immediately when `end == total`.
- [ ] **Step 5: Run the focused tests and confirm they pass; then run the existing suite to detect model regressions.**
- [ ] **Step 6: Commit with `git commit -m "feat: add exact overlapping chunk calculations"`.**
### Task 2: Rich FFprobe source metadata
**Files:**
- Modify: `src/voice_transcriptor/models.py`
- Modify: `src/voice_transcriptor/services/media_probe.py`
- Modify: `tests/test_media_probe.py`
**Interfaces:**
- Extends: `MediaInfo(..., duration_text: str | None, has_audio: bool, has_video: bool)` while retaining existing fields.
- `parse_probe_output` selects the first audio codec and preserves the raw positive duration string for exact chunk math.
- [ ] **Step 1: Add failing tests for audio-only, video-with-audio, and video-without-audio FFprobe JSON.** Use complete payloads with `format.duration` and stream `codec_type`/`codec_name`; assert exact `duration_text`, `has_audio`, and `has_video` literals.
- [ ] **Step 2: Run `python -m pytest tests/test_media_probe.py -v` and confirm missing-field/constructor failures.**
- [ ] **Step 3: Extend `MediaInfo`, parser, and FFprobe `-show_entries` without changing shell/timeout/error safeguards.**
- [ ] **Step 4: Run media-probe tests and the full suite.**
- [ ] **Step 5: Commit with `git commit -m "feat: expose audio and video probe metadata"`.**
### Task 3: Preprocessing manifests and FFmpeg command contract
**Files:**
- Modify: `src/voice_transcriptor/models.py`
- Create: `src/voice_transcriptor/services/preprocessing.py`
- Create: `tests/test_preprocessing.py`
**Interfaces:**
- Produces: `PreprocessingOptions(chunk_duration_seconds=900, overlap_seconds=15, retain_temporary_files=False)`.
- Produces: `CancellationToken.cancel()`, `.cancelled`, and `.raise_if_cancelled()`.
- Produces: `PreprocessingProgress(percent: int, phase: str, message: str)`.
- Produces: `PreprocessingResult(job_directory: Path, manifest_path: Path, retained: bool)` with `.cleanup()` and context-manager behavior.
- Produces: `PreprocessingService(ffmpeg_path, probe_service, temporary_root=None, process_factory=subprocess.Popen).preprocess(source, options, token=None, progress=None) -> PreprocessingResult`.
- [ ] **Step 1: Write failing tests for extension/settings/source validation.** Assert all seven extensions are accepted case-insensitively and unsupported extension, absent audio, absent/non-positive duration, and invalid overlap fail before process creation.
- [ ] **Step 2: Write a failing command-contract test with a controlled process fake.** Assert list arguments include `-map 0:a:0`, `-vn`, `-sn`, `-dn`, exact `-ss`/`-t`, `-ac 1`, `-ar 24000`, `-c:a aac`, `-b:a 64k`, `-progress pipe:1`, and a job-owned `.m4a` path; assert one process per boundary.
- [ ] **Step 3: Write failing manifest tests.** Read real JSON written under `tmp_path`; assert exact string timestamps, relative chunk paths, source/output metadata, completed count, and atomic terminal `completed` state.
- [ ] **Step 4: Run `python -m pytest tests/test_preprocessing.py -v` and confirm import/interface failure.**
- [ ] **Step 5: Implement validation, job creation, manifest serialization, atomic writes, command construction, and sequential process execution.** Use `mkdtemp`, `Popen` without a shell, text line buffering, `CREATE_NO_WINDOW` when available, and sanitized typed exceptions.
- [ ] **Step 6: Run focused and full tests; refactor manifest helpers only while green.**
- [ ] **Step 7: Commit with `git commit -m "feat: add streaming FFmpeg preprocessing"`.**
### Task 4: Progress, cancellation, and safe cleanup
**Files:**
- Modify: `src/voice_transcriptor/services/preprocessing.py`
- Modify: `tests/test_preprocessing.py`
**Interfaces:**
- Progress callback receives monotonic `PreprocessingProgress` values.
- Cancellation is reported with `PreprocessingCancelled`, distinct from `PreprocessingError`.
- Cleanup only removes the resolved directory identity stored at creation.
- [ ] **Step 1: Add failing progress tests.** Feed fake `out_time_us=450000000`, `progress=continue`, and `progress=end` lines; assert percentages are monotonic, current-chunk time is weighted against total planned chunk duration, and 100 occurs only after completion.
- [ ] **Step 2: Add failing cancellation tests.** Cancel while the fake process emits progress; assert `terminate()` is called, later process factories are untouched, partial output is removed, and manifest state is `cancelled`.
- [ ] **Step 3: Add failing lifecycle tests.** Assert default cleanup removes the exact job directory, retained mode preserves it, context-manager exit cleans it, and a tampered cleanup target raises rather than deleting another path.
- [ ] **Step 4: Run the focused tests and confirm expected failures.**
- [ ] **Step 5: Implement line-by-line progress parsing, cooperative termination/kill fallback, terminal manifest transitions, and identity-checked cleanup.** Never call `communicate()` in a way that buffers an unbounded stream.
- [ ] **Step 6: Run focused and full tests.**
- [ ] **Step 7: Commit with `git commit -m "feat: add preprocessing cancellation and cleanup"`.**
### Task 5: Persist advanced preprocessing settings
**Files:**
- Modify: `src/voice_transcriptor/models.py`
- Modify: `src/voice_transcriptor/services/settings.py`
- Modify: `tests/test_settings.py`
**Interfaces:**
- Extends: `AppSettings(model, language, output_directory, chunk_duration_seconds=900, chunk_overlap_seconds=15, retain_temporary_files=False)`.
- Existing three-argument construction remains valid through defaults.
- [ ] **Step 1: Add failing tests for defaults, new-value round trip, legacy three-key JSON, invalid overlap fallback, and continued absence of API-key fields.** Expected JSON literals include the three new snake-case keys.
- [ ] **Step 2: Run `python -m pytest tests/test_settings.py -v` and confirm constructor/serialization failures.**
- [ ] **Step 3: Implement backward-compatible loading and strict type/range validation.** Reject booleans where integer seconds are expected and require `0 <= overlap < duration`.
- [ ] **Step 4: Run settings and full tests.**
- [ ] **Step 5: Commit with `git commit -m "feat: persist preprocessing settings"`.**
### Task 6: Advanced settings UI
**Files:**
- Modify: `src/voice_transcriptor/ui/settings_dialog.py`
- Modify: `tests/ui/test_settings_dialog.py`
**Interfaces:**
- Adds object-named widgets `advancedToggle`, `advancedPanel`, `chunkDurationInput`, `chunkOverlapInput`, and `retainTemporaryFilesInput`.
- `settings_saved` emits the complete updated `AppSettings`.
- [ ] **Step 1: Add failing Qt tests for initially hidden advanced panel, toggle visibility, populated defaults, complete save, and overlap-equals-duration warning without persistence.**
- [ ] **Step 2: Run the focused UI test and confirm widgets are missing.**
- [ ] **Step 3: Implement a checkable Advanced button, panel, two `QSpinBox` controls, and `QCheckBox`; preserve existing API-key handling and directory validation.**
- [ ] **Step 4: Run focused UI and full tests.**
- [ ] **Step 5: Commit with `git commit -m "feat: add advanced chunk settings"`.**
### Task 7: Main window and asynchronous preprocessing worker
**Files:**
- Create: `src/voice_transcriptor/ui/main_window.py`
- Create: `tests/ui/test_main_window.py`
**Interfaces:**
- Produces: `MainWindow(media_probe, preprocessing_service, settings_repository, credentials)`.
- Produces internal `PreprocessingWorker(QRunnable)` and signal object carrying progress/result/error/cancelled.
- Provides `select_file(path: Path)`, `start_preprocessing()`, and `cancel_preprocessing()`.
- [ ] **Step 1: Write failing Qt tests for controls and supported file selection.** Assert Browse, Settings, Prepare, progress bar, Cancel, metadata, and log exist; unsupported/multi-file inputs are rejected.
- [ ] **Step 2: Write failing worker-state tests with a blocking fake service.** Start work, assert selection/settings/prepare disabled and Cancel enabled; emit progress and assert displayed value/message; cancel and assert the same token is cancelled; finish each terminal signal and assert controls restore.
- [ ] **Step 3: Write failing stale-probe and close-during-work tests.** Ensure an older selection cannot overwrite newer metadata and close requests cancellation.
- [ ] **Step 4: Run `python -m pytest tests/ui/test_main_window.py -v` and confirm import failure.**
- [ ] **Step 5: Build the minimal native layout and async QRunnable orchestration.** Keep service calls off the GUI thread, connect queued signals, retain worker/signal references through completion, and clean results according to retention policy.
- [ ] **Step 6: Run focused UI and full tests.**
- [ ] **Step 7: Commit with `git commit -m "feat: add preprocessing desktop workflow"`.**
### Task 8: Application wiring and end-to-end verification
**Files:**
- Create: `src/voice_transcriptor/app.py`
- Create: `src/voice_transcriptor/__main__.py`
- Create: `tests/ui/test_app.py`
- Modify: `README.md`
**Interfaces:**
- Produces: `create_main_window() -> MainWindow` and `main() -> int`.
- Preserves console entry point `voice-transcriptor = voice_transcriptor.app:main`.
- [ ] **Step 1: Write a failing offscreen smoke test that creates, shows, processes events for, and closes the concrete window.**
- [ ] **Step 2: Run `python -m pytest tests/ui/test_app.py -v` and confirm bootstrap import failure.**
- [ ] **Step 3: Wire detected FFmpeg/FFprobe paths, probe/preprocessing/settings/credential services, guarded startup messages, and `python -m voice_transcriptor`.** Missing tools must leave the window usable with preprocessing disabled.
- [ ] **Step 4: Document supported formats, defaults, duration-based chunking, FFmpeg requirement, temporary retention, cancellation, and the absence of transcription API calls.**
- [ ] **Step 5: Run `python -m pytest -v`, `python -m compileall -q src tests`, and `git diff --check`.**
- [ ] **Step 6: Run `ffmpeg -version` and `ffprobe -version`. If present, generate a short sine/speech-like fixture under a job-owned temporary directory and run the real service, then inspect its manifest and clean the fixture; report absence separately.**
- [ ] **Step 7: Review every specification requirement against code/tests and scan tracked text for credential-like values.**
- [ ] **Step 8: Commit with `git commit -m "docs: document preprocessing workflow"`.**

199
docs/superpowers/plans/2026-08-30-openai-transcription.md

@ -1,199 +0,0 @@
# OpenAI Transcription Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add independent OpenAI transcription of prepared chunks with Brazilian-Portuguese prompting, durable incremental manifests, resumability, finite retries, cancellation, and responsive GUI progress.
**Architecture:** Extend the existing preprocessing manifest into a versioned durable job stored beneath the configured output directory. A pure manifest repository owns atomic state changes; an OpenAI adapter owns one SDK request; a transcription orchestrator owns resume, retries, progress, cancellation, and ordered transcript assembly; Qt only adapts worker callbacks into UI signals.
**Tech Stack:** Python 3.12+, OpenAI Python SDK 3.6.0-compatible interface, PySide6, keyring/Windows Credential Manager, pytest, pytest-qt, FFmpeg/ffprobe.
**Spec:** `docs/superpowers/specs/2026-08-30-openai-transcription-design.md`
## Global Constraints
- Use `client.audio.transcriptions.create`, not the translation endpoint.
- Default model is exactly `gpt-transcribe`; never silently substitute another model.
- Seed model choices with `gpt-transcribe`, `gpt-4o-transcribe`, and `gpt-4o-mini-transcribe`, while accepting future strings.
- Present Brazilian Portuguese as `pt-BR` and send documented ISO-639-1 `pt` to the API.
- Read the API key from the exact Windows Credential Manager target `OPENAI_API_KEY`; never persist or log it elsewhere.
- Every successful chunk must be atomically recorded before the next request.
- Never retranscribe a completed chunk except through an explicit reset/retranscribe operation.
- No overlap deduplication or translation in this milestone.
- All network and job work remains outside the GUI thread.
- Use test-driven development: add one failing behavior test, verify RED, implement minimally, verify GREEN, then refactor.
---
## File Map
- `src/voice_transcriptor/models.py`: add context/vocabulary to `AppSettings`.
- `src/voice_transcriptor/services/settings.py`: persist new defaults and migrate older files.
- `src/voice_transcriptor/services/credentials.py`: target-based Windows credential lookup/update.
- `src/voice_transcriptor/services/job_manifest.py`: atomic schema-2 job persistence, recovery, resume, and transcript assembly.
- `src/voice_transcriptor/services/preprocessing.py`: durable output-root jobs and per-chunk preprocessing status updates.
- `src/voice_transcriptor/services/transcription.py`: prompt normalization, SDK adapter, error classification, retries, cancellation, and orchestration.
- `src/voice_transcriptor/ui/settings_dialog.py`: editable model combo and context editor.
- `src/voice_transcriptor/ui/main_window.py`: combined job worker, progress labels, elapsed time, API errors, resume and cancellation.
- `src/voice_transcriptor/app.py`: concrete OpenAI/job service wiring.
- `tests/test_settings.py`, `tests/test_credentials.py`: settings and exact credential-target coverage.
- `tests/test_job_manifest.py`: durable state and resume invariants.
- `tests/test_preprocessing.py`: durable job integration.
- `tests/test_transcription.py`: API arguments, retry/error behavior, persistence, resume, and cancellation.
- `tests/ui/test_settings_dialog.py`, `tests/ui/test_main_window.py`, `tests/ui/test_app.py`: GUI settings, worker progress, errors, cancellation, and startup.
- `README.md`: API credential, model/context, job/resume, and validation documentation.
### Task 1: Settings and Existing Windows Credential Target
**Files:**
- Modify: `src/voice_transcriptor/models.py`
- Modify: `src/voice_transcriptor/services/settings.py`
- Modify: `src/voice_transcriptor/services/credentials.py`
- Modify: `tests/test_settings.py`
- Modify: `tests/test_credentials.py`
**Interfaces:**
- Produces: `DEFAULT_CONTEXT`, `SUPPORTED_MODEL_SUGGESTIONS`, and `AppSettings(..., context_vocabulary: str = DEFAULT_CONTEXT)`.
- Produces: `CredentialService.get_api_key() -> str | None`, `has_api_key() -> bool`, and `set_api_key(value: str) -> None` using target `OPENAI_API_KEY` and backend `get_credential(service_name, username=None)`.
- Preserves all existing public settings and credential methods.
- [ ] **Step 1: Add failing settings tests** asserting the default model is `gpt-transcribe`, language remains `pt-BR`, the default context contains Brazilian guidance and vocabulary examples, older JSON files receive that context, and save/load round-trips it without an API key.
- [ ] **Step 2: Run `python -m pytest tests/test_settings.py -v`** and verify failures report the old model or missing `context_vocabulary` field.
- [ ] **Step 3: Implement the minimal settings/model changes** by appending the dataclass field (to preserve positional callers), updating defaults, parsing optional string context, and serializing it.
- [ ] **Step 4: Re-run `python -m pytest tests/test_settings.py -v`** and confirm GREEN.
- [ ] **Step 5: Add failing credential tests** with a fake backend whose `get_credential("OPENAI_API_KEY", None)` returns an object containing `username` and `password`; assert retrieval, absence, sanitized backend failures, update preserving username, and creation using username `OPENAI_API_KEY`.
- [ ] **Step 6: Run `python -m pytest tests/test_credentials.py -v`** and verify RED because the current service queries `voice-transcriptor` / `openai-api-key` with `get_password`.
- [ ] **Step 7: Implement exact-target lookup/update** using `TARGET_NAME = "OPENAI_API_KEY"`, `get_credential`, and `set_password(TARGET_NAME, resolved_username, value)` without exposing backend details.
- [ ] **Step 8: Run `python -m pytest tests/test_settings.py tests/test_credentials.py -v`** and confirm GREEN.
- [ ] **Step 9: Commit** with `git add src/voice_transcriptor/models.py src/voice_transcriptor/services/settings.py src/voice_transcriptor/services/credentials.py tests/test_settings.py tests/test_credentials.py && git commit -m "feat: configure transcription models and credential target"`.
### Task 2: Atomic Resumable Job Manifest
**Files:**
- Create: `src/voice_transcriptor/services/job_manifest.py`
- Create: `tests/test_job_manifest.py`
**Interfaces:**
- Produces: `ChunkStatus(StrEnum)` with `PENDING`, `PROCESSING`, `COMPLETED`, `FAILED`.
- Produces: `JobManifestRepository.create(job_directory: Path, source: dict, settings: dict, chunks: Sequence[dict]) -> dict`.
- Produces: `load(path: Path) -> dict`, `save(path: Path, manifest: dict) -> None`, `recover_for_resume(path: Path) -> dict`, `mark_processing(path, index)`, `mark_completed(path, index, text)`, `mark_failed(path, index, message)`, `mark_job_state(path, state, error=None)`, and `reset_completed(path, indexes=None)`.
- Produces: `assemble_transcript(manifest_path: Path) -> Path` writing sibling `transcript.txt` atomically in chunk-index order.
- [ ] **Step 1: Write failing manifest tests** for schema 2 creation, exact decimal timing preservation, four chunk statuses, safe relative paths, atomic `.tmp` replacement, completed counters, and rejection of malformed/out-of-directory chunk paths.
- [ ] **Step 2: Run `python -m pytest tests/test_job_manifest.py -v`** and verify import failure.
- [ ] **Step 3: Implement creation, validation, load, and atomic save** with UTF-8 JSON and ISO-8601 UTC timestamps.
- [ ] **Step 4: Re-run the focused tests** and confirm the first state tests pass.
- [ ] **Step 5: Add failing transition tests** asserting each mutator immediately changes disk state, `processing` recovers to `pending`, `completed` survives recovery, explicit resume resets `failed`, and ordinary transitions refuse to overwrite completed chunks.
- [ ] **Step 6: Run the focused tests** and verify RED on missing transition methods.
- [ ] **Step 7: Implement transition and recovery methods** with index validation, attempt counts, sanitized error strings, and completed-count recalculation.
- [ ] **Step 8: Add failing assembly tests** asserting completed texts are newline-concatenated in index order, partial transcripts contain only completed chunks, and source timing remains unchanged.
- [ ] **Step 9: Implement atomic transcript assembly** and explicit completed-reset behavior.
- [ ] **Step 10: Run `python -m pytest tests/test_job_manifest.py -v`** and confirm GREEN.
- [ ] **Step 11: Commit** with `git add src/voice_transcriptor/services/job_manifest.py tests/test_job_manifest.py && git commit -m "feat: add durable resumable job manifests"`.
### Task 3: Durable Preprocessing Jobs
**Files:**
- Modify: `src/voice_transcriptor/services/preprocessing.py`
- Modify: `tests/test_preprocessing.py`
**Interfaces:**
- Consumes: `JobManifestRepository` and its schema-2 chunk records.
- Changes: `PreprocessingOptions` gains `job_root: Path | None = None` or `PreprocessingService.preprocess(..., job_root: Path | None = None)`; the implementation must receive the configured persistent root rather than use a system temporary directory for transcribable jobs.
- Produces: `PreprocessingResult` whose directory remains durable and whose manifest contains independently reusable encoded chunks.
- [ ] **Step 1: Add failing preprocessing tests** asserting jobs are created below `<output>/voice-transcriptor-jobs`, a pending transcript status is written after every successful encoded chunk, cancellation/failure retains the job, and existing valid chunks can be reused on resume.
- [ ] **Step 2: Run `python -m pytest tests/test_preprocessing.py -v`** and verify RED against temporary-directory deletion/schema 1 behavior.
- [ ] **Step 3: Inject `JobManifestRepository` and durable root selection** while preserving the existing FFmpeg command, exact boundaries, and atomic per-chunk writes.
- [ ] **Step 4: Implement preprocessing resume** so valid existing chunk files are skipped, missing files are regenerated, completed transcript text is never cleared, and partial current outputs are removed after FFmpeg failure.
- [ ] **Step 5: Run `python -m pytest tests/test_preprocessing.py tests/test_chunking.py -v`** and confirm GREEN.
- [ ] **Step 6: Commit** with `git add src/voice_transcriptor/services/preprocessing.py tests/test_preprocessing.py && git commit -m "feat: persist and resume preprocessing jobs"`.
### Task 4: OpenAI Adapter, Prompting, Retry Policy, and Orchestrator
**Files:**
- Replace: `src/voice_transcriptor/services/transcription.py`
- Replace: `tests/test_transcription.py`
**Interfaces:**
- Produces: `normalize_language(value: str) -> str` mapping case-insensitive `pt-BR`/`pt_BR` to `pt` and otherwise using the primary ISO language subtag.
- Produces: `build_prompt(context: str) -> str` combining the fixed no-translation Brazilian instruction with trimmed optional context.
- Produces: `OpenAITranscriptionClient(client)` with `transcribe(path: Path, model: str, language: str, prompt: str) -> str` calling `client.audio.transcriptions.create` exactly once.
- Produces: `RetryPolicy(max_attempts=5, initial_delay_seconds=1.0, max_delay_seconds=30.0, jitter_ratio=0.2)`.
- Produces: `TranscriptionProgress(completed: int, total: int, current_chunk: int | None, elapsed_seconds: float, phase: str, message: str, api_error: str | None = None)`.
- Produces: `TranscriptionService.run(manifest_path, settings, token, progress) -> Path` returning `transcript.txt`.
- Produces typed `TranscriptionError`, `PermanentTranscriptionError`, and `RetryExhaustedError` with sanitized messages.
- [ ] **Step 1: Write failing pure tests** for language normalization, fixed no-translation prompt content, optional vocabulary trimming, and arbitrary models.
- [ ] **Step 2: Run `python -m pytest tests/test_transcription.py -v`** and verify RED because the service is still a milestone stub.
- [ ] **Step 3: Implement only the pure helpers** and re-run those tests to GREEN.
- [ ] **Step 4: Add a failing SDK adapter test** using a fake endpoint; assert the opened binary file, exact model, `language="pt"`, combined `prompt`, `response_format="json"`, and returned `.text`. Assert no translation method is touched.
- [ ] **Step 5: Implement `OpenAITranscriptionClient`** with the verified installed-SDK call signature and a context manager for the chunk file.
- [ ] **Step 6: Add failing retry-classification tests** for `RateLimitError`, `APIConnectionError`, `APITimeoutError`, 408/409/429/5xx, authentication/permission/bad request, maximum attempts, exponential delays, jitter bounds, and capped `Retry-After`.
- [ ] **Step 7: Implement deterministic injectable retry dependencies** (`sleep`, monotonic clock, random source) and classification by SDK class/status code. Disable SDK retries during app wiring so only this policy applies.
- [ ] **Step 8: Add failing orchestration tests** asserting one independent call per unfinished chunk, immediate disk completion after each success, skip of completed chunks, recovery of `processing`, explicit retry of failed chunks, timing retention, ordered assembly, cancellation before a request and during backoff, progress fields, and sanitized errors without key/header-like content.
- [ ] **Step 9: Implement orchestration minimally** around `JobManifestRepository`, checking the token before each call and via cancellation-aware backoff waits.
- [ ] **Step 10: Run `python -m pytest tests/test_transcription.py tests/test_job_manifest.py -v`** and confirm GREEN.
- [ ] **Step 11: Commit** with `git add src/voice_transcriptor/services/transcription.py tests/test_transcription.py && git commit -m "feat: transcribe chunks with retries and resume"`.
### Task 5: Settings GUI for Model and Context
**Files:**
- Modify: `src/voice_transcriptor/ui/settings_dialog.py`
- Modify: `tests/ui/test_settings_dialog.py`
**Interfaces:**
- Changes: `model_input` becomes an editable `QComboBox` seeded from `SUPPORTED_MODEL_SUGGESTIONS` while preserving the object name `modelInput`.
- Produces: multiline `context_input: QPlainTextEdit` with object name `contextVocabularyInput`.
- Emits the extended `AppSettings` value without exposing the API key.
- [ ] **Step 1: Add failing Qt tests** for the three seeded model suggestions, editable future model value, populated default context, round-trip user context, and validation of a blank model.
- [ ] **Step 2: Run `python -m pytest tests/ui/test_settings_dialog.py -v`** and verify RED on the missing combo/context controls.
- [ ] **Step 3: Implement the editable combo and context field** and adapt save logic to `currentText()` and `toPlainText()`.
- [ ] **Step 4: Run `python -m pytest tests/ui/test_settings_dialog.py tests/test_settings.py -v`** and confirm GREEN.
- [ ] **Step 5: Commit** with `git add src/voice_transcriptor/ui/settings_dialog.py tests/ui/test_settings_dialog.py && git commit -m "feat: add transcription model and vocabulary settings"`.
### Task 6: GUI Job Worker, Progress, Resume, Errors, and Cancellation
**Files:**
- Modify: `src/voice_transcriptor/ui/main_window.py`
- Modify: `tests/ui/test_main_window.py`
**Interfaces:**
- Consumes: preprocessing service, transcription service, settings repository, credentials.
- Produces: a `JobWorker(QRunnable)` that preprocesses/resumes then transcribes and emits typed progress, completion, cancellation, and sanitized failure signals.
- Produces visible widgets `chunk_progress_label`, `elapsed_label`, `current_chunk_label`, and existing read-only log entries for API errors.
- Changes primary action text to `Transcribe`; retains `Cancel` and thread-pool execution.
- [ ] **Step 1: Add failing widget tests** for Transcribe action text, `0 / 0` chunk progress, elapsed display, current chunk display, and API error log surface.
- [ ] **Step 2: Run `python -m pytest tests/ui/test_main_window.py -v`** and verify RED.
- [ ] **Step 3: Add the progress widgets and slots** without changing worker behavior; re-run widget tests to GREEN.
- [ ] **Step 4: Add failing worker-flow tests** with fake services asserting the click returns without executing network work on the GUI thread, progress updates all required fields, completed results report the transcript path, and controls restore on success/failure/cancellation.
- [ ] **Step 5: Implement `JobWorker` and connect signals**; retrieve the credential before constructing work, pass no key through signals/logs, and reuse the existing cancellation token.
- [ ] **Step 6: Add failing resume tests** asserting a retained matching manifest is offered/reused and completed chunks are not reset; test that no implicit retranscribe path exists.
- [ ] **Step 7: Implement resume selection** using a deterministic matching-manifest service method; retain all job data on terminal cancellation/failure.
- [ ] **Step 8: Run `python -m pytest tests/ui/test_main_window.py -v`** and confirm GREEN.
- [ ] **Step 9: Commit** with `git add src/voice_transcriptor/ui/main_window.py tests/ui/test_main_window.py && git commit -m "feat: add responsive transcription job interface"`.
### Task 7: Application Wiring and Integrated Validation
**Files:**
- Modify: `src/voice_transcriptor/app.py`
- Modify: `tests/ui/test_app.py`
- Modify: `README.md`
**Interfaces:**
- `create_main_window()` wires `OpenAI(api_key=credential, max_retries=0)`, `JobManifestRepository`, durable preprocessing, and `TranscriptionService` without making an API request at startup.
- Startup remains usable for selecting media and editing settings when the credential is absent; Transcribe reports a concise configuration error.
- [ ] **Step 1: Add a failing app-wiring test** with injected factories asserting `max_retries=0`, no startup API call, exact credential retrieval, and all concrete services reach `MainWindow`.
- [ ] **Step 2: Run `python -m pytest tests/ui/test_app.py -v`** and verify RED.
- [ ] **Step 3: Refactor app wiring for injectable factories** and construct the concrete SDK client only when a transcription job starts.
- [ ] **Step 4: Run `python -m pytest tests/ui/test_app.py tests/ui/test_main_window.py -v`** and confirm GREEN.
- [ ] **Step 5: Update README** with Windows Credential Manager target `OPENAI_API_KEY`, model behavior (including the documented-model caveat for `gpt-transcribe`), Brazilian context, durable job directory, resume semantics, retries/rate limits, cancellation, and test commands.
- [ ] **Step 6: Run the complete automated suite:** `python -m pytest -v`.
- [ ] **Step 7: Run static/startup validation:** `python -m compileall -q src tests`, then `$env:QT_QPA_PLATFORM='offscreen'; python -c "from PySide6.QtCore import QTimer; from PySide6.QtWidgets import QApplication; from voice_transcriptor.app import create_main_window; app=QApplication([]); window=create_main_window(); window.show(); QTimer.singleShot(100, app.quit); app.exec(); window.close()"`.
- [ ] **Step 8: Run local media-tool checks:** `ffmpeg -version` and `ffprobe -version`; if available, execute the existing generated-fixture preprocessing smoke path without a live OpenAI call.
- [ ] **Step 9: Audit the result:** `git diff --check`, `git status --short`, and `rg -n "Authorization|Bearer |sk-[A-Za-z0-9]" src tests README.md docs` to confirm no credential/header leakage.
- [ ] **Step 10: Use `superpowers:verification-before-completion`** and report the exact test, compile, startup, FFmpeg, and audit outputs before claiming completion.
- [ ] **Step 11: Commit** with `git add src/voice_transcriptor/app.py tests/ui/test_app.py README.md && git commit -m "docs: wire and validate OpenAI transcription"`.

168
docs/superpowers/plans/2026-08-30-transcription-desktop-milestone-1.md

@ -1,168 +0,0 @@
# Audio Transcription Desktop Milestone 1 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Deliver a Windows desktop GUI that selects and inspects media, manages secure settings, and exposes a clearly stubbed transcription action.
**Architecture:** A `src/voice_transcriptor` package separates typed models and services from PySide6 UI modules. FFprobe and settings logic are synchronous, testable services; the UI runs media probing in a Qt thread pool and ignores stale results.
**Tech Stack:** Python 3.12+, PySide6, keyring, OpenAI Python SDK, pytest, pytest-qt, FFmpeg/ffprobe.
**Spec:** `docs/superpowers/specs/2026-08-29-transcription-desktop-milestone-1-design.md`
## Global Constraints
- Target Windows 11 and Python 3.12 or newer.
- Store API keys only through `keyring`; never serialize or log them.
- Default language is `pt-BR`.
- Transcription remains stubbed in milestone 1.
- Keep UI imports out of business services and retain PyInstaller-compatible entry points.
- Preserve the empty tracked `README.md` by filling it with project documentation.
---
## File Map
- `pyproject.toml`: package metadata, entry point, and pytest settings.
- `requirements.txt`: runtime and test dependencies for straightforward Windows setup.
- `.gitignore`: Python, test, environment, editor, and PyInstaller artifacts.
- `src/voice_transcriptor/models.py`: immutable shared data values.
- `src/voice_transcriptor/formatting.py`: file-size and duration formatting.
- `src/voice_transcriptor/services/media_probe.py`: tool discovery, file validation, ffprobe execution and parsing.
- `src/voice_transcriptor/services/settings.py`: defaults and atomic non-secret JSON persistence.
- `src/voice_transcriptor/services/credentials.py`: keyring-only API-key access.
- `src/voice_transcriptor/services/transcription.py`: explicit milestone stub.
- `src/voice_transcriptor/ui/settings_dialog.py`: settings form and validation.
- `src/voice_transcriptor/ui/main_window.py`: selection/drop UI, async probing, metadata and status presentation.
- `src/voice_transcriptor/app.py`, `__main__.py`: dependency wiring and executable entry points.
- `tests/`: focused unit tests and an offscreen Qt smoke test.
- `README.md`: Windows setup, execution, security, testing, and milestone limitations.
### Task 1: Typed core, formatting, and project scaffolding
**Files:**
- Create: `pyproject.toml`, `requirements.txt`, `.gitignore`
- Create: `src/voice_transcriptor/__init__.py`, `src/voice_transcriptor/models.py`, `src/voice_transcriptor/formatting.py`
- Test: `tests/test_formatting.py`
**Interfaces:**
- Produces: `AppSettings(model: str, language: str, output_directory: Path)`, `ToolStatus(ffmpeg_path: Path | None, ffprobe_path: Path | None)`, `MediaInfo(path: Path, size_bytes: int, duration_seconds: float | None, audio_codec: str | None)`, `format_file_size(int) -> str`, and `format_duration(float | None) -> str`.
- [ ] **Step 1: Write failing formatting tests** for byte units, multi-hour duration, and unavailable duration.
- [ ] **Step 2: Run `python -m pytest tests/test_formatting.py -v`** and confirm collection/import failure.
- [ ] **Step 3: Add package configuration, typed dataclasses, and minimal pure formatting functions.** Size uses binary units and duration renders `HH:MM:SS`; unavailable duration renders `Unknown`.
- [ ] **Step 4: Run `python -m pytest tests/test_formatting.py -v`** and confirm all cases pass.
- [ ] **Step 5: Commit with `git commit -m "build: scaffold typed application core"`.**
### Task 2: Media inspection service
**Files:**
- Create: `src/voice_transcriptor/services/__init__.py`, `src/voice_transcriptor/services/media_probe.py`
- Test: `tests/test_media_probe.py`
**Interfaces:**
- Consumes: `MediaInfo`, `ToolStatus`.
- Produces: `detect_tools() -> ToolStatus`, `validate_media_path(path: Path) -> Path`, `parse_probe_output(path: Path, size_bytes: int, payload: str) -> MediaInfo`, `MediaProbeService.probe(path: Path) -> MediaInfo`, and typed `MediaProbeError` subclasses.
- [ ] **Step 1: Write failing tests** for executable discovery, path rejection, audio-only and video-with-audio JSON, missing fields, malformed JSON, nonzero subprocess result, and timeout mapping.
- [ ] **Step 2: Run `python -m pytest tests/test_media_probe.py -v`** and confirm import failure.
- [ ] **Step 3: Implement safe probing** with `subprocess.run([...], capture_output=True, text=True, timeout=30, check=False, creationflags=CREATE_NO_WINDOW on Windows)` and no shell.
- [ ] **Step 4: Run the focused tests** and confirm they pass.
- [ ] **Step 5: Commit with `git commit -m "feat: add ffprobe media inspection"`.**
### Task 3: Settings and credential persistence
**Files:**
- Create: `src/voice_transcriptor/services/settings.py`, `src/voice_transcriptor/services/credentials.py`
- Test: `tests/test_settings.py`, `tests/test_credentials.py`
**Interfaces:**
- Consumes: `AppSettings`.
- Produces: `SettingsRepository(path: Path | None = None)`, `.load() -> tuple[AppSettings, str | None]`, `.save(settings: AppSettings) -> None`; `CredentialService(backend=keyring)`, `.get_api_key() -> str | None`, `.has_api_key() -> bool`, `.set_api_key(value: str) -> None`; `SettingsError`, `CredentialError`.
- [ ] **Step 1: Write failing tests** for defaults, round trip, malformed recovery warning, atomic replacement, absence of API-key fields, and mocked keyring calls/failures.
- [ ] **Step 2: Run both focused test files** and confirm import failure.
- [ ] **Step 3: Implement per-user configuration** under `%APPDATA%/VoiceTranscriptor/settings.json`, using UTF-8 JSON and `Path.replace` from a sibling temporary file.
- [ ] **Step 4: Implement the keyring adapter** with stable service `voice-transcriptor` and account `openai-api-key`; reject blank keys and sanitize exceptions.
- [ ] **Step 5: Run focused tests** and confirm they pass.
- [ ] **Step 6: Commit with `git commit -m "feat: persist settings and credentials securely"`.**
### Task 4: Transcription boundary
**Files:**
- Create: `src/voice_transcriptor/services/transcription.py`
- Test: `tests/test_transcription.py`
**Interfaces:**
- Produces: `TranscriptionNotImplementedError` and `TranscriptionService.transcribe(media: MediaInfo, settings: AppSettings) -> None`.
- [ ] **Step 1: Write a failing test** asserting the service raises the specific milestone exception with a user-readable message.
- [ ] **Step 2: Run the focused test** and confirm import failure.
- [ ] **Step 3: Implement the minimal typed stub.**
- [ ] **Step 4: Run the focused test** and confirm it passes.
- [ ] **Step 5: Commit with `git commit -m "feat: define transcription service boundary"`.**
### Task 5: Settings dialog
**Files:**
- Create: `src/voice_transcriptor/ui/__init__.py`, `src/voice_transcriptor/ui/settings_dialog.py`
- Test: `tests/ui/test_settings_dialog.py`
**Interfaces:**
- Consumes: `AppSettings`, `SettingsRepository`, `CredentialService`.
- Produces: `SettingsDialog(settings, repository, credentials, parent=None)` and `settings_saved(AppSettings)` signal.
- [ ] **Step 1: Write offscreen Qt tests** for populated defaults, masked key entry, preserving an existing credential when blank, saving a supplied key, output-directory validation, and no secret in widget/log output after save.
- [ ] **Step 2: Run the focused UI tests** and confirm import failure.
- [ ] **Step 3: Build a native `QDialog` form** with editable model/language, folder picker, password echo mode, Save/Cancel buttons, and concise message boxes on service failure.
- [ ] **Step 4: Run the focused tests** and confirm they pass.
- [ ] **Step 5: Commit with `git commit -m "feat: add secure settings dialog"`.**
### Task 6: Main window and asynchronous probe flow
**Files:**
- Create: `src/voice_transcriptor/ui/main_window.py`
- Test: `tests/ui/test_main_window.py`
**Interfaces:**
- Consumes all service interfaces, formatting helpers, and shared models.
- Produces: `MainWindow(media_probe, settings_repository, credentials, transcription_service)` with `select_file(path: Path) -> None`; internal `ProbeSignals` and `ProbeWorker(QRunnable)` communicate results without blocking.
- [ ] **Step 1: Write offscreen Qt tests** for the visible controls, supported single-file drops, rejected URLs/multiple files, populated metadata, missing-tool status, stale probe suppression, settings opening, validation on Transcribe, and stub status logging.
- [ ] **Step 2: Run focused UI tests** and confirm import failure.
- [ ] **Step 3: Implement the native main-window layout** with a drop frame, Browse and Settings buttons, four metadata rows, tool status, prominent Transcribe button, and read-only log.
- [ ] **Step 4: Implement QRunnable probing** tagged with monotonically increasing request IDs; callbacks update only when the ID matches the current selection.
- [ ] **Step 5: Implement button actions and readable error handling** without logging credentials or full tracebacks in the GUI.
- [ ] **Step 6: Run focused UI tests** and confirm they pass.
- [ ] **Step 7: Commit with `git commit -m "feat: add media selection desktop interface"`.**
### Task 7: Application wiring and startup smoke test
**Files:**
- Create: `src/voice_transcriptor/app.py`, `src/voice_transcriptor/__main__.py`
- Test: `tests/ui/test_app.py`
**Interfaces:**
- Produces: `create_main_window() -> MainWindow` and `main() -> int`; console entry point `voice-transcriptor`.
- [ ] **Step 1: Write a failing offscreen smoke test** that constructs, shows, processes events for, and closes the window.
- [ ] **Step 2: Run the focused test** and confirm import failure.
- [ ] **Step 3: Wire concrete services and add guarded startup error reporting.** Reuse an existing `QApplication` in tests and create one in `main()`.
- [ ] **Step 4: Run the smoke test and `python -m compileall -q src tests`.**
- [ ] **Step 5: Commit with `git commit -m "feat: wire desktop application startup"`.**
### Task 8: Documentation and full validation
**Files:**
- Modify: `README.md`, `requirements.txt`
**Interfaces:**
- Documents the `python -m voice_transcriptor` and `voice-transcriptor` entry points and environment setup.
- [ ] **Step 1: Write README instructions** for Python 3.12 virtual environments, dependency installation, FFmpeg installation/PATH checks, running, testing, Windows Credential Manager security, known milestone limitations, and future PyInstaller packaging.
- [ ] **Step 2: Install dependencies with `python -m pip install -r requirements.txt`** when they are not already available.
- [ ] **Step 3: Run `python -m pytest -v` and `python -m compileall -q src tests`.**
- [ ] **Step 4: Run `ffmpeg -version` and `ffprobe -version`** and record availability without treating absence as a test failure.
- [ ] **Step 5: Launch an offscreen controlled startup** that opens the main window, processes events, and closes it via a timer.
- [ ] **Step 6: Review `git diff --check`, `git status --short`, and scan tracked project text for accidental API-key-like values.**
- [ ] **Step 7: Commit with `git commit -m "docs: add Windows setup and validation guide"`.**

70
docs/superpowers/specs/2026-08-30-audio-preprocessing-chunking-design.md

@ -1,70 +0,0 @@
# Audio Preprocessing and Chunking Design
## Goal and Scope
Add a cancellable, duration-based FFmpeg pipeline for `m4a`, `mp3`, `wav`, `mp4`, `mov`, `webm`, and `mkv`. It converts speech into compact overlapping chunks, records exact source timestamps, and exposes settings and progress through the desktop GUI. Transcription API calls remain out of scope.
The repository currently lacks its planned main window and bootstrap, so this change also adds the minimal host UI needed to select a recording, start preprocessing outside the GUI thread, show progress, and cancel a job.
## Architecture
- `models.py` defines preprocessing settings and immutable chunk, manifest, and progress values.
- `services/chunking.py` contains pure validation, boundary, overlap, duration, and offset calculations.
- `services/preprocessing.py` owns job directories, FFprobe validation, incremental FFmpeg execution, progress parsing, cancellation, manifests, and cleanup.
- `services/settings.py` and `ui/settings_dialog.py` persist and edit the advanced values.
- `ui/main_window.py` adapts service callbacks to Qt signals and runs preprocessing on a worker thread.
- `app.py` and `__main__.py` wire the runnable application.
Services never import PySide6.
## Inspection and Validation
Every job uses the existing shell-free FFprobe service. Its query is extended to identify audio and video streams. Preprocessing rejects unreadable paths, unsupported extensions, missing tools, sources without audio, unavailable or non-positive durations, non-positive chunk duration, negative overlap, and overlap greater than or equal to chunk duration. Typed service errors expose sanitized GUI messages.
## Exact Chunk Math
Defaults are `900` seconds per chunk and `15` seconds overlap. Chunking is duration-based and assumes no universal upload-size limit.
For source duration `D`, chunk duration `C`, and overlap `O`, the stride is `C - O`. The first chunk is `[0, min(C, D)]`. Each next chunk begins one stride after the previous start and ends at `min(start + C, D)`. Generation stops when a chunk reaches `D`. Thus adjacent full chunks share exactly `O`, and the last end equals `D`.
Calculations use `Decimal` values derived from FFprobe strings to prevent accumulated binary-float drift. JSON timestamps are decimal seconds serialized as strings. Each chunk records its zero-based index, relative path, exact source start/end, and duration. A future local transcription timestamp maps to the source timeline by adding the chunk's source start.
## Incremental FFmpeg Processing
One FFmpeg process is launched per boundary. It selects `0:a:0`, disables video/subtitle/data output, seeks to the exact start, limits output to the exact duration, downmixes to mono, resamples to 24 kHz, and encodes AAC-LC at 64 kbps in `.m4a`. Machine-readable `-progress pipe:1 -nostats` output drives progress.
Mono AAC-LC at 24 kHz and 64 kbps retains strong speech-recognition quality while keeping a 15-minute chunk near 7.2 MB. Video audio goes directly into each chunk without a large intermediate file. FFmpeg streams input and output; Python retains only progress lines and metadata, so multi-hour sources do not enter RAM.
## Manifest and Temporary Lifecycle
Each run creates a unique `voice-transcriptor-<job-id>-*` system-temporary directory containing `manifest.json` and `chunks/`. The manifest records schema/job identifiers, source metadata, effective settings, output codec settings, state (`running`, `completed`, `cancelled`, or `failed`), every planned chunk and timestamp, completed count, and an optional sanitized failure message.
The manifest is atomically written before chunking and after every chunk or terminal transition. It never contains credentials.
The service returns a context-managed job result. The GUI closes it after success, cancellation, or failure. Cleanup removes only the exact verified job directory unless `retain_temporary_files` is enabled. Retained paths are logged for debugging; cleanup failures are reported because they may leave material data behind.
## Cancellation and Progress
A thread-safe token is checked before probing, before every chunk, and while consuming FFmpeg progress. Cancellation terminates the active process, waits briefly, and kills it only if necessary. It prevents later chunks, marks the manifest `cancelled`, and removes partial current output when possible.
Progress carries percentage, phase, and message. FFmpeg `out_time` is combined with completed planned work, remains monotonic, and reaches 100 only after the completed manifest is written. The Qt worker emits progress, completion, cancellation, and sanitized failure signals. The window disables conflicting controls during a job, enables Cancel, restores controls on all terminal paths, and requests cancellation when closing.
## Settings and GUI
`AppSettings` gains `chunk_duration_seconds=900`, `chunk_overlap_seconds=15`, and `retain_temporary_files=false`. Older settings files load these defaults; invalid persisted values use the existing warning/fallback behavior. Saving stays atomic and excludes API keys.
An initially collapsed Advanced group provides integer-second duration/overlap controls and a retain-files checkbox. Validation requires overlap smaller than duration.
The main window provides supported-format selection, metadata, a preprocessing action, progress bar, Cancel button, and ordered status messages. Completion reports prepared chunks without invoking the transcription stub or any API.
## Testing and Verification
Pure tests cover sub-chunk sources, exact/fractional final boundaries, multi-hour duration, default and zero overlap, total planned duration including overlaps, source timestamp offsets, and invalid values.
Service tests use controlled process fakes to verify commands, incremental progress, manifest transitions, cancellation, and cleanup without requiring FFmpeg. Probe tests cover audio/video metadata and missing audio. Settings and Qt tests cover defaults, round trips, advanced validation, progress, controls, and cancellation.
Final verification runs the entire pytest suite, Python compilation, and whitespace checks. If FFmpeg and FFprobe are installed, a generated fixture is processed end-to-end; missing binaries are reported separately rather than failing portable unit tests.
## Success Criteria
Supported inputs are inspected and converted incrementally into overlapping speech-oriented chunks; each chunk has exact source timestamps in an atomic manifest; multi-hour operation is memory-bounded; cleanup honors configuration; GUI progress and cancellation do not block; advanced settings persist; no transcription API is called; and all tests pass.

125
docs/superpowers/specs/2026-08-30-openai-transcription-design.md

@ -1,125 +0,0 @@
# OpenAI Transcription and Resumable Jobs Design
## Goal and Scope
Add production-quality transcription of preprocessed chunks through the current OpenAI Audio Transcriptions API. Jobs must survive crashes and network loss, resume without repeating successful work, remain cancellable, and expose useful progress and sanitized API failures in the desktop GUI.
Each chunk is submitted independently. Completed chunk transcripts are concatenated in source order without overlap deduplication. Translation, diarization, overlap reconciliation, and word-level timestamp inference are out of scope.
## Verified OpenAI Interface
The implementation targets the installed OpenAI Python SDK 3.6.0 interface verified on 2026-08-30:
```python
client.audio.transcriptions.create(
file=audio_file,
model=model,
language="pt",
prompt=prompt,
response_format="json",
)
```
The method accepts arbitrary model strings and returns a transcription object with a `text` value for JSON responses. The SDK exposes distinct `RateLimitError`, `APIConnectionError`, `APITimeoutError`, and `APIStatusError` classes.
The application default is the explicitly requested `gpt-transcribe`. The editable selector is seeded with `gpt-transcribe`, `gpt-4o-transcribe`, and `gpt-4o-mini-transcribe`, while allowing future compatible identifiers. Current official OpenAI documentation lists the two GPT-4o transcription models but does not list `gpt-transcribe`; therefore rejection of that identifier is treated as a permanent, clearly displayed API error rather than silently substituting a model.
The app uses the transcription endpoint, never the translation endpoint. It sends `language="pt"`, the ISO-639-1 code documented by the API, while the prompt explicitly requests Brazilian Portuguese conventions.
## Architecture
- `models.py` extends application settings with an optional context/vocabulary string.
- `services/settings.py` persists the non-secret model, language, output directory, preprocessing values, and context.
- `services/credentials.py` reads and updates the existing Windows Credential Manager generic credential whose target name is exactly `OPENAI_API_KEY`.
- `services/job_manifest.py` owns schema validation, atomic manifest writes, resumable state transitions, and transcript assembly.
- `services/transcription.py` adapts the OpenAI SDK, classifies failures, implements bounded exponential backoff, and orchestrates one request per unfinished chunk.
- `services/preprocessing.py` creates durable job directories inside the configured output directory and produces chunk records suitable for transcription.
- `ui/settings_dialog.py` provides an editable model selector and multiline Context / Vocabulary field.
- `ui/main_window.py` runs the job service in a Qt worker and displays progress, cancellation, elapsed time, current chunk, and API errors.
- `app.py` wires the credential, OpenAI client, preprocessing, manifest, and transcription services.
Services do not import PySide6. Network calls, retries, waits, manifest I/O, and preprocessing run outside the GUI thread.
## API Credential Integration
The OpenAI API key already exists in Windows Credential Manager under the generic credential target name `OPENAI_API_KEY`. The credential service uses the Windows `keyring` backend's credential lookup for that exact target and accepts the username stored with the matching credential; it does not require or assume the previous `voice-transcriptor` / `openai-api-key` service-account pair.
Saving a replacement key updates the same `OPENAI_API_KEY` target while preserving the credential's existing username when one is available. If no matching credential exists, the settings dialog reports that the API key is not configured and may create the target using `OPENAI_API_KEY` as the stable username. The key is passed directly to the OpenAI client and is never copied into application settings, job manifests, GUI logs, exception messages, or transcript files.
## Durable Job Layout and Manifest
Every run has a stable directory below `<output_directory>/voice-transcriptor-jobs/<job-id>/`:
```text
manifest.json
chunks/chunk-00000.m4a
chunks/chunk-00001.m4a
transcript.txt
```
The manifest remains the source of truth and is replaced atomically after every state transition. It contains no API key, authorization header, or sensitive request metadata. Schema version 2 records job identity and overall state; canonical source metadata; effective preprocessing and transcription settings; timestamps; and each chunk's index, relative audio path, exact source timing, status, attempt count, transcript, sanitized last error, and completion timestamp.
Chunk statuses are `pending`, `processing`, `completed`, and `failed`. The overall job may additionally be `preprocessing`, `transcribing`, `completed`, `cancelled`, or `failed`.
A successfully encoded chunk is recorded immediately. A successfully transcribed chunk is changed to `completed`, its transcript is stored, counters are updated, and the manifest is atomically replaced before the next API call. `transcript.txt` is then atomically regenerated from all completed chunk texts in index order. This ordering ensures a crash cannot make an unrecorded success appear complete.
## Resume Semantics
Starting a job for a source first looks for an existing non-complete manifest whose source identity and effective settings match. The GUI may also resume a retained job directly from its manifest.
On resume:
- `completed` chunks are never submitted again;
- interrupted `processing` chunks return to `pending` because no durable successful response exists;
- `failed` chunks return to `pending` for an explicit user resume action;
- existing valid encoded chunk files are reused;
- missing or invalid encoded files are regenerated without affecting completed transcript records;
- an explicit retranscribe action is the only operation allowed to clear completed transcript state.
Cancellation stops before the next request or retry, marks the overall job `cancelled`, and retains the durable directory. Cancellation during an active synchronous SDK request takes effect immediately after that request returns; the UI remains responsive because the call is on a worker thread.
## Prompting and Language Preservation
The effective prompt combines a fixed Brazilian-Portuguese instruction with optional user context:
> Conversa em português brasileiro. Preserve a língua falada; não traduza. Preserve ortografia e pontuação brasileiras, números, nomes próprios, termos técnicos e siglas com máxima fidelidade.
When supplied, the Context / Vocabulary value follows this instruction. It is trimmed, user-editable, and persisted as non-secret configuration. The default example is:
> Brazilian Portuguese conversation. Preserve Brazilian spelling and punctuation. Vocabulary may include AWS, Kubernetes, OpenAI, PostgreSQL, Brasília, Banco do Brasil.
No previous chunk transcript is injected into a later request in this milestone, keeping chunks independent and avoiding accidental propagation of errors.
## Retry and Error Classification
The OpenAI client is configured with SDK retries disabled so application behavior is deterministic and visible. The job service uses cancellation-aware exponential backoff with jitter and a finite maximum attempt count.
Retryable failures include rate limits, connection failures, timeouts, HTTP 408, HTTP 409, HTTP 429, and HTTP 5xx responses. A server-provided `Retry-After` delay is honored when available, subject to a reasonable cap. The GUI reports rate limiting and the next retry without exposing headers.
Permanent failures include authentication and permission errors, invalid requests, unsupported models, missing files, and other non-retryable 4xx responses. They immediately mark the current chunk `failed` and the job `failed`. Retries never continue indefinitely.
All logged errors use exception type, status code when safe, request ID when available, chunk index, and a sanitized message. API keys, authorization values, request headers, and raw SDK request objects are never logged or included in manifests.
## Progress and GUI Behavior
The main action prepares and transcribes the recording as one worker-owned job. Visible state includes chunks completed / total, elapsed wall-clock time, current chunk, phase, retry/rate-limit notices, and sanitized API errors.
The progress bar reflects completed chunks during transcription and preprocessing percentage before API work begins. Conflicting file/settings actions are disabled while active. Cancel remains available. All worker terminal paths restore controls, and closing the window requests cancellation.
The settings dialog uses an editable combo box for model selection and a multiline Context / Vocabulary editor. The language remains editable for future languages, defaults to `pt-BR` in presentation and persistence, and is normalized to `pt` for the API when Brazilian Portuguese is selected.
## Transcript Timing Metadata
Every chunk transcript retains the preprocessing manifest's exact decimal `source_start_seconds`, `source_end_seconds`, and `duration_seconds`. These values identify the source interval covered by the raw text. The plain `transcript.txt` contains text only, separated by newlines in chunk order; timing-rich data remains in `manifest.json` for future structured export and overlap deduplication.
## Testing and Validation
Development follows test-driven cycles. Unit tests use a fake transcription endpoint and deterministic sleeper/random sources; they never require a real API key or network access.
Coverage includes exact SDK arguments and independent chunk calls; lookup and update of the exact `OPENAI_API_KEY` Windows credential target; Brazilian language/prompt composition; arbitrary model identifiers; retry, rate-limit, exhaustion, and permanent failures; atomic persistence; crash recovery and resume skipping; cancellation; timing retention and ordered concatenation; error sanitization; GUI worker progress; and settings migration/UI behavior.
Final validation runs the complete pytest suite, Python compilation, whitespace checks, an offscreen GUI startup probe, and an optional local FFmpeg preprocessing smoke test when tools are installed. No live OpenAI request is required for automated validation.
## Success Criteria
The application preprocesses and independently transcribes every unfinished chunk through the verified OpenAI Python SDK interface; defaults to the requested model and Brazilian-Portuguese guidance; persists each success and exact source timing immediately; resumes without retranscribing completed chunks; uses finite, graceful retry behavior; remains responsive and cancellable; exposes complete job progress and safe errors; concatenates raw chunk text without deduplication; and passes all automated and startup validation.

27
pyproject.toml

@ -1,27 +0,0 @@
[build-system]
requires = ["setuptools>=75"]
build-backend = "setuptools.build_meta"
[project]
name = "voice-transcriptor"
version = "0.1.0"
description = "Windows desktop media transcription client"
requires-python = ">=3.12"
dependencies = [
"openai>=1.0",
"PySide6>=6.7",
"keyring>=25.0",
]
[project.scripts]
voice-transcriptor = "voice_transcriptor.app:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
addopts = "-ra"
testpaths = ["tests"]
pythonpath = ["src"]
qt_api = "pyside6"

6
requirements.txt

@ -1,6 +0,0 @@
openai>=1.0
PySide6>=6.7
keyring>=25.0
pytest>=8.0
pytest-qt>=4.4

4
src/voice_transcriptor/__init__.py

@ -1,4 +0,0 @@
"""Voice Transcriptor desktop application."""
__version__ = "0.1.0"

3
src/voice_transcriptor/__main__.py

@ -1,3 +0,0 @@
from voice_transcriptor.app import main
raise SystemExit(main())

40
src/voice_transcriptor/app.py

@ -1,40 +0,0 @@
from __future__ import annotations
import sys
from PySide6.QtWidgets import QApplication
from openai import OpenAI
from voice_transcriptor.services.credentials import CredentialService
from voice_transcriptor.services.media_probe import MediaProbeService, detect_tools
from voice_transcriptor.services.preprocessing import PreprocessingService
from voice_transcriptor.services.job_manifest import JobManifestRepository
from voice_transcriptor.services.settings import SettingsRepository
from voice_transcriptor.services.transcription import OpenAITranscriptionClient, TranscriptionService
from voice_transcriptor.ui.main_window import MainWindow
def create_main_window() -> MainWindow:
tools = detect_tools()
probe = MediaProbeService(tools.ffprobe_path) if tools.ffprobe_path else _UnavailableProbe()
manifests = JobManifestRepository()
preprocessing = PreprocessingService(tools.ffmpeg_path, probe, manifest_repository=manifests) if tools.ffmpeg_path else _UnavailablePreprocessing()
def transcription_factory(api_key: str) -> TranscriptionService:
client = OpenAI(api_key=api_key, max_retries=0)
return TranscriptionService(OpenAITranscriptionClient(client), manifests)
return MainWindow(probe, preprocessing, SettingsRepository(), CredentialService(), transcription_factory)
class _UnavailableProbe:
def probe(self, path): raise RuntimeError("FFprobe is required to inspect media.")
class _UnavailablePreprocessing:
def preprocess(self, *args, **kwargs): raise RuntimeError("FFmpeg is required to preprocess media.")
def main() -> int:
application = QApplication.instance() or QApplication(sys.argv)
window = create_main_window(); window.show(); return application.exec()

20
src/voice_transcriptor/formatting.py

@ -1,20 +0,0 @@
def format_file_size(size_bytes: int) -> str:
if size_bytes < 0:
raise ValueError("File size cannot be negative.")
if size_bytes < 1024:
return f"{size_bytes} B"
value = float(size_bytes)
for unit in ("KiB", "MiB", "GiB", "TiB"):
value /= 1024
if value < 1024 or unit == "TiB":
return f"{value:.2f} {unit}"
raise AssertionError("unreachable")
def format_duration(duration_seconds: float | None) -> str:
if duration_seconds is None:
return "Unknown"
seconds = max(0, round(duration_seconds))
hours, remainder = divmod(seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"

52
src/voice_transcriptor/models.py

@ -1,52 +0,0 @@
from dataclasses import dataclass
from decimal import Decimal
from pathlib import Path
DEFAULT_CONTEXT = (
"Brazilian Portuguese conversation. Preserve Brazilian spelling and punctuation. "
"Vocabulary may include AWS, Kubernetes, OpenAI, PostgreSQL, Brasília, Banco do Brasil."
)
@dataclass(frozen=True, slots=True)
class AppSettings:
model: str
language: str
output_directory: Path
chunk_duration_seconds: int = 900
chunk_overlap_seconds: int = 15
retain_temporary_files: bool = False
context_vocabulary: str = DEFAULT_CONTEXT
@dataclass(frozen=True, slots=True)
class ToolStatus:
ffmpeg_path: Path | None
ffprobe_path: Path | None
@property
def available(self) -> bool:
return self.ffmpeg_path is not None and self.ffprobe_path is not None
@dataclass(frozen=True, slots=True)
class MediaInfo:
path: Path
size_bytes: int
duration_seconds: float | None
audio_codec: str | None
duration_text: str | None = None
has_audio: bool = False
has_video: bool = False
@dataclass(frozen=True, slots=True)
class ChunkBoundary:
index: int
start_seconds: Decimal
end_seconds: Decimal
@property
def duration_seconds(self) -> Decimal:
return self.end_seconds - self.start_seconds

2
src/voice_transcriptor/services/__init__.py

@ -1,2 +0,0 @@
"""Application services independent from the GUI."""

54
src/voice_transcriptor/services/chunking.py

@ -1,54 +0,0 @@
from __future__ import annotations
from decimal import Decimal, InvalidOperation
from typing import Iterable
from voice_transcriptor.models import ChunkBoundary
def _decimal(value: Decimal | str | int | float) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError) as exc:
raise ValueError("Time values must be finite decimal numbers.") from exc
if not result.is_finite():
raise ValueError("Time values must be finite decimal numbers.")
return result
def calculate_chunk_boundaries(
total_duration: Decimal | str | int | float,
chunk_duration: Decimal | str | int | float = Decimal("900"),
overlap: Decimal | str | int | float = Decimal("15"),
) -> tuple[ChunkBoundary, ...]:
total = _decimal(total_duration)
chunk = _decimal(chunk_duration)
shared = _decimal(overlap)
if total <= 0 or chunk <= 0:
raise ValueError("Duration and chunk duration must be greater than zero.")
if shared < 0 or shared >= chunk:
raise ValueError("Overlap must be non-negative and smaller than chunk duration.")
result: list[ChunkBoundary] = []
start = Decimal("0")
stride = chunk - shared
while start < total:
end = min(start + chunk, total)
result.append(ChunkBoundary(len(result), start, end))
if end == total:
break
start += stride
return tuple(result)
def total_chunk_duration(boundaries: Iterable[ChunkBoundary]) -> Decimal:
return sum((item.duration_seconds for item in boundaries), Decimal("0"))
def source_timestamp(
boundary: ChunkBoundary, local_seconds: Decimal | str | int | float
) -> Decimal:
local = _decimal(local_seconds)
if local < 0 or local > boundary.duration_seconds:
raise ValueError("Local timestamp is outside the chunk.")
return boundary.start_seconds + local

39
src/voice_transcriptor/services/credentials.py

@ -1,39 +0,0 @@
"""Secure API-key storage backed by the operating system keyring."""
from __future__ import annotations
import keyring
TARGET_NAME = "OPENAI_API_KEY"
class CredentialError(Exception):
"""A credential operation failed without exposing backend details."""
class CredentialService:
"""Read and write the application's API key through a keyring backend."""
def __init__(self, backend=keyring) -> None:
self._backend = backend
def get_api_key(self) -> str | None:
try:
credential = self._backend.get_credential(TARGET_NAME, None)
return credential.password if credential is not None else None
except Exception:
raise CredentialError("Unable to access the saved API key.") from None
def has_api_key(self) -> bool:
return bool(self.get_api_key())
def set_api_key(self, value: str) -> None:
if not isinstance(value, str) or not value.strip():
raise CredentialError("An API key is required.")
try:
credential = self._backend.get_credential(TARGET_NAME, None)
username = credential.username if credential is not None else TARGET_NAME
self._backend.set_password(TARGET_NAME, username, value)
except Exception:
raise CredentialError("Unable to save the API key.") from None

199
src/voice_transcriptor/services/job_manifest.py

@ -1,199 +0,0 @@
from __future__ import annotations
import json
from datetime import datetime, timezone
from enum import StrEnum
from pathlib import Path, PurePosixPath
from typing import Iterable, Sequence
class JobManifestError(RuntimeError):
pass
class ChunkStatus(StrEnum):
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
class JobManifestRepository:
def create(
self,
job_directory: Path,
source: dict,
settings: dict,
chunks: Sequence[dict],
) -> dict:
job_directory = job_directory.resolve()
job_directory.mkdir(parents=True, exist_ok=True)
(job_directory / "chunks").mkdir(exist_ok=True)
timestamp = _now()
items = [self._new_chunk(item) for item in chunks]
manifest_path = job_directory / "manifest.json"
manifest = {
"schema_version": 2,
"job_id": job_directory.name,
"state": "preprocessing",
"source": dict(source),
"settings": dict(settings),
"created_at": timestamp,
"updated_at": timestamp,
"started_at": None,
"completed_at": None,
"chunks": items,
"completed_chunks": 0,
"total_chunks": len(items),
"error": None,
}
self.save(manifest_path, manifest)
return {**manifest, "manifest_path": str(manifest_path)}
def load(self, path: Path) -> dict:
try:
manifest = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise JobManifestError("Could not load the job manifest.") from exc
self._validate(manifest)
return manifest
def save(self, path: Path, manifest: dict) -> None:
self._validate(manifest)
manifest["updated_at"] = _now()
temporary = path.with_suffix(".tmp")
try:
temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
temporary.replace(path)
except OSError as exc:
raise JobManifestError("Could not save the job manifest.") from exc
def mark_processing(self, path: Path, index: int) -> dict:
manifest = self.load(path)
item = self._item(manifest, index)
if item["status"] == ChunkStatus.COMPLETED:
raise JobManifestError("A completed chunk cannot be processed again.")
item["status"] = ChunkStatus.PROCESSING
item["attempt_count"] += 1
item["last_error"] = None
if manifest["started_at"] is None:
manifest["started_at"] = _now()
manifest["state"] = "transcribing"
self.save(path, manifest)
return manifest
def mark_completed(self, path: Path, index: int, text: str) -> dict:
manifest = self.load(path)
item = self._item(manifest, index)
item["status"] = ChunkStatus.COMPLETED
item["transcript"] = text
item["last_error"] = None
item["completed_at"] = _now()
self._recount(manifest)
self.save(path, manifest)
return manifest
def mark_failed(self, path: Path, index: int, message: str) -> dict:
manifest = self.load(path)
item = self._item(manifest, index)
if item["status"] == ChunkStatus.COMPLETED:
raise JobManifestError("A completed chunk cannot be marked failed.")
item["status"] = ChunkStatus.FAILED
item["last_error"] = self._sanitize(message)
manifest["state"] = "failed"
manifest["error"] = item["last_error"]
self.save(path, manifest)
return manifest
def mark_job_state(self, path: Path, state: str, error: str | None = None) -> dict:
manifest = self.load(path)
manifest["state"] = state
manifest["error"] = self._sanitize(error) if error else None
if state == "completed":
manifest["completed_at"] = _now()
self.save(path, manifest)
return manifest
def recover_for_resume(self, path: Path) -> dict:
manifest = self.load(path)
for item in manifest["chunks"]:
if item["status"] in (ChunkStatus.PROCESSING, ChunkStatus.FAILED):
item["status"] = ChunkStatus.PENDING
item["last_error"] = None
manifest["state"] = "transcribing"
manifest["error"] = None
self._recount(manifest)
self.save(path, manifest)
return manifest
def reset_completed(self, path: Path, indexes: Iterable[int] | None = None) -> dict:
manifest = self.load(path)
selected = set(indexes) if indexes is not None else {item["index"] for item in manifest["chunks"]}
for item in manifest["chunks"]:
if item["index"] in selected and item["status"] == ChunkStatus.COMPLETED:
item.update(status=ChunkStatus.PENDING, transcript=None, completed_at=None, last_error=None)
self._recount(manifest)
self.save(path, manifest)
return manifest
def assemble_transcript(self, manifest_path: Path) -> Path:
manifest = self.load(manifest_path)
texts = [
item["transcript"]
for item in sorted(manifest["chunks"], key=lambda value: value["index"])
if item["status"] == ChunkStatus.COMPLETED and item["transcript"] is not None
]
target = manifest_path.parent / "transcript.txt"
temporary = target.with_suffix(".tmp")
temporary.write_text("".join(f"{text}\n" for text in texts), encoding="utf-8")
temporary.replace(target)
return target
@staticmethod
def _new_chunk(source: dict) -> dict:
relative = PurePosixPath(str(source.get("path", "")))
if relative.is_absolute() or ".." in relative.parts or not relative.parts or relative.parts[0] != "chunks":
raise JobManifestError("Invalid chunk path in job manifest.")
return {
**source,
"path": relative.as_posix(),
"status": ChunkStatus.PENDING,
"attempt_count": 0,
"transcript": None,
"last_error": None,
"completed_at": None,
}
@staticmethod
def _validate(manifest: object) -> None:
if not isinstance(manifest, dict) or manifest.get("schema_version") != 2:
raise JobManifestError("Unsupported job manifest schema.")
chunks = manifest.get("chunks")
if not isinstance(chunks, list):
raise JobManifestError("Invalid chunks in job manifest.")
for item in chunks:
if not isinstance(item, dict):
raise JobManifestError("Invalid chunk in job manifest.")
relative = PurePosixPath(str(item.get("path", "")))
if relative.is_absolute() or ".." in relative.parts or not relative.parts or relative.parts[0] != "chunks":
raise JobManifestError("Invalid chunk path in job manifest.")
@staticmethod
def _item(manifest: dict, index: int) -> dict:
for item in manifest["chunks"]:
if item.get("index") == index:
return item
raise JobManifestError("Unknown chunk index.")
@staticmethod
def _recount(manifest: dict) -> None:
manifest["completed_chunks"] = sum(item["status"] == ChunkStatus.COMPLETED for item in manifest["chunks"])
@staticmethod
def _sanitize(message: str) -> str:
value = str(message).replace("\r", " ").replace("\n", " ")
return value[:500]

108
src/voice_transcriptor/services/media_probe.py

@ -1,108 +0,0 @@
from __future__ import annotations
import json
import shutil
import subprocess
from pathlib import Path
from typing import Any
from voice_transcriptor.models import MediaInfo, ToolStatus
class MediaProbeError(RuntimeError):
"""A media file could not be validated or inspected."""
class InvalidMediaPathError(MediaProbeError):
"""The selected path is not a readable local file."""
class ProbeOutputError(MediaProbeError):
"""FFprobe returned output that cannot be interpreted."""
class ProbeExecutionError(MediaProbeError):
"""FFprobe could not start or inspect the selected media."""
class ProbeTimeoutError(MediaProbeError):
"""FFprobe did not finish before its configured timeout."""
def detect_tools() -> ToolStatus:
ffmpeg = shutil.which("ffmpeg")
ffprobe = shutil.which("ffprobe")
return ToolStatus(Path(ffmpeg) if ffmpeg else None, Path(ffprobe) if ffprobe else None)
def validate_media_path(path: Path) -> Path:
path = path.expanduser()
if not path.exists():
raise InvalidMediaPathError("The selected file does not exist.")
if not path.is_file():
raise InvalidMediaPathError("The selected path is not a file.")
try:
path.stat()
except OSError as exc:
raise InvalidMediaPathError("The selected file cannot be read.") from exc
return path.resolve()
def parse_probe_output(path: Path, size_bytes: int, payload: str) -> MediaInfo:
try:
data: Any = json.loads(payload)
except (json.JSONDecodeError, TypeError) as exc:
raise ProbeOutputError("FFprobe returned invalid data.") from exc
if not isinstance(data, dict):
raise ProbeOutputError("FFprobe returned invalid data.")
format_data = data.get("format", {})
streams = data.get("streams", [])
if (
not isinstance(format_data, dict)
or not isinstance(streams, list)
or not all(isinstance(stream, dict) for stream in streams)
):
raise ProbeOutputError("FFprobe returned invalid data.")
duration: float | None = None
raw_duration = format_data.get("duration")
try:
if raw_duration is not None:
duration = float(raw_duration)
except (TypeError, ValueError):
duration = None
codec = next(
(stream.get("codec_name") for stream in streams if stream.get("codec_type") == "audio"),
None,
)
duration_text = str(raw_duration) if raw_duration is not None else None
return MediaInfo(
path=path, size_bytes=size_bytes, duration_seconds=duration, audio_codec=codec,
duration_text=duration_text, has_audio=codec is not None,
has_video=any(stream.get("codec_type") == "video" for stream in streams),
)
class MediaProbeService:
def __init__(self, ffprobe_path: Path, timeout_seconds: int = 30) -> None:
self.ffprobe_path = ffprobe_path
self.timeout_seconds = timeout_seconds
def probe(self, path: Path) -> MediaInfo:
media_path = validate_media_path(path)
command = [
str(self.ffprobe_path), "-v", "error", "-show_entries",
"format=duration:stream=codec_type,codec_name", "-of", "json", str(media_path),
]
startup_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
try:
result = subprocess.run(
command, capture_output=True, text=True, timeout=self.timeout_seconds,
check=False, creationflags=startup_flags,
)
except subprocess.TimeoutExpired as exc:
raise ProbeTimeoutError("FFprobe timed out while inspecting the file.") from exc
except OSError as exc:
raise ProbeExecutionError("FFprobe could not be started.") from exc
if result.returncode != 0:
raise ProbeExecutionError("FFprobe could not inspect this media file.")
return parse_probe_output(media_path, media_path.stat().st_size, result.stdout)

164
src/voice_transcriptor/services/preprocessing.py

@ -1,164 +0,0 @@
from __future__ import annotations
import json
import shutil
import subprocess
import tempfile
import threading
import uuid
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Callable
from voice_transcriptor.services.chunking import calculate_chunk_boundaries, total_chunk_duration
from voice_transcriptor.services.job_manifest import JobManifestRepository
SUPPORTED_EXTENSIONS = frozenset({".m4a", ".mp3", ".wav", ".mp4", ".mov", ".webm", ".mkv"})
class PreprocessingError(RuntimeError): pass
class PreprocessingCancelled(PreprocessingError): pass
@dataclass(frozen=True, slots=True)
class PreprocessingOptions:
chunk_duration_seconds: int = 900
overlap_seconds: int = 15
retain_temporary_files: bool = False
@dataclass(frozen=True, slots=True)
class PreprocessingProgress:
percent: int
phase: str
message: str
class CancellationToken:
def __init__(self) -> None: self._event = threading.Event()
def cancel(self) -> None: self._event.set()
@property
def cancelled(self) -> bool: return self._event.is_set()
def raise_if_cancelled(self) -> None:
if self.cancelled: raise PreprocessingCancelled("Preprocessing was cancelled.")
@dataclass(slots=True)
class PreprocessingResult:
job_directory: Path
manifest_path: Path
retained: bool
_identity: Path
def cleanup(self) -> None:
if self.retained or not self.job_directory.exists(): return
if self.job_directory.resolve() != self._identity:
raise PreprocessingError("Refusing to clean an unexpected temporary directory.")
shutil.rmtree(self.job_directory)
def __enter__(self): return self
def __exit__(self, *args): self.cleanup()
class PreprocessingService:
def __init__(self, ffmpeg_path: Path, probe_service, temporary_root: Path | None = None, process_factory=subprocess.Popen, manifest_repository: JobManifestRepository | None = None) -> None:
self.ffmpeg_path = ffmpeg_path
self.probe_service = probe_service
self.temporary_root = temporary_root
self.process_factory = process_factory
self.manifest_repository = manifest_repository or JobManifestRepository()
def preprocess(self, source: Path, options: PreprocessingOptions, token: CancellationToken | None = None, progress: Callable[[PreprocessingProgress], None] | None = None, durable_root: Path | None = None, transcription_settings: dict | None = None) -> PreprocessingResult:
token = token or CancellationToken()
source = source.resolve()
if source.suffix.lower() not in SUPPORTED_EXTENSIONS: raise PreprocessingError("Unsupported media format.")
token.raise_if_cancelled()
info = self.probe_service.probe(source)
if not info.has_audio: raise PreprocessingError("The selected media has no audio stream.")
raw_duration = info.duration_text if info.duration_text is not None else info.duration_seconds
try: duration = Decimal(str(raw_duration))
except (InvalidOperation, ValueError): raise PreprocessingError("The media duration is unavailable.")
boundaries = calculate_chunk_boundaries(duration, options.chunk_duration_seconds, options.overlap_seconds)
durable = durable_root is not None
if durable:
root = (durable_root / "voice-transcriptor-jobs").resolve()
root.mkdir(parents=True, exist_ok=True)
job = (root / f"voice-transcriptor-{uuid.uuid4().hex[:8]}").resolve()
job.mkdir()
else:
root = self.temporary_root
if root is not None: root.mkdir(parents=True, exist_ok=True)
job = Path(tempfile.mkdtemp(prefix=f"voice-transcriptor-{uuid.uuid4().hex[:8]}-", dir=root)).resolve()
chunks_dir = job / "chunks"; chunks_dir.mkdir()
manifest_path = job / "manifest.json"
chunk_items = [{"index": b.index, "path": f"chunks/chunk-{b.index:05d}.m4a", "source_start_seconds": str(b.start_seconds), "source_end_seconds": str(b.end_seconds), "duration_seconds": str(b.duration_seconds)} for b in boundaries]
manifest = {"schema_version": 1, "job_id": job.name, "state": "running", "source": {"path": str(source), "size_bytes": info.size_bytes, "duration_seconds": str(duration), "audio_codec": info.audio_codec, "has_video": info.has_video}, "settings": {"chunk_duration_seconds": options.chunk_duration_seconds, "overlap_seconds": options.overlap_seconds}, "output": {"codec": "aac", "bitrate": "64k", "sample_rate": 24000, "channels": 1, "container": "m4a"}, "chunks": chunk_items, "completed_chunks": 0}
if durable:
created = self.manifest_repository.create(
job,
manifest["source"],
{**manifest["settings"], **(transcription_settings or {})},
chunk_items,
)
manifest_path = Path(created["manifest_path"])
else:
self._write_manifest(manifest_path, manifest)
total_work = total_chunk_duration(boundaries)
completed = Decimal("0"); last_percent = 0
try:
for boundary, item in zip(boundaries, chunk_items):
token.raise_if_cancelled()
output = job / item["path"]
command = self._command(source, output, boundary.start_seconds, boundary.duration_seconds)
flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
process = self.process_factory(command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1, creationflags=flags)
assert process.stdout is not None
for line in process.stdout:
if token.cancelled:
process.terminate()
try: process.wait(timeout=3)
except subprocess.TimeoutExpired: process.kill()
raise PreprocessingCancelled("Preprocessing was cancelled.")
if line.startswith("out_time_us="):
try: current = Decimal(line.partition("=")[2].strip()) / Decimal("1000000")
except InvalidOperation: continue
percent = min(99, int((completed + min(current, boundary.duration_seconds)) * 100 / total_work))
if percent >= last_percent:
last_percent = percent
if progress: progress(PreprocessingProgress(percent, "encoding", f"Preparing chunk {boundary.index + 1} of {len(boundaries)}"))
if process.wait() != 0: raise PreprocessingError("FFmpeg could not preprocess the media.")
completed += boundary.duration_seconds
if durable:
durable_manifest = self.manifest_repository.load(manifest_path)
durable_manifest["chunks"][boundary.index]["encoded"] = True
self.manifest_repository.save(manifest_path, durable_manifest)
else:
manifest["completed_chunks"] = boundary.index + 1
self._write_manifest(manifest_path, manifest)
if durable:
self.manifest_repository.mark_job_state(manifest_path, "transcribing")
else:
manifest["state"] = "completed"; self._write_manifest(manifest_path, manifest)
if progress: progress(PreprocessingProgress(100, "completed", "Preprocessing complete."))
return PreprocessingResult(job, manifest_path, durable or options.retain_temporary_files, job)
except PreprocessingCancelled:
if durable: self.manifest_repository.mark_job_state(manifest_path, "cancelled")
else: manifest["state"] = "cancelled"; self._write_manifest(manifest_path, manifest)
if not durable and not options.retain_temporary_files: shutil.rmtree(job, ignore_errors=True)
raise
except Exception as exc:
if durable: self.manifest_repository.mark_job_state(manifest_path, "failed", str(exc))
else: manifest["state"] = "failed"; manifest["error"] = str(exc); self._write_manifest(manifest_path, manifest)
if not durable and not options.retain_temporary_files: shutil.rmtree(job, ignore_errors=True)
if isinstance(exc, PreprocessingError): raise
raise PreprocessingError("Preprocessing failed.") from exc
def _command(self, source: Path, output: Path, start: Decimal, duration: Decimal) -> list[str]:
return [str(self.ffmpeg_path), "-hide_banner", "-y", "-ss", str(start), "-i", str(source), "-t", str(duration), "-map", "0:a:0", "-vn", "-sn", "-dn", "-ac", "1", "-ar", "24000", "-c:a", "aac", "-b:a", "64k", "-progress", "pipe:1", "-nostats", str(output)]
@staticmethod
def _write_manifest(path: Path, manifest: dict) -> None:
temporary = path.with_suffix(".tmp")
temporary.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
temporary.replace(path)

123
src/voice_transcriptor/services/settings.py

@ -1,123 +0,0 @@
"""Per-user persistence for non-secret application settings."""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
from voice_transcriptor.models import DEFAULT_CONTEXT, AppSettings
DEFAULT_MODEL = "gpt-transcribe"
DEFAULT_LANGUAGE = "pt-BR"
SUPPORTED_MODEL_SUGGESTIONS = (
"gpt-transcribe",
"gpt-4o-transcribe",
"gpt-4o-mini-transcribe",
)
class SettingsError(Exception):
"""Settings could not be written."""
class SettingsRepository:
"""Load and atomically save settings without storing credentials."""
def __init__(self, path: Path | None = None) -> None:
self.path = path if path is not None else self.default_path()
@staticmethod
def default_path() -> Path:
app_data = os.environ.get("APPDATA")
base_directory = Path(app_data) if app_data else Path.home() / "AppData" / "Roaming"
return base_directory / "VoiceTranscriptor" / "settings.json"
@staticmethod
def default_settings() -> AppSettings:
home = Path.home()
documents = home / "Documents"
return AppSettings(
DEFAULT_MODEL,
DEFAULT_LANGUAGE,
documents if documents.is_dir() else home,
context_vocabulary=DEFAULT_CONTEXT,
)
def load(self) -> tuple[AppSettings, str | None]:
if not self.path.exists():
return self.default_settings(), None
try:
payload = json.loads(self.path.read_text(encoding="utf-8"))
return self._settings_from_payload(payload), None
except (OSError, TypeError, ValueError, json.JSONDecodeError):
return self.default_settings(), "Could not load settings; defaults are being used."
def save(self, settings: AppSettings) -> None:
payload = {
"model": settings.model,
"language": settings.language,
"output_directory": str(settings.output_directory),
"chunk_duration_seconds": settings.chunk_duration_seconds,
"chunk_overlap_seconds": settings.chunk_overlap_seconds,
"retain_temporary_files": settings.retain_temporary_files,
"context_vocabulary": settings.context_vocabulary,
}
temporary_path: Path | None = None
try:
self.path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(
dir=self.path.parent, prefix=f".{self.path.name}.", suffix=".tmp"
)
temporary_path = Path(temporary_name)
with os.fdopen(descriptor, "w", encoding="utf-8") as temporary_file:
json.dump(payload, temporary_file, ensure_ascii=False)
temporary_path.replace(self.path)
except OSError as exc:
raise SettingsError("Unable to save settings.") from exc
finally:
if temporary_path is not None and temporary_path.exists():
try:
temporary_path.unlink(missing_ok=True)
except OSError:
pass
@staticmethod
def _settings_from_payload(payload: object) -> AppSettings:
if not isinstance(payload, dict):
raise ValueError("Settings payload must be an object.")
model = payload.get("model")
language = payload.get("language")
output_directory = payload.get("output_directory")
if (
not isinstance(model, str)
or not model.strip()
or not isinstance(language, str)
or not language.strip()
or not isinstance(output_directory, str)
or not output_directory.strip()
):
raise ValueError("Settings payload is invalid.")
chunk_duration = payload.get("chunk_duration_seconds", 900)
overlap = payload.get("chunk_overlap_seconds", 15)
retain = payload.get("retain_temporary_files", False)
context = payload.get("context_vocabulary", DEFAULT_CONTEXT)
if (
isinstance(chunk_duration, bool) or not isinstance(chunk_duration, int)
or isinstance(overlap, bool) or not isinstance(overlap, int)
or not isinstance(retain, bool) or not isinstance(context, str) or chunk_duration <= 0
or overlap < 0 or overlap >= chunk_duration
):
raise ValueError("Preprocessing settings are invalid.")
return AppSettings(
model,
language,
Path(output_directory),
chunk_duration,
overlap,
retain,
context,
)

183
src/voice_transcriptor/services/transcription.py

@ -1,183 +0,0 @@
from __future__ import annotations
import random
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Callable
import openai
from voice_transcriptor.models import AppSettings
from voice_transcriptor.services.job_manifest import ChunkStatus, JobManifestRepository
from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled
BRAZILIAN_PROMPT = (
"Conversa em português brasileiro. Preserve a língua falada; não traduza. "
"Preserve ortografia e pontuação brasileiras, números, nomes próprios, "
"termos técnicos e siglas com máxima fidelidade."
)
class TranscriptionError(RuntimeError): pass
class PermanentTranscriptionError(TranscriptionError): pass
class RetryExhaustedError(TranscriptionError): pass
@dataclass(frozen=True, slots=True)
class RetryPolicy:
max_attempts: int = 5
initial_delay_seconds: float = 1.0
max_delay_seconds: float = 30.0
jitter_ratio: float = 0.2
@dataclass(frozen=True, slots=True)
class TranscriptionProgress:
completed: int
total: int
current_chunk: int | None
elapsed_seconds: float
phase: str
message: str
api_error: str | None = None
def normalize_language(value: str) -> str:
normalized = value.strip().replace("_", "-")
return normalized.split("-", 1)[0].lower()
def build_prompt(context: str) -> str:
extra = context.strip()
return f"{BRAZILIAN_PROMPT}\n\n{extra}" if extra else BRAZILIAN_PROMPT
def find_resumable_manifest(output_directory: Path, source: Path, settings: AppSettings) -> Path | None:
root = output_directory / "voice-transcriptor-jobs"
candidates = root.rglob("manifest.json") if root.is_dir() else output_directory.rglob("manifest.json")
expected_source = str(source.resolve())
expected_settings = {
"model": settings.model,
"language": settings.language,
"context_vocabulary": settings.context_vocabulary,
}
matches: list[tuple[str, Path]] = []
for path in candidates:
try:
manifest = JobManifestRepository().load(path)
except Exception:
continue
actual = manifest.get("settings", {})
if manifest.get("state") == "completed" or manifest.get("source", {}).get("path") != expected_source:
continue
if all(actual.get(key) == value for key, value in expected_settings.items()):
matches.append((str(manifest.get("updated_at", "")), path))
return max(matches, default=("", None), key=lambda value: value[0])[1]
class OpenAITranscriptionClient:
def __init__(self, client) -> None: self.client = client
def transcribe(self, path: Path, model: str, language: str, prompt: str) -> str:
with path.open("rb") as audio_file:
response = self.client.audio.transcriptions.create(
file=audio_file,
model=model,
language=normalize_language(language),
prompt=prompt,
response_format="json",
)
text = getattr(response, "text", None)
if not isinstance(text, str):
raise PermanentTranscriptionError("The transcription API returned no text.")
return text
class TranscriptionService:
def __init__(
self,
client: OpenAITranscriptionClient,
manifests: JobManifestRepository,
retry_policy: RetryPolicy | None = None,
sleep: Callable[[float], None] = time.sleep,
monotonic: Callable[[], float] = time.monotonic,
random_value: Callable[[], float] = random.random,
) -> None:
self.client = client; self.manifests = manifests
self.retry_policy = retry_policy or RetryPolicy()
self.sleep = sleep; self.monotonic = monotonic; self.random_value = random_value
def run(
self,
manifest_path: Path,
settings: AppSettings,
token: CancellationToken | None = None,
progress: Callable[[TranscriptionProgress], None] | None = None,
) -> Path:
token = token or CancellationToken(); started = self.monotonic()
try:
token.raise_if_cancelled()
manifest = self.manifests.recover_for_resume(manifest_path)
total = manifest["total_chunks"]
for item in manifest["chunks"]:
if item["status"] == ChunkStatus.COMPLETED: continue
token.raise_if_cancelled(); index = item["index"]
text = self._transcribe_with_retry(manifest_path, item, settings, token, progress, started, total)
self.manifests.mark_completed(manifest_path, index, text)
self.manifests.assemble_transcript(manifest_path)
completed = self.manifests.load(manifest_path)["completed_chunks"]
self._emit(progress, completed, total, index + 1, started, "transcribing", f"Completed chunk {index + 1} of {total}")
self.manifests.mark_job_state(manifest_path, "completed")
return self.manifests.assemble_transcript(manifest_path)
except PreprocessingCancelled:
self.manifests.mark_job_state(manifest_path, "cancelled")
raise
def _transcribe_with_retry(self, manifest_path: Path, item: dict, settings: AppSettings, token: CancellationToken, progress, started: float, total: int) -> str:
policy = self.retry_policy; index = item["index"]
for attempt in range(1, policy.max_attempts + 1):
token.raise_if_cancelled(); self.manifests.mark_processing(manifest_path, index)
self._emit(progress, self.manifests.load(manifest_path)["completed_chunks"], total, index + 1, started, "transcribing", f"Transcribing chunk {index + 1} of {total}")
try:
return self.client.transcribe(manifest_path.parent / item["path"], settings.model, settings.language, build_prompt(settings.context_vocabulary))
except PreprocessingCancelled: raise
except Exception as exc:
status = getattr(exc, "status_code", None)
retryable = self._retryable(exc, status)
safe = self._safe_error(status, retryable)
if not retryable:
self.manifests.mark_failed(manifest_path, index, safe)
raise PermanentTranscriptionError(safe) from None
if attempt >= policy.max_attempts:
self.manifests.mark_failed(manifest_path, index, safe)
raise RetryExhaustedError(f"{safe} Retry limit reached.") from None
delay = self._delay(exc, attempt)
self._emit(progress, self.manifests.load(manifest_path)["completed_chunks"], total, index + 1, started, "retrying", f"API temporarily unavailable; retrying in {delay:g} seconds.", safe)
token.raise_if_cancelled(); self.sleep(delay); token.raise_if_cancelled()
raise AssertionError("unreachable")
def _delay(self, exc: Exception, attempt: int) -> float:
headers = getattr(getattr(exc, "response", None), "headers", {}) or {}
retry_after = headers.get("retry-after") or headers.get("Retry-After")
try: server_delay = float(retry_after)
except (TypeError, ValueError): server_delay = 0
base = max(server_delay, self.retry_policy.initial_delay_seconds * (2 ** (attempt - 1)))
base = min(base, self.retry_policy.max_delay_seconds)
jitter = base * self.retry_policy.jitter_ratio * ((self.random_value() * 2) - 1)
return max(0, min(self.retry_policy.max_delay_seconds, base + jitter))
@staticmethod
def _retryable(exc: Exception, status: int | None) -> bool:
transient_types = (openai.RateLimitError, openai.APIConnectionError, openai.APITimeoutError)
return isinstance(exc, transient_types) or status in (408, 409, 429) or (isinstance(status, int) and status >= 500)
@staticmethod
def _safe_error(status: int | None, retryable: bool) -> str:
kind = "transient" if retryable else "permanent"
suffix = f" (HTTP {status})" if isinstance(status, int) else ""
return f"OpenAI API {kind} error{suffix}."
def _emit(self, callback, completed: int, total: int, current: int | None, started: float, phase: str, message: str, api_error: str | None = None) -> None:
if callback: callback(TranscriptionProgress(completed, total, current, max(0, self.monotonic() - started), phase, message, api_error))

1
src/voice_transcriptor/ui/__init__.py

@ -1 +0,0 @@
"""Native user-interface components for Voice Transcriptor."""

163
src/voice_transcriptor/ui/main_window.py

@ -1,163 +0,0 @@
from __future__ import annotations
from pathlib import Path
from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import QFileDialog, QHBoxLayout, QLabel, QMainWindow, QPlainTextEdit, QProgressBar, QPushButton, QVBoxLayout, QWidget
from voice_transcriptor.formatting import format_duration, format_file_size
from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled, PreprocessingOptions, PreprocessingProgress, SUPPORTED_EXTENSIONS
from voice_transcriptor.services.transcription import TranscriptionProgress, find_resumable_manifest
from voice_transcriptor.ui.settings_dialog import SettingsDialog
class WorkerSignals(QObject):
progress = Signal(object)
completed = Signal(object)
cancelled = Signal()
failed = Signal(str)
class PreprocessingWorker(QRunnable):
def __init__(self, service, source: Path, options: PreprocessingOptions, token: CancellationToken) -> None:
super().__init__(); self.service = service; self.source = source; self.options = options; self.token = token; self.signals = WorkerSignals()
@Slot()
def run(self) -> None:
try:
result = self.service.preprocess(self.source, self.options, self.token, self.signals.progress.emit)
except PreprocessingCancelled:
self.signals.cancelled.emit()
except Exception as exc:
self.signals.failed.emit(str(exc))
else:
self.signals.completed.emit(result)
class JobWorker(QRunnable):
def __init__(self, preprocessing_service, transcription_factory, credentials, source: Path, settings, token: CancellationToken) -> None:
super().__init__(); self.preprocessing_service = preprocessing_service; self.transcription_factory = transcription_factory; self.credentials = credentials; self.source = source; self.settings = settings; self.token = token; self.signals = WorkerSignals()
@Slot()
def run(self) -> None:
try:
self.token.raise_if_cancelled()
api_key = self.credentials.get_api_key()
if not api_key: raise RuntimeError("OpenAI API key is not configured in Windows Credential Manager.")
manifest_path = find_resumable_manifest(self.settings.output_directory, self.source, self.settings)
if manifest_path is None:
options = PreprocessingOptions(self.settings.chunk_duration_seconds, self.settings.chunk_overlap_seconds, True)
result = self.preprocessing_service.preprocess(
self.source, options, self.token, self.signals.progress.emit,
durable_root=self.settings.output_directory,
transcription_settings={"model": self.settings.model, "language": self.settings.language, "context_vocabulary": self.settings.context_vocabulary},
)
manifest_path = result.manifest_path
transcript = self.transcription_factory(api_key).run(manifest_path, self.settings, self.token, self.signals.progress.emit)
except PreprocessingCancelled:
self.signals.cancelled.emit()
except Exception as exc:
self.signals.failed.emit(str(exc))
else:
self.signals.completed.emit(transcript)
class MainWindow(QMainWindow):
def __init__(self, media_probe, preprocessing_service, settings_repository, credentials, transcription_factory=None, parent=None) -> None:
super().__init__(parent)
self.media_probe = media_probe; self.preprocessing_service = preprocessing_service
self.settings_repository = settings_repository; self.credentials = credentials
self.transcription_factory = transcription_factory
self.settings, warning = settings_repository.load()
self.selected_media = None; self._worker = None; self._cancellation_token = CancellationToken()
self.setWindowTitle("Voice Transcriptor"); self.setMinimumSize(650, 430); self.setAcceptDrops(True)
central = QWidget(self); layout = QVBoxLayout(central)
self.file_label = QLabel("No media selected", central); layout.addWidget(self.file_label)
row = QHBoxLayout(); self.browse_button = QPushButton("Browse…", central); self.settings_button = QPushButton("Settings", central); row.addWidget(self.browse_button); row.addWidget(self.settings_button); layout.addLayout(row)
self.metadata_label = QLabel("Select a supported audio or video file.", central); layout.addWidget(self.metadata_label)
self.progress_bar = QProgressBar(central); self.progress_bar.setRange(0, 100); self.progress_bar.setValue(0); layout.addWidget(self.progress_bar)
progress_details = QHBoxLayout(); self.chunk_progress_label = QLabel("0 / 0 chunks", central); self.current_chunk_label = QLabel("Current chunk: —", central); self.elapsed_label = QLabel("Elapsed: 00:00:00", central); progress_details.addWidget(self.chunk_progress_label); progress_details.addWidget(self.current_chunk_label); progress_details.addWidget(self.elapsed_label); layout.addLayout(progress_details)
actions = QHBoxLayout(); self.prepare_button = QPushButton("Transcribe", central); self.prepare_button.setEnabled(False); self.cancel_button = QPushButton("Cancel", central); self.cancel_button.setEnabled(False); actions.addWidget(self.prepare_button); actions.addWidget(self.cancel_button); layout.addLayout(actions)
self.log = QPlainTextEdit(central); self.log.setReadOnly(True); layout.addWidget(self.log); self.setCentralWidget(central)
self.browse_button.clicked.connect(self.browse); self.settings_button.clicked.connect(self.open_settings); self.prepare_button.clicked.connect(self.start_preprocessing); self.cancel_button.clicked.connect(self.cancel_preprocessing)
if warning: self.log.appendPlainText(warning)
def browse(self) -> None:
filters = "Media files (*.m4a *.mp3 *.wav *.mp4 *.mov *.webm *.mkv)"
filename, _ = QFileDialog.getOpenFileName(self, "Choose recording", "", filters)
if filename: self.select_file(Path(filename))
def select_file(self, path: Path) -> None:
if path.suffix.lower() not in SUPPORTED_EXTENSIONS:
self.log.appendPlainText("Unsupported media format."); return
try: media = self.media_probe.probe(path)
except Exception as exc:
self.log.appendPlainText(str(exc)); self.prepare_button.setEnabled(False); return
if not media.has_audio:
self.log.appendPlainText("The selected media has no audio stream."); return
self.selected_media = media
self.file_label.setText(media.path.name)
self.metadata_label.setText(f"{format_file_size(media.size_bytes)} · {format_duration(media.duration_seconds)} · {media.audio_codec or 'Unknown codec'}")
self.prepare_button.setEnabled(True)
def start_preprocessing(self) -> None:
if self.selected_media is None: return
self._begin_busy_state()
options = PreprocessingOptions(self.settings.chunk_duration_seconds, self.settings.chunk_overlap_seconds, self.settings.retain_temporary_files)
worker = JobWorker(self.preprocessing_service, self.transcription_factory, self.credentials, self.selected_media.path, self.settings, self._cancellation_token) if self.transcription_factory else PreprocessingWorker(self.preprocessing_service, self.selected_media.path, options, self._cancellation_token)
self._worker = worker
worker.signals.progress.connect(self._on_progress); worker.signals.completed.connect(self._on_completed); worker.signals.cancelled.connect(self._on_cancelled); worker.signals.failed.connect(self._on_failed)
QThreadPool.globalInstance().start(worker)
def _begin_busy_state(self) -> None:
self._cancellation_token = CancellationToken(); self.prepare_button.setEnabled(False); self.browse_button.setEnabled(False); self.settings_button.setEnabled(False); self.cancel_button.setEnabled(True); self.progress_bar.setValue(0)
def cancel_preprocessing(self) -> None:
self._cancellation_token.cancel(); self.cancel_button.setEnabled(False); self.log.appendPlainText("Cancelling preprocessing…")
@Slot(object)
def _on_progress(self, progress: PreprocessingProgress | TranscriptionProgress) -> None:
if isinstance(progress, TranscriptionProgress):
percent = int(progress.completed * 100 / progress.total) if progress.total else 0
self.progress_bar.setValue(percent)
self.chunk_progress_label.setText(f"{progress.completed} / {progress.total} chunks")
self.current_chunk_label.setText(f"Current chunk: {progress.current_chunk if progress.current_chunk is not None else ''}")
self.elapsed_label.setText(f"Elapsed: {format_duration(progress.elapsed_seconds)}")
self.log.appendPlainText(progress.message)
if progress.api_error: self.log.appendPlainText(progress.api_error)
else:
self.progress_bar.setValue(progress.percent); self.log.appendPlainText(progress.message)
@Slot(object)
def _on_completed(self, result) -> None:
self.progress_bar.setValue(100)
if isinstance(result, Path): self.log.appendPlainText(f"Transcription complete: {result}")
elif result.retained: self.log.appendPlainText(f"Prepared chunks retained at {result.job_directory}")
else: result.cleanup(); self.log.appendPlainText("Audio preprocessing complete.")
self._finish_busy_state()
@Slot()
def _on_cancelled(self) -> None: self.log.appendPlainText("Preprocessing cancelled."); self._finish_busy_state()
@Slot(str)
def _on_failed(self, message: str) -> None: self.log.appendPlainText(message); self._finish_busy_state()
def _finish_busy_state(self) -> None:
self.browse_button.setEnabled(True); self.settings_button.setEnabled(True); self.prepare_button.setEnabled(self.selected_media is not None); self.cancel_button.setEnabled(False); self._worker = None
def open_settings(self) -> None:
dialog = SettingsDialog(self.settings, self.settings_repository, self.credentials, self)
dialog.settings_saved.connect(self._set_settings); dialog.exec()
@Slot(object)
def _set_settings(self, settings) -> None: self.settings = settings
def dragEnterEvent(self, event) -> None:
urls = event.mimeData().urls()
if len(urls) == 1 and urls[0].isLocalFile(): event.acceptProposedAction()
def dropEvent(self, event) -> None:
urls = event.mimeData().urls()
if len(urls) == 1: self.select_file(Path(urls[0].toLocalFile()))
def closeEvent(self, event) -> None:
if self.cancel_button.isEnabled(): self.cancel_preprocessing()
super().closeEvent(event)

203
src/voice_transcriptor/ui/settings_dialog.py

@ -1,203 +0,0 @@
"""Modal editor for non-secret settings and the stored API key."""
from __future__ import annotations
import os
from pathlib import Path
from PySide6.QtCore import Signal
from PySide6.QtWidgets import (
QDialog,
QDialogButtonBox,
QCheckBox,
QComboBox,
QFileDialog,
QFormLayout,
QHBoxLayout,
QLineEdit,
QMessageBox,
QPlainTextEdit,
QPushButton,
QSpinBox,
QToolButton,
QVBoxLayout,
QWidget,
)
from voice_transcriptor.models import AppSettings
from voice_transcriptor.services.credentials import CredentialError, CredentialService
from voice_transcriptor.services.settings import SUPPORTED_MODEL_SUGGESTIONS, SettingsError, SettingsRepository
class EditableModelCombo(QComboBox):
"""Editable model picker with the previous line-edit convenience API."""
def text(self) -> str:
return self.currentText()
def setText(self, value: str) -> None:
self.setCurrentText(value)
class SettingsDialog(QDialog):
"""Edit application settings while keeping the API key out of normal storage."""
settings_saved = Signal(AppSettings)
def __init__(
self,
settings: AppSettings,
repository: SettingsRepository,
credentials: CredentialService,
parent=None,
) -> None:
super().__init__(parent)
self._repository = repository
self._credentials = credentials
self.setWindowTitle("Settings")
self.model_input = EditableModelCombo(self)
self.model_input.setEditable(True)
self.model_input.addItems(SUPPORTED_MODEL_SUGGESTIONS)
self.model_input.setCurrentText(settings.model)
self.model_input.setObjectName("modelInput")
self.language_input = QLineEdit(settings.language, self)
self.language_input.setObjectName("languageInput")
self.api_key_input = QLineEdit(self)
self.api_key_input.setObjectName("apiKeyInput")
self.api_key_input.setEchoMode(QLineEdit.EchoMode.Password)
self.output_directory_input = QLineEdit(str(settings.output_directory), self)
self.output_directory_input.setObjectName("outputDirectoryInput")
self.context_input = QPlainTextEdit(settings.context_vocabulary, self)
self.context_input.setObjectName("contextVocabularyInput")
self.context_input.setPlaceholderText("Names, acronyms, technical terms, and conversation context")
self.choose_directory_button = QPushButton("Browse…", self)
self.choose_directory_button.setObjectName("chooseDirectoryButton")
self.choose_directory_button.clicked.connect(self.choose_output_directory)
output_directory_layout = QHBoxLayout()
output_directory_layout.addWidget(self.output_directory_input)
output_directory_layout.addWidget(self.choose_directory_button)
form = QFormLayout()
form.addRow("OpenAI API key", self.api_key_input)
form.addRow("Model", self.model_input)
form.addRow("Language", self.language_input)
form.addRow("Context / Vocabulary", self.context_input)
form.addRow("Output directory", output_directory_layout)
self.advanced_toggle = QToolButton(self)
self.advanced_toggle.setText("Advanced")
self.advanced_toggle.setCheckable(True)
self.advanced_toggle.setObjectName("advancedToggle")
self.advanced_panel = QWidget(self)
self.advanced_panel.setObjectName("advancedPanel")
advanced_form = QFormLayout(self.advanced_panel)
self.chunk_duration_input = QSpinBox(self.advanced_panel)
self.chunk_duration_input.setObjectName("chunkDurationInput")
self.chunk_duration_input.setRange(1, 86400)
self.chunk_duration_input.setSuffix(" seconds")
self.chunk_duration_input.setValue(settings.chunk_duration_seconds)
self.chunk_overlap_input = QSpinBox(self.advanced_panel)
self.chunk_overlap_input.setObjectName("chunkOverlapInput")
self.chunk_overlap_input.setRange(0, 3600)
self.chunk_overlap_input.setSuffix(" seconds")
self.chunk_overlap_input.setValue(settings.chunk_overlap_seconds)
self.retain_temporary_files_input = QCheckBox("Retain temporary files for debugging", self.advanced_panel)
self.retain_temporary_files_input.setObjectName("retainTemporaryFilesInput")
self.retain_temporary_files_input.setChecked(settings.retain_temporary_files)
advanced_form.addRow("Chunk duration", self.chunk_duration_input)
advanced_form.addRow("Overlap", self.chunk_overlap_input)
advanced_form.addRow(self.retain_temporary_files_input)
self.advanced_panel.hide()
self.advanced_toggle.toggled.connect(self.advanced_panel.setVisible)
self.button_box = QDialogButtonBox(
QDialogButtonBox.StandardButton.Save | QDialogButtonBox.StandardButton.Cancel,
parent=self,
)
self.save_button = self.button_box.button(QDialogButtonBox.StandardButton.Save)
self.save_button.setObjectName("saveButton")
self.cancel_button = self.button_box.button(QDialogButtonBox.StandardButton.Cancel)
self.cancel_button.setObjectName("cancelButton")
self.button_box.accepted.connect(self.save)
self.button_box.rejected.connect(self.reject)
layout = QVBoxLayout(self)
layout.addLayout(form)
layout.addWidget(self.advanced_toggle)
layout.addWidget(self.advanced_panel)
layout.addWidget(self.button_box)
def choose_output_directory(self) -> None:
directory = QFileDialog.getExistingDirectory(
self,
"Choose output directory",
self.output_directory_input.text(),
)
if directory:
self.output_directory_input.setText(directory)
def save(self) -> None:
"""Persist valid settings and, only when supplied, replace the API key."""
api_key = self.api_key_input.text().strip()
self.api_key_input.clear()
output_directory = self._output_directory()
if output_directory is None:
QMessageBox.warning(
self,
"Invalid output directory",
"Choose an accessible output directory.",
)
return
model = self.model_input.currentText().strip()
language = self.language_input.text().strip()
if not model or not language:
QMessageBox.warning(self, "Invalid settings", "Model and language are required.")
return
chunk_duration = self.chunk_duration_input.value()
overlap = self.chunk_overlap_input.value()
if overlap >= chunk_duration:
QMessageBox.warning(self, "Invalid settings", "Overlap must be smaller than chunk duration.")
return
updated_settings = AppSettings(
model, language, output_directory, chunk_duration, overlap,
self.retain_temporary_files_input.isChecked(), self.context_input.toPlainText().strip(),
)
try:
self._repository.save(updated_settings)
except SettingsError:
QMessageBox.warning(self, "Could not save settings", "Settings could not be saved.")
return
if api_key:
try:
self._credentials.set_api_key(api_key)
except CredentialError:
QMessageBox.warning(
self,
"Could not save settings",
"The API key could not be saved.",
)
return
self.settings_saved.emit(updated_settings)
self.accept()
def _output_directory(self) -> Path | None:
value = self.output_directory_input.text().strip()
if not value:
return None
directory = Path(value)
try:
if not directory.is_dir():
return None
if not os.access(directory, os.R_OK | os.W_OK):
return None
except OSError:
return None
return directory

50
tests/test_chunking.py

@ -1,50 +0,0 @@
from decimal import Decimal
import pytest
from voice_transcriptor.services.chunking import (
calculate_chunk_boundaries,
source_timestamp,
total_chunk_duration,
)
@pytest.mark.parametrize(
("total", "chunk", "overlap", "expected"),
[
("600", "900", "15", [("0", "600")]),
("900", "900", "15", [("0", "900")]),
("1800", "900", "15", [("0", "900"), ("885", "1785"), ("1770", "1800")]),
("901.25", "900", "15", [("0", "900"), ("885", "901.25")]),
("1800", "900", "0", [("0", "900"), ("900", "1800")]),
],
)
def test_chunk_boundaries_keep_context_and_exact_final_end(
total: str, chunk: str, overlap: str, expected: list[tuple[str, str]]
) -> None:
boundaries = calculate_chunk_boundaries(total, chunk, overlap)
assert [(str(item.start_seconds), str(item.end_seconds)) for item in boundaries] == expected
def test_overlap_is_included_in_total_generated_audio_duration() -> None:
boundaries = calculate_chunk_boundaries("1800", "900", "15")
assert total_chunk_duration(boundaries) == Decimal("1830")
def test_local_timestamp_maps_to_exact_source_offset() -> None:
boundary = calculate_chunk_boundaries("1800", "900", "15")[1]
assert source_timestamp(boundary, "12.5") == Decimal("897.5")
def test_many_hour_recording_ends_exactly_at_source_duration() -> None:
boundaries = calculate_chunk_boundaries("28800.125", "900", "15")
assert boundaries[-1].end_seconds == Decimal("28800.125")
@pytest.mark.parametrize(
("total", "chunk", "overlap"),
[("0", "900", "15"), ("10", "0", "0"), ("10", "10", "10"), ("10", "10", "-1")],
)
def test_invalid_chunk_configuration_is_rejected(total: str, chunk: str, overlap: str) -> None:
with pytest.raises(ValueError):
calculate_chunk_boundaries(total, chunk, overlap)

88
tests/test_credentials.py

@ -1,88 +0,0 @@
import pytest
from types import SimpleNamespace
from voice_transcriptor.services.credentials import CredentialError, CredentialService
class FakeKeyring:
def __init__(self) -> None:
self.credential = None
self.calls: list[tuple[str, str, str | None]] = []
def get_credential(self, service: str, account: str | None):
self.calls.append(("get", service, account))
return self.credential
def set_password(self, service: str, account: str, value: str) -> None:
self.calls.append(("set", service, account, value))
self.credential = SimpleNamespace(username=account, password=value)
def test_get_api_key_uses_existing_windows_credential_target() -> None:
backend = FakeKeyring()
backend.credential = SimpleNamespace(username="stored-user", password="secret")
assert CredentialService(backend).get_api_key() == "secret"
assert backend.calls == [("get", "OPENAI_API_KEY", None)]
def test_has_api_key_reflects_keyring_value() -> None:
backend = FakeKeyring()
service = CredentialService(backend)
assert service.has_api_key() is False
backend.credential = SimpleNamespace(username="stored-user", password="secret")
assert service.has_api_key() is True
def test_set_api_key_rejects_blank_values() -> None:
backend = FakeKeyring()
with pytest.raises(CredentialError, match="API key"):
CredentialService(backend).set_api_key(" \t")
assert backend.calls == []
def test_set_api_key_persists_nonblank_value() -> None:
backend = FakeKeyring()
backend.credential = SimpleNamespace(username="stored-user", password="old")
CredentialService(backend).set_api_key("secret")
assert backend.calls == [
("get", "OPENAI_API_KEY", None),
("set", "OPENAI_API_KEY", "stored-user", "secret"),
]
def test_set_api_key_creates_target_with_stable_username_when_missing() -> None:
backend = FakeKeyring()
CredentialService(backend).set_api_key("secret")
assert backend.calls == [
("get", "OPENAI_API_KEY", None),
("set", "OPENAI_API_KEY", "OPENAI_API_KEY", "secret"),
]
@pytest.mark.parametrize("method", ["get_credential", "set_password"])
def test_backend_failures_are_sanitized(method: str) -> None:
class FailingKeyring:
def get_credential(self, service: str, account: str | None):
raise RuntimeError("backend leaked secret")
def set_password(self, service: str, account: str, value: str) -> None:
raise RuntimeError("backend leaked secret")
service = CredentialService(FailingKeyring())
with pytest.raises(CredentialError) as caught:
if method == "get_credential":
service.get_api_key()
else:
service.set_api_key("secret")
assert "secret" not in str(caught.value).lower()
assert caught.value.__cause__ is None

16
tests/test_formatting.py

@ -1,16 +0,0 @@
from unittest import TestCase
from voice_transcriptor.formatting import format_duration, format_file_size
class FormattingTests(TestCase):
def test_formats_binary_file_sizes(self) -> None:
self.assertEqual(format_file_size(0), "0 B")
self.assertEqual(format_file_size(1536), "1.50 KiB")
self.assertEqual(format_file_size(2 * 1024**3), "2.00 GiB")
def test_formats_multi_hour_duration(self) -> None:
self.assertEqual(format_duration(3 * 3600 + 5 * 60 + 9.8), "03:05:10")
def test_formats_unavailable_duration(self) -> None:
self.assertEqual(format_duration(None), "Unknown")

113
tests/test_job_manifest.py

@ -1,113 +0,0 @@
import json
from pathlib import Path
import pytest
from voice_transcriptor.services.job_manifest import JobManifestError, JobManifestRepository
def chunks() -> list[dict]:
return [
{
"index": 0,
"path": "chunks/chunk-00000.m4a",
"source_start_seconds": "0",
"source_end_seconds": "900.125",
"duration_seconds": "900.125",
},
{
"index": 1,
"path": "chunks/chunk-00001.m4a",
"source_start_seconds": "885.125",
"source_end_seconds": "1000.5",
"duration_seconds": "115.375",
},
]
def create_manifest(tmp_path: Path) -> tuple[JobManifestRepository, Path]:
job = tmp_path / "job"
repository = JobManifestRepository()
manifest = repository.create(
job,
source={"path": str(tmp_path / "source.mp3"), "size_bytes": 10},
settings={"model": "gpt-transcribe", "language": "pt-BR"},
chunks=chunks(),
)
return repository, Path(manifest["manifest_path"])
def test_create_persists_schema_timing_and_pending_statuses(tmp_path: Path) -> None:
repository, path = create_manifest(tmp_path)
manifest = repository.load(path)
assert manifest["schema_version"] == 2
assert manifest["state"] == "preprocessing"
assert manifest["completed_chunks"] == 0
assert manifest["total_chunks"] == 2
assert [item["status"] for item in manifest["chunks"]] == ["pending", "pending"]
assert manifest["chunks"][1]["source_start_seconds"] == "885.125"
assert not path.with_suffix(".tmp").exists()
def test_create_rejects_chunk_path_outside_job(tmp_path: Path) -> None:
invalid = chunks()
invalid[0]["path"] = "../secret.m4a"
with pytest.raises(JobManifestError, match="chunk path"):
JobManifestRepository().create(tmp_path / "job", {}, {}, invalid)
def test_completed_chunk_is_written_immediately_and_cannot_be_reprocessed(tmp_path: Path) -> None:
repository, path = create_manifest(tmp_path)
repository.mark_processing(path, 0)
repository.mark_completed(path, 0, "Primeiro trecho")
on_disk = json.loads(path.read_text(encoding="utf-8"))
assert on_disk["chunks"][0]["status"] == "completed"
assert on_disk["chunks"][0]["transcript"] == "Primeiro trecho"
assert on_disk["completed_chunks"] == 1
with pytest.raises(JobManifestError, match="completed"):
repository.mark_processing(path, 0)
def test_resume_recovers_processing_and_failed_but_preserves_completed(tmp_path: Path) -> None:
repository, path = create_manifest(tmp_path)
repository.mark_processing(path, 0)
repository.mark_completed(path, 0, "Feito")
repository.mark_processing(path, 1)
repository.mark_failed(path, 1, "temporary failure")
manifest = repository.recover_for_resume(path)
assert manifest["chunks"][0]["status"] == "completed"
assert manifest["chunks"][0]["transcript"] == "Feito"
assert manifest["chunks"][1]["status"] == "pending"
assert manifest["chunks"][1]["last_error"] is None
def test_assemble_transcript_uses_only_completed_chunks_in_index_order(tmp_path: Path) -> None:
repository, path = create_manifest(tmp_path)
repository.mark_processing(path, 1)
repository.mark_completed(path, 1, "Segundo")
repository.mark_processing(path, 0)
repository.mark_completed(path, 0, "Primeiro")
transcript_path = repository.assemble_transcript(path)
assert transcript_path.read_text(encoding="utf-8") == "Primeiro\nSegundo\n"
assert repository.load(path)["chunks"][1]["source_end_seconds"] == "1000.5"
def test_explicit_reset_is_required_to_clear_completed_work(tmp_path: Path) -> None:
repository, path = create_manifest(tmp_path)
repository.mark_processing(path, 0)
repository.mark_completed(path, 0, "Feito")
repository.reset_completed(path, indexes=[0])
item = repository.load(path)["chunks"][0]
assert item["status"] == "pending"
assert item["transcript"] is None

124
tests/test_media_probe.py

@ -1,124 +0,0 @@
import json
import subprocess
from pathlib import Path
import pytest
from voice_transcriptor.services.media_probe import (
InvalidMediaPathError,
MediaProbeError,
MediaProbeService,
ProbeExecutionError,
ProbeOutputError,
ProbeTimeoutError,
detect_tools,
parse_probe_output,
validate_media_path,
)
def test_detects_ffmpeg_and_ffprobe_executables(monkeypatch: pytest.MonkeyPatch) -> None:
discovered = {"ffmpeg": r"C:\\tools\\ffmpeg.exe", "ffprobe": r"C:\\tools\\ffprobe.exe"}
monkeypatch.setattr("voice_transcriptor.services.media_probe.shutil.which", discovered.get)
status = detect_tools()
assert status.ffmpeg_path == Path(r"C:\tools\ffmpeg.exe")
assert status.ffprobe_path == Path(r"C:\tools\ffprobe.exe")
def test_rejects_directory_path(tmp_path: Path) -> None:
with pytest.raises(InvalidMediaPathError, match="not a file"):
validate_media_path(tmp_path)
def test_parses_audio_only_media(tmp_path: Path) -> None:
media = tmp_path / "memo.m4a"
payload = json.dumps({
"format": {"duration": "4.5"},
"streams": [{"codec_type": "audio", "codec_name": "aac"}],
})
info = parse_probe_output(media, 3, payload)
assert info.duration_seconds == 4.5
assert info.audio_codec == "aac"
assert info.duration_text == "4.5"
assert info.has_audio is True
assert info.has_video is False
def test_parses_video_with_audio(tmp_path: Path) -> None:
media = tmp_path / "meeting.mov"
media.write_bytes(b"x" * 12)
payload = json.dumps({
"format": {"duration": "3661.25"},
"streams": [{"codec_type": "video", "codec_name": "h264"}, {"codec_type": "audio", "codec_name": "aac"}],
})
info = parse_probe_output(media, 12, payload)
assert info.duration_seconds == 3661.25
assert info.audio_codec == "aac"
assert info.has_video is True
def test_missing_metadata_is_unknown(tmp_path: Path) -> None:
media = tmp_path / "memo.m4a"
info = parse_probe_output(media, 0, '{"format": {}, "streams": []}')
assert info.duration_seconds is None
assert info.audio_codec is None
def test_rejects_malformed_probe_json(tmp_path: Path) -> None:
with pytest.raises(MediaProbeError, match="invalid data"):
parse_probe_output(tmp_path / "bad.mov", 0, "not json")
@pytest.mark.parametrize("payload", ["[]", '{"format": []}', '{"streams": [null]}'])
def test_rejects_structurally_invalid_probe_json(tmp_path: Path, payload: str) -> None:
with pytest.raises(ProbeOutputError, match="invalid data"):
parse_probe_output(tmp_path / "bad.mov", 0, payload)
def test_rejects_missing_file(tmp_path: Path) -> None:
with pytest.raises(MediaProbeError, match="does not exist"):
validate_media_path(tmp_path / "missing.mov")
def test_maps_nonzero_ffprobe_result(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
media = tmp_path / "bad.mov"
media.write_bytes(b"bad")
monkeypatch.setattr(subprocess, "run", lambda *a, **k: subprocess.CompletedProcess(a, 1, "", "invalid input"))
with pytest.raises(MediaProbeError, match="could not inspect"):
MediaProbeService(Path("ffprobe")).probe(media)
def test_maps_ffprobe_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
media = tmp_path / "slow.mov"
media.write_bytes(b"slow")
def timeout(*args: object, **kwargs: object) -> None:
raise subprocess.TimeoutExpired("ffprobe", 30)
monkeypatch.setattr(subprocess, "run", timeout)
with pytest.raises(MediaProbeError, match="timed out"):
MediaProbeService(Path("ffprobe")).probe(media)
def test_maps_probe_failures_to_typed_errors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing = tmp_path / "missing.mov"
with pytest.raises(InvalidMediaPathError):
validate_media_path(missing)
with pytest.raises(ProbeOutputError):
parse_probe_output(tmp_path / "bad.mov", 0, "not json")
media = tmp_path / "bad.mov"
media.write_bytes(b"bad")
monkeypatch.setattr(subprocess, "run", lambda *a, **k: subprocess.CompletedProcess(a, 1, "", "invalid input"))
with pytest.raises(ProbeExecutionError):
MediaProbeService(Path("ffprobe")).probe(media)
def timeout(*args: object, **kwargs: object) -> None:
raise subprocess.TimeoutExpired("ffprobe", 30)
monkeypatch.setattr(subprocess, "run", timeout)
with pytest.raises(ProbeTimeoutError):
MediaProbeService(Path("ffprobe")).probe(media)

100
tests/test_preprocessing.py

@ -1,100 +0,0 @@
import json
from pathlib import Path
import pytest
from voice_transcriptor.models import MediaInfo
from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled, PreprocessingOptions, PreprocessingService
from voice_transcriptor.services.job_manifest import JobManifestRepository
class FakeProbe:
def __init__(self, media: MediaInfo): self.media = media
def probe(self, path: Path) -> MediaInfo: return self.media
class FakeProcess:
def __init__(self, command):
self.stdout = iter(["out_time_us=450000000\n", "progress=continue\n", "progress=end\n"])
self.returncode = 0
self.terminated = False
def wait(self, timeout=None): return self.returncode
def poll(self): return self.returncode
def terminate(self): self.terminated = True
def kill(self): self.terminated = True
def make_media(path: Path, duration="1800") -> MediaInfo:
path.write_bytes(b"source")
return MediaInfo(path, 6, float(duration), "aac", duration, True, path.suffix in {".mp4", ".mov", ".mkv", ".webm"})
def test_preprocess_builds_streaming_commands_and_exact_manifest(tmp_path: Path) -> None:
source = tmp_path / "recording.mp4"
commands = []
def factory(command, **kwargs):
commands.append(command); Path(command[-1]).write_bytes(b"chunk"); return FakeProcess(command)
progress = []
service = PreprocessingService(Path("ffmpeg"), FakeProbe(make_media(source)), tmp_path / "jobs", factory)
result = service.preprocess(source, PreprocessingOptions(), progress=progress.append)
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
assert len(commands) == 3
assert [commands[0][commands[0].index(f) + 1] for f in ("-ss", "-t", "-ac", "-ar", "-c:a", "-b:a")] == ["0", "900", "1", "24000", "aac", "64k"]
assert all("0:a:0" in command and "-progress" in command for command in commands)
assert manifest["state"] == "completed"
assert manifest["chunks"][1]["source_start_seconds"] == "885"
assert manifest["chunks"][-1]["source_end_seconds"] == "1800"
assert manifest["completed_chunks"] == 3
assert progress[-1].percent == 100
assert [p.percent for p in progress] == sorted(p.percent for p in progress)
def test_cleanup_and_retention_are_job_scoped(tmp_path: Path) -> None:
source = tmp_path / "recording.wav"
def factory(command, **kwargs): Path(command[-1]).write_bytes(b"x"); return FakeProcess(command)
service = PreprocessingService(Path("ffmpeg"), FakeProbe(make_media(source, "10")), tmp_path / "jobs", factory)
result = service.preprocess(source, PreprocessingOptions()); job = result.job_directory; result.cleanup(); assert not job.exists()
retained = service.preprocess(source, PreprocessingOptions(retain_temporary_files=True)); retained.cleanup(); assert retained.job_directory.exists()
def test_durable_preprocessing_creates_schema_2_pending_transcription_job(tmp_path: Path) -> None:
source = tmp_path / "recording.wav"
output = tmp_path / "output"
def factory(command, **kwargs):
Path(command[-1]).write_bytes(b"encoded")
return FakeProcess(command)
service = PreprocessingService(
Path("ffmpeg"),
FakeProbe(make_media(source, "10")),
process_factory=factory,
manifest_repository=JobManifestRepository(),
)
result = service.preprocess(source, PreprocessingOptions(), durable_root=output)
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
assert result.job_directory.parent == (output / "voice-transcriptor-jobs").resolve()
assert result.retained is True
assert manifest["schema_version"] == 2
assert manifest["state"] == "transcribing"
assert manifest["completed_chunks"] == 0
assert manifest["chunks"][0]["status"] == "pending"
assert manifest["chunks"][0]["encoded"] is True
result.cleanup()
assert result.job_directory.exists()
def test_cancelled_job_stops_before_process(tmp_path: Path) -> None:
source = tmp_path / "recording.m4a"; token = CancellationToken(); token.cancel()
service = PreprocessingService(Path("ffmpeg"), FakeProbe(make_media(source, "10")), tmp_path / "jobs", lambda *a, **k: pytest.fail("started"))
with pytest.raises(PreprocessingCancelled): service.preprocess(source, PreprocessingOptions(), token)
@pytest.mark.parametrize("suffix", [".m4a", ".mp3", ".wav", ".mp4", ".mov", ".webm", ".mkv"])
def test_supported_extensions(tmp_path: Path, suffix: str) -> None:
source = tmp_path / f"recording{suffix}"
def factory(command, **kwargs): Path(command[-1]).write_bytes(b"x"); return FakeProcess(command)
result = PreprocessingService(Path("ffmpeg"), FakeProbe(make_media(source, "1")), tmp_path / "jobs", factory).preprocess(source, PreprocessingOptions())
assert result.manifest_path.exists()

144
tests/test_settings.py

@ -1,144 +0,0 @@
import json
from pathlib import Path
import pytest
from voice_transcriptor.models import AppSettings
from voice_transcriptor.services.settings import DEFAULT_CONTEXT, SettingsError, SettingsRepository
def test_load_missing_file_returns_documented_defaults(monkeypatch, tmp_path: Path) -> None:
documents = tmp_path / "Documents"
documents.mkdir()
monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
settings, warning = SettingsRepository(tmp_path / "settings.json").load()
assert settings == AppSettings("gpt-transcribe", "pt-BR", documents)
assert "AWS, Kubernetes, OpenAI" in settings.context_vocabulary
assert warning is None
def test_load_missing_file_uses_home_when_documents_is_unavailable(
monkeypatch, tmp_path: Path
) -> None:
monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
settings, warning = SettingsRepository(tmp_path / "settings.json").load()
assert settings.output_directory == tmp_path
assert warning is None
def test_save_and_load_round_trip_utf8_settings(tmp_path: Path) -> None:
path = tmp_path / "settings.json"
expected = AppSettings("gpt-4o-transcribe", "pt-BR", tmp_path / "Transcrições")
repository = SettingsRepository(path)
repository.save(expected)
actual, warning = repository.load()
assert actual == expected
assert warning is None
assert json.loads(path.read_text(encoding="utf-8")) == {
"model": "gpt-4o-transcribe",
"language": "pt-BR",
"output_directory": str(tmp_path / "Transcrições"),
"chunk_duration_seconds": 900,
"chunk_overlap_seconds": 15,
"retain_temporary_files": False,
"context_vocabulary": DEFAULT_CONTEXT,
}
def test_load_malformed_file_recovers_defaults_with_nonfatal_warning(tmp_path: Path) -> None:
path = tmp_path / "settings.json"
path.write_text("not json", encoding="utf-8")
settings, warning = SettingsRepository(path).load()
assert settings.model == "gpt-transcribe"
assert warning is not None
assert "settings" in warning.lower()
def test_save_replaces_existing_file_atomically(monkeypatch, tmp_path: Path) -> None:
path = tmp_path / "settings.json"
path.write_text("old", encoding="utf-8")
calls: list[tuple[Path, Path]] = []
original_replace = Path.replace
def recording_replace(source: Path, destination: Path) -> Path:
calls.append((source, destination))
return original_replace(source, destination)
monkeypatch.setattr(Path, "replace", recording_replace)
SettingsRepository(path).save(AppSettings("model", "pt-BR", tmp_path / "out"))
assert calls
source, destination = calls[0]
assert source.parent == path.parent
assert destination == path
assert json.loads(path.read_text(encoding="utf-8"))["model"] == "model"
def test_save_preserves_settings_error_when_replace_and_cleanup_fail(
monkeypatch, tmp_path: Path
) -> None:
path = tmp_path / "settings.json"
replacement_failure = OSError("target is locked")
def failing_replace(source: Path, destination: Path) -> Path:
raise replacement_failure
def failing_unlink(path: Path, missing_ok: bool = False) -> None:
raise OSError("temporary file is locked")
monkeypatch.setattr(Path, "replace", failing_replace)
monkeypatch.setattr(Path, "unlink", failing_unlink)
with pytest.raises(SettingsError) as caught:
SettingsRepository(path).save(AppSettings("model", "pt-BR", tmp_path / "out"))
assert caught.value.__cause__ is replacement_failure
def test_saved_settings_never_include_api_key_fields(tmp_path: Path) -> None:
path = tmp_path / "settings.json"
SettingsRepository(path).save(AppSettings("model", "pt-BR", tmp_path / "out"))
serialized = path.read_text(encoding="utf-8").lower()
assert "api" not in serialized
assert "key" not in serialized
def test_preprocessing_settings_round_trip(tmp_path: Path) -> None:
path = tmp_path / "settings.json"
expected = AppSettings("model", "pt-BR", tmp_path, 1200, 20, True)
repository = SettingsRepository(path)
repository.save(expected)
actual, warning = repository.load()
assert actual == expected
assert warning is None
def test_legacy_settings_receive_preprocessing_defaults(tmp_path: Path) -> None:
path = tmp_path / "settings.json"
path.write_text(json.dumps({"model": "model", "language": "pt-BR", "output_directory": str(tmp_path)}), encoding="utf-8")
settings, warning = SettingsRepository(path).load()
assert (settings.chunk_duration_seconds, settings.chunk_overlap_seconds, settings.retain_temporary_files) == (900, 15, False)
assert settings.context_vocabulary == DEFAULT_CONTEXT
assert warning is None
def test_context_vocabulary_round_trips_as_non_secret_setting(tmp_path: Path) -> None:
path = tmp_path / "settings.json"
expected = AppSettings("future-transcribe", "pt-BR", tmp_path, context_vocabulary="Brasília, Pix")
SettingsRepository(path).save(expected)
actual, warning = SettingsRepository(path).load()
assert actual == expected
assert warning is None

93
tests/test_transcription.py

@ -1,93 +0,0 @@
from pathlib import Path
from types import SimpleNamespace
import pytest
from voice_transcriptor.models import AppSettings
from voice_transcriptor.services.job_manifest import JobManifestRepository
from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled
from voice_transcriptor.services.transcription import OpenAITranscriptionClient, PermanentTranscriptionError, RetryPolicy, TranscriptionService, build_prompt, find_resumable_manifest, normalize_language
class Endpoint:
def __init__(self, responses): self.responses = list(responses); self.calls = []
def create(self, **kwargs):
self.calls.append(kwargs); result = self.responses.pop(0)
if isinstance(result, Exception): raise result
return SimpleNamespace(text=result)
class StatusFailure(Exception):
def __init__(self, status_code: int):
super().__init__(f"sensitive sk-leaked-key status {status_code}")
self.status_code = status_code; self.response = SimpleNamespace(headers={})
def make_job(tmp_path: Path) -> tuple[JobManifestRepository, Path]:
job = tmp_path / "job"; (job / "chunks").mkdir(parents=True)
for index in range(2): (job / "chunks" / f"chunk-{index:05d}.m4a").write_bytes(b"audio")
repository = JobManifestRepository()
created = repository.create(job, {"path": str(tmp_path / "source.mp3")}, {"model": "gpt-transcribe"}, [
{"index": 0, "path": "chunks/chunk-00000.m4a", "source_start_seconds": "0", "source_end_seconds": "10", "duration_seconds": "10"},
{"index": 1, "path": "chunks/chunk-00001.m4a", "source_start_seconds": "9", "source_end_seconds": "20", "duration_seconds": "11"},
])
return repository, Path(created["manifest_path"])
def test_brazilian_language_and_prompt_preserve_spoken_language() -> None:
prompt = build_prompt(" AWS, PostgreSQL, Brasília ")
assert normalize_language("pt-BR") == "pt"; assert normalize_language("pt_BR") == "pt"; assert normalize_language("en-US") == "en"
assert "não traduza" in prompt.lower(); assert "números" in prompt.lower(); assert prompt.endswith("AWS, PostgreSQL, Brasília")
def test_adapter_calls_verified_audio_transcriptions_interface(tmp_path: Path) -> None:
audio = tmp_path / "chunk.m4a"; audio.write_bytes(b"audio")
endpoint = Endpoint(["Olá, Brasília."]); client = SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))
text = OpenAITranscriptionClient(client).transcribe(audio, "future-compatible-model", "pt-BR", "vocabulário")
assert text == "Olá, Brasília."
call = endpoint.calls[0]
assert call["model"] == "future-compatible-model"; assert call["language"] == "pt"; assert call["prompt"] == "vocabulário"; assert call["response_format"] == "json"; assert call["file"].closed is True
def test_run_saves_each_chunk_and_resume_skips_completed(tmp_path: Path) -> None:
repository, manifest_path = make_job(tmp_path); first_endpoint = Endpoint(["Primeiro", "Segundo"])
service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=first_endpoint))), repository)
settings = AppSettings("gpt-transcribe", "pt-BR", tmp_path, context_vocabulary="Pix")
transcript = service.run(manifest_path, settings)
assert transcript.read_text(encoding="utf-8") == "Primeiro\nSegundo\n"
manifest = repository.load(manifest_path)
assert manifest["state"] == "completed"; assert [item["status"] for item in manifest["chunks"]] == ["completed", "completed"]; assert manifest["chunks"][1]["source_start_seconds"] == "9"
resume_endpoint = Endpoint([])
TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=resume_endpoint))), repository).run(manifest_path, settings)
assert resume_endpoint.calls == []
def test_transient_failure_retries_with_exponential_backoff(tmp_path: Path) -> None:
repository, manifest_path = make_job(tmp_path); endpoint = Endpoint([StatusFailure(429), StatusFailure(503), "Primeiro", "Segundo"]); delays = []
service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository, retry_policy=RetryPolicy(max_attempts=3, initial_delay_seconds=1, max_delay_seconds=10, jitter_ratio=0), sleep=delays.append)
service.run(manifest_path, AppSettings("gpt-transcribe", "pt-BR", tmp_path))
assert delays == [1, 2]; assert len(endpoint.calls) == 4
def test_permanent_api_failure_is_not_retried_and_is_sanitized(tmp_path: Path) -> None:
repository, manifest_path = make_job(tmp_path); endpoint = Endpoint([StatusFailure(400)])
service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository)
with pytest.raises(PermanentTranscriptionError) as caught: service.run(manifest_path, AppSettings("bad-model", "pt-BR", tmp_path))
assert len(endpoint.calls) == 1; assert "sk-leaked-key" not in str(caught.value); assert "sk-leaked-key" not in manifest_path.read_text(encoding="utf-8")
def test_cancelled_job_makes_no_api_request(tmp_path: Path) -> None:
repository, manifest_path = make_job(tmp_path); endpoint = Endpoint(["unexpected"]); token = CancellationToken(); token.cancel()
with pytest.raises(PreprocessingCancelled): TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository).run(manifest_path, AppSettings("gpt-transcribe", "pt-BR", tmp_path), token=token)
assert endpoint.calls == []; assert repository.load(manifest_path)["state"] == "cancelled"
def test_find_resumable_manifest_matches_source_and_effective_settings(tmp_path: Path) -> None:
repository, manifest_path = make_job(tmp_path)
manifest = repository.load(manifest_path)
manifest["settings"] = {"model": "gpt-transcribe", "language": "pt-BR", "context_vocabulary": "Pix"}
repository.save(manifest_path, manifest)
settings = AppSettings("gpt-transcribe", "pt-BR", tmp_path, context_vocabulary="Pix")
assert find_resumable_manifest(tmp_path, Path(manifest["source"]["path"]), settings) == manifest_path
assert find_resumable_manifest(tmp_path, Path(manifest["source"]["path"]), AppSettings("other", "pt-BR", tmp_path, context_vocabulary="Pix")) is None

26
tests/ui/test_app.py

@ -1,26 +0,0 @@
def test_concrete_window_starts_offscreen(qtbot, monkeypatch) -> None:
from voice_transcriptor import app
window = app.create_main_window()
qtbot.addWidget(window)
window.show()
assert window.windowTitle() == "Voice Transcriptor"
def test_openai_client_is_created_lazily_with_sdk_retries_disabled(qtbot, monkeypatch) -> None:
from voice_transcriptor import app
calls = []
class FakeOpenAI:
def __init__(self, **kwargs):
calls.append(kwargs)
self.audio = type("Audio", (), {"transcriptions": object()})()
monkeypatch.setattr(app, "OpenAI", FakeOpenAI)
window = app.create_main_window()
qtbot.addWidget(window)
assert calls == []
service = window.transcription_factory("secret-value")
assert calls == [{"api_key": "secret-value", "max_retries": 0}]
assert service.client.client.audio.transcriptions is not None

91
tests/ui/test_main_window.py

@ -1,91 +0,0 @@
from pathlib import Path
from types import SimpleNamespace
from voice_transcriptor.models import AppSettings, MediaInfo
from voice_transcriptor.ui.main_window import MainWindow
from voice_transcriptor.services.transcription import TranscriptionProgress
class Repository:
def __init__(self, root: Path): self.settings = AppSettings("model", "pt-BR", root)
def load(self): return self.settings, None
def save(self, settings): self.settings = settings
class Credentials:
def set_api_key(self, value): pass
def get_api_key(self): return "test-key"
class Probe:
def probe(self, path): return MediaInfo(path, 10, 60.0, "aac", "60", True, False)
class Preprocessor:
def preprocess(self, *args, **kwargs): raise AssertionError("not started")
def test_main_window_exposes_progress_and_cancel_controls(qtbot, tmp_path: Path) -> None:
window = MainWindow(Probe(), Preprocessor(), Repository(tmp_path), Credentials())
qtbot.addWidget(window)
assert window.prepare_button.text() == "Transcribe"
assert window.cancel_button.text() == "Cancel"
assert window.cancel_button.isEnabled() is False
assert window.progress_bar.value() == 0
assert window.chunk_progress_label.text() == "0 / 0 chunks"
assert window.current_chunk_label.text() == "Current chunk: —"
assert window.elapsed_label.text() == "Elapsed: 00:00:00"
def test_select_file_populates_media_and_enables_preprocessing(qtbot, tmp_path: Path) -> None:
source = tmp_path / "recording.mp3"; source.write_bytes(b"x")
window = MainWindow(Probe(), Preprocessor(), Repository(tmp_path), Credentials())
qtbot.addWidget(window)
window.select_file(source)
assert window.selected_media.path == source
assert window.prepare_button.isEnabled() is True
assert "recording.mp3" in window.file_label.text()
def test_cancel_button_cancels_active_token(qtbot, tmp_path: Path) -> None:
window = MainWindow(Probe(), Preprocessor(), Repository(tmp_path), Credentials())
qtbot.addWidget(window)
window._begin_busy_state()
window.cancel_preprocessing()
assert window._cancellation_token.cancelled is True
def test_transcription_progress_updates_required_status_fields(qtbot, tmp_path: Path) -> None:
window = MainWindow(Probe(), Preprocessor(), Repository(tmp_path), Credentials())
qtbot.addWidget(window)
window._on_progress(TranscriptionProgress(2, 5, 3, 65.0, "retrying", "Rate limited", "OpenAI API transient error (HTTP 429)."))
assert window.chunk_progress_label.text() == "2 / 5 chunks"
assert window.current_chunk_label.text() == "Current chunk: 3"
assert window.elapsed_label.text() == "Elapsed: 00:01:05"
assert "HTTP 429" in window.log.toPlainText()
def test_transcription_job_runs_through_worker_and_restores_controls(qtbot, tmp_path: Path) -> None:
source = tmp_path / "recording.mp3"; source.write_bytes(b"x")
transcript = tmp_path / "transcript.txt"
class DurablePreprocessor:
def preprocess(self, *args, **kwargs):
manifest = tmp_path / "manifest.json"; manifest.write_text("{}", encoding="utf-8")
return SimpleNamespace(manifest_path=manifest)
class Transcriber:
def run(self, manifest_path, settings, token, progress):
progress(TranscriptionProgress(1, 1, 1, 2.0, "transcribing", "Completed chunk 1 of 1"))
transcript.write_text("Olá\n", encoding="utf-8")
return transcript
window = MainWindow(Probe(), DurablePreprocessor(), Repository(tmp_path), Credentials(), lambda key: Transcriber())
qtbot.addWidget(window); window.select_file(source); window.prepare_button.click()
qtbot.waitUntil(lambda: "Transcription complete" in window.log.toPlainText())
assert window.progress_bar.value() == 100
assert window.browse_button.isEnabled()
assert not window.cancel_button.isEnabled()

191
tests/ui/test_settings_dialog.py

@ -1,191 +0,0 @@
from __future__ import annotations
from pathlib import Path
import pytest
from PySide6.QtWidgets import QComboBox, QLineEdit, QMessageBox
from voice_transcriptor.models import AppSettings
from voice_transcriptor.services.credentials import CredentialError
from voice_transcriptor.services.settings import SettingsError
from voice_transcriptor.services.settings import SUPPORTED_MODEL_SUGGESTIONS
from voice_transcriptor.ui.settings_dialog import SettingsDialog
class FakeRepository:
def __init__(self) -> None:
self.saved: list[AppSettings] = []
def save(self, settings: AppSettings) -> None:
self.saved.append(settings)
class FakeCredentials:
def __init__(self, api_key: str | None = None) -> None:
self.api_key = api_key
self.saved_values: list[str] = []
def get_api_key(self) -> str | None:
return self.api_key
def has_api_key(self) -> bool:
return bool(self.api_key)
def set_api_key(self, value: str) -> None:
self.saved_values.append(value)
self.api_key = value
@pytest.fixture
def settings(tmp_path: Path) -> AppSettings:
output_directory = tmp_path / "output"
output_directory.mkdir()
return AppSettings("gpt-4o-transcribe", "pt-BR", output_directory)
def test_populates_fields_from_settings(qtbot, settings: AppSettings) -> None:
dialog = SettingsDialog(settings, FakeRepository(), FakeCredentials())
qtbot.addWidget(dialog)
assert dialog.model_input.text() == settings.model
assert dialog.language_input.text() == settings.language
assert dialog.output_directory_input.text() == str(settings.output_directory)
assert dialog.api_key_input.text() == ""
assert dialog.advanced_panel.isVisible() is False
assert dialog.chunk_duration_input.value() == 900
assert dialog.chunk_overlap_input.value() == 15
assert isinstance(dialog.model_input, QComboBox)
assert dialog.model_input.isEditable()
assert tuple(dialog.model_input.itemText(i) for i in range(dialog.model_input.count())) == SUPPORTED_MODEL_SUGGESTIONS
assert dialog.context_input.toPlainText() == settings.context_vocabulary
def test_advanced_settings_are_saved(qtbot, settings: AppSettings) -> None:
repository = FakeRepository()
dialog = SettingsDialog(settings, repository, FakeCredentials())
qtbot.addWidget(dialog)
dialog.chunk_duration_input.setValue(1200)
dialog.chunk_overlap_input.setValue(20)
dialog.retain_temporary_files_input.setChecked(True)
dialog.save()
assert repository.saved[0].chunk_duration_seconds == 1200
assert repository.saved[0].chunk_overlap_seconds == 20
assert repository.saved[0].retain_temporary_files is True
def test_api_key_input_uses_password_echo_mode(qtbot, settings: AppSettings) -> None:
dialog = SettingsDialog(settings, FakeRepository(), FakeCredentials())
qtbot.addWidget(dialog)
assert dialog.api_key_input.echoMode() is QLineEdit.EchoMode.Password
def test_blank_api_key_preserves_existing_credential(qtbot, settings: AppSettings) -> None:
credentials = FakeCredentials(api_key="stored-token")
repository = FakeRepository()
dialog = SettingsDialog(settings, repository, credentials)
qtbot.addWidget(dialog)
dialog.save()
assert credentials.api_key == "stored-token"
assert credentials.saved_values == []
assert repository.saved == [settings]
def test_save_persists_new_api_key_and_emits_saved_settings(qtbot, settings: AppSettings) -> None:
credentials = FakeCredentials()
repository = FakeRepository()
dialog = SettingsDialog(settings, repository, credentials)
qtbot.addWidget(dialog)
saved_with = qtbot.waitSignal(dialog.settings_saved)
supplied_key = "test-token-for-dialog"
dialog.api_key_input.setText(supplied_key)
dialog.model_input.setText("gpt-4o-mini-transcribe")
dialog.context_input.setPlainText("Pix, Banco do Brasil")
dialog.save()
assert saved_with.args == [
AppSettings("gpt-4o-mini-transcribe", "pt-BR", settings.output_directory, context_vocabulary="Pix, Banco do Brasil")
]
assert credentials.api_key == supplied_key
assert repository.saved == saved_with.args
def test_save_rejects_nonexistent_output_directory(qtbot, settings: AppSettings, tmp_path: Path) -> None:
repository = FakeRepository()
dialog = SettingsDialog(settings, repository, FakeCredentials())
qtbot.addWidget(dialog)
warnings: list[tuple[str, str]] = []
monkeypatch = pytest.MonkeyPatch()
monkeypatch.setattr(
QMessageBox,
"warning",
lambda parent, title, text: warnings.append((title, text)),
)
try:
dialog.output_directory_input.setText(str(tmp_path / "missing"))
dialog.save()
finally:
monkeypatch.undo()
assert repository.saved == []
assert warnings == [("Invalid output directory", "Choose an accessible output directory.")]
assert dialog.isVisible() is False
def test_save_clears_key_field_and_never_writes_key_to_widgets(qtbot, settings: AppSettings) -> None:
dialog = SettingsDialog(settings, FakeRepository(), FakeCredentials())
qtbot.addWidget(dialog)
supplied_key = "test-token-for-dialog"
dialog.api_key_input.setText(supplied_key)
dialog.save()
assert dialog.api_key_input.text() == ""
assert all(supplied_key not in widget.text() for widget in dialog.findChildren(QLineEdit))
def test_service_failures_show_sanitized_message_and_keep_dialog_usable(
qtbot, settings: AppSettings, monkeypatch
) -> None:
class FailingCredentials(FakeCredentials):
def set_api_key(self, value: str) -> None:
raise CredentialError("backend detail must not be shown")
dialog = SettingsDialog(settings, FakeRepository(), FailingCredentials())
qtbot.addWidget(dialog)
warnings: list[tuple[str, str]] = []
monkeypatch.setattr(
QMessageBox,
"warning",
lambda parent, title, text: warnings.append((title, text)),
)
dialog.api_key_input.setText("test-token-for-dialog")
dialog.save()
assert warnings == [("Could not save settings", "The API key could not be saved.")]
assert dialog.result() == 0
assert dialog.api_key_input.text() == ""
def test_settings_repository_failure_is_sanitized(qtbot, settings: AppSettings, monkeypatch) -> None:
class FailingRepository(FakeRepository):
def save(self, settings: AppSettings) -> None:
raise SettingsError("filesystem detail must not be shown")
dialog = SettingsDialog(settings, FailingRepository(), FakeCredentials())
qtbot.addWidget(dialog)
warnings: list[tuple[str, str]] = []
monkeypatch.setattr(
QMessageBox,
"warning",
lambda parent, title, text: warnings.append((title, text)),
)
dialog.save()
assert warnings == [("Could not save settings", "Settings could not be saved.")]
assert dialog.result() == 0
Loading…
Cancel
Save