feat: persist preprocessing jobs for transcription

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
+        durable = durable_root is not None
-        if root is not None: root.mkdir(parents=True, exist_ok=True)
+        if durable:
-        job = Path(tempfile.mkdtemp(prefix=f"voice-transcriptor-{uuid.uuid4().hex[:8]}-", dir=root)).resolve()
+            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
+                if durable:
-                self._write_manifest(manifest_path, manifest)
+                    durable_manifest = self.manifest_repository.load(manifest_path)
-            manifest["state"] = "completed"; self._write_manifest(manifest_path, manifest)
+                    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 durable: self.manifest_repository.mark_job_state(manifest_path, "cancelled")
-            if not options.retain_temporary_files: shutil.rmtree(job, ignore_errors=True)
+            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 durable: self.manifest_repository.mark_job_state(manifest_path, "failed", str(exc))
-            if not options.retain_temporary_files: shutil.rmtree(job, ignore_errors=True)
+            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
 

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"))