feat: abschliessen.py – fertige Abschrift nach fertig/, Arbeitsordner löschen
Kopiert die korrigierte(n) Abschrift(en) eines Arbeitsordners mit Quellenkopf (Titel, Kanal, Datum, URL, Abschlussdatum) nach fertig/, listet Einträge des Video-Glossars, die in keinem gemeinsamen Glossar stehen, und löscht den Ordner nach Rückfrage (--ja ohne). Schutz: Abbruch bei offenen Prüfmarken (--trotzdem) und ungeklärten *.neu.korrigiert.txt; fertig/ wird nie überschrieben; nicht gelöscht werden Ordner mit fremden Dateien, das Aufrufverzeichnis und alles in testdaten/, glossare/, tests/, fertig/. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
3aa04fff4b
commit
ebaec63963
2 changed files with 403 additions and 0 deletions
241
abschliessen.py
Executable file
241
abschliessen.py
Executable file
|
|
@ -0,0 +1,241 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Schließt die Abschrift eines Videos ab: Die korrigierte Textdatei kommt mit einem
|
||||
Quellenkopf nach fertig/, der Arbeitsordner abschriften/<Name>/ wird nach Rückfrage
|
||||
gelöscht."""
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
import argparse
|
||||
import json
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
from abschrift_korrigieren import read_glossary, video_glossary
|
||||
from aenderungen import REVIEW_MARK
|
||||
|
||||
FINISHED_DIR = Path("fertig")
|
||||
GLOSSARY_DIR = Path("glossare")
|
||||
# What a working folder may hold; anything else stops the deletion
|
||||
WORK_SUFFIXES = {".txt", ".vtt", ".html", ".json"}
|
||||
# Never deleted, nor anything inside them
|
||||
PROTECTED = {"testdaten", "glossare", "tests", "fertig"}
|
||||
SHOWN_MARKS = 3
|
||||
|
||||
|
||||
def results(folder):
|
||||
"""Corrected texts of the folder (several for excerpts), without conflicts."""
|
||||
return sorted(
|
||||
p for p in folder.glob("*.korrigiert.txt")
|
||||
if not p.name.endswith(".neu.korrigiert.txt")
|
||||
)
|
||||
|
||||
|
||||
def finished_name(result):
|
||||
"""rede_00-05-00_bis_00-06-00.korrigiert.txt -> rede_00-05-00_bis_00-06-00.txt"""
|
||||
return result.name.removesuffix(".korrigiert.txt") + ".txt"
|
||||
|
||||
|
||||
def read_source(folder):
|
||||
"""Contents of the folder's <Rohtext>.quelle.json, None without one."""
|
||||
for path in sorted(folder.glob("*.quelle.json")):
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
return None
|
||||
|
||||
|
||||
def format_date(text):
|
||||
""""20260328" -> "2026-03-28"; anything else unchanged."""
|
||||
if text and len(text) == 8 and text.isdigit():
|
||||
return f"{text[:4]}-{text[4:6]}-{text[6:]}"
|
||||
return text
|
||||
|
||||
|
||||
def header(source, result, today):
|
||||
"""Source lines in front of the finished text, then a blank line."""
|
||||
lines = []
|
||||
if source:
|
||||
for label, value in (
|
||||
("Titel", source.get("titel")),
|
||||
("Kanal", source.get("kanal")),
|
||||
("Datum", format_date(source.get("datum"))),
|
||||
("Quelle", source.get("url")),
|
||||
):
|
||||
if value:
|
||||
lines.append(f"{label}: {value}")
|
||||
else:
|
||||
lines.append(f"Datei: {result.name}")
|
||||
lines.append(f"Abgeschlossen: {today}")
|
||||
return "\n".join(lines) + "\n\n"
|
||||
|
||||
|
||||
def already_finished(target, text):
|
||||
"""True if the target holds this text below a header."""
|
||||
if not target.exists():
|
||||
return False
|
||||
_, _, body = target.read_text(encoding="utf-8").partition("\n\n")
|
||||
return body == text
|
||||
|
||||
|
||||
def open_marks(text):
|
||||
"""(line number, line) of every line with a review mark."""
|
||||
return [
|
||||
(number, line.strip())
|
||||
for number, line in enumerate(text.splitlines(), start=1)
|
||||
if REVIEW_MARK.search(line)
|
||||
]
|
||||
|
||||
|
||||
def unshared_entries(glossary, glossary_dir):
|
||||
"""Entries of the video glossary that are in no shared glossary."""
|
||||
shared_terms, shared_replacements = set(), set()
|
||||
for path in sorted(glossary_dir.glob("*.glossar.txt")):
|
||||
terms, replacements = read_glossary(path)
|
||||
shared_terms.update(terms)
|
||||
shared_replacements.update(replacements)
|
||||
terms, replacements = read_glossary(glossary)
|
||||
return [
|
||||
f"{wrong} => {right}" for wrong, right in replacements
|
||||
if (wrong, right) not in shared_replacements
|
||||
] + [term for term in terms if term not in shared_terms]
|
||||
|
||||
|
||||
def deletion_blockers(folder):
|
||||
"""Reasons not to delete the folder; empty for a plain working folder."""
|
||||
resolved = folder.resolve()
|
||||
cwd = Path.cwd().resolve()
|
||||
if resolved == cwd or resolved in cwd.parents:
|
||||
return ["Er ist das Aufrufverzeichnis oder liegt darüber."]
|
||||
for path in (resolved, *resolved.parents):
|
||||
if path == cwd:
|
||||
break
|
||||
if path.name in PROTECTED:
|
||||
return [f"Er ist oder liegt in „{path.name}“, das nie gelöscht wird."]
|
||||
return [
|
||||
f"Fremde Datei: {entry.name}"
|
||||
for entry in sorted(folder.iterdir())
|
||||
if not (
|
||||
(entry.is_dir() and entry.name.endswith(".parts"))
|
||||
or (entry.is_file() and entry.suffix.lower() in WORK_SUFFIXES)
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def size(path):
|
||||
if path.is_file():
|
||||
return path.stat().st_size
|
||||
return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
|
||||
|
||||
|
||||
def confirm(question):
|
||||
try:
|
||||
return input(question).strip().lower() in {"j", "ja", "y", "yes"}
|
||||
except EOFError:
|
||||
return False
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Schließt die Abschrift eines Videos ab: Ergebnis mit Quellenkopf "
|
||||
"nach fertig/, Arbeitsordner nach Rückfrage löschen."
|
||||
)
|
||||
parser.add_argument("ordner", type=Path, help="Arbeitsordner, z. B. abschriften/<Name>")
|
||||
parser.add_argument(
|
||||
"--ziel", type=Path, default=FINISHED_DIR,
|
||||
help=f"Verzeichnis für fertige Abschriften (Standard: {FINISHED_DIR})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--glossare", type=Path, default=GLOSSARY_DIR,
|
||||
help=f"Verzeichnis der gemeinsamen Glossare (Standard: {GLOSSARY_DIR})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ja", action="store_true", help="Arbeitsordner ohne Rückfrage löschen"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--trotzdem", action="store_true",
|
||||
help="auch abschließen, wenn noch Prüfmarken [prüfen: …] im Text stehen",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
if not args.ordner.is_dir():
|
||||
parser.error(f"Ordner nicht gefunden: {args.ordner}")
|
||||
return args
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
folder = args.ordner
|
||||
|
||||
found = results(folder)
|
||||
if not found:
|
||||
raise SystemExit(f"Keine korrigierte Abschrift (*.korrigiert.txt) in {folder}.")
|
||||
conflicts = sorted(folder.glob("*.neu.korrigiert.txt"))
|
||||
if conflicts:
|
||||
names = ", ".join(p.name for p in conflicts)
|
||||
raise SystemExit(
|
||||
f"Ungeklärter Konflikt: {names}. Erst die Änderungen übertragen bzw. die "
|
||||
"Datei löschen, dann abschließen."
|
||||
)
|
||||
|
||||
texts = {result: result.read_text(encoding="utf-8") for result in found}
|
||||
marked = {result: open_marks(text) for result, text in texts.items()}
|
||||
if any(marked.values()):
|
||||
for result, marks in marked.items():
|
||||
if marks:
|
||||
print(f"{result.name}: {len(marks)} Zeile(n) mit Prüfmarken, z. B.")
|
||||
for number, line in marks[:SHOWN_MARKS]:
|
||||
print(f" Zeile {number}: {line[:120]}")
|
||||
if not args.trotzdem:
|
||||
raise SystemExit(
|
||||
"Noch offene Prüfmarken. Erst klären, oder mit --trotzdem abschließen."
|
||||
)
|
||||
|
||||
targets = {result: args.ziel / finished_name(result) for result in found}
|
||||
# A finished text with the same body is from an earlier run (e.g. answered
|
||||
# "n" when asked to delete); a different one is never overwritten
|
||||
done = {result for result, target in targets.items() if already_finished(target, texts[result])}
|
||||
existing = [str(t) for r, t in targets.items() if t.exists() and r not in done]
|
||||
if existing:
|
||||
raise SystemExit(
|
||||
f"Gibt es schon in {args.ziel} mit anderem Text: {', '.join(existing)}. "
|
||||
"Nichts überschrieben; vorhandene Datei umbenennen oder entfernen."
|
||||
)
|
||||
|
||||
source = read_source(folder)
|
||||
if source is None and len(done) < len(targets):
|
||||
print("Hinweis: keine Quellenangabe (*.quelle.json); der Kopf nennt nur die Datei.")
|
||||
today = date.today().isoformat()
|
||||
args.ziel.mkdir(parents=True, exist_ok=True)
|
||||
for result, target in targets.items():
|
||||
if result in done:
|
||||
print(f"Bereits fertig: {target}")
|
||||
continue
|
||||
target.write_text(header(source, result, today) + texts[result], encoding="utf-8")
|
||||
print(f"Fertig: {target}")
|
||||
|
||||
glossary = video_glossary(folder)
|
||||
if glossary is not None:
|
||||
missing = unshared_entries(glossary, args.glossare)
|
||||
if missing:
|
||||
print(
|
||||
f"\nDiese Einträge aus {glossary} stehen in keinem Glossar in "
|
||||
f"{args.glossare}/; vor dem Löschen ggf. in ein gemeinsames übernehmen:"
|
||||
)
|
||||
for entry in missing:
|
||||
print(f" {entry}")
|
||||
|
||||
blockers = deletion_blockers(folder)
|
||||
if blockers:
|
||||
print(f"\nArbeitsordner {folder} bleibt erhalten:")
|
||||
for reason in blockers:
|
||||
print(f" {reason}")
|
||||
return
|
||||
|
||||
print(f"\nArbeitsordner {folder}:")
|
||||
for entry in sorted(folder.iterdir()):
|
||||
print(f" {entry.name}{'/' if entry.is_dir() else ''} ({size(entry) / 1024:.0f} KB)")
|
||||
if not args.ja and not confirm(f"Arbeitsordner {folder} löschen? [j/N] "):
|
||||
print("Nicht gelöscht.")
|
||||
return
|
||||
shutil.rmtree(folder)
|
||||
print(f"Gelöscht: {folder}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
162
tests/test_abschliessen.py
Normal file
162
tests/test_abschliessen.py
Normal file
|
|
@ -0,0 +1,162 @@
|
|||
"""Tests for abschliessen.py (finish a video: result to fertig/, delete working folder)."""
|
||||
import json
|
||||
import sys
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
|
||||
import abschliessen as ab
|
||||
|
||||
TEXT = "Der Kreml ist in Moskau.\n\nDas ist bekannt.\n"
|
||||
SOURCE = {
|
||||
"titel": "Eine Rede", "kanal": "Kanal", "datum": "20260328",
|
||||
"url": "https://www.youtube.com/watch?v=abc123", "id": "abc123",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def work(tmp_path, monkeypatch):
|
||||
"""abschriften/rede/ with everything a run leaves behind; cwd is tmp_path."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
folder = tmp_path / "abschriften" / "rede"
|
||||
parts = folder / "rede.korrigiert.parts"
|
||||
parts.mkdir(parents=True)
|
||||
(parts / "part001-abc.txt").write_text("x", encoding="utf-8")
|
||||
(folder / "rede.txt").write_text("der kremmel ist in moskau\n", encoding="utf-8")
|
||||
(folder / "rede.korrigiert.txt").write_text(TEXT, encoding="utf-8")
|
||||
(folder / "rede.aenderungen.html").write_text("<p>", encoding="utf-8")
|
||||
(folder / "rede.quelle.json").write_text(json.dumps(SOURCE), encoding="utf-8")
|
||||
return folder
|
||||
|
||||
|
||||
def run_main(monkeypatch, *args, answer=None):
|
||||
monkeypatch.setattr(sys, "argv", ["abschliessen.py", *map(str, args)])
|
||||
if answer is not None:
|
||||
monkeypatch.setattr("builtins.input", lambda question: answer)
|
||||
ab.main()
|
||||
|
||||
|
||||
def test_finish_writes_header_and_deletes(work, tmp_path, monkeypatch, capsys):
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
|
||||
finished = (tmp_path / "fertig" / "rede.txt").read_text(encoding="utf-8")
|
||||
assert finished == (
|
||||
"Titel: Eine Rede\nKanal: Kanal\nDatum: 2026-03-28\n"
|
||||
"Quelle: https://www.youtube.com/watch?v=abc123\n"
|
||||
f"Abgeschlossen: {date.today().isoformat()}\n\n" + TEXT
|
||||
)
|
||||
assert not work.exists()
|
||||
out = capsys.readouterr().out
|
||||
assert "rede.korrigiert.parts/" in out and f"Gelöscht: {work}" in out
|
||||
|
||||
|
||||
def test_answer_no_keeps_folder_and_rerun_deletes(work, tmp_path, monkeypatch, capsys):
|
||||
run_main(monkeypatch, work, answer="n")
|
||||
assert work.is_dir()
|
||||
assert (tmp_path / "fertig" / "rede.txt").is_file()
|
||||
assert "Nicht gelöscht." in capsys.readouterr().out
|
||||
|
||||
# Later: the result is already in fertig/, only the deletion is left
|
||||
run_main(monkeypatch, work, answer="j")
|
||||
assert not work.exists()
|
||||
assert "Bereits fertig" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_open_review_marks_stop(work, tmp_path, monkeypatch, capsys):
|
||||
(work / "rede.korrigiert.txt").write_text(
|
||||
"Der Kreml [prüfen: Kremmel] ist in Moskau.\n", encoding="utf-8"
|
||||
)
|
||||
with pytest.raises(SystemExit, match="offene Prüfmarken"):
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
assert "Zeile 1: Der Kreml [prüfen: Kremmel]" in capsys.readouterr().out
|
||||
assert not (tmp_path / "fertig").exists() and work.is_dir()
|
||||
|
||||
run_main(monkeypatch, work, "--ja", "--trotzdem")
|
||||
assert (tmp_path / "fertig" / "rede.txt").is_file()
|
||||
|
||||
|
||||
def test_conflict_stops(work, monkeypatch):
|
||||
(work / "rede.neu.korrigiert.txt").write_text(TEXT, encoding="utf-8")
|
||||
with pytest.raises(SystemExit, match="Ungeklärter Konflikt: rede.neu.korrigiert.txt"):
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
assert work.is_dir()
|
||||
|
||||
|
||||
def test_no_result_stops(work, monkeypatch):
|
||||
(work / "rede.korrigiert.txt").unlink()
|
||||
with pytest.raises(SystemExit, match="Keine korrigierte Abschrift"):
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
|
||||
|
||||
def test_existing_different_target_is_never_overwritten(work, tmp_path, monkeypatch):
|
||||
(tmp_path / "fertig").mkdir()
|
||||
(tmp_path / "fertig" / "rede.txt").write_text("Anderer Text\n", encoding="utf-8")
|
||||
with pytest.raises(SystemExit, match="mit anderem Text"):
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
assert (tmp_path / "fertig" / "rede.txt").read_text(encoding="utf-8") == "Anderer Text\n"
|
||||
assert work.is_dir()
|
||||
|
||||
|
||||
def test_without_source_header_names_the_file(work, tmp_path, monkeypatch, capsys):
|
||||
(work / "rede.quelle.json").unlink()
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
finished = (tmp_path / "fertig" / "rede.txt").read_text(encoding="utf-8")
|
||||
assert finished.startswith("Datei: rede.korrigiert.txt\nAbgeschlossen: ")
|
||||
assert "keine Quellenangabe" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_every_excerpt_is_finished(work, tmp_path, monkeypatch):
|
||||
(work / "rede_ab_00-05-00.korrigiert.txt").write_text("[00:05:00 - 00:06:00]\n\nB.\n")
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
assert (tmp_path / "fertig" / "rede.txt").is_file()
|
||||
assert (tmp_path / "fertig" / "rede_ab_00-05-00.txt").read_text().endswith(
|
||||
"[00:05:00 - 00:06:00]\n\nB.\n"
|
||||
)
|
||||
|
||||
|
||||
def test_foreign_file_prevents_deletion(work, tmp_path, monkeypatch, capsys):
|
||||
(work / "notizen.docx").write_bytes(b"x")
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
assert work.is_dir()
|
||||
assert (tmp_path / "fertig" / "rede.txt").is_file()
|
||||
assert "Fremde Datei: notizen.docx" in capsys.readouterr().out
|
||||
|
||||
|
||||
@pytest.mark.parametrize("protected", ["testdaten", "glossare", "tests", "fertig"])
|
||||
def test_protected_folders_are_never_deleted(tmp_path, monkeypatch, capsys, protected):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
folder = tmp_path / protected / "rede"
|
||||
folder.mkdir(parents=True)
|
||||
(folder / "rede.korrigiert.txt").write_text(TEXT, encoding="utf-8")
|
||||
run_main(monkeypatch, folder, "--ja", "--ziel", tmp_path / "ziel")
|
||||
assert folder.is_dir()
|
||||
assert f"„{protected}“, das nie gelöscht wird" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_current_directory_is_never_deleted(tmp_path, monkeypatch, capsys):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / "rede.korrigiert.txt").write_text(TEXT, encoding="utf-8")
|
||||
run_main(monkeypatch, ".", "--ja")
|
||||
assert (tmp_path / "rede.korrigiert.txt").is_file()
|
||||
assert "Aufrufverzeichnis" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_video_glossary_entries_not_shared_are_listed(work, tmp_path, monkeypatch, capsys):
|
||||
(work / "rede.glossar.txt").write_text(
|
||||
"Kremmel => Kreml\nNetano => Netanjahu\nEgon Bahr\nMoskau\n", encoding="utf-8"
|
||||
)
|
||||
(tmp_path / "glossare").mkdir()
|
||||
(tmp_path / "glossare" / "gemeinsam.glossar.txt").write_text(
|
||||
"Kremmel => Kreml\nMoskau\n", encoding="utf-8"
|
||||
)
|
||||
run_main(monkeypatch, work, "--ja")
|
||||
out = capsys.readouterr().out
|
||||
listed = out.split("gemeinsames übernehmen:")[1].split("\n\n")[0]
|
||||
assert "Netano => Netanjahu" in listed and "Egon Bahr" in listed
|
||||
assert "Kremmel" not in listed and "Moskau" not in listed
|
||||
|
||||
|
||||
def test_missing_folder(tmp_path, monkeypatch, capsys):
|
||||
with pytest.raises(SystemExit):
|
||||
run_main(monkeypatch, tmp_path / "fehlt")
|
||||
assert "Ordner nicht gefunden" in capsys.readouterr().err
|
||||
Loading…
Reference in a new issue