285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""Zentrales Datenmodell der Wissens-Pipeline.
|
|
|
|
Alles dreht sich um: Tool -> Quelle(n) mit Strategie + Scope -> Dokument(e).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import date
|
|
from enum import Enum
|
|
|
|
# Default-Kontaktadresse (Ansprechpartner), die herausgegeben werden darf.
|
|
# Bewusst getrennt von 'owners' (= intern Verantwortliche).
|
|
DEFAULT_CONTACT = "einfachbahn@deutschebahn.com"
|
|
|
|
_PAGE_ID_RE = re.compile(r"/pages/(\d+)|pageId=(\d+)")
|
|
|
|
|
|
def page_identity(url: str) -> str:
|
|
"""Stabile Identitaet einer Quelle fuer Dedup.
|
|
|
|
Dieselbe Confluence-Seite kann aus mehreren Quellen kommen (z.B. ein breiter
|
|
confluence_tree UND eine dedizierte confluence_faq-Quelle) und in unterschiedlichen
|
|
URL-Schreibweisen (`/spaces/.../pages/<id>/...` vs. `pageId=<id>`). Wir reduzieren
|
|
daher auf die Confluence-Page-ID; sonst auf die normalisierte URL. So zaehlt eine
|
|
Seite als EINE Einheit -> Dedup im Review-Gate.
|
|
"""
|
|
if not url:
|
|
return ""
|
|
m = _PAGE_ID_RE.search(url)
|
|
if m:
|
|
return "confluence:" + (m.group(1) or m.group(2))
|
|
return url.rstrip("/")
|
|
|
|
|
|
class Scope(str, Enum):
|
|
"""Sichtbarkeit eines Wissensdokuments.
|
|
|
|
ALLGEMEIN gilt sowohl fuer intern als auch extern (Schnittmenge).
|
|
"""
|
|
|
|
INTERN = "intern"
|
|
EXTERN = "extern"
|
|
ALLGEMEIN = "allgemein"
|
|
|
|
@property
|
|
def visible_extern(self) -> bool:
|
|
return self in (Scope.EXTERN, Scope.ALLGEMEIN)
|
|
|
|
@property
|
|
def visible_intern(self) -> bool:
|
|
return self in (Scope.INTERN, Scope.ALLGEMEIN)
|
|
|
|
|
|
class Strategy(str, Enum):
|
|
"""Herangehensweise pro Quelle (austauschbar)."""
|
|
|
|
CONFLUENCE_PAGE = "confluence_page" # nur genau diese Seite
|
|
CONFLUENCE_TREE = "confluence_tree" # Seite inkl. aller Unterseiten
|
|
CONFLUENCE_FAQ = "confluence_faq" # FAQs erzeugen (z.B. aus Tabellen)
|
|
CRAWLER = "crawler" # allgemeiner rekursiver Web-Crawl
|
|
SITEMAP = "sitemap" # Detailseiten aus einer (gz-)Sitemap
|
|
GITLAB_MD = "gitlab_md" # Markdown-Dateien aus einem GitLab-Repo
|
|
FILE = "file" # Dateien (md/pdf) aus dem Repo-Ordner files/
|
|
PDF = "pdf" # PDF-Links einer Seite -> Markdown
|
|
|
|
|
|
class ComponentType(str, Enum):
|
|
PAGE = "page"
|
|
FAQ = "faq"
|
|
PDF = "pdf"
|
|
|
|
|
|
@dataclass
|
|
class Source:
|
|
"""Eine einzelne Wissensquelle eines Tools (1..n pro Tool).
|
|
|
|
scopes: eine oder mehrere Sichtbarkeiten. 'intern,extern' bedeutet: die ganze
|
|
Seite wird fuer BEIDE genutzt (dupliziert, KEINE Inhaltstrennung pro Seite).
|
|
"""
|
|
|
|
url: str
|
|
strategy: Strategy
|
|
scopes: list[Scope] = field(default_factory=lambda: [Scope.INTERN])
|
|
tags: list[str] = field(default_factory=list)
|
|
options: dict = field(default_factory=dict)
|
|
domain: str = "allgemein"
|
|
trusted: bool = False
|
|
owners: list[str] = field(default_factory=list) # intern Verantwortliche (Accountability)
|
|
contact: str = "" # herausgebbare Kontaktadresse (Ansprechpartner)
|
|
|
|
@property
|
|
def scope(self) -> Scope:
|
|
return self.scopes[0]
|
|
|
|
@property
|
|
def contact_address(self) -> str:
|
|
"""Kontaktadresse mit Default einfachbahn@deutschebahn.com."""
|
|
return self.contact or DEFAULT_CONTACT
|
|
|
|
@staticmethod
|
|
def _parse_scopes(raw) -> list[Scope]:
|
|
if raw is None:
|
|
return [Scope.INTERN]
|
|
items = raw if isinstance(raw, list) else str(raw).split(",")
|
|
scopes = [Scope(s.strip()) for s in items if str(s).strip()]
|
|
return scopes or [Scope.INTERN]
|
|
|
|
@staticmethod
|
|
def _parse_list(raw) -> list[str]:
|
|
if not raw:
|
|
return []
|
|
items = raw if isinstance(raw, list) else str(raw).split(",")
|
|
return [str(s).strip() for s in items if str(s).strip()]
|
|
|
|
@staticmethod
|
|
def from_dict(d: dict) -> "Source":
|
|
return Source(
|
|
url=d["url"],
|
|
strategy=Strategy(d["strategy"]),
|
|
scopes=Source._parse_scopes(d.get("scope")),
|
|
tags=list(d.get("tags", [])),
|
|
options=dict(d.get("options", {})),
|
|
domain=d.get("domain", "allgemein"),
|
|
trusted=bool(d.get("trusted", False)),
|
|
owners=Source._parse_list(d.get("owners")),
|
|
contact=str(d.get("contact", "")).strip(),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Tool:
|
|
"""Eine Fachanwendung / ein Thema im 1st-Level-Support."""
|
|
|
|
id: str
|
|
name: str
|
|
domain: str = "allgemein"
|
|
scope: str = "intern" # tool-weit: intern|extern|allgemein|mixed
|
|
owners: list[str] = field(default_factory=list) # intern Verantwortliche
|
|
contact: str = "" # herausgebbare Kontaktadresse
|
|
sources: list[Source] = field(default_factory=list)
|
|
|
|
@staticmethod
|
|
def from_dict(d: dict) -> "Tool":
|
|
domain = d.get("domain", "allgemein")
|
|
tool_scope = d.get("scope", "intern") # intern|extern|allgemein|mixed
|
|
tool_owners = Source._parse_list(d.get("owners"))
|
|
tool_contact = str(d.get("contact", "")).strip()
|
|
sources: list[Source] = []
|
|
for sd in d.get("sources", []):
|
|
s = Source.from_dict(sd)
|
|
if not s.domain or s.domain == "allgemein":
|
|
s.domain = domain
|
|
if "scope" not in sd and tool_scope in ("intern", "extern", "allgemein"):
|
|
s.scopes = [Scope(tool_scope)]
|
|
if not s.owners:
|
|
s.owners = list(tool_owners)
|
|
if not s.contact:
|
|
s.contact = tool_contact
|
|
sources.append(s)
|
|
return Tool(id=d["id"], name=d.get("name", d["id"]), domain=domain,
|
|
scope=tool_scope, owners=tool_owners, contact=tool_contact,
|
|
sources=sources)
|
|
|
|
|
|
class ReviewStatus(str, Enum):
|
|
PENDING = "pending"
|
|
APPROVED = "approved"
|
|
|
|
|
|
@dataclass
|
|
class Document:
|
|
"""Ein aufbereitetes Wissensdokument (wird zu einer Markdown-Datei)."""
|
|
|
|
tool: str # Tool-Id oder "allgemein"
|
|
scope: Scope
|
|
title: str
|
|
body_md: str
|
|
url: str
|
|
domain: str = "allgemein"
|
|
component_type: ComponentType = ComponentType.PAGE
|
|
tags: list[str] = field(default_factory=list)
|
|
owners: list[str] = field(default_factory=list) # intern Verantwortliche
|
|
contact: str = DEFAULT_CONTACT # herausgebbare Kontaktadresse
|
|
source_system: str = "web"
|
|
source_version: str = "" # z.B. Confluence-Version (fuer inkrementelles Crawlen)
|
|
meta_fingerprint: str = "" # Hash der Metadaten (tags/owners/contact) -> inkrementelles Re-Tagging
|
|
part: str = "" # optionaler Teil-Bezeichner (z.B. INB-Abschnitt)
|
|
last_updated: str = field(default_factory=lambda: date.today().isoformat())
|
|
review_status: ReviewStatus = ReviewStatus.PENDING
|
|
review_notes: list[str] = field(default_factory=list)
|
|
# Chunking (nur bei Chunks gesetzt; Voll-Dokument: kind="document")
|
|
kind: str = "document" # "document" | "chunk"
|
|
parent_url: str = "" # URL/Quelle des Voll-Dokuments
|
|
parent_hash: str = "" # content_hash des Voll-Dokuments
|
|
section: str = "" # Abschnitts-Ueberschrift des Chunks
|
|
ziffer: str = "" # erkannte Gliederungsziffer (z.B. 7.3.1.1.1.1)
|
|
chunk_index: int = 0 # laufende Nummer innerhalb des Voll-Dokuments
|
|
chunk_fingerprint: str = "" # Hash der wirksamen Chunk-Optionen (-> inkrementelles Re-Chunking)
|
|
attachment_name: str = "" # Dateiname des Anhangs (nur bei kind="attachment")
|
|
|
|
@property
|
|
def content_hash(self) -> str:
|
|
return hashlib.sha256(self.body_md.encode("utf-8")).hexdigest()[:16]
|
|
|
|
def slug(self) -> str:
|
|
base = f"{self.title} {self.part}".strip() or self.url
|
|
keep = [c.lower() if c.isalnum() else "-" for c in base]
|
|
s = "".join(keep)
|
|
while "--" in s:
|
|
s = s.replace("--", "-")
|
|
return s.strip("-")[:80] or self.content_hash
|
|
|
|
def to_markdown(self) -> str:
|
|
fm = [
|
|
"---",
|
|
f'domain: "{self.domain}"',
|
|
f'tool: "{self.tool}"',
|
|
f'scope: "{self.scope.value}"',
|
|
f"tags: [{', '.join(self._q(t) for t in self.tags)}]",
|
|
f"owners: [{', '.join(self._q(o) for o in self.owners)}]",
|
|
f'contact: "{self.contact}"',
|
|
f'component_type: "{self.component_type.value}"',
|
|
f'source: "{self.source_system}"',
|
|
f'source_version: "{self.source_version}"',
|
|
f'meta_fingerprint: "{self.meta_fingerprint}"',
|
|
f'url: "{self.url}"',
|
|
]
|
|
if self.part:
|
|
fm.append(f'part: "{self.part}"')
|
|
if self.kind == "chunk":
|
|
fm += [
|
|
'kind: "chunk"',
|
|
f'parent_url: "{self.parent_url}"',
|
|
f'parent_hash: "{self.parent_hash}"',
|
|
f'section: "{self.section}"',
|
|
f'ziffer: "{self.ziffer}"',
|
|
f'chunk_index: {self.chunk_index}',
|
|
f'chunk_fingerprint: "{self.chunk_fingerprint}"',
|
|
]
|
|
if self.kind == "attachment":
|
|
fm += [
|
|
'kind: "attachment"',
|
|
f'parent_url: "{self.parent_url}"',
|
|
f'attachment_name: "{self.attachment_name}"',
|
|
]
|
|
fm += [
|
|
f'last_updated: "{self.last_updated}"',
|
|
f'review_status: "{self.review_status.value}"',
|
|
f'review_notes: "{self._notes_str()}"',
|
|
f'content_hash: "{self.content_hash}"',
|
|
"---",
|
|
"",
|
|
self.body_md.strip(),
|
|
"",
|
|
]
|
|
return "\n".join(fm)
|
|
|
|
def _notes_str(self) -> str:
|
|
joined = " | ".join(self.review_notes)
|
|
return joined.replace('"', "'")
|
|
|
|
@staticmethod
|
|
def _q(s: str) -> str:
|
|
return f'"{s}"'
|
|
|
|
|
|
def source_meta_fingerprint(source: "Source") -> str:
|
|
"""Stabiler Hash der herausgebbaren Metadaten einer Quelle.
|
|
|
|
Bewusst NUR Felder, die NICHT den Ablagepfad bestimmen: tags, owners, contact.
|
|
(scope/domain steuern den Pfad und werden separat behandelt.) Aendert sich einer
|
|
dieser Werte in der Config, weicht der Fingerprint vom gespeicherten ab und die
|
|
Pipeline aktualisiert die betroffenen Dokumente inkrementell.
|
|
"""
|
|
payload = {
|
|
"tags": sorted(t.strip().lower() for t in source.tags if t.strip()),
|
|
"owners": sorted(o.strip().lower() for o in source.owners if o.strip()),
|
|
"contact": source.contact_address.strip().lower(),
|
|
}
|
|
raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
|