diff --git a/README.md b/README.md index 4a18f47..633bfc1 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,8 @@ Zwischenschritt im temporären Verzeichnis. Die Spurwahl ist dieselbe wie bei # → John_Mearsheimer_Offensive_Realism_Episode_1_Origins_of_the_War_In_Ukraine_2026-07-30.txt ./extract_text.py -u URL -l de -o testdaten/rede.txt # Sprache und Ziel ./extract_text.py -u URL -f -o rede.txt # lesbarer Fließtext -./extract_text.py -u URL -s 3:17 -e 5:25 # nur ein Ausschnitt +./extract_text.py -u URL -s 0:03:17 -e 0:05:25 # nur ein Ausschnitt +./extract_text.py -u URL -z "00:05-00:06 00:10-00:12 00:25" # mehrere Bereiche ./rede_korrigieren.py -i rede.txt # → rede.korrigiert.txt ``` @@ -77,8 +78,8 @@ Zwischenschritt im temporären Verzeichnis. Die Spurwahl ist dieselbe wie bei | `-u`, `--url` | YouTube-URL, Pflicht | | `-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`) | +| `-s`, `--start` / `-e`, `--ende` / `-z`, `--zeitraeume` | nur Untertitel in diesen Zeiträumen, siehe [Zeitauswahl](#zeitauswahl--s--e--z) | +| `--exakt` | genau an den Zeiten schneiden statt an Satzgrenzen | | `-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 @@ -94,8 +95,9 @@ Ohne `-o` bzw. `-n` bilden `extract_text.py`, `yt_untertitel.py` und Wörter werden mit `_` verbunden. Bindestriche und Punkte im Wort bleiben („Krone-Schmalz“), andere Zeichen fallen weg. Der Kanal wird auf 40, der Titel auf 60 Zeichen an einer Wortgrenze gekürzt. Ohne verwertbaren Kanal und Titel gilt die -Video-ID. Mit `-s`/`-e` kommt der Zeitraum dazu, z. B. -`…_2026-07-30_00-03-17_bis_00-05-25` (nur `-s`: `_ab_…`, nur `-e`: `_bis_…`). +Video-ID. Mit `-s`/`-e`/`-z` kommen die Zeiträume dazu, z. B. +`…_2026-07-30_00-03-17_bis_00-05-25` (nur `-s`: `_ab_…`, nur `-e`: `_bis_…`; +mehrere Bereiche mit `+` verbunden: `…_00-05-00_bis_00-06-00+ab_00-25-00`). Gleiches Video ergibt immer denselben Namen; eine vorhandene Datei wird überschrieben (ein mit `-u` gespeicherter, abweichender Rohtext aber nie). @@ -145,22 +147,62 @@ zweimal, bleibt sie zweimal erhalten. ./vtt2txt.py -i "ramstein.de*.vtt" # Muster löst das Programm auf (Windows) ./vtt2txt.py -i rede.de.vtt -o out/rede.txt # eigene Ausgabedatei ./vtt2txt.py -i *.vtt -o out/ # alle in ein Verzeichnis -./vtt2txt.py -i rede.de.vtt -s 3:17 -e 5:25 # nur ein Ausschnitt → rede.de_00-03-17_bis_00-05-25.txt +./vtt2txt.py -i rede.de.vtt -s 0:03:17 -e 0:05:25 # Ausschnitt → rede.de_00-03-17_bis_00-05-25.txt +./vtt2txt.py -i rede.de.vtt -z "00:05-00:06 00:25" # mehrere Bereiche in einer Datei ``` | Option | Bedeutung | |---|---| | `-i`, `--input` | eine oder mehrere VTT-Dateien oder Muster, Pflicht | -| `-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 | +| `-o`, `--output` | Ausgabedatei (nur bei einer Eingabedatei) oder Verzeichnis (vorhanden oder mit `/` am Ende); Standard: `.txt` neben der Eingabe, mit `-s`/`-e`/`-z` zusätzlich die Zeiträume im Namen | +| `-s`, `--start` | nur Untertitel ab dieser Zeit | +| `-e`, `--ende` | nur Untertitel vor dieser Zeit | +| `-z`, `--zeitraeume` | mehrere Bereiche statt `-s`/`-e` | +| `--exakt` | genau an den Zeiten 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. +#### Zeitauswahl (`-s`, `-e`, `-z`) + +Gilt gleich für `vtt2txt.py`, `extract_text.py` und `rede_korrigieren.py`. + +| Zeit | Bedeutung | +|---|---| +| `00:05:10` | Stunden:Minuten:Sekunden (auch `00:5:10`, Sekunden mit Nachkommastellen) | +| `00:20` | Stunden:Minuten, also Minute 20 | +| `25` | Minute 25 | + +| Aufruf | Text | +|---|---| +| `-s 00:05:10` | ab 5:10 bis zum Ende des Videos | +| `-e 00:20` | vom Anfang bis Minute 20 | +| `-s 00:05 -e 00:06` | von Minute 5 bis Minute 6 | +| `-z "00:5:00-00:6:00 00:10-00:12:00 00:25"` | Minute 5–6, Minute 10–12 und alles ab Minute 25 | + +Bei `-z` werden die Bereiche durch Leerzeichen getrennt: `A-B`, `A` (ab A bis +zum Ende) oder `-B` (vom Anfang bis B). Sie dürfen sich nicht überschneiden und +werden nach der Zeit sortiert. + +Alle Bereiche landen in **einer** Datei. Vor jedem Bereich steht eine Überschrift +mit seinen tatsächlichen Zeiten: + +```text +[00:04:57 - 00:06:04] +Ihr seht, es gibt keinen einzigen Dialog … + +[00:09:58 - 00:12:03] +Also, wenn ihr Programme baut mit Claude Code, … +``` + +Angeschnittene Sätze werden an beiden Enden vollständig aufgenommen. Der +Ausschnitt beginnt nach dem letzten Satzende vor dem Start und endet mit dem +ersten Satzende nach dem Ende, höchstens 30 Sekunden darüber hinaus +(`MAX_SNAP_SECONDS`). Liegt kein Satzende in Reichweite, wird an dieser Seite +genau geschnitten (`--exakt` schneidet immer genau). + +Die Zeiten der Überschrift sind die Start- und Endzeit der ersten bzw. letzten +Untertitelzeile, also auf wenige Sekunden genau. Überschneiden sich zwei +Bereiche erst durch die Satz-Erweiterung, werden sie zu einem Block +zusammengefasst. Bereiche ohne Untertitel werden mit einem Hinweis übersprungen. +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. @@ -175,7 +217,8 @@ werden trotzdem umgewandelt; der Exit-Code ist dann 1. ./rede_korrigieren.py -i rede.de.vtt # VTT direkt → rede.de.korrigiert.txt ./rede_korrigieren.py -u URL # YouTube → __.korrigiert.txt ./rede_korrigieren.py -i speech.txt -l en # englische Rede -./rede_korrigieren.py -u URL -s 3:17 -e 5:25 # nur ein Ausschnitt des Videos +./rede_korrigieren.py -u URL -s 0:03:17 -e 0:05:25 # nur ein Ausschnitt des Videos +./rede_korrigieren.py -u URL -z "00:05-00:06 00:25" # mehrere Bereiche, eine Datei ``` | Option | Bedeutung | @@ -183,9 +226,16 @@ werden trotzdem umgewandelt; der Exit-Code ist dann 1. | `-i`, `--input` | Eingabedatei: Rohabschrift (`.txt`) oder Untertitel (`.vtt`, wird wie mit `vtt2txt.py` umgewandelt) | | `-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. 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 | +| `-o`, `--output` | Ausgabedatei; Standard: `.korrigiert.txt` neben der Eingabe, mit `-u` `__.korrigiert.txt`; mit `-s`/`-e`/`-z` zusätzlich die Zeiträume im Namen | +| `-s`, `--start` / `-e`, `--ende` / `-z`, `--zeitraeume` | nur diese Zeiträume korrigieren, siehe [Zeitauswahl](#zeitauswahl--s--e--z); nur mit `.vtt`-Eingabe oder `-u`, da reiner Text keine Zeitstempel hat | +| `--exakt` | genau an den Zeiten schneiden statt an Satzgrenzen | + +Mit Zeitauswahl wird jeder Bereich als eigener Block korrigiert und gegliedert; +die Überschriften `[00:04:57 - 00:06:04]` bekommt das Modell nicht zu sehen, sie +stehen unverändert als eigene Zeilen im Ergebnis. Zusätzlich weist der Prompt das +Modell an, an den Rändern der Ausschnitte nichts zu ergänzen. Eine Rohabschrift +(`.txt`) mit solchen Überschriften, etwa von `vtt2txt.py -z`, wird ebenso in +Blöcken korrigiert. | `-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 0bdaf3d..85a7542 100755 --- a/extract_text.py +++ b/extract_text.py @@ -7,7 +7,10 @@ import shutil import tempfile import textwrap -from vtt2txt import add_time_options, check_time_range, time_suffix, vtt_to_lines +from vtt2txt import ( + add_time_options, join_segments, resolve_time_ranges, split_segments, time_suffix, + vtt_to_text, +) from yt_untertitel import ( YT_DLP, choose_track, default_name, download, fetch_info, no_track_message, original_language, @@ -28,7 +31,7 @@ def parse_args(): parser.add_argument( "-o", "--output", type=Path, help="Ausgabedatei (Standard: __.txt, " - "mit -s/-e zusätzlich der Zeitraum)", + "mit -s/-e/-z zusätzlich die Zeiträume)", ) parser.add_argument( "-f", "--fliesstext", action="store_true", @@ -36,7 +39,7 @@ def parse_args(): ) add_time_options(parser) args = parser.parse_args() - check_time_range(parser, args) + resolve_time_ranges(parser, args) if args.output is not None and args.output.is_dir(): parser.error(f"-o/--output ist ein Verzeichnis: {args.output}") @@ -45,12 +48,12 @@ def parse_args(): return args -def fetch_text(url, lang=None, start=None, end=None, exact=False): - """Download the subtitles of a video; return (video info, text lines). +def fetch_text(url, lang=None, ranges=(), exact=False): + """Download the subtitles of a video; return (video info, text). - Without `lang` the language of the original audio is used. `start`/`end` - (seconds) keep only the captions of that time range, completed to whole - sentences unless `exact`. + Without `lang` the language of the original audio is used. With time + `ranges` (seconds) the text has one block per range under a header with + its actual times, see vtt2txt.vtt_to_text. """ info = fetch_info(url) print(f"Video: {info.get('title', '?')}") @@ -70,35 +73,40 @@ def fetch_text(url, lang=None, start=None, end=None, exact=False): 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, exact) + text = vtt_to_text(vtt.read_text(encoding="utf-8-sig"), ranges, exact) - if not lines: - if start is None and end is None: - raise SystemExit("Die Untertitel enthalten keinen Text.") - raise SystemExit("Die Untertitel enthalten im gewählten Zeitraum keinen Text.") - return info, lines + if not text: + if ranges: + raise SystemExit("Die Untertitel enthalten im gewählten Zeitraum keinen Text.") + raise SystemExit("Die Untertitel enthalten keinen Text.") + return info, text def main(): args = parse_args() - info, lines = fetch_text(args.url, args.lang, args.start, args.ende, args.exakt) - suffix = time_suffix(args.start, args.ende) + info, text = fetch_text(args.url, args.lang, args.bereiche, args.exakt) + suffix = time_suffix(args.bereiche) output = args.output or Path(f"{default_name(info)}{suffix}.txt") + segments = split_segments(text) if args.fliesstext: - # Never split words, not even at hyphens ("EU-Gipfel") - text = textwrap.fill( - " ".join(lines), width=FLOW_WIDTH, - break_long_words=False, break_on_hyphens=False, - ) - else: - text = "\n".join(lines) + # Each block on its own, headers stay lines of their own; never split + # words, not even at hyphens ("EU-Gipfel") + segments = [ + (header, textwrap.fill( + " ".join(body.split()), width=FLOW_WIDTH, + break_long_words=False, break_on_hyphens=False, + )) + for header, body in segments + ] + text = join_segments(segments) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(text + "\n", encoding="utf-8") print( f"Gespeichert: {output} " - f"({len(text.splitlines())} Zeilen, {len(text.split())} Wörter)" + f"({len(text.splitlines())} Zeilen, " + f"{sum(len(body.split()) for _, body in segments)} Wörter)" ) diff --git a/rede_korrigieren.py b/rede_korrigieren.py index 8857a8c..e34c19b 100755 --- a/rede_korrigieren.py +++ b/rede_korrigieren.py @@ -13,7 +13,10 @@ import urllib.request from aenderungen import REVIEW_MARK, print_summary, report_path, write_report from extract_text import fetch_text -from vtt2txt import add_time_options, check_time_range, time_suffix, vtt_to_lines +from vtt2txt import ( + add_time_options, join_segments, resolve_time_ranges, split_segments, time_suffix, + vtt_to_text, +) from yt_untertitel import YT_DLP, base_language, default_name, original_language MODEL = "qwen3:32b" @@ -507,8 +510,7 @@ def parse_args(): ) add_time_options(parser) args = parser.parse_args() - check_time_range(parser, args) - timed = args.start is not None or args.ende is not None + resolve_time_ranges(parser, args) for glossary in args.glossar: if not glossary.is_file(): @@ -525,10 +527,10 @@ def parse_args(): if not args.input.is_file(): parser.error(f"Eingabedatei nicht gefunden: {args.input}") - if timed and args.input.suffix.lower() != ".vtt": - parser.error("-s/-e brauchen Zeitstempel: nur mit .vtt-Eingabe oder -u.") + if args.bereiche and args.input.suffix.lower() != ".vtt": + parser.error("-s/-e/-z brauchen Zeitstempel: nur mit .vtt-Eingabe oder -u.") if args.output is None: - suffix = time_suffix(args.start, args.ende) + suffix = time_suffix(args.bereiche) args.output = args.input.with_name(f"{args.input.stem}{suffix}.korrigiert.txt") if args.output.resolve() == args.input.resolve(): parser.error("Ausgabedatei darf nicht die Eingabedatei sein.") @@ -555,21 +557,21 @@ 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, args.exakt) + info, text = fetch_text(args.url, args.lang, args.bereiche, args.exakt) lang = base_language(args.lang or original_language(info) or DEFAULT_LANGUAGE) if lang not in PROMPTS: raise SystemExit( f"Die Rede ist auf „{lang}“; unterstützt: {', '.join(PROMPTS)} (-l)." ) - suffix = time_suffix(args.start, args.ende) + suffix = time_suffix(args.bereiche) target = args.output or Path(f"{default_name(info)}{suffix}.korrigiert.txt") - raw = "\n".join(lines) + "\n" + raw = text + "\n" save_raw_text(raw_text_path(target), raw) return raw, target, lang 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, args.exakt)) + raw = vtt_to_text(raw, args.bereiche, args.exakt) return raw, args.output, base_language(args.lang or DEFAULT_LANGUAGE) @@ -594,58 +596,74 @@ def main(): ) terms += more_terms replacements += more_replacements - excerpt = args.start is not None or args.ende is not None + # Time headers ("[00:04:57 - 00:06:04]") split the text into blocks that are + # corrected one by one; the model never sees the headers. Subtitle line + # breaks carry no meaning: send plain running text + segments = [(header, " ".join(body.split())) for header, body in split_segments(raw)] + segments = [(header, body) for header, body in segments if body] + # The change report compares against this, so glossary effects show up too + raw_text = join_segments(segments) + excerpt = any(header is not None for header, _ in segments) system = build_system(terms, lang, excerpt) - # Subtitle line breaks carry no meaning; send plain running text - text = " ".join(raw.split()) - # The change report compares against this, so glossary effects show up too - raw_text = text - if replacements: - text = apply_replacements(text, replacements) + # One pass over all blocks (one per line) gives one count per rule + bodies = apply_replacements( + "\n".join(body for _, body in segments), replacements + ).split("\n") + segments = [(header, body) for (header, _), body in zip(segments, bodies)] - chunks = split_chunks(text) + blocks = [(header, split_chunks(body)) for header, body in segments] + total = sum(len(chunks) for _, chunks in blocks) - if not chunks: - if args.start is None and args.ende is None: - raise SystemExit("Die Eingabedatei ist leer.") - raise SystemExit("Die Eingabedatei enthält im gewählten Zeitraum keinen Text.") + if not total: + if args.bereiche: + raise SystemExit("Die Eingabedatei enthält im gewählten Zeitraum keinen Text.") + raise SystemExit("Die Eingabedatei ist leer.") parts_dir.mkdir(parents=True, exist_ok=True) - results = [] + corrected_blocks = [] warnings = [] + index = 0 - for index, chunk in enumerate(chunks, start=1): - # Key covers model, prompt and chunk: any change invalidates the cache - key = hashlib.sha256( - f"{MODEL}\n{system}\n{chunk}".encode("utf-8") - ).hexdigest()[:12] - part = parts_dir / f"part{index:03d}-{key}.txt" + for header, chunks in blocks: + if header is not None: + print(f"Block {header}") + results = [] + for chunk in chunks: + index += 1 + # Key covers model, prompt and chunk: any change invalidates the cache + key = hashlib.sha256( + f"{MODEL}\n{system}\n{chunk}".encode("utf-8") + ).hexdigest()[:12] + part = parts_dir / f"part{index:03d}-{key}.txt" - if part.exists(): - print(f"Abschnitt {index}/{len(chunks)} bereits vorhanden, übersprungen.") - corrected = part.read_text(encoding="utf-8") - else: - print(f"Bearbeite Abschnitt {index}/{len(chunks)} …", flush=True) - corrected = correct_chunk(chunk, index, system) - part.write_text(corrected, encoding="utf-8") + if part.exists(): + print(f"Abschnitt {index}/{total} bereits vorhanden, übersprungen.") + corrected = part.read_text(encoding="utf-8") + else: + print(f"Bearbeite Abschnitt {index}/{total} …", flush=True) + corrected = correct_chunk(chunk, index, system) + part.write_text(corrected, encoding="utf-8") - before = count_words(chunk) - after = count_words(corrected) - deviation = abs(after - before) / before - if deviation > MAX_WORD_DEVIATION: - message = ( - f"Abschnitt {index}: {before} → {after} Wörter " - f"({deviation:.0%} Abweichung), auf Auslassungen/Ergänzungen " - f"prüfen: {part}" - ) - print(f"WARNUNG: {message}") - warnings.append(message) + before = count_words(chunk) + after = count_words(corrected) + deviation = abs(after - before) / before + if deviation > MAX_WORD_DEVIATION: + message = ( + f"Abschnitt {index}: {before} → {after} Wörter " + f"({deviation:.0%} Abweichung), auf Auslassungen/Ergänzungen " + f"prüfen: {part}" + ) + print(f"WARNUNG: {message}") + warnings.append(message) - results.append(corrected) + results.append(corrected) - text = build_paragraphs(" ".join(results), parts_dir, lang) + paragraphs = build_paragraphs(" ".join(results), parts_dir, lang) + corrected_blocks.append((header, paragraphs)) + + text = join_segments(corrected_blocks, header_gap="\n\n") target.write_text(text + "\n", encoding="utf-8") print(f"Fertig: {target}") diff --git a/tests/test_extract_text.py b/tests/test_extract_text.py index cb89cd9..4b8b801 100644 --- a/tests/test_extract_text.py +++ b/tests/test_extract_text.py @@ -130,10 +130,12 @@ def test_cli_errors(tmp_path, monkeypatch, capsys, args, message): def test_time_range_default_name(fake_yt_dlp, tmp_path, monkeypatch): monkeypatch.setattr(yt.subprocess, "run", fake_yt_dlp(GERMAN_VIDEO, vtt=VTT)) monkeypatch.chdir(tmp_path) - run_main(monkeypatch, "-u", URL, "-s", "2", "-e", "0:03") + run_main(monkeypatch, "-u", URL, "-s", "0:00:02", "-e", "0:00:03") name = "Kanal_Eine_Rede_2026-03-28_00-00-02_bis_00-00-03.txt" - assert (tmp_path / name).read_text(encoding="utf-8") == "Schön, dass Sie da sind.\n" + assert (tmp_path / name).read_text(encoding="utf-8") == ( + "[00:00:02 - 00:00:03]\nSchön, dass Sie da sind.\n" + ) def test_time_range_without_text(fake_yt_dlp, tmp_path, monkeypatch): @@ -157,11 +159,32 @@ CUT_VTT = ( @pytest.mark.parametrize("exact, text", [ - ([], "Guten Abend und\nwillkommen. Schön, dass\nSie da sind.\n"), - (["--exakt"], "willkommen. Schön, dass\n"), + ([], "[00:00:01 - 00:00:04]\nGuten Abend und\nwillkommen. Schön, dass\nSie da sind.\n"), + (["--exakt"], "[00:00:02 - 00:00:03]\nwillkommen. 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) + run_main(monkeypatch, "-u", URL, "-s", "0:00:02", "-e", "0:00:03", "-o", target, *exact) assert target.read_text(encoding="utf-8") == text + + +MULTI_VTT = ( + "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nErster Satz der Rede.\n\n" + "00:01:00.000 --> 00:01:01.000\nZweiter Satz.\n\n" + "00:02:00.000 --> 00:02:01.000\nDritter Satz, der etwas länger ist.\n" +) + + +def test_several_ranges_flow_text_keeps_headers(fake_yt_dlp, tmp_path, monkeypatch, capsys): + monkeypatch.setattr(yt.subprocess, "run", fake_yt_dlp(GERMAN_VIDEO, vtt=MULTI_VTT)) + monkeypatch.setattr(extract_text, "FLOW_WIDTH", 20) + monkeypatch.chdir(tmp_path) + run_main(monkeypatch, "-u", URL, "-z", "0:00:00-0:00:30 0:02", "-f") + + name = "Kanal_Eine_Rede_2026-03-28_00-00-00_bis_00-00-30+ab_00-02-00.txt" + assert (tmp_path / name).read_text(encoding="utf-8") == ( + "[00:00:01 - 00:00:02]\nErster Satz der\nRede.\n\n" + "[00:02:00 - 00:02:01]\nDritter Satz, der\netwas länger ist.\n" + ) + assert "(7 Zeilen, 10 Wörter)" in capsys.readouterr().out diff --git a/tests/test_rede_korrigieren.py b/tests/test_rede_korrigieren.py index bb786aa..1850248 100644 --- a/tests/test_rede_korrigieren.py +++ b/tests/test_rede_korrigieren.py @@ -523,12 +523,13 @@ def test_cli_errors(sample_copy, monkeypatch, capsys, args, message): def test_main_url_time_range_default_names(from_url, tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) - run_main(monkeypatch, "-u", URL, "-s", "0:02") + run_main(monkeypatch, "-u", URL, "-s", "0:00:02") name = "Kanal_Rede_2026-03-28_ab_00-00-02" raw = (tmp_path / f"{name}.txt").read_text(encoding="utf-8") - assert raw == "Das ist bekannt.\n" - assert (tmp_path / f"{name}.korrigiert.txt").is_file() + assert raw == "[00:00:02 - 00:00:03]\nDas ist bekannt.\n" + result = (tmp_path / f"{name}.korrigiert.txt").read_text(encoding="utf-8") + assert result == "[00:00:02 - 00:00:03]\n\nDas ist bekannt.\n" def test_main_url_time_range_without_text(from_url, tmp_path, monkeypatch): @@ -541,10 +542,12 @@ def test_main_vtt_time_range(tmp_path, monkeypatch): vtt.write_text(VTT, encoding="utf-8") monkeypatch.setattr(rk, "correct_chunk", identity_correct) monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5)) - run_main(monkeypatch, "-i", vtt, "-e", "2") + run_main(monkeypatch, "-i", vtt, "-e", "0:00:02") result = tmp_path / "rede.de_bis_00-00-02.korrigiert.txt" - assert result.read_text(encoding="utf-8") == "Der Kremmel ist in Moskau.\n" + assert result.read_text(encoding="utf-8") == ( + "[00:00:01 - 00:00:02]\n\nDer Kremmel ist in Moskau.\n" + ) def test_main_vtt_time_range_without_text(tmp_path, monkeypatch): @@ -559,6 +562,7 @@ def test_main_vtt_time_range_without_text(tmp_path, monkeypatch): (["-i", "{sample}", "-s", "1:00"], "nur mit .vtt-Eingabe oder -u"), (["-i", "{sample}", "-s", "x"], "ungültige Zeit"), (["-i", "{sample}", "-s", "2:00", "-e", "1:00"], "muss vor -e/--ende liegen"), + (["-i", "{sample}", "-z", "0:05 0:10"], "überschneiden"), ]) def test_cli_time_errors(sample_copy, monkeypatch, capsys, args, message): args = [a.format(sample=sample_copy) for a in args] @@ -637,14 +641,81 @@ CUT_VTT = ( @pytest.mark.parametrize("exact, raw", [ - ([], "Der Kremmel ist\nin Moskau. Das ist\nbekannt.\n"), - (["--exakt"], "in Moskau. Das ist\n"), + ([], "[00:00:01 - 00:00:04]\nDer Kremmel ist\nin Moskau. Das ist\nbekannt.\n"), + (["--exakt"], "[00:00:02 - 00:00:03]\nin 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) + run_main(monkeypatch, "-u", URL, "-s", "0:00:02", "-e", "0:00:03", "-o", target, *exact) assert (tmp_path / "rede.txt").read_text(encoding="utf-8") == raw assert rk.PROMPTS["de"]["excerpt"] in systems[0] + + +# === Several time ranges (-z): blocks with headers === + +MULTI_VTT = ( + "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nDer Kremmel ist in Moskau.\n\n" + "00:01:00.000 --> 00:01:01.000\nZwischendurch etwas anderes.\n\n" + "00:02:00.000 --> 00:02:01.000\nDas ist bekannt.\n" +) + + +def chunk_recorder(monkeypatch): + """Fake models; returns the chunks and system prompts sent for correction.""" + sent = [] + + def correct(chunk, index, system): + sent.append((chunk, system)) + return chunk.strip() + + monkeypatch.setattr(rk, "correct_chunk", correct) + monkeypatch.setattr(rk, "find_paragraph_starts", every_nth_start(5)) + return sent + + +def test_main_url_several_ranges(fake_yt_dlp, yt_dlp_installed, tmp_path, monkeypatch, capsys): + monkeypatch.setattr(subprocess, "run", fake_yt_dlp(VIDEO, vtt=MULTI_VTT)) + sent = chunk_recorder(monkeypatch) + monkeypatch.chdir(tmp_path) + run_main(monkeypatch, "-u", URL, "-z", "0:00:00-0:00:30 0:02", "-g", SHARED_GLOSSARY) + + name = "Kanal_Rede_2026-03-28_00-00-00_bis_00-00-30+ab_00-02-00" + assert (tmp_path / f"{name}.txt").read_text(encoding="utf-8") == ( + "[00:00:01 - 00:00:02]\nDer Kremmel ist in Moskau.\n\n" + "[00:02:00 - 00:02:01]\nDas ist bekannt.\n" + ) + assert (tmp_path / f"{name}.korrigiert.txt").read_text(encoding="utf-8") == ( + "[00:00:01 - 00:00:02]\n\nDer Kreml ist in Moskau.\n\n" + "[00:02:00 - 00:02:01]\n\nDas ist bekannt.\n" + ) + # One chunk per block, never a header; excerpt note in the prompt + assert [chunk for chunk, _ in sent] == ["Der Kreml ist in Moskau.", "Das ist bekannt."] + assert all(rk.PROMPTS["de"]["excerpt"] in system for _, system in sent) + out = capsys.readouterr().out + assert "Abschnitt 2/2" in out + # Headers are no change; only the glossary replacement is + assert "Änderungen gegenüber dem Rohtext: 1" in out + + +def test_main_txt_with_headers_corrected_in_blocks(tmp_path, monkeypatch): + raw = tmp_path / "rede.txt" + raw.write_text( + "[00:00:01 - 00:00:02]\nErster\nBlock.\n\n[00:02:00 - 00:02:01]\nZweiter Block.\n", + encoding="utf-8", + ) + sent = chunk_recorder(monkeypatch) + run_main(monkeypatch, "-i", raw) + + assert [chunk for chunk, _ in sent] == ["Erster Block.", "Zweiter Block."] + assert (tmp_path / "rede.korrigiert.txt").read_text(encoding="utf-8") == ( + "[00:00:01 - 00:00:02]\n\nErster Block.\n\n[00:02:00 - 00:02:01]\n\nZweiter Block.\n" + ) + + +def test_main_without_headers_has_no_excerpt_note(sample_copy, monkeypatch): + sent = chunk_recorder(monkeypatch) + run_main(monkeypatch, "-i", sample_copy) + assert rk.PROMPTS["de"]["excerpt"] not in sent[0][1] diff --git a/tests/test_vtt2txt.py b/tests/test_vtt2txt.py index e586798..d5201b4 100644 --- a/tests/test_vtt2txt.py +++ b/tests/test_vtt2txt.py @@ -164,51 +164,215 @@ def test_cli_errors(sample_copy, monkeypatch, capsys, args, message): assert message in capsys.readouterr().err +# === Time format and ranges === + @pytest.mark.parametrize("text, seconds", [ - ("00:03:17", 197), ("3:17", 197), ("197", 197), ("00:03:17.5", 197.5), - ("00:03:17,5", 197.5), ("1:02:03", 3723), + ("00:05:10", 310), ("00:20", 1200), ("1:30", 5400), ("25", 1500), ("00:5:00", 300), + ("00:03:17.5", 197.5), ("00:03:17,5", 197.5), ("1:02:03", 3723), ]) -def test_parse_time(text, seconds): - assert vtt2txt.parse_time(text) == seconds +def test_parse_user_time(text, seconds): + assert vtt2txt.parse_user_time(text) == seconds -@pytest.mark.parametrize("text", ["", "abc", "1:2:3:4", "-5"]) -def test_parse_time_rejects_invalid(text): +@pytest.mark.parametrize("text", ["", "abc", "1:2:3:4", "-5", "1:75", "0:10:60", "1:30.5"]) +def test_parse_user_time_rejects_invalid(text): with pytest.raises(ValueError): - vtt2txt.parse_time(text) + vtt2txt.parse_user_time(text) -@pytest.mark.parametrize("start, end, suffix", [ - (None, None, ""), - (197, 325, "_00-03-17_bis_00-05-25"), - (197, None, "_ab_00-03-17"), - (None, 325, "_bis_00-05-25"), +@pytest.mark.parametrize("text, seconds", [ + ("00:03:17.500", 197.5), ("03:17.500", 197.5), ("1:02:03.000", 3723), ]) -def test_time_suffix(start, end, suffix): - assert vtt2txt.time_suffix(start, end) == suffix +def test_parse_cue_time_keeps_minutes_seconds(text, seconds): + assert vtt2txt.parse_cue_time(text) == seconds -def test_vtt_to_lines_exact_time_range(): +@pytest.mark.parametrize("text, ranges", [ + ("00:5:00-00:6:00 00:10-00:12:00 00:25", [(300, 360), (600, 720), (1500, None)]), + ("-00:20", [(None, 1200)]), + ("00:25- 00:10-00:12 -00:05", [(None, 300), (600, 720), (1500, None)]), + ("00:05-00:06 00:06-00:07", [(300, 360), (360, 420)]), +]) +def test_parse_ranges(text, ranges): + assert vtt2txt.parse_ranges(text) == ranges + + +@pytest.mark.parametrize("text, message", [ + ("", "keine Bereiche"), + ("-", "ungültiger Bereich"), + ("00:06-00:05", "Start muss vor Ende"), + ("00:05-00:10 00:08-00:12", "überschneiden"), + ("00:05 00:10", "überschneiden"), + ("-00:05 -00:10", "überschneiden"), + ("00:05-x", "ungültiger Bereich"), +]) +def test_parse_ranges_rejects(text, message): + with pytest.raises(ValueError, match=message): + vtt2txt.parse_ranges(text) + + +@pytest.mark.parametrize("ranges, suffix", [ + ([], ""), + ([(197, 325)], "_00-03-17_bis_00-05-25"), + ([(197, None)], "_ab_00-03-17"), + ([(None, 325)], "_bis_00-05-25"), + ([(300, 360), (600, 720), (1500, None)], + "_00-05-00_bis_00-06-00+00-10-00_bis_00-12-00+ab_00-25-00"), +]) +def test_time_suffix(ranges, suffix): + assert vtt2txt.time_suffix(ranges) == suffix + + +# === Excerpts === + +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) + + +def block(vtt, start=None, end=None, exact=False): + """Lines of the single excerpt of one time range ([] without captions).""" + blocks = vtt2txt.excerpts(vtt2txt.timed_lines(vtt), [(start, end)], exact) + return blocks[0][2] if blocks else [] + + +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_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, 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"] + assert block(vtt, 2, 4, exact=True) == ["B", "C"] + assert block(vtt, 3, exact=True) == ["C", "D"] + assert block(vtt, end=2, exact=True) == ["A"] -def test_vtt_to_lines_time_range_skips_window_from_before_start(): +def test_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, exact=True) == ["C"] + assert block(vtt, 2, exact=True) == ["C"] +def test_snap_extends_both_sides_to_sentences(): + # Range holds only "kommt der dritte" and "und endet. Der vierte" + assert block(SPEECH, 106, 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 block(SPEECH, 102, 104) == ["Der zweite Satz", "beginnt vorher."] + assert block(SPEECH, 100, 102) == ["Erster Satz ist hier."] + + +def test_snap_reaches_start_and_end_of_captions(): + assert block(SPEECH, 114) == ["Schluss", "folgt."] + speech = captions((0, "ohne Satzzeichen"), (2, "am Anfang. Rest")) + assert block(speech, 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 block(speech, far, far + 1) == ["Satz ohne"] + + +def test_snap_exact_keeps_cut(): + assert block(SPEECH, 106, 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 block(speech, 2, 4) == ["Und", "dann", "ging er."] + + +def test_vtt_to_text_headers_with_actual_times(): + text = vtt2txt.vtt_to_text(SPEECH, [(102, 104), (106, 110)]) + assert text == ( + "[00:01:42 - 00:01:45]\nDer zweite Satz\nbeginnt vorher.\n\n" + "[00:01:44 - 00:01:53]\nDann\nkommt der dritte\nund endet. Der vierte\n" + "läuft weiter\nund endet auch." + ) + + +def test_vtt_to_text_merges_blocks_overlapping_after_snap(): + # 106-110 grows to "und endet auch." (cue 112), 110-114 back to "Der vierte" (cue 108) + text = vtt2txt.vtt_to_text(SPEECH, [(106, 110), (110, 114)]) + assert text == ( + "[00:01:44 - 00:01:55]\nDann\nkommt der dritte\nund endet. Der vierte\n" + "läuft weiter\nund endet auch. Schluss\nfolgt." + ) + + +def test_vtt_to_text_shared_line_split_at_sentence_end(): + # Both blocks get their part of "und endet. Der vierte", without overlap + text = vtt2txt.vtt_to_text(SPEECH, [(106, 108), (110, 112)]) + assert vtt2txt.split_segments(text) == [ + ("[00:01:44 - 00:01:49]", "Dann\nkommt der dritte\nund endet."), + ("[00:01:48 - 00:01:53]", "Der vierte\nläuft weiter\nund endet auch."), + ] + + +def test_vtt_to_text_skips_empty_range(capsys): + text = vtt2txt.vtt_to_text(SPEECH, [(0, 50), (100, 102), (200, None)]) + assert text == "[00:01:40 - 00:01:41]\nErster Satz ist hier." + out = capsys.readouterr().out + assert "keine Untertitel im Bereich 00:00:00 – 00:00:50" in out + assert "keine Untertitel im Bereich 00:03:20 – Ende" in out + assert vtt2txt.vtt_to_text(SPEECH, [(200, None)]) == "" + + +def test_vtt_to_text_without_ranges_has_no_header(): + assert vtt2txt.vtt_to_text(SPEECH).splitlines()[0] == "Erster Satz ist hier." + + +def test_split_segments(): + text = "Vorspann\n\n[00:01:00 - 00:02:00]\nA\nB\n\n[applaus]\n[00:03:00 - 00:04:00]\n\nC\n" + assert vtt2txt.split_segments(text) == [ + (None, "Vorspann"), + ("[00:01:00 - 00:02:00]", "A\nB\n\n[applaus]"), + ("[00:03:00 - 00:04:00]", "C"), + ] + assert vtt2txt.split_segments("ohne\nÜberschrift") == [(None, "ohne\nÜberschrift")] + assert vtt2txt.split_segments(" \n") == [] + + +def test_split_segments_inverts_join_segments(): + text = vtt2txt.vtt_to_text(SPEECH, [(102, 104), (112, None)]) + assert vtt2txt.join_segments(vtt2txt.split_segments(text)) == text + + +# === Command line === + def test_main_time_range_default_name(sample_copy, monkeypatch): - run_main(monkeypatch, "-i", sample_copy, "-s", "0:30", "-e", "1:00") + run_main(monkeypatch, "-i", sample_copy, "-s", "0:00:30", "-e", "0:01") target = sample_copy.with_name("ramstein.de_00-00-30_bis_00-01-00.txt") full = SAMPLE_TXT.read_text(encoding="utf-8").splitlines() part = target.read_text(encoding="utf-8").splitlines() + assert vtt2txt.HEADER.fullmatch(part[0]) assert 0 < len(part) < len(full) +def test_main_several_ranges(sample_copy, monkeypatch): + run_main(monkeypatch, "-i", sample_copy, "-z", "0:01-0:02 0:05") + target = sample_copy.with_name("ramstein.de_00-01-00_bis_00-02-00+ab_00-05-00.txt") + segments = vtt2txt.split_segments(target.read_text(encoding="utf-8")) + assert [header for header, _ in segments] == [ + "[00:00:56 - 00:02:09]", "[00:04:57 - 00:48:07]", + ] + + def test_main_time_range_without_captions(sample_copy, monkeypatch, capsys): with pytest.raises(SystemExit) as exc: run_main(monkeypatch, "-i", sample_copy, "-s", "99:00:00") @@ -219,6 +383,9 @@ def test_main_time_range_without_captions(sample_copy, monkeypatch, capsys): @pytest.mark.parametrize("args, message", [ (["-s", "abc"], "ungültige Zeit"), (["-s", "1:00", "-e", "0:30"], "muss vor -e/--ende liegen"), + (["-z", "0:05-0:04"], "Start muss vor Ende"), + (["-z", "0:05 0:10"], "überschneiden"), + (["-z", "0:05", "-s", "0:01"], "nicht zusammen mit -s/-e"), ]) def test_cli_time_errors(sample_copy, monkeypatch, capsys, args, message): with pytest.raises(SystemExit) as exc: @@ -227,68 +394,13 @@ def test_cli_time_errors(sample_copy, monkeypatch, capsys, args, message): 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") + run_main(monkeypatch, "-i", source, "-s", "0:01:46", "-e", "0:01:50", "--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") == ( + "[00:01:46 - 00:01:49]\nkommt der dritte\nund endet. Der vierte\n" + ) + run_main(monkeypatch, "-i", source, "-s", "0:01:46", "-e", "0:01:50") assert target.read_text(encoding="utf-8").endswith("und endet auch.\n") diff --git a/vtt2txt.py b/vtt2txt.py index 03eecec..04f88d6 100755 --- a/vtt2txt.py +++ b/vtt2txt.py @@ -11,30 +11,89 @@ import sys 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*-->") +CUE_TIME = re.compile(r"(?:(\d+):)??(?:(\d+):)?(\d+(?:[.,]\d+)?)") +CUE_TIMES = re.compile(r"^\s*([\d:.,]+)\s*-->\s*([\d:.,]+)") +USER_TIME = re.compile(r"(\d+)(?::(\d+)(?::(\d+(?:[.,]\d+)?))?)?") +HEADER = re.compile(r"\[(\d{2,}:\d{2}:\d{2}) - (\d{2,}:\d{2}:\d{2})\]") 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 +TIME_EXAMPLES = "z. B. 00:05:10, 00:20 (Minute 20) oder 25 (Minute 25)" -def parse_time(text): - """Seconds from "00:03:17", "3:17", "197" or "00:03:17.5".""" - match = TIME.fullmatch(text.strip()) +def parse_cue_time(text): + """Seconds from a VTT time: "00:03:17.500" or, without hours, "03:17.500".""" + match = CUE_TIME.fullmatch(text.strip()) if not match: raise ValueError(f"Ungültige Zeit: {text}") hours, minutes, seconds = match.groups() return int(hours or 0) * 3600 + int(minutes or 0) * 60 + float(seconds.replace(",", ".")) +def parse_user_time(text): + """Seconds from "00:05:10" (h:m:s), "00:20" (h:m, minute 20) or "25" (minutes).""" + match = USER_TIME.fullmatch(text.strip()) + if not match: + raise ValueError(f"Ungültige Zeit: {text}") + first, minutes, seconds = match.groups() + if minutes is None: + return int(first) * 60 + seconds = float(seconds.replace(",", ".")) if seconds is not None else 0 + if int(minutes) >= 60 or seconds >= 60: + raise ValueError(f"Ungültige Zeit: {text}") + return int(first) * 3600 + int(minutes) * 60 + seconds + + def time_option(text): """argparse type for -s/--start and -e/--ende.""" try: - return parse_time(text) + return parse_user_time(text) except ValueError: raise argparse.ArgumentTypeError( - f"ungültige Zeit „{text}“ (z. B. 00:03:17, 3:17 oder 197)" + f"ungültige Zeit „{text}“ ({TIME_EXAMPLES})" + ) from None + + +def parse_ranges(text): + """Sorted (start, end) seconds from "00:05:00-00:06:00 00:10-00:12 00:25". + + "A" or "A-" runs to the end, "-B" from the beginning; None marks an open side. + """ + ranges = [] + for item in text.split(): + start, dash, end = item.partition("-") + try: + span = ( + parse_user_time(start) if start else None, + parse_user_time(end) if end else None, + ) + except ValueError: + raise ValueError(f"ungültiger Bereich „{item}“") from None + if span == (None, None): + raise ValueError(f"ungültiger Bereich „{item}“") + if None not in span and span[0] >= span[1]: + raise ValueError(f"Bereich „{item}“: Start muss vor Ende liegen") + ranges.append(span) + if not ranges: + raise ValueError("keine Bereiche angegeben") + + # Open start first, then by start time + ranges.sort(key=lambda span: (span[0] is not None, span[0] or 0)) + for (_, end), (start, _) in zip(ranges, ranges[1:]): + if end is None or start is None or start < end: + raise ValueError("Bereiche überschneiden sich") + return ranges + + +def ranges_option(text): + """argparse type for -z/--zeitraeume.""" + try: + return parse_ranges(text) + except ValueError as error: + raise argparse.ArgumentTypeError( + f"{error} (Bereiche durch Leerzeichen getrennt, je A-B, A = ab A, " + f"-B = bis B; Zeiten {TIME_EXAMPLES})" ) from None @@ -45,59 +104,84 @@ def format_time(seconds, separator=":"): ) -def time_suffix(start, end): - """Part of a default file name for a time range: "_00-03-17_bis_00-05-25".""" - if start is None and end is None: - return "" +def format_header(start, end): + """Header line in front of a block of text: "[00:04:57 - 00:06:04]".""" + return f"[{format_time(start)} - {format_time(end)}]" + + +def range_name(start, end): if end is None: - return f"_ab_{format_time(start, '-')}" + return f"ab_{format_time(start, '-')}" if start is None: - return f"_bis_{format_time(end, '-')}" - return f"_{format_time(start, '-')}_bis_{format_time(end, '-')}" + return f"bis_{format_time(end, '-')}" + return f"{format_time(start, '-')}_bis_{format_time(end, '-')}" + + +def time_suffix(ranges): + """Part of a default file name: "_00-03-17_bis_00-05-25", several joined with "+".""" + if not ranges: + return "" + return "_" + "+".join(range_name(start, end) for start, end in ranges) def add_time_options(parser): parser.add_argument( "-s", "--start", type=time_option, metavar="ZEIT", - help="nur Untertitel ab dieser Zeit, z. B. 00:03:17", + help=f"nur Untertitel ab dieser Zeit, {TIME_EXAMPLES}", ) parser.add_argument( "-e", "--ende", type=time_option, metavar="ZEIT", - help="nur Untertitel vor dieser Zeit, z. B. 00:05:25", + help="nur Untertitel vor dieser Zeit (Format wie -s)", + ) + parser.add_argument( + "-z", "--zeitraeume", type=ranges_option, metavar="BEREICHE", + help="mehrere Bereiche statt -s/-e, durch Leerzeichen getrennt: A-B, " + "A (ab A bis zum Ende) oder -B (vom Anfang bis B), " + "z. B. \"00:05:00-00:06:00 00:10-00:12 00:25\"", ) parser.add_argument( "--exakt", action="store_true", - help="genau an -s/-e schneiden, statt angeschnittene Sätze vollständig " + help="genau an den Zeiten schneiden, statt angeschnittene Sätze vollständig " f"aufzunehmen (bis {MAX_SNAP_SECONDS} s über den Zeitraum hinaus)", ) -def check_time_range(parser, args): - if args.start is not None and args.ende is not None and args.start >= args.ende: - parser.error("-s/--start muss vor -e/--ende liegen.") +def resolve_time_ranges(parser, args): + """Check the time options and set args.bereiche: list of (start, end), maybe empty.""" + single = args.start is not None or args.ende is not None + if args.zeitraeume is not None: + if single: + parser.error("-z/--zeitraeume nicht zusammen mit -s/-e angeben.") + args.bereiche = args.zeitraeume + elif single: + if args.start is not None and args.ende is not None and args.start >= args.ende: + parser.error("-s/--start muss vor -e/--ende liegen.") + args.bereiche = [(args.start, args.ende)] + else: + args.bereiche = [] def read_cues(vtt): - """Yield (start in seconds, non-empty tag-free text lines) of each cue.""" + """Yield (start, end in seconds, non-empty tag-free text lines) of each cue.""" cue = None - start = None + start = end = None for line in vtt.splitlines(): if "-->" in line: if cue is not None: - yield start, cue - match = CUE_START.match(line) + yield start, end, cue + match = CUE_TIMES.match(line) try: - start = parse_time(match.group(1)) if match else None + start, end = map(parse_cue_time, match.groups()) if match else (None, None) except ValueError: - start = None + start = end = None cue = [] continue # Only a truly empty line ends a cue; YouTube puts lines with a single # space inside cues, followed by more text if not line: if cue is not None: - yield start, cue + yield start, end, cue cue = None continue # Header, NOTE/STYLE blocks and cue identifiers @@ -109,7 +193,7 @@ def read_cues(vtt): cue.append(text) if cue is not None: - yield start, cue + yield start, end, cue def overlap(previous, cue): @@ -122,12 +206,12 @@ def overlap(previous, cue): def in_range(time, start, end): if time is None: - return start is None and end is None + return False return (start is None or time >= start) and (end is None or time < end) def timed_lines(vtt): - """Return (cue start in seconds, line) for the new caption lines of a VTT document. + """Return (cue start, cue end, 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 @@ -136,93 +220,173 @@ def timed_lines(vtt): lines = [] previous = [] - for time, cue in read_cues(vtt): + for start, end, cue in read_cues(vtt): # Empty cues (YouTube transitions) say nothing about the window if not cue: continue - lines.extend((time, line) for line in cue[overlap(previous, cue):]) + lines.extend((start, end, line) for line in cue[overlap(previous, cue):]) previous = cue return lines +# Positions in the caption text are (line index, character offset); an end +# position is exclusive. + 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]): + """Position where the sentence cut by `start` begins, line `first` if none is cut.""" + if start is None or first == 0 or LINE_ENDS_SENTENCE.search(timed[first - 1][2]): + return first, 0 + for index in range(first - 1, -1, -1): + time, _, line = timed[index] if time is None or time < start - MAX_SNAP_SECONDS: - return [] + return first, 0 ends = list(SENTENCE_END.finditer(line)) if ends: - rest = line[ends[-1].end():].strip() - return ([rest] if rest else []) + extra - extra.insert(0, line) + rest = line[ends[-1].end():] + if not rest.strip(): + return index + 1, 0 + return index, len(line) - len(rest.lstrip()) # Reached the first caption: the speech starts there - return extra + return 0, 0 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:]: + """Position after the sentence cut by `end`, after line `last` if none is cut.""" + if end is None or LINE_ENDS_SENTENCE.search(timed[last][2]): + return last + 1, 0 + for index in range(last + 1, len(timed)): + time, _, line = timed[index] if time is None or time >= end + MAX_SNAP_SECONDS: - return [] + return last + 1, 0 match = SENTENCE_END.search(line) if match: - return extra + [line[:match.end()]] - extra.append(line) + if match.end() == len(line): + return index + 1, 0 + return index, match.end() # Reached the last caption: the speech ends there - return extra + return len(timed), 0 -def snap_to_sentences(timed, start, end): - """Lines in the range, extended so that no sentence is cut at either side. +def excerpt_bounds(timed, start, end, exact=False): + """(begin, end) positions of a time range, None without captions in it. - Without a sentence end within MAX_SNAP_SECONDS that side is cut exactly. + Unless `exact`, the range grows to whole sentences; 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)] + inside = [i for i, (time, _, _) in enumerate(timed) if in_range(time, start, end)] if not inside: - return [] + return None 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) + if exact: + return (first, 0), (last + 1, 0) + return sentence_start(timed, first, start), 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. +def lines_between(timed, begin, end): + (first, begin_offset), (last, end_offset) = begin, end + lines = [] + for index in range(first, min(last + 1, len(timed))): + line = timed[index][2] + part = line[begin_offset if index == first else 0:end_offset if index == last else None] + if part.strip(): + lines.append(part.strip()) + return lines - 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. + +def range_label(start, end): + left = "Anfang" if start is None else format_time(start) + right = "Ende" if end is None else format_time(end) + return f"{left} – {right}" + + +def excerpts(timed, ranges, exact=False): + """Return (start, end in seconds, lines) of the text block of each time range. + + Blocks that overlap after growing to whole sentences are merged; ranges + without captions are skipped with a note. + """ + bounds = [] + for start, end in ranges: + found = excerpt_bounds(timed, start, end, exact) + if found is None: + print(f"Hinweis: keine Untertitel im Bereich {range_label(start, end)}") + continue + if bounds and found[0] < bounds[-1][1]: + bounds[-1] = (bounds[-1][0], max(bounds[-1][1], found[1])) + else: + bounds.append(found) + + blocks = [] + for begin, end in bounds: + last = end[0] if end[1] else end[0] - 1 + blocks.append((timed[begin[0]][0], timed[last][1], lines_between(timed, begin, end))) + return blocks + + +def join_segments(segments, header_gap="\n"): + """Text of (header or None, body) blocks, separated by blank lines.""" + return "\n\n".join( + body if header is None else f"{header}{header_gap}{body}" for header, body in segments + ) + + +def split_segments(text): + """Split text at header lines into [(header or None, body)]; see join_segments.""" + segments = [] + header, body = None, [] + + def close(): + if header is not None or "".join(body).strip(): + segments.append((header, "\n".join(body).strip())) + + for line in text.splitlines(): + if HEADER.fullmatch(line.strip()): + close() + header, body = line.strip(), [] + else: + body.append(line) + close() + return segments + + +def vtt_to_lines(vtt): + """Return the caption text lines of a VTT document without rolling duplicates.""" + return [line for _, _, line in timed_lines(vtt)] + + +def vtt_to_text(vtt, ranges=(), exact=False): + """Caption text of a VTT document; empty if there is none. + + With time ranges (seconds) one block per range under a header with its + actual times, completed to whole sentences unless `exact`. The rolling + window is followed through all cues, so a line repeated from before a + range 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) + if not ranges: + return "\n".join(line for _, _, line in timed) + return join_segments( + (format_header(start, end), "\n".join(lines)) + for start, end, lines in excerpts(timed, ranges, exact) + ) -def convert(source, target, start=None, end=None, exact=False): +def convert(source, target, ranges=(), 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, exact) - if not lines: - where = "" if start is None and end is None else " im gewählten Zeitraum" + text = vtt_to_text(vtt, ranges, exact) + if not text: + where = " im gewählten Zeitraum" if ranges else "" return f"Keine Untertitel{where} gefunden: {source}" target.parent.mkdir(parents=True, exist_ok=True) - target.write_text("\n".join(lines) + "\n", encoding="utf-8") - print(f"Erstellt: {target} ({len(lines)} Zeilen)") + target.write_text(text + "\n", encoding="utf-8") + print(f"Erstellt: {target} ({len(text.splitlines())} Zeilen)") return None @@ -259,12 +423,12 @@ def parse_args(): "-o", "--output", help="Ausgabedatei (nur bei einer Eingabedatei) oder Verzeichnis " "(vorhanden oder mit / am Ende); Standard: .txt, " - "mit -s/-e zusätzlich der Zeitraum im Namen", + "mit -s/-e/-z zusätzlich die Zeiträume im Namen", ) add_time_options(parser) args = parser.parse_args() - check_time_range(parser, args) - suffix = time_suffix(args.start, args.ende) + resolve_time_ranges(parser, args) + suffix = time_suffix(args.bereiche) sources, unmatched = expand_inputs(args.input) if unmatched is not None: @@ -304,7 +468,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, args.exakt) + error = convert(source, target, args.bereiche, args.exakt) if error: print(f"FEHLER: {error}", file=sys.stderr) failed += 1