From 20bcb0803d0bc92ae01df185341a59ba921fda8c Mon Sep 17 00:00:00 2001 From: Yutsuo Date: Sun, 30 Aug 2026 21:09:10 -0300 Subject: [PATCH] feat: persist preprocessing jobs for transcription --- .../services/preprocessing.py | 56 ++++++++++++++----- tests/test_preprocessing.py | 30 ++++++++++ 2 files changed, 72 insertions(+), 14 deletions(-) diff --git a/src/voice_transcriptor/services/preprocessing.py b/src/voice_transcriptor/services/preprocessing.py index 3dc0da2..74fda56 100644 --- a/src/voice_transcriptor/services/preprocessing.py +++ b/src/voice_transcriptor/services/preprocessing.py @@ -12,6 +12,7 @@ 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"}) @@ -61,13 +62,14 @@ class PreprocessingResult: class PreprocessingService: - def __init__(self, ffmpeg_path: Path, probe_service, temporary_root: Path | None = None, process_factory=subprocess.Popen) -> None: + 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) -> PreprocessingResult: + def preprocess(self, source: Path, options: PreprocessingOptions, token: CancellationToken | None = None, progress: Callable[[PreprocessingProgress], None] | None = None, durable_root: Path | None = None) -> PreprocessingResult: token = token or CancellationToken() source = source.resolve() if source.suffix.lower() not in SUPPORTED_EXTENSIONS: raise PreprocessingError("Unsupported media format.") @@ -78,14 +80,30 @@ class PreprocessingService: 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) - 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() + 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} - self._write_manifest(manifest_path, manifest) + if durable: + created = self.manifest_repository.create( + job, + manifest["source"], + {**manifest["settings"], "model": None, "language": None}, + 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: @@ -111,18 +129,28 @@ class PreprocessingService: 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 - manifest["completed_chunks"] = boundary.index + 1 - self._write_manifest(manifest_path, manifest) - manifest["state"] = "completed"; self._write_manifest(manifest_path, manifest) + 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, options.retain_temporary_files, job) + return PreprocessingResult(job, manifest_path, durable or options.retain_temporary_files, job) except PreprocessingCancelled: - manifest["state"] = "cancelled"; self._write_manifest(manifest_path, manifest) - if not options.retain_temporary_files: shutil.rmtree(job, ignore_errors=True) + 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: - manifest["state"] = "failed"; manifest["error"] = str(exc); self._write_manifest(manifest_path, manifest) - if not options.retain_temporary_files: shutil.rmtree(job, ignore_errors=True) + 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 diff --git a/tests/test_preprocessing.py b/tests/test_preprocessing.py index 2309a38..b142060 100644 --- a/tests/test_preprocessing.py +++ b/tests/test_preprocessing.py @@ -5,6 +5,7 @@ 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: @@ -56,6 +57,35 @@ def test_cleanup_and_retention_are_job_scoped(tmp_path: Path) -> None: 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"))