"""Konfigurations-Loader (TOML).""" from __future__ import annotations import tomllib from dataclasses import dataclass, field from pathlib import Path from typing import Any class ConfigError(RuntimeError): """Konfigurationsdatei ist unvollständig oder fehlerhaft.""" @dataclass class Paths: incoming: Path outgoing: Path working: Path error: Path @dataclass class OcrConfig: languages: str = "deu+eng" jobs: int = 4 skip_text: bool = True oversample: int = 300 # Default bewusst leer: mit Ghostscript 10.0.0-10.02.0 (Debian-12-Default) # lehnt ocrmypdf die Kombination pdfa_level + skip_text ab (Issue #3). # ACHTUNG, kein Freibrief: pdfa_level = "" allein schützt nur zusammen mit # ocrmypdf >= 17. Bis 16.x läuft dieselbe Prüfung auch ohne PDF/A und # blockiert dann JEDE Datei — deshalb pinnt requirements.txt 17.x und der # Preflight prüft die installierte ocrmypdf-Version mit (siehe # service._gs_block_reason). pdfa_level: str = "" deskew: bool = True clean: bool = False max_workers: int = 2 # Max. Sekunden, die Tesseract pro Seite laufen darf (0 = kein eigenes Limit) timeout: int = 300 @dataclass class OutputConfig: # "prefix" | "suffix" | "none" name_mode: str = "prefix" # Tag-String, verbatim eingefügt (Leerstring = kein Tag) name_tag: str = "OCR_" # "delete" | "archive" original_on_success: str = "delete" # Absoluter Pfad; Pflicht wenn original_on_success == "archive" archive_dir: str = "" @dataclass class VeraPdfConfig: enabled: bool = False binary: str = "/opt/verapdf/verapdf" flavour: str = "1b" @dataclass class FolderUpload: enabled: bool = True target: str = "" @dataclass class NextcloudUpload: enabled: bool = False url: str = "" username: str = "" password: str = "" remote_path: str = "" verify_ssl: bool = True @dataclass class SftpUpload: enabled: bool = False host: str = "" port: int = 22 username: str = "" key_file: str = "" password: str = "" remote_path: str = "" @dataclass class EmailNotify: enabled: bool = False smtp_host: str = "" smtp_port: int = 587 smtp_user: str = "" smtp_password: str = "" use_starttls: bool = True from_addr: str = "" to_addrs: list[str] = field(default_factory=list) on: str = "errors" # always | errors | never @dataclass class Config: paths: Paths ocr: OcrConfig output: OutputConfig verapdf: VeraPdfConfig folder: FolderUpload nextcloud: NextcloudUpload sftp: SftpUpload email: EmailNotify log_level: str = "INFO" # Einträge der TOML, die zu keiner Dataclass gehören (Tippfehler oder # Optionen aus älteren Versionen). load_config() sammelt sie hier ein, # statt sie stumm zu verwerfen — geloggt wird erst weiter oben, damit # load_config() ohne konfiguriertes Logging benutzbar bleibt. unknown_keys: list[str] = field(default_factory=list) # Bekannte Sektionen — alles andere landet in Config.unknown_keys _KNOWN_SECTIONS = ("paths", "ocr", "output", "verapdf", "upload", "notify", "logging") _KNOWN_UPLOAD_TARGETS = ("folder", "nextcloud", "sftp") _KNOWN_NOTIFY_TARGETS = ("email",) _KNOWN_LOGGING_KEYS = ("level",) def _section(data: dict[str, Any], *keys: str) -> dict[str, Any]: cur: Any = data for k in keys: cur = cur.get(k, {}) if isinstance(cur, dict) else {} return cur if isinstance(cur, dict) else {} def _require_absolute(value: str, label: str, cfg_path: Path, beispiel: str) -> None: """Weist relative Pfadangaben zurück. Ein relativer Pfad wird gegen das Arbeitsverzeichnis des Prozesses aufgelöst — bei der systemd-Unit also gegen `WorkingDirectory` (/opt/pdf-ocr-hotfolder). `incoming = "in"` legte damit stillschweigend /opt/pdf-ocr-hotfolder/in an: der Scanner schreibt woanders hin als der Dienst schaut, und niemand sieht einen Fehler. Absolute Pfade sind die einzige sinnvolle Angabe; install.sh erzeugt ohnehin nur solche. """ if not value or Path(value).is_absolute(): return raise ConfigError( f"{cfg_path}: {label} = {value!r} ist ein relativer Pfad. Hier sind " f"nur absolute Pfade zulässig — ein relativer würde gegen das " f"Arbeitsverzeichnis des Dienstes aufgelöst " f"(WorkingDirectory, also z.B. /opt/pdf-ocr-hotfolder/{value}) und " f"nicht gegen das Verzeichnis, in dem die Config liegt. " f'Bitte absolut angeben, z.B. "{beispiel}".' ) def _require_path(p: dict[str, Any], key: str, cfg_path: Path) -> Path: """Holt einen Pflicht-Pfad aus der [paths]-Sektion. Wirft ConfigError mit klarer Meldung statt eines nackten KeyError. """ value = p.get(key) if value is None or (isinstance(value, str) and not value.strip()): raise ConfigError( f"{cfg_path}: In der Sektion [paths] fehlt der Eintrag '{key}' " f"(oder er ist leer). Bitte ergänzen, z.B. " f'{key} = "/var/lib/pdf-ocr-hotfolder/{key}" ' f"— siehe config.example.toml." ) value = str(value) _require_absolute(value, f"In der Sektion [paths] der Eintrag '{key}'", cfg_path, f"/var/lib/pdf-ocr-hotfolder/{key}") return Path(value) def _unknown_in(data: dict[str, Any], keys: tuple[str, ...], known: tuple[str, ...]) -> list[str]: """Listet alle Keys einer Sektion auf, die nicht in `known` stehen.""" label = ".".join(keys) return [f"[{label}].{k}" for k in _section(data, *keys) if k not in known] def _collect_unknown_keys(data: dict[str, Any]) -> list[str]: """Sammelt alle TOML-Einträge, die nirgends ausgewertet werden. Dazu zählen Tippfehler (`[ocr].langauges`), Optionen aus älteren Versionen und komplett unbekannte Sektionen. Die Reihenfolge entspricht der Datei, damit die Meldung reproduzierbar bleibt. """ unknown: list[str] = [] for name, value in data.items(): if name not in _KNOWN_SECTIONS: unknown.append(f"[{name}]" if isinstance(value, dict) else name) unknown += _unknown_in(data, ("paths",), tuple(Paths.__annotations__)) unknown += _unknown_in(data, ("ocr",), tuple(OcrConfig.__annotations__)) unknown += _unknown_in(data, ("output",), tuple(OutputConfig.__annotations__)) unknown += _unknown_in(data, ("verapdf",), tuple(VeraPdfConfig.__annotations__)) for sub in _section(data, "upload"): if sub not in _KNOWN_UPLOAD_TARGETS: unknown.append(f"[upload.{sub}]") unknown += _unknown_in(data, ("upload", "folder"), tuple(FolderUpload.__annotations__)) unknown += _unknown_in(data, ("upload", "nextcloud"), tuple(NextcloudUpload.__annotations__)) unknown += _unknown_in(data, ("upload", "sftp"), tuple(SftpUpload.__annotations__)) for sub in _section(data, "notify"): if sub not in _KNOWN_NOTIFY_TARGETS: unknown.append(f"[notify.{sub}]") unknown += _unknown_in(data, ("notify", "email"), tuple(EmailNotify.__annotations__)) unknown += _unknown_in(data, ("logging",), _KNOWN_LOGGING_KEYS) return unknown def load_config(path: str | Path) -> Config: path = Path(path) with path.open("rb") as f: data = tomllib.load(f) if not isinstance(data.get("paths"), dict): raise ConfigError( f"{path}: Die Sektion [paths] fehlt (oder ist keine Tabelle). " "Sie muss die Einträge incoming, outgoing, working und error " "enthalten — siehe config.example.toml." ) p = _section(data, "paths") paths = Paths( incoming=_require_path(p, "incoming", path), outgoing=_require_path(p, "outgoing", path), working=_require_path(p, "working", path), error=_require_path(p, "error", path), ) ocr = OcrConfig(**{k: v for k, v in _section(data, "ocr").items() if k in OcrConfig.__annotations__}) output = OutputConfig(**{k: v for k, v in _section(data, "output").items() if k in OutputConfig.__annotations__}) verapdf = VeraPdfConfig(**{k: v for k, v in _section(data, "verapdf").items() if k in VeraPdfConfig.__annotations__}) folder = FolderUpload(**{k: v for k, v in _section(data, "upload", "folder").items() if k in FolderUpload.__annotations__}) nextcloud = NextcloudUpload(**{k: v for k, v in _section(data, "upload", "nextcloud").items() if k in NextcloudUpload.__annotations__}) sftp = SftpUpload(**{k: v for k, v in _section(data, "upload", "sftp").items() if k in SftpUpload.__annotations__}) email = EmailNotify(**{k: v for k, v in _section(data, "notify", "email").items() if k in EmailNotify.__annotations__}) # Dieselbe Regel wie für [paths]: beides sind Verzeichnisse, in die der # Dienst schreibt, und beide wären relativ aufgelöst schlicht falsch. _require_absolute(str(output.archive_dir), "[output].archive_dir", path, "/var/lib/pdf-ocr-hotfolder/archive") _require_absolute(str(folder.target), "[upload.folder].target", path, "/srv/scans/fertig") log_level = _section(data, "logging").get("level", "INFO") return Config( paths=paths, ocr=ocr, output=output, verapdf=verapdf, folder=folder, nextcloud=nextcloud, sftp=sftp, email=email, log_level=log_level, unknown_keys=_collect_unknown_keys(data), ) # ---- Legacy- und Plausibilitätswarnungen ---- # [ocr].timeout war vor 0.4.0 ein (wirkungsloses) Gesamt-Timeout mit Default # 1800. Ab diesem Wert gehen wir von einem Altwert aus. LEGACY_TIMEOUT_THRESHOLD = 900 # Richtwert für das Seiten-Timeout seit 0.4.0 RECOMMENDED_PAGE_TIMEOUT = 300 def legacy_warnings(cfg: Config) -> list[str]: """Warnt vor Einträgen, deren Bedeutung sich geändert hat. Die Meldungen werden sowohl beim Dienststart ins Log geschrieben als auch von `--check-config` ausgegeben — deshalb steht der Text nur hier. """ out: list[str] = [] if cfg.ocr.timeout >= LEGACY_TIMEOUT_THRESHOLD: out.append( f"[ocr].timeout = {cfg.ocr.timeout}: Seit Version 0.4.0 sind das " "Sekunden PRO SEITE (vorher ein wirkungsloses Gesamt-Timeout mit " f"Default 1800). Ein Wert >= {LEGACY_TIMEOUT_THRESHOLD} stammt fast " "sicher aus einer alten Config und lässt eine einzelne Seite " f"unnötig lange laufen. Richtwert: {RECOMMENDED_PAGE_TIMEOUT}." ) if cfg.ocr.pdfa_level: out.append( f"[ocr].pdfa_level = {cfg.ocr.pdfa_level!r}: PDF/A-Ausgabe ist " "aktiv. Ghostscript 10.0.0-10.02.0 (Debian-12-Default) hat einen " "Bug, wegen dem ocrmypdf die Kombination mit skip_text ablehnt " "(Issue #3). Der Preflight bricht ab, falls die installierte " "Ghostscript-Version betroffen ist; ab 10.02.1 ist alles in " "Ordnung." ) return out def unknown_key_warnings(cfg: Config) -> list[str]: """Macht die beim Laden verworfenen Einträge sichtbar.""" return [ f"Unbekannter Config-Eintrag {key} — wird ignoriert (Tippfehler oder " "Option aus einer älteren Version?)" for key in cfg.unknown_keys ] def config_warnings(cfg: Config) -> list[str]: """Alle Warnungen zu einer geladenen Config (Legacy + unbekannte Keys).""" return legacy_warnings(cfg) + unknown_key_warnings(cfg)