Orchestrator/bahn/wissensdatenbank/src/main.py

"""Orchestrator der Wissens-ETL-Pipeline.

Ablauf:  Config laden -> pro Tool/General-Quelle Connector ausfuehren ->
Tags vergeben -> filtern -> Review-Gate (processed/ vs staging/).

Lokaler Lauf:
    python -m src.main --config config/tools.yaml --data output --staging staging
"""
from __future__ import annotations

import argparse
from pathlib import Path

from .config_loader import load_approvals, load_filter_rules, load_tools
from .connectors import get_connector, set_data_dir
from .model import Document, ReviewStatus, Source, Tool, source_meta_fingerprint
from .review.staging import ReviewGate
from .store import append_run_log, prune_duplicate_files, prune_scope_orphans, write_run_meta
from .transformers.content_filter import filter_document
from .transformers.tagger import apply_tags


def process_source(source: Source, tool: Tool | None, rules: dict, gate: ReviewGate,
                   approvals: dict | None = None) -> dict:
    """Verarbeitet eine Quelle. Rueckgabe: Ergebnis-Dict fuers Lauf-Log."""
    label = f"{source.domain}/{tool.id if tool else 'allgemein'}"
    print(f"-> {label} | {source.strategy.value} | {source.scope.value} | {source.url}")
    result = {"source": label, "strategy": source.strategy.value, "scope": source.scope.value,
              "url": source.url, "docs": 0, "doc_errors": 0, "error": None}
    try:
        connector = get_connector(source.strategy)
        docs: list[Document] = connector.fetch(source, tool)
    except Exception as e:  # noqa: BLE001
        print(f"   Fehler beim Abruf: {e}")
        result["error"] = str(e)
        return result

    fingerprint = source_meta_fingerprint(source)
    produced: dict[tuple[str, str], set[str]] = {}
    doc_errors = 0
    for doc in docs:
        try:
            apply_tags(doc)
            doc.meta_fingerprint = fingerprint
            filter_document(doc, rules, trusted=source.trusted, approvals=approvals,
                            redact=source.options.get("redact", True))
            gate.write(doc)
            if doc.review_status == ReviewStatus.APPROVED:
                produced.setdefault((doc.domain, doc.url), set()).add(doc.scope.value)
        except Exception as e:  # noqa: BLE001 - ein defektes Dokument darf den Lauf nicht stoppen
            print(f"   Fehler bei Dokument ({getattr(doc, 'url', '?')}): {e} -> uebersprungen")
            doc_errors += 1
    print(f"   {len(docs)} Dokument(e) verarbeitet")

    # Nach erfolgreichem Schreiben: alte Ablage entfernen, wenn der Scope gewechselt hat
    if produced:
        removed = prune_scope_orphans(gate.output_dir, produced)
        for p in removed:
            print(f"   [orphan] entfernt (Scope-Wechsel): {p}")

    result["docs"] = len(docs)
    result["doc_errors"] = doc_errors
    return result


def run(config_path: str, data_dir: str, filter_path: str, only: str | None = None,
        staging_dir: str = "staging") -> int:
    tools, general = load_tools(config_path)
    rules = load_filter_rules(filter_path)
    approvals = load_approvals(Path(config_path).parent / "approvals.yaml")
    set_data_dir(data_dir, staging_dir)
    gate = ReviewGate(data_dir, staging_dir)

    def matches(source: Source, tool: Tool | None) -> bool:
        if not only:
            return True
        hay = f"{tool.id if tool else 'allgemein'} {source.domain} {source.url}"
        return only.lower() in hay.lower()

    results: list[dict] = []
    for tool in tools:
        for source in tool.sources:
            if matches(source, tool):
                results.append(process_source(source, tool, rules, gate, approvals))
    for source in general:
        if matches(source, None):
            results.append(process_source(source, None, rules, gate, approvals))

    total = sum(r["docs"] for r in results)

    # Alt-Duplikate derselben Seite (Hash-Suffix) aufraeumen; im Voll-Lauf, da es den
    # Gesamtbestand betrifft (das Gate verhindert NEUE Duplikate bereits beim Schreiben).
    if not only:
        for p in prune_duplicate_files(data_dir):
            print(f"   [dedup] entferntes Duplikat: {p}")

    report = gate.write_report()
    meta_path = write_run_meta(data_dir, gate.summary())

    # Lauf-Log: append-only Historie (output/run_log.jsonl) inkl. Fehler je Quelle,
    # damit man (auch nachtraeglich) sieht, ob ein Lauf sauber durchlief.
    errors = [{"source": r["source"], "strategy": r["strategy"], "error": r["error"]}
              for r in results if r["error"]]
    log_entry = {
        "documents": total,
        "by_status": gate.summary(),
        "sources_total": len(results),
        "sources_failed": len(errors),
        "doc_errors": sum(r["doc_errors"] for r in results),
        "errors": errors,
    }
    log_path = append_run_log(data_dir, log_entry)

    # Chunking (inkrementell, offline): haelt output/chunks synchron zum freigegebenen
    # Bestand. Default aus - nur Quellen mit options.chunk != off. Bei Vorschaulaeufen
    # (--only) uebersprungen, da nur ein Teilbestand vorliegt. Fehler hier duerfen den
    # ETL-Lauf NICHT abbrechen (Chunks sind ein abgeleitetes Zusatz-Artefakt).
    if not only:
        try:
            from .chunk import run as chunk_run
            print("\n=== Chunking ===")
            chunk_run(config_path, data_dir)
        except Exception as e:  # noqa: BLE001 - bewusst breit: Chunking ist nicht kritisch
            print(f"   Chunking uebersprungen (Fehler, nicht kritisch): {e}")

    print("\n=== Zusammenfassung ===")
    print(f"Dokumente gesamt: {total}")
    for status, count in sorted(gate.summary().items()):
        print(f"  {status:10s}: {count}")
    if errors:
        print(f"Quellen mit Fehler: {len(errors)}")
        for e in errors:
            print(f"  ! {e['source']} ({e['strategy']}): {e['error']}")
    print(f"Report: {report}")
    print(f"Meta:   {meta_path}")
    print(f"Log:    {log_path}")
    return total


def main() -> None:
    parser = argparse.ArgumentParser(description="Wissens-ETL-Pipeline")
    parser.add_argument("--config", default="config/tools.yaml")
    parser.add_argument("--data", default="output",
                        help="Output-Verzeichnis (Konsumenten-Feed): processed/, chunks/, _index.json, _meta.json, run_log.jsonl")
    parser.add_argument("--staging", default="staging",
                        help="Staging-Verzeichnis (intern): pending/, review_report.json")
    parser.add_argument("--filter", default="config/filter_rules.json")
    parser.add_argument("--only", default=None,
                        help="nur Quellen verarbeiten, die diesen Text (Tool/Domaene/URL) enthalten (Vorschau)")
    args = parser.parse_args()

    if not Path(args.config).exists():
        raise SystemExit(f"Config nicht gefunden: {args.config}")
    run(args.config, args.data, args.filter, only=args.only, staging_dir=args.staging)


if __name__ == "__main__":
    main()