cluster-knaster/core/src/knaster/ontology/model.py

"""Ontologie als Quelle der Wahrheit: Kern (Klassen, Beziehungen), Vokabulare, Abbildung auf Tabellen.

Drei Ebenen mit unterschiedlicher Pflege (SPEC-ontology §1):
- Kern: bewusst gepflegt, Änderungen per Commit.
- Vokabulare: wachsen über Vorschläge aus der Service-Routine (beobachtet → vorgeschlagen → bestätigt).
- Instanzen: leben in den Daten, nicht in dieser Datei.
"""

from __future__ import annotations

from pathlib import Path
from typing import Literal

import yaml
from pydantic import BaseModel, Field

from ..util import load_yaml

RelationKind = Literal["key", "soft", "external", "subclass"]


class OntologyClass(BaseModel):
    label: str
    module: str
    definition: str = ""
    table: str | None = None  # None = implizit (nur als Wert in Spalten) oder externes System
    key: str = ""
    sensitive: str = ""
    abstract: bool = False
    subclass_of: str | None = None
    attributes: dict[str, str] = Field(default_factory=dict)  # Spalte → SQL-Typ (Tabellenklassen)


class Relation(BaseModel):
    source: str
    name: str
    target: str
    via: str = ""
    kind: RelationKind = "key"


class Concept(BaseModel):
    label: str = ""
    alt: list[str] = Field(default_factory=list)  # weitere Schreibweisen, die auf dieses Konzept zeigen
    note: str = ""
    technical: bool = False  # kein fachlicher Wert (z. B. Rückfall „default“); zählt nicht als „fehlt“


class Vocabulary(BaseModel):
    label: str
    class_: str | None = Field(default=None, alias="class")
    columns: list[str] = Field(default_factory=list)  # "tabelle.spalte", in denen Werte dieses Vokabulars stehen
    concepts: dict[str, Concept] = Field(default_factory=dict)
    rejected: list[str] = Field(default_factory=list)  # bewusst verworfene Werte (kein Konzept)

    model_config = {"populate_by_name": True}

    def resolve(self, value: str) -> str | None:
        """Wert → Konzept-ID (über ID, Label oder alternative Schreibweise), sonst None."""
        v = value.strip().lower()
        for cid, c in self.concepts.items():
            if v == cid.lower() or v == c.label.lower() or v in (a.lower() for a in c.alt):
                return cid
        return None

    def is_rejected(self, value: str) -> bool:
        return value.strip().lower() in (r.lower() for r in self.rejected)


class Ontology(BaseModel):
    id: str = "know"
    label: str = ""
    namespace: str = "urn:example:onto#"
    version: int = 1
    modules: dict[str, str] = Field(default_factory=dict)
    classes: dict[str, OntologyClass] = Field(default_factory=dict)
    relations: list[Relation] = Field(default_factory=list)
    vocabularies: dict[str, Vocabulary] = Field(default_factory=dict)
    ignore: list[str] = Field(default_factory=list)  # bewusst hingenommene Befunde (Schlüssel aus `check`)

    @classmethod
    def from_file(cls, path: str | Path) -> Ontology:
        return cls(**load_yaml(path))

    def save(self, path: str | Path, header: str = "") -> None:
        data = {"id": self.id, "version": self.version, **self.model_dump(by_alias=True, exclude_defaults=True)}
        text = yaml.safe_dump(data, sort_keys=False, allow_unicode=True, width=120)
        Path(path).write_text(header + text, encoding="utf-8")

    # --- Abfragen ---------------------------------------------------------------------------
    def class_for_table(self, table: str) -> str | None:
        return next((cid for cid, c in self.classes.items() if c.table == table), None)

    def tables(self) -> dict[str, str]:
        return {c.table: cid for cid, c in self.classes.items() if c.table}

    def vocabularies_for(self, table: str, column: str) -> list[str]:
        ref = f"{table}.{column}"
        return [vid for vid, v in self.vocabularies.items() if ref in v.columns]

    def vocab_columns(self) -> dict[str, list[str]]:
        out: dict[str, list[str]] = {}
        for vid, v in self.vocabularies.items():
            for ref in v.columns:
                out.setdefault(ref, []).append(vid)
        return out

    # --- Prüfung ----------------------------------------------------------------------------
    def validate_model(self) -> list[str]:
        errors: list[str] = []
        for cid, c in self.classes.items():
            if c.module not in self.modules:
                errors.append(f"Klasse {cid}: Modul {c.module} unbekannt")
            if c.subclass_of and c.subclass_of not in self.classes:
                errors.append(f"Klasse {cid}: Oberklasse {c.subclass_of} unbekannt")
            if c.attributes and not c.table:
                errors.append(f"Klasse {cid}: Attribute ohne Tabelle")
        seen_tables: dict[str, str] = {}
        for cid, c in self.classes.items():
            if c.table:
                if c.table in seen_tables:
                    errors.append(f"Tabelle {c.table} ist {seen_tables[c.table]} und {cid} zugeordnet")
                seen_tables[c.table] = cid
        for r in self.relations:
            for end in (r.source, r.target):
                if end not in self.classes:
                    errors.append(f"Beziehung {r.source} {r.name} {r.target}: Klasse {end} unbekannt")
        for vid, v in self.vocabularies.items():
            if v.class_ and v.class_ not in self.classes:
                errors.append(f"Vokabular {vid}: Klasse {v.class_} unbekannt")
            for ref in v.columns:
                table, _, column = ref.partition(".")
                owner = self.class_for_table(table)
                if owner is None:
                    errors.append(f"Vokabular {vid}: Tabelle {table} keiner Klasse zugeordnet")
                elif column not in self.classes[owner].attributes:
                    errors.append(f"Vokabular {vid}: Spalte {ref} fehlt in Klasse {owner}")
            labels: dict[str, str] = {}
            for cid, c in v.concepts.items():
                for name in [cid, c.label, *c.alt]:
                    key = name.strip().lower()
                    if key and key in labels and labels[key] != cid:
                        errors.append(f"Vokabular {vid}: „{name}“ zeigt auf {labels[key]} und {cid}")
                    labels.setdefault(key, cid)
            for r in v.rejected:
                if v.resolve(r):
                    errors.append(f"Vokabular {vid}: „{r}“ ist zugleich verworfen und Konzept {v.resolve(r)}")
        return errors