diff --git a/src/ccn_transcribe/backends/openvino_backend.py b/src/ccn_transcribe/backends/openvino_backend.py index 85488db..0fef42c 100644 --- a/src/ccn_transcribe/backends/openvino_backend.py +++ b/src/ccn_transcribe/backends/openvino_backend.py @@ -206,6 +206,8 @@ class _OpenvinoBackend: model_revision=self.revision, audio_s=audio_s, wall_s=wall, + task=request.task, + num_beams=request.num_beams, ) def close(self) -> None: diff --git a/src/ccn_transcribe/formats.py b/src/ccn_transcribe/formats.py index 8962638..67f26db 100644 --- a/src/ccn_transcribe/formats.py +++ b/src/ccn_transcribe/formats.py @@ -61,6 +61,8 @@ def to_json(result: TranscriptResult, source: Mapping[str, object] | None = None "backend": result.backend, "device": result.device, "model": {"id": result.model_id, "revision": result.model_revision}, + "task": result.task, + "num_beams": result.num_beams, "audio_s": result.audio_s, "wall_s": result.wall_s, "rtf": result.rtf, diff --git a/src/ccn_transcribe/jobs/recover.py b/src/ccn_transcribe/jobs/recover.py new file mode 100644 index 0000000..9689ade --- /dev/null +++ b/src/ccn_transcribe/jobs/recover.py @@ -0,0 +1,120 @@ +"""Rebuilding a job record when state.json is gone or unreadable. + +Artifacts are the source of truth, so most of the record can be read back off +the filesystem. Two things the filesystem cannot answer are recovered from the +files that do record them: *why* an artifact is absent comes from events.jsonl, +and how the transcript was produced comes from transcript.json. Without this a +damaged state.json costs a full re-download and re-transcription of work that is +sitting on disk. +""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING, Any + +from ccn_transcribe import errors, formats +from ccn_transcribe.jobs import store +from ccn_transcribe.jobs.state import Artifact, ArtifactStatus, JobState, Outputs +from ccn_transcribe.jobs.state import TranscriptionRecord as Record +from ccn_transcribe.media import ffmpeg +from ccn_transcribe.stages import download as download_stage + +if TYPE_CHECKING: + from pathlib import Path + + from ccn_transcribe.paths import JobPaths + +log = logging.getLogger(__name__) + + +def rebuild(job: JobPaths, *, job_id: str, source_url: str) -> JobState: + """Reconstruct as much of the record as the job directory can support.""" + state = JobState.new(job_id=job_id, source_url=source_url) + _scan_media(job, state) + _scan_outputs(job, state) + _restore_transcription(job, state) + _replay_events(job, state) + return state + + +def _safe_duration(path: Path) -> float | None: + try: + return ffmpeg.probe_duration(path) + except errors.ProbeError: + return None + + +def _found(job: JobPaths, path: Path) -> Artifact: + return Artifact( + status=ArtifactStatus.PRESENT, + path=str(path.relative_to(job.root)), + size=path.stat().st_size, + duration_s=_safe_duration(path), + ) + + +def _scan_media(job: JobPaths, state: JobState) -> None: + video = download_stage.find_video(job) + if video is not None: + state.video = _found(job, video) + state.duration_s = state.video.duration_s + if job.audio_file.exists(): + state.audio = _found(job, job.audio_file) + + +def _scan_outputs(job: JobPaths, state: JobState) -> None: + written = { + path.suffix.lstrip("."): str(path.relative_to(job.root)) + for path in sorted(job.out_dir.glob("transcript.*")) + if path.suffix.lstrip(".") in formats.FORMATS + } + if written: + state.outputs = Outputs(status=ArtifactStatus.PRESENT, paths=written) + + +def _restore_transcription(job: JobPaths, state: JobState) -> None: + path = job.transcript_file("json") + if not path.exists(): + return + try: + doc: dict[str, Any] = json.loads(path.read_text(encoding="utf-8")) + model: dict[str, Any] = doc["model"] + state.transcription = Record( + model_id=model["id"], + model_revision=model.get("revision"), + backend=doc["backend"], + device=doc["device"], + language=doc.get("language"), + # Absent in transcripts written before these were recorded. Defaulting + # can only disagree with what the user asks for next, and a disagreement + # costs a re-transcription rather than a wrong answer. + task=doc.get("task", "transcribe"), + num_beams=doc.get("num_beams", 1), + segments=len(doc.get("segments", ())), + audio_s=doc.get("audio_s"), + wall_s=doc.get("wall_s"), + rtf=doc.get("rtf"), + ) + except (OSError, ValueError, KeyError, TypeError): + log.warning("%s is unreadable; the transcription record cannot be recovered", path.name) + return + + state.duration_s = state.duration_s or state.transcription.audio_s + source: dict[str, Any] = doc.get("source") or {} + state.title = source.get("title") or state.title + + +def _replay_events(job: JobPaths, state: JobState) -> None: + for event in store.read_events(job): + if event.get("event") != "deleted": + continue + artifact = {"video": state.video, "audio": state.audio}.get(str(event.get("artifact"))) + # A deletion explains an absence. If the file is back, it was re-fetched + # since, and the event is history rather than the current truth. + if artifact is None or artifact.status is ArtifactStatus.PRESENT: + continue + artifact.status = ArtifactStatus.DELETED_BY_POLICY + artifact.reason = event.get("reason") + artifact.deleted_at = event.get("at") diff --git a/src/ccn_transcribe/jobs/runner.py b/src/ccn_transcribe/jobs/runner.py index 3fe59f7..c700634 100644 --- a/src/ccn_transcribe/jobs/runner.py +++ b/src/ccn_transcribe/jobs/runner.py @@ -8,7 +8,7 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from ccn_transcribe import errors -from ccn_transcribe.jobs import store, verify +from ccn_transcribe.jobs import recover, store, verify from ccn_transcribe.jobs.plan import Stage, Verdicts, plan_stages from ccn_transcribe.jobs.state import Artifact, ArtifactStatus, JobState, Outputs from ccn_transcribe.jobs.state import TranscriptionRecord as Record @@ -96,7 +96,7 @@ def run_job( ) -> JobOutcome: """Run whatever stages this job still needs.""" job.ensure() - state = store.load(job) or JobState.new(job_id=job.root.name, source_url=url) + state = store.load(job) or recover.rebuild(job, job_id=job.root.name, source_url=url) if entry: state.title = entry.get("title") or state.title duration = entry.get("duration") @@ -256,6 +256,8 @@ def _rerender(job: JobPaths, config: RunConfig, state: JobState) -> dict[str, st model_revision=record.model_revision, audio_s=record.audio_s or 0.0, wall_s=record.wall_s or 0.0, + task=record.task, + num_beams=record.num_beams, ) return outputs_stage.write(job, result, config.formats, source=_source(state)) diff --git a/src/ccn_transcribe/transcript.py b/src/ccn_transcribe/transcript.py index bce6d57..c5850bb 100644 --- a/src/ccn_transcribe/transcript.py +++ b/src/ccn_transcribe/transcript.py @@ -36,6 +36,8 @@ class TranscriptResult: model_revision: str | None audio_s: float wall_s: float + task: str = "transcribe" + num_beams: int = 1 @property def text(self) -> str: diff --git a/tests/test_jobs_recover.py b/tests/test_jobs_recover.py new file mode 100644 index 0000000..2e6d9e7 --- /dev/null +++ b/tests/test_jobs_recover.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from ccn_transcribe import paths +from ccn_transcribe.jobs import recover, store +from ccn_transcribe.jobs.state import ArtifactStatus + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def job(tmp_path: Path) -> paths.JobPaths: + j = paths.Workspace(tmp_path).job("youtube-abc") + j.ensure() + return j + + +def _transcript(job: paths.JobPaths, **over: object) -> None: + doc: dict[str, object] = { + "text": "hello", + "language": "en", + "backend": "openvino", + "device": "GPU", + "model": {"id": "OpenVINO/whisper", "revision": "abc123"}, + "task": "translate", + "num_beams": 3, + "audio_s": 212.8, + "wall_s": 52.9, + "rtf": 4.02, + "segments": [{"start": 0.0, "end": 1.0, "text": "hello"}], + "source": {"id": "youtube-abc", "title": "A Title", "url": "https://a.test/v"}, + } + doc.update(over) + job.transcript_file("json").write_text(json.dumps(doc), encoding="utf-8") + + +class TestRebuildFromArtifacts: + def test_an_empty_job_recovers_an_empty_record(self, job: paths.JobPaths) -> None: + state = recover.rebuild(job, job_id="youtube-abc", source_url="https://a.test/v") + assert state.video.status is ArtifactStatus.MISSING + assert state.audio.status is ArtifactStatus.MISSING + assert state.transcription is None + + def test_a_downloaded_video_is_found(self, job: paths.JobPaths, sine_video: Path) -> None: + # Without this a corrupt state.json re-downloads a file already on disk. + (job.media_dir / "video.mkv").write_bytes(sine_video.read_bytes()) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.video.status is ArtifactStatus.PRESENT + assert state.video.path == "media/video.mkv" + assert state.video.size == sine_video.stat().st_size + assert state.video.duration_s == pytest.approx(1.0, abs=0.2) + + def test_an_unmerged_stream_is_not_mistaken_for_the_video( + self, job: paths.JobPaths, sine_video: Path + ) -> None: + (job.media_dir / "video.f251.webm").write_bytes(sine_video.read_bytes()) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.video.status is ArtifactStatus.MISSING + + def test_extracted_audio_is_found(self, job: paths.JobPaths, sine_wav: Path) -> None: + from ccn_transcribe.media import ffmpeg + + ffmpeg.extract_flac(sine_wav, job.audio_file) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.audio.status is ArtifactStatus.PRESENT + assert state.audio.path == "media/audio.flac" + assert state.audio.duration_s == pytest.approx(1.0, abs=0.2) + + def test_unreadable_media_recovers_without_a_duration(self, job: paths.JobPaths) -> None: + (job.media_dir / "video.mkv").write_bytes(b"not a video") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.video.status is ArtifactStatus.PRESENT + assert state.video.duration_s is None + + def test_written_outputs_are_found(self, job: paths.JobPaths) -> None: + for name in ("txt", "srt", "json"): + job.transcript_file(name).write_text("x", encoding="utf-8") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.outputs.status is ArtifactStatus.PRESENT + assert state.outputs.paths == { + "json": "out/transcript.json", + "srt": "out/transcript.srt", + "txt": "out/transcript.txt", + } + + def test_a_stray_temp_file_is_not_an_output(self, job: paths.JobPaths) -> None: + job.transcript_file("txt").with_suffix(".txt.tmp").write_text("x", encoding="utf-8") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.outputs.paths == {} + + +class TestRestoreTranscription: + def test_the_record_comes_back_from_the_transcript(self, job: paths.JobPaths) -> None: + # Without it params_changed sees no record, and a completed job is + # re-transcribed from scratch despite every segment being on disk. + _transcript(job) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.transcription is not None + assert state.transcription.model_id == "OpenVINO/whisper" + assert state.transcription.model_revision == "abc123" + assert state.transcription.language == "en" + assert state.transcription.task == "translate" + assert state.transcription.num_beams == 3 + assert state.transcription.device == "GPU" + assert state.transcription.segments == 1 + assert state.transcription.rtf == pytest.approx(4.02) + + def test_the_title_comes_back_for_out_dir_naming(self, job: paths.JobPaths) -> None: + _transcript(job) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.title == "A Title" + assert state.duration_s == pytest.approx(212.8) + + def test_a_transcript_from_before_task_was_recorded(self, job: paths.JobPaths) -> None: + # Older transcripts omit task/num_beams; defaulting them is the safe + # direction, since a mismatch only costs a re-transcription. + _transcript(job) + doc = json.loads(job.transcript_file("json").read_text(encoding="utf-8")) + del doc["task"], doc["num_beams"] + job.transcript_file("json").write_text(json.dumps(doc), encoding="utf-8") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.transcription is not None + assert state.transcription.task == "transcribe" + assert state.transcription.num_beams == 1 + + def test_an_unreadable_transcript_leaves_no_record(self, job: paths.JobPaths) -> None: + job.transcript_file("json").write_text("{ truncated", encoding="utf-8") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.transcription is None + + def test_a_transcript_missing_its_model_block(self, job: paths.JobPaths) -> None: + _transcript(job, model="not a mapping") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.transcription is None + + +def _deleted(job: paths.JobPaths, which: str, reason: str) -> None: + store.append_event(job, {"event": "deleted", "artifact": which, "reason": reason}) + + +class TestReplayEvents: + def test_a_policy_deletion_is_recovered(self, job: paths.JobPaths) -> None: + # The whole point of events.jsonl: without it this reads as data loss and + # triggers a re-download of media the user asked us to delete. + _deleted(job, "video", "--no-retain-video") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.video.status is ArtifactStatus.DELETED_BY_POLICY + assert state.video.reason == "--no-retain-video" + assert state.video.deleted_at is not None + + def test_audio_deletions_too(self, job: paths.JobPaths) -> None: + _deleted(job, "audio", "--no-retain-audio") + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.audio.status is ArtifactStatus.DELETED_BY_POLICY + + def test_a_superseded_deletion_does_not_hide_a_present_file( + self, job: paths.JobPaths, sine_video: Path + ) -> None: + # Deleted once, downloaded again later. The event is history, not truth: + # the file on disk wins. + _deleted(job, "video", "--no-retain-video") + (job.media_dir / "video.mkv").write_bytes(sine_video.read_bytes()) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.video.status is ArtifactStatus.PRESENT + + def test_unrelated_events_are_ignored(self, job: paths.JobPaths) -> None: + store.append_event(job, {"event": "started"}) + store.append_event(job, {"event": "deleted", "artifact": "outputs"}) + state = recover.rebuild(job, job_id="youtube-abc", source_url="u") + assert state.video.status is ArtifactStatus.MISSING + assert state.audio.status is ArtifactStatus.MISSING