Zum Inhalt

transcript_editor.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
"""Transcript editor helpers — load, save, regenerate, invalidate.

Pure Python, no FastAPI dependency. Tested standalone in
tests/test_transcript_editor.py.
"""

import json
import shutil
from pathlib import Path
from typing import Any

from transcript_align import propagate_text_to_words


def load_segments(json_path: str) -> list[dict[str, Any]]:
    """Return the list of segments from a WhisperX JSON file.

    Each segment dict carries: start, end, text, speaker, words (list),
    _edited (bool, defaults False if absent).
    """
    data = json.loads(Path(json_path).read_text(encoding="utf-8"))
    segments: list[dict[str, Any]] = data.get("segments", [])
    return segments


def _backup_path_for(json_path: Path) -> Path:
    """Given <stem>.whisperx.json, return <stem>.whisperx.original.json."""
    # json_path.stem strips only the LAST .json suffix → keep ".whisperx" intact
    return json_path.with_name(json_path.stem + ".original.json")


def save_edits(json_path: str, new_texts: list[str]) -> dict[str, Any]:
    """Update each segment's text from new_texts (parallel list).

    On first save, creates a one-time backup at <stem>.whisperx.original.json.
    Sets `_edited: true` on segments where the new text differs from the existing.
    Returns: {"total_segments": int, "edited_count": int, "backup_created": bool}.
    """
    path = Path(json_path)
    backup_path = _backup_path_for(path)
    backup_created = False
    if not backup_path.exists():
        shutil.copy2(path, backup_path)
        backup_created = True

    data = json.loads(path.read_text(encoding="utf-8"))
    segments = data.get("segments", [])
    if len(new_texts) != len(segments):
        raise ValueError(f"new_texts length {len(new_texts)} != segments length {len(segments)}")
    edited_count = 0
    for seg, new_text in zip(segments, new_texts, strict=True):
        if seg.get("text") != new_text:
            seg["text"] = new_text
            seg["_edited"] = True
            edited_count += 1
            # The renderer reads words, not text — carry the edit across or the
            # correction never reaches the MP4.
            propagate_text_to_words(seg)
    path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
    regenerate_srt_txt(str(path))
    return {
        "total_segments": len(segments),
        "edited_count": edited_count,
        "backup_created": backup_created,
    }


def has_been_edited(json_path: str) -> bool:
    """Return True if any segment in the JSON has _edited: true."""
    path = Path(json_path)
    if not path.exists():
        return False
    data = json.loads(path.read_text(encoding="utf-8"))
    return any(seg.get("_edited") for seg in data.get("segments", []))


def _format_srt_time(seconds: float) -> str:
    """HH:MM:SS,mmm — matches transcribe.format_srt_time byte-for-byte."""
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    ms = int((seconds % 1) * 1000)
    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"


def regenerate_srt_txt(json_path: str) -> tuple[str, str]:
    """Rewrite <stem>.srt and <stem>.txt siblings from the JSON segments.

    Same format as transcribe.py — SRT carries [SPEAKER_id] prefix per cue,
    TXT groups consecutive same-speaker segments under a speaker header.
    Returns (srt_path, txt_path) as strings.
    """
    path = Path(json_path)
    data = json.loads(path.read_text(encoding="utf-8"))
    segments = data.get("segments", [])
    # Strip ".whisperx" from path.stem to recover the base stem ("ep01.whisperx" → "ep01")
    base_stem = path.stem
    if base_stem.endswith(".whisperx"):
        base_stem = base_stem[: -len(".whisperx")]
    srt_path = path.parent / f"{base_stem}.srt"
    txt_path = path.parent / f"{base_stem}.txt"

    with srt_path.open("w", encoding="utf-8") as f:
        for i, seg in enumerate(segments, 1):
            speaker = seg.get("speaker", "SPEAKER_00")
            text = f"[{speaker}] {seg['text'].strip()}"
            f.write(
                f"{i}\n"
                f"{_format_srt_time(seg['start'])} --> {_format_srt_time(seg['end'])}\n"
                f"{text}\n\n"
            )

    with txt_path.open("w", encoding="utf-8") as f:
        current_speaker = None
        for seg in segments:
            speaker = seg.get("speaker", "SPEAKER_00")
            if speaker != current_speaker:
                current_speaker = speaker
                f.write(f"\n{speaker}:\n")
            f.write(seg["text"].strip() + " ")
        f.write("\n")

    return str(srt_path), str(txt_path)


def invalidate_downstream(run_state_path: str) -> list[str]:
    """Reset meta + render phases to 'pending' if they were 'done'.

    Returns list of invalidated phase names. Upload is intentionally left
    alone — V1 treats edits as affecting transcript-derived artifacts only.
    """
    path = Path(run_state_path)
    if not path.exists():
        return []
    state = json.loads(path.read_text(encoding="utf-8"))
    phases = state.setdefault("phases", {})
    invalidated: list[str] = []
    for phase in ("meta", "render"):
        entry = phases.get(phase, {})
        if entry.get("status") == "done":
            phases[phase] = {"status": "pending"}
            invalidated.append(phase)
    if invalidated:
        path.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
    return invalidated