pdf-ocr-hotfolder/pdf_ocr_hotfolder/config.py

"""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: pdfa_level + skip_text zerschießt OCR mit
    # Ghostscript 10.0.0-10.02.0 (Debian-12-Default), siehe Issue #3
    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_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."
        )
    return Path(str(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__})

    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, der zusammen mit skip_text das OCR blockiert (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)