feat: add local transcription fallback when captions are unavailable (#42)

* feat: add local Whisper transcription fallback

* feat: use local transcription when captions are unavailable

* test: cover local transcription fallback lifecycle

* test: verify preparation fallback selection and provenance

* docs: document local transcription fallback contract

* test: align preparation expectations with local fallback

* test: preserve source provenance on reused preparation

Author: gcomneno

docs/local-transcription-fallback.md

@@ -0,0 +1,72 @@
+# Local transcription fallback
+
+GYTE prefers source-provided captions. Local transcription is a fallback only when inspection found no usable caption.
+
+## Selection order
+
+```text
+usable source caption
+  -> caption acquisition
+otherwise
+  -> private audio acquisition with yt-dlp
+  -> local Whisper transcription
+  -> stable private transcript
+  -> normal normalization/reflow/preparation
+```
+
+The fallback does not turn generated transcription into source-provided evidence. Pipeline provenance records `evidence_origin: local-transcription`; caption-backed runs record `evidence_origin: source-caption`.
+
+## Private artifacts
+
+Fallback artifacts stay inside the private workspace:
+
+- `transcription-source.<ext>` — minimum acquired audio used for local transcription;
+- `.whisper-output/` — transient/local Whisper output directory;
+- `transcription.local.txt` — stable locally generated transcription evidence;
+- the normal `transcript.raw.txt`, normalized and analysis artifacts derived afterwards.
+
+These files must not be committed or published automatically.
+
+## Whisper configuration
+
+The default executable is `whisper` from `PATH` and the default model is `base`.
+
+Optional overrides:
+
+```text
+GYTE_WHISPER_COMMAND=/path/to/whisper
+GYTE_WHISPER_MODEL=tiny|base|small|...
+```
+
+The command override may be an executable path or a command name available on `PATH`.
+
+## Restart and retry semantics
+
+- a completed non-empty `transcription.local.txt` plus its private audio is reused unless `--force` is requested;
+- if audio acquisition completed but transcription did not, the existing audio is reused on retry;
+- partial `yt-dlp` files are not treated as completed evidence;
+- `prepare` is not advanced when local transcription fails;
+- completed preparation preserves the original `source_mode`, `evidence_origin` and local-transcription details when stable outputs are reused.
+
+`--force` requests a fresh transcription attempt while still allowing already acquired private audio to be reused.
+
+## Failure contract
+
+Local fallback failures are explicit and classified internally as:
+
+- `configuration` — required executable/path is unavailable;
+- `download` — private audio acquisition failed or produced no usable audio;
+- `transcription` — Whisper execution failed;
+- `output` — Whisper completed without a usable UTF-8 transcript.
+
+These failures surface as preparation errors and do not silently advance later pipeline stages.
+
+## Authority boundary
+
+```text
+source caption != locally generated transcription
+local transcription != reviewed source lesson
+local transcription != factual authority
+```
+
+Local Whisper output remains derived evidence that may contain recognition errors. It must pass the same later preparation/editorial boundaries as caption-derived material.

src/gyte_study_tools/preparation.py

@@ -18,6 +18,11 @@
     atomic_write_text,
     load_state,
 )
+from gyte_study_tools.transcription import (
+    LocalTranscriptionError,
+    LocalTranscriptionResult,
+    transcribe_locally,
+)
 
 
 RAW_FILENAME = "transcript.raw.txt"
@@ -291,11 +296,21 @@ def build_analysis_markdown(
     return "\n".join(lines)
 
 
+def previous_transcription_record(state_path: Path) -> dict[str, Any]:
+    state = load_state(state_path)
+    stages = state.get("stages")
+    if not isinstance(stages, dict):
+        return {}
+    transcribe = stages.get("transcribe")
+    return transcribe if isinstance(transcribe, dict) else {}
+
+
 def update_pipeline_state(
     state_path: Path,
     video_id: str,
     source_transcript_path: Path,
     source_mode: str,
+    evidence_origin: str,
     raw_path: Path,
     normalized_path: Path,
     analysis_text_path: Path,
@@ -304,17 +319,23 @@ def update_pipeline_state(
     normalized_words: int,
     analysis_words: int,
     reused: bool,
+    local_transcription: dict[str, Any] | None = None,
 ) -> None:
     now = datetime.now(timezone.utc).isoformat()
     state = load_state(state_path)
     stages = state.setdefault("stages", {})
 
-    stages["transcribe"] = {
+    transcribe_stage: dict[str, Any] = {
         "status": "complete",
         "completed_at": now,
         "source_mode": source_mode,
+        "evidence_origin": evidence_origin,
         "source_transcript": source_transcript_path.name,
     }
+    if local_transcription is not None:
+        transcribe_stage["local_transcription"] = local_transcription
+
+    stages["transcribe"] = transcribe_stage
     stages["prepare"] = {
         "status": "complete",
         "completed_at": now,
@@ -387,6 +408,7 @@ def prepare_transcript(
         if isinstance(caption, dict)
         else None
     )
+    previous_transcription = previous_transcription_record(state_path)
 
     if (
         not force
@@ -414,11 +436,27 @@ def prepare_transcript(
                 f"analysis={analysis_words}."
             )
 
+        previous_mode = previous_transcription.get("source_mode")
+        source_mode = (
+            previous_mode
+            if isinstance(previous_mode, str) and previous_mode
+            else "adopted-existing"
+        )
+        previous_origin = previous_transcription.get("evidence_origin")
+        evidence_origin = (
+            previous_origin
+            if isinstance(previous_origin, str) and previous_origin
+            else ("source-caption" if caption is not None else "local-transcription")
+        )
+        previous_local = previous_transcription.get("local_transcription")
+        local_details = previous_local if isinstance(previous_local, dict) else None
+
         update_pipeline_state(
             state_path=state_path,
             video_id=video_id,
             source_transcript_path=source_transcript_path,
-            source_mode="adopted-existing",
+            source_mode=source_mode,
+            evidence_origin=evidence_origin,
             raw_path=raw_path,
             normalized_path=normalized_path,
             analysis_text_path=analysis_text_path,
@@ -427,6 +465,7 @@ def prepare_transcript(
             normalized_words=normalized_words,
             analysis_words=analysis_words,
             reused=True,
+            local_transcription=local_details,
         )
 
         return PreparationResult(
@@ -439,16 +478,31 @@ def prepare_transcript(
             raw_words=raw_words,
             normalized_words=normalized_words,
             analysis_words=analysis_words,
-            source_mode="adopted-existing",
+            source_mode=source_mode,
             reused=True,
         )
 
     source_transcript_path: Path | None = None
     source_mode = ""
+    evidence_origin = ""
+    local_details: dict[str, Any] | None = None
 
     if not force and is_nonempty_file(raw_path):
         source_transcript_path = raw_path
-        source_mode = "stable-existing"
+        previous_mode = previous_transcription.get("source_mode")
+        source_mode = (
+            previous_mode
+            if isinstance(previous_mode, str) and previous_mode
+            else "stable-existing"
+        )
+        previous_origin = previous_transcription.get("evidence_origin")
+        evidence_origin = (
+            previous_origin
+            if isinstance(previous_origin, str) and previous_origin
+            else ("source-caption" if caption is not None else "local-transcription")
+        )
+        previous_local = previous_transcription.get("local_transcription")
+        local_details = previous_local if isinstance(previous_local, dict) else None
     elif isinstance(language, str) and language:
         source_transcript_path = locate_caption_transcript(
             workdir,
@@ -464,15 +518,37 @@ def prepare_transcript(
                 language,
             )
             source_mode = "generated-caption"
+        evidence_origin = "source-caption"
     else:
-        raise PreparationError(
-            "Nessuna caption utilizzabile e fallback audio "
-            "non ancora implementato."
+        try:
+            local_result: LocalTranscriptionResult = transcribe_locally(
+                url,
+                workdir,
+                force=force,
+            )
+        except LocalTranscriptionError as error:
+            raise PreparationError(
+                "Fallback di trascrizione locale fallito "
+                f"({error.kind}): {error}"
+            ) from error
+
+        source_transcript_path = local_result.transcript_path
+        source_mode = (
+            "reused-local-transcription"
+            if local_result.reused_transcript
+            else "generated-local-transcription"
         )
+        evidence_origin = "local-transcription"
+        local_details = {
+            "audio": local_result.audio_path.name,
+            "model": local_result.model,
+            "reused_audio": local_result.reused_audio,
+            "reused_transcript": local_result.reused_transcript,
+        }
 
     if source_transcript_path is None:
         raise PreparationError(
-            "gyte-transcript non ha prodotto un transcript individuabile."
+            "Nessuna sorgente transcript utilizzabile è stata prodotta."
         )
 
     if source_transcript_path != raw_path:
@@ -508,6 +584,7 @@ def prepare_transcript(
         video_id=video_id,
         source_transcript_path=source_transcript_path,
         source_mode=source_mode,
+        evidence_origin=evidence_origin,
         raw_path=raw_path,
         normalized_path=normalized_path,
         analysis_text_path=analysis_text_path,
@@ -516,6 +593,7 @@ def prepare_transcript(
         normalized_words=normalized_words,
         analysis_words=analysis_words,
         reused=False,
+        local_transcription=local_details,
     )
 
     return PreparationResult(