diff --git a/README.md b/README.md index 2b10208..4a18f47 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,7 @@ Zwischenschritt im temporären Verzeichnis. Die Spurwahl ist dieselbe wie bei | `-l`, `--lang` | Sprache; Standard: Sprache des Original-Audios (`-orig`) | | `-o`, `--output` | Ausgabedatei (Verzeichnisse werden angelegt, vorhandene Datei überschrieben); Standard: `__.txt`, siehe unten | | `-s`, `--start` / `-e`, `--ende` | nur Untertitel in diesem Zeitraum (`00:03:17`, `3:17` oder `197` Sekunden), wie bei `vtt2txt.py` | +| `--exakt` | genau an `-s`/`-e` schneiden statt an Satzgrenzen (siehe `vtt2txt.py`) | | `-f`, `--fliesstext` | Fließtext, auf 80 Zeichen umbrochen (Wörter werden nie getrennt), statt der kurzen Untertitelzeilen; beides versteht `rede_korrigieren.py` | ### Standard-Dateinamen für YouTube-Videos @@ -153,6 +154,13 @@ zweimal, bleibt sie zweimal erhalten. | `-o`, `--output` | Ausgabedatei (nur bei einer Eingabedatei) oder Verzeichnis (vorhanden oder mit `/` am Ende); Standard: `.txt` neben der Eingabe, mit `-s`/`-e` zusätzlich der Zeitraum im Namen | | `-s`, `--start` | nur Untertitel ab dieser Zeit (`00:03:17`, `3:17` oder `197` Sekunden) | | `-e`, `--ende` | nur Untertitel vor dieser Zeit (gleiches Format) | +| `--exakt` | genau an `-s`/`-e` schneiden, auch mitten im Satz | + +Mit `-s`/`-e` werden angeschnittene Sätze an beiden Enden vollständig aufgenommen: +Der Ausschnitt beginnt nach dem letzten Satzende vor `-s` und endet mit dem ersten +Satzende nach `-e`, höchstens 30 Sekunden darüber hinaus (`MAX_SNAP_SECONDS`). +Liegt kein Satzende in Reichweite, wird an dieser Seite genau geschnitten. Der +Dateiname enthält immer die angegebenen Zeiten. Eine fehlerhafte Datei (keine Untertitel, nicht UTF-8) wird gemeldet, die übrigen werden trotzdem umgewandelt; der Exit-Code ist dann 1. @@ -176,7 +184,8 @@ werden trotzdem umgewandelt; der Exit-Code ist dann 1. | `-u`, `--url` | statt `-i`: YouTube-URL; Untertitel wie bei `extract_text.py`. Der Rohtext wird neben der Ausgabe gespeichert (`rede.korrigiert.txt` → `rede.txt`, sonst `.roh.txt`), eine vorhandene, abweichende Datei aber nie überschrieben | | `-l`, `--lang` | Sprache der Rede: `de` oder `en` (auch mit Region, z. B. `en-US`); wählt Prompts und Prüfmarke. Mit `-u` zugleich die Sprache der Untertitel. Standard: `de`, mit `-u` die Sprache des Original-Audios | | `-o`, `--output` | Ausgabedatei; Standard: `.korrigiert.txt` neben der Eingabe, mit `-u` `__.korrigiert.txt`; mit `-s`/`-e` zusätzlich der Zeitraum im Namen | -| `-s`, `--start` / `-e`, `--ende` | nur diesen Zeitraum korrigieren (`00:03:17`, `3:17` oder `197` Sekunden); nur mit `.vtt`-Eingabe oder `-u`, da reiner Text keine Zeitstempel hat | +| `-s`, `--start` / `-e`, `--ende` | nur diesen Zeitraum korrigieren (`00:03:17`, `3:17` oder `197` Sekunden); nur mit `.vtt`-Eingabe oder `-u`, da reiner Text keine Zeitstempel hat. Angeschnittene Sätze werden vollständig aufgenommen (siehe `vtt2txt.py`); zusätzlich weist der Prompt das Modell an, an den Rändern nichts zu ergänzen | +| `--exakt` | genau an `-s`/`-e` schneiden statt an Satzgrenzen | | `-g`, `--glossar` | Glossar mit festen Ersetzungen und Schreibweisen; mehrfach angeben zum Kombinieren (Ersetzungen in Reihenfolge der Angabe) | | `-m`, `--model` | Ollama-Modell; Standard: `qwen3:32b` (Konstante `MODEL`) | | `--kein-bericht` | keinen Änderungsbericht schreiben | diff --git a/extract_text.py b/extract_text.py index 662e611..0bdaf3d 100755 --- a/extract_text.py +++ b/extract_text.py @@ -45,11 +45,12 @@ def parse_args(): return args -def fetch_text(url, lang=None, start=None, end=None): +def fetch_text(url, lang=None, start=None, end=None, exact=False): """Download the subtitles of a video; return (video info, text lines). Without `lang` the language of the original audio is used. `start`/`end` - (seconds) keep only the captions of that time range. + (seconds) keep only the captions of that time range, completed to whole + sentences unless `exact`. """ info = fetch_info(url) print(f"Video: {info.get('title', '?')}") @@ -69,7 +70,7 @@ def fetch_text(url, lang=None, start=None, end=None): vtt = Path(f"{target_base}.{track}.vtt") if not vtt.is_file(): raise SystemExit("yt-dlp hat keine Untertitel-Datei geschrieben.") - lines = vtt_to_lines(vtt.read_text(encoding="utf-8-sig"), start, end) + lines = vtt_to_lines(vtt.read_text(encoding="utf-8-sig"), start, end, exact) if not lines: if start is None and end is None: @@ -80,7 +81,7 @@ def fetch_text(url, lang=None, start=None, end=None): def main(): args = parse_args() - info, lines = fetch_text(args.url, args.lang, args.start, args.ende) + info, lines = fetch_text(args.url, args.lang, args.start, args.ende, args.exakt) suffix = time_suffix(args.start, args.ende) output = args.output or Path(f"{default_name(info)}{suffix}.txt") diff --git a/rede_korrigieren.py b/rede_korrigieren.py index 2c6c19f..8857a8c 100755 --- a/rede_korrigieren.py +++ b/rede_korrigieren.py @@ -142,6 +142,10 @@ PROMPTS = { "glossary": "Glossar: korrekte Schreibweisen von Namen und Begriffen, die in " "dieser Rede vorkommen. Falsch erkannte Varianten davon in der Abschrift " "stillschweigend korrigieren:", + "excerpt": "Der Text ist ein Ausschnitt aus einer längeren Rede und kann mitten " + "im Satz beginnen oder enden. Vervollständige angeschnittene Sätze nicht und " + "ergänze kein Wort, auch nicht in einer Markierung; lass einen unvollständigen " + "Anfang oder Schluss unverändert stehen.", }, "en": { "system": SYSTEM_EN, @@ -150,6 +154,10 @@ PROMPTS = { "glossary": "Glossary: correct spellings of names and terms that occur in " "this speech. Silently correct misrecognized variants of them in the " "transcript:", + "excerpt": "The text is an excerpt from a longer speech and may begin or end " + "in the middle of a sentence. Do not complete cut-off sentences and do not " + "add any word, not even in a mark; leave an incomplete beginning or ending " + "unchanged.", }, } DEFAULT_LANGUAGE = "de" @@ -163,13 +171,16 @@ PARAGRAPH_FORMAT = { } -def build_system(glossary, lang=DEFAULT_LANGUAGE): - """Append glossary terms (correct spellings) to the system prompt.""" +def build_system(glossary, lang=DEFAULT_LANGUAGE, excerpt=False): + """Append the excerpt note and glossary terms (correct spellings) to the system prompt.""" prompts = PROMPTS[lang] + system = prompts["system"] + if excerpt: + system += f"\n{prompts['excerpt']}\n" if not glossary: - return prompts["system"] + return system terms = "\n".join(f"- {term}" for term in glossary) - return f"{prompts['system']}\n{prompts['glossary']}\n{terms}\n" + return f"{system}\n{prompts['glossary']}\n{terms}\n" def read_glossary(path): @@ -544,7 +555,7 @@ def save_raw_text(path, raw): def read_source(args): """Return the raw transcript, the output path and the language of the speech.""" if args.url is not None: - info, lines = fetch_text(args.url, args.lang, args.start, args.ende) + info, lines = fetch_text(args.url, args.lang, args.start, args.ende, args.exakt) lang = base_language(args.lang or original_language(info) or DEFAULT_LANGUAGE) if lang not in PROMPTS: raise SystemExit( @@ -558,7 +569,7 @@ def read_source(args): raw = args.input.read_text(encoding="utf-8-sig") if args.input.suffix.lower() == ".vtt": - raw = "\n".join(vtt_to_lines(raw, args.start, args.ende)) + raw = "\n".join(vtt_to_lines(raw, args.start, args.ende, args.exakt)) return raw, args.output, base_language(args.lang or DEFAULT_LANGUAGE) @@ -583,7 +594,8 @@ def main(): ) terms += more_terms replacements += more_replacements - system = build_system(terms, lang) + excerpt = args.start is not None or args.ende is not None + system = build_system(terms, lang, excerpt) # Subtitle line breaks carry no meaning; send plain running text text = " ".join(raw.split()) diff --git a/tests/test_extract_text.py b/tests/test_extract_text.py index 5f89306..cb89cd9 100644 --- a/tests/test_extract_text.py +++ b/tests/test_extract_text.py @@ -147,3 +147,21 @@ def test_time_range_order_error(monkeypatch, capsys): run_main(monkeypatch, "-u", URL, "-s", "2:00", "-e", "1:00") assert exc.value.code == 2 assert "muss vor -e/--ende liegen" in capsys.readouterr().err + + +CUT_VTT = ( + "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nGuten Abend und\n\n" + "00:00:02.000 --> 00:00:03.000\nwillkommen. Schön, dass\n\n" + "00:00:03.000 --> 00:00:04.000\nSie da sind.\n" +) + + +@pytest.mark.parametrize("exact, text", [ + ([], "Guten Abend und\nwillkommen. Schön, dass\nSie da sind.\n"), + (["--exakt"], "willkommen. Schön, dass\n"), +]) +def test_time_range_sentence_bounds(fake_yt_dlp, tmp_path, monkeypatch, exact, text): + monkeypatch.setattr(yt.subprocess, "run", fake_yt_dlp(GERMAN_VIDEO, vtt=CUT_VTT)) + target = tmp_path / "out.txt" + run_main(monkeypatch, "-u", URL, "-s", "2", "-e", "3", "-o", target, *exact) + assert target.read_text(encoding="utf-8") == text diff --git a/tests/test_rede_korrigieren.py b/tests/test_rede_korrigieren.py index 33ff12f..bb786aa 100644 --- a/tests/test_rede_korrigieren.py +++ b/tests/test_rede_korrigieren.py @@ -615,3 +615,36 @@ def test_ollama_open_passes_http_errors(monkeypatch): monkeypatch.setattr(rk.urllib.request, "urlopen", not_found) with pytest.raises(rk.urllib.error.HTTPError): rk.ollama_open(rk.OLLAMA_URL) + + +# === Excerpts: sentence bounds and prompt note === + +@pytest.mark.parametrize("lang", ["de", "en"]) +def test_build_system_excerpt_note(lang): + prompts = rk.PROMPTS[lang] + assert prompts["excerpt"] not in rk.build_system(["X"], lang) + system = rk.build_system(["X"], lang, excerpt=True) + assert system.startswith(prompts["system"]) + assert system.index(prompts["excerpt"]) < system.index(prompts["glossary"]) + assert rk.build_system([], lang, excerpt=True).endswith(prompts["excerpt"] + "\n") + + +CUT_VTT = ( + "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nDer Kremmel ist\n\n" + "00:00:02.000 --> 00:00:03.000\nin Moskau. Das ist\n\n" + "00:00:03.000 --> 00:00:04.000\nbekannt.\n" +) + + +@pytest.mark.parametrize("exact, raw", [ + ([], "Der Kremmel ist\nin Moskau. Das ist\nbekannt.\n"), + (["--exakt"], "in Moskau. Das ist\n"), +]) +def test_main_url_excerpt(fake_yt_dlp, yt_dlp_installed, tmp_path, monkeypatch, exact, raw): + monkeypatch.setattr(subprocess, "run", fake_yt_dlp(VIDEO, vtt=CUT_VTT)) + systems = recording_model(monkeypatch) + target = tmp_path / "rede.korrigiert.txt" + run_main(monkeypatch, "-u", URL, "-s", "2", "-e", "3", "-o", target, *exact) + + assert (tmp_path / "rede.txt").read_text(encoding="utf-8") == raw + assert rk.PROMPTS["de"]["excerpt"] in systems[0] diff --git a/tests/test_vtt2txt.py b/tests/test_vtt2txt.py index 27cd30e..e586798 100644 --- a/tests/test_vtt2txt.py +++ b/tests/test_vtt2txt.py @@ -188,17 +188,17 @@ def test_time_suffix(start, end, suffix): assert vtt2txt.time_suffix(start, end) == suffix -def test_vtt_to_lines_time_range(): +def test_vtt_to_lines_exact_time_range(): vtt = "WEBVTT\n\n" + cue(1, "A") + cue(2, "A", "B") + cue(3, "B", "C") + cue(4, "C", "D") - assert vtt2txt.vtt_to_lines(vtt, start=2, end=4) == ["B", "C"] - assert vtt2txt.vtt_to_lines(vtt, start=3) == ["C", "D"] - assert vtt2txt.vtt_to_lines(vtt, end=2) == ["A"] + assert vtt2txt.vtt_to_lines(vtt, start=2, end=4, exact=True) == ["B", "C"] + assert vtt2txt.vtt_to_lines(vtt, start=3, exact=True) == ["C", "D"] + assert vtt2txt.vtt_to_lines(vtt, end=2, exact=True) == ["A"] def test_vtt_to_lines_time_range_skips_window_from_before_start(): # "B" was said before the range; its repetition in the first cue must not count vtt = "WEBVTT\n\n" + cue(1, "A", "B") + cue(2, "B", "C") - assert vtt2txt.vtt_to_lines(vtt, start=2) == ["C"] + assert vtt2txt.vtt_to_lines(vtt, start=2, exact=True) == ["C"] def test_main_time_range_default_name(sample_copy, monkeypatch): @@ -225,3 +225,70 @@ def test_cli_time_errors(sample_copy, monkeypatch, capsys, args, message): run_main(monkeypatch, "-i", sample_copy, *args) assert exc.value.code == 2 assert message in capsys.readouterr().err + + +def timed_cue(seconds, *lines): + return f"{vtt2txt.format_time(seconds)}.000 --> {vtt2txt.format_time(seconds + 1)}.000\n" \ + + "\n".join(lines) + "\n\n" + + +def captions(*cues): + """VTT with one new line per cue: cues are (seconds, text).""" + return "WEBVTT\n\n" + "".join(timed_cue(t, text) for t, text in cues) + + +SPEECH = captions( + (100, "Erster Satz ist hier."), (102, "Der zweite Satz"), (104, "beginnt vorher. Dann"), + (106, "kommt der dritte"), (108, "und endet. Der vierte"), (110, "läuft weiter"), + (112, "und endet auch. Schluss"), (114, "folgt."), +) + + +def test_snap_extends_both_sides_to_sentences(): + # Range holds only "kommt der dritte" and "und endet. Der vierte" + assert vtt2txt.vtt_to_lines(SPEECH, start=106, end=110) == [ + "Dann", "kommt der dritte", "und endet. Der vierte", "läuft weiter", + "und endet auch.", + ] + + +def test_snap_keeps_range_already_at_sentence_bounds(): + assert vtt2txt.vtt_to_lines(SPEECH, start=102, end=104) == [ + "Der zweite Satz", "beginnt vorher.", + ] + assert vtt2txt.vtt_to_lines(SPEECH, start=100, end=102) == ["Erster Satz ist hier."] + + +def test_snap_reaches_start_and_end_of_captions(): + assert vtt2txt.vtt_to_lines(SPEECH, start=114) == ["Schluss", "folgt."] + speech = captions((0, "ohne Satzzeichen"), (2, "am Anfang. Rest")) + assert vtt2txt.vtt_to_lines(speech, start=2) == ["ohne Satzzeichen", "am Anfang. Rest"] + + +def test_snap_cuts_exactly_without_sentence_end_in_reach(): + far = vtt2txt.MAX_SNAP_SECONDS + 5 + speech = captions( + (0, "Satz."), (2, "ein endloser"), (far, "Satz ohne"), (2 * far, "Ende"), (3 * far, "jetzt."), + ) + assert vtt2txt.vtt_to_lines(speech, start=far, end=far + 1) == ["Satz ohne"] + + +def test_snap_exact_keeps_cut(): + assert vtt2txt.vtt_to_lines(SPEECH, start=106, end=110, exact=True) == [ + "kommt der dritte", "und endet. Der vierte", + ] + + +def test_snap_closing_quote_belongs_to_sentence(): + speech = captions((0, "Er sagte „Nein.“ Und"), (2, "dann"), (4, "ging er.")) + assert vtt2txt.vtt_to_lines(speech, start=2, end=4) == ["Und", "dann", "ging er."] + + +def test_main_exact_option(tmp_path, monkeypatch): + source = tmp_path / "rede.vtt" + source.write_text(SPEECH, encoding="utf-8") + run_main(monkeypatch, "-i", source, "-s", "106", "-e", "110", "--exakt") + target = tmp_path / "rede_00-01-46_bis_00-01-50.txt" + assert target.read_text(encoding="utf-8") == "kommt der dritte\nund endet. Der vierte\n" + run_main(monkeypatch, "-i", source, "-s", "106", "-e", "110") + assert target.read_text(encoding="utf-8").endswith("und endet auch.\n") diff --git a/vtt2txt.py b/vtt2txt.py index 5336192..03eecec 100755 --- a/vtt2txt.py +++ b/vtt2txt.py @@ -13,6 +13,10 @@ TAG = re.compile(r"<[^>]*>") GLOB_CHARS = re.compile(r"[*?\[]") TIME = re.compile(r"(?:(\d+):)??(?:(\d+):)?(\d+(?:[.,]\d+)?)") CUE_START = re.compile(r"^\s*([\d:.,]+)\s*-->") +SENTENCE_END = re.compile(r"[.!?…][\"'»«“”)\]]*(?=\s|$)") +LINE_ENDS_SENTENCE = re.compile(r"[.!?…][\"'»«“”)\]]*$") + +MAX_SNAP_SECONDS = 30 # so weit darf ein Zeitraum bis zur nächsten Satzgrenze wachsen def parse_time(text): @@ -61,6 +65,11 @@ def add_time_options(parser): "-e", "--ende", type=time_option, metavar="ZEIT", help="nur Untertitel vor dieser Zeit, z. B. 00:05:25", ) + parser.add_argument( + "--exakt", action="store_true", + help="genau an -s/-e schneiden, statt angeschnittene Sätze vollständig " + f"aufzunehmen (bis {MAX_SNAP_SECONDS} s über den Zeitraum hinaus)", + ) def check_time_range(parser, args): @@ -117,16 +126,12 @@ def in_range(time, start, end): return (start is None or time >= start) and (end is None or time < end) -def vtt_to_lines(vtt, start=None, end=None): - """Return the caption text lines of a VTT document without rolling duplicates. +def timed_lines(vtt): + """Return (cue start in seconds, line) for the new caption lines of a VTT document. Auto captions show a rolling window: each cue repeats the end of the previous cue before adding new text. Only the lines after that overlap are new, so a line the speaker really says twice in a row is kept. - - With `start`/`end` (seconds) only the new lines of cues beginning in that - range are kept. The window is followed through all cues, so a line repeated - from before `start` does not slip in. """ lines = [] previous = [] @@ -135,21 +140,82 @@ def vtt_to_lines(vtt, start=None, end=None): # Empty cues (YouTube transitions) say nothing about the window if not cue: continue - if in_range(time, start, end): - lines.extend(cue[overlap(previous, cue):]) + lines.extend((time, line) for line in cue[overlap(previous, cue):]) previous = cue return lines -def convert(source, target, start=None, end=None): +def sentence_start(timed, first, start): + """Lines before index `first` back to the start of the sentence cut by `start`.""" + if start is None or first == 0 or LINE_ENDS_SENTENCE.search(timed[first - 1][1]): + return [] + extra = [] + for time, line in reversed(timed[:first]): + if time is None or time < start - MAX_SNAP_SECONDS: + return [] + ends = list(SENTENCE_END.finditer(line)) + if ends: + rest = line[ends[-1].end():].strip() + return ([rest] if rest else []) + extra + extra.insert(0, line) + # Reached the first caption: the speech starts there + return extra + + +def sentence_end(timed, last, end): + """Lines after index `last` up to the end of the sentence cut by `end`.""" + if end is None or LINE_ENDS_SENTENCE.search(timed[last][1]): + return [] + extra = [] + for time, line in timed[last + 1:]: + if time is None or time >= end + MAX_SNAP_SECONDS: + return [] + match = SENTENCE_END.search(line) + if match: + return extra + [line[:match.end()]] + extra.append(line) + # Reached the last caption: the speech ends there + return extra + + +def snap_to_sentences(timed, start, end): + """Lines in the range, extended so that no sentence is cut at either side. + + Without a sentence end within MAX_SNAP_SECONDS that side is cut exactly. + """ + inside = [i for i, (time, _) in enumerate(timed) if in_range(time, start, end)] + if not inside: + return [] + first, last = inside[0], inside[-1] + lines = [line for _, line in timed[first:last + 1]] + return sentence_start(timed, first, start) + lines + sentence_end(timed, last, end) + + +def vtt_to_lines(vtt, start=None, end=None, exact=False): + """Return the caption text lines of a VTT document without rolling duplicates. + + With `start`/`end` (seconds) only the new lines of cues beginning in that + range are kept, completed to whole sentences unless `exact`. The window is + followed through all cues, so a line repeated from before `start` does not + slip in. + """ + timed = timed_lines(vtt) + if start is None and end is None: + return [line for _, line in timed] + if exact: + return [line for time, line in timed if in_range(time, start, end)] + return snap_to_sentences(timed, start, end) + + +def convert(source, target, start=None, end=None, exact=False): """Convert one file; return an error message or None on success.""" try: vtt = source.read_text(encoding="utf-8-sig") except UnicodeDecodeError: return f"Keine UTF-8-Datei: {source}" - lines = vtt_to_lines(vtt, start, end) + lines = vtt_to_lines(vtt, start, end, exact) if not lines: where = "" if start is None and end is None else " im gewählten Zeitraum" return f"Keine Untertitel{where} gefunden: {source}" @@ -238,7 +304,7 @@ def main(): failed = 0 # Keep going on errors so one bad file does not stop a batch for source, target in args.jobs: - error = convert(source, target, args.start, args.ende) + error = convert(source, target, args.start, args.ende, args.exakt) if error: print(f"FEHLER: {error}", file=sys.stderr) failed += 1