241 lines
8.5 KiB
Python
241 lines
8.5 KiB
Python
|
#!/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())
|