breakpilot-core/rag-service/legal_metadata.py

"""Legal-Metadaten-Normalisierung fuer zitierfaehige Chunks.

Vertrag: docs-src/development/rag_reingest_spec.md (§2 Payload-Feld-Vertrag, §3 Transform).
Transformiert die Roh-Ausgabe des Legal-Chunkers (section="§ 38" / "Artikel 13",
section_title, paragraph="(1)") in die consumer-facing Felder
(article, citation_style, article_label, paragraph, sub, section_header, is_recital).

Pure Funktionen (nur stdlib) -> lokal + in CI testbar, haelt documents.py schlank.
"""

import hashlib
import re
import uuid

# Namespace fuer deterministische Point-IDs (Qdrant braucht UUID/uint, kein roher sha1).
_ID_NS = uuid.UUID("6ba7b811-9dad-11d1-80b4-00c04fd430c8")

# Strikt: nur echte Legal-Header am Anfang des Section-Strings (optional in [..]).
# Lehnt Datums-/Seiten-/Fundstellen-Rauschen ab ("4.5.2016", "L 119/9", "Begriffsbestimmungen").
_LEGAL_HEADER_RE = re.compile(r"^\s*\[?\s*(§|art\.?|artikel)\s*(\d+)\s*([a-z])?", re.IGNORECASE)
_SUB_RE = re.compile(r"(lit\.\s*[a-z]+|Satz\s*\d+|Nr\.\s*\d+)", re.IGNORECASE)

# Urteile/Beschluesse: KEINE §-/Art.-Fundstelle bilden. Der Entscheidungstext zitiert
# §§ fremder Gesetze (Querverweise) -> wuerde ein falsches Label "AZ § 87" erzeugen.
# Zitiert wird das Aktenzeichen (display_name / regulation_short).
_RULING_TYPES = {"urteil", "ruling", "court_decision", "beschluss"}

# Akronyme, deren KORREKTE Schreibweise GROSS ist. Der Ingest leitet regulation_short
# z.T. via title()-Casing aus Dateinamen ab ("dsgvo" -> "Dsgvo", "osha otm" -> "Osha Otm").
# Token-basiert: NUR Tokens, deren Upper-Form hier steht, werden gross geschrieben — damit
# bleiben legitime Mixed-Case-Kuerzel (GeschGehG, MuSchG, GoBD, MiCA, eIDAS, EuGH) unberuehrt.
# Bewusst NICHT enthalten: EUGH/BVERFG/BVERWG (korrekt sind EuGH/BVerfG/BVerwG).
# Konservativ + erweiterbar.
_REG_SHORT_ACRONYMS = {
    "DSGVO", "GDPR", "DSA", "DMA", "DORA", "AMLR", "GPSR", "DPF", "IFRS", "DSM", "MDR",
    "CRA", "NIS2", "EU", "US", "PPE", "ICS", "SCADA", "OSHA", "OTM", "NIST", "NISTIR",
    "ENISA", "EDPB", "EDPS", "DSK", "BFDI", "OWASP", "API", "GPAI", "EUCC", "ECCG",
    "CISA", "CVSS", "CVD", "SRP", "PF", "SCHUFA", "SLSA", "BGH", "OGH", "BAG",
}
# Ganz-Wert-Overrides (Token-Logik kann diese nicht herleiten).
_REG_SHORT_OVERRIDES = {
    "Dataact": "Data Act",
    "Ecommerce": "E-Commerce",
}


def normalize_regulation_short(name: str) -> str:
    """Korrigiert title-caste Akronyme im druckbaren reg-short ('Dsgvo'->'DSGVO',
    'Osha Otm ...'->'OSHA OTM ...', 'EU Mdr'->'EU MDR') und erhaelt legitimes Mixed-Case
    (GeschGehG, MuSchG, MiCA, eIDAS, EuGH). Nur Tokens in _REG_SHORT_ACRONYMS werden
    gross geschrieben; alles andere bleibt unveraendert."""
    s = (name or "").strip()
    if not s:
        return s
    if s in _REG_SHORT_OVERRIDES:
        return _REG_SHORT_OVERRIDES[s]
    return " ".join(
        tok.upper() if tok.upper() in _REG_SHORT_ACRONYMS else tok
        for tok in s.split()
    )


def detect_citation_style(section: str) -> str:
    """'§ 38' -> 'paragraph' (DE-Gesetze); 'Artikel/Art. 13' -> 'article' (EU-VO)."""
    s = (section or "").lower()
    if "§" in s:
        return "paragraph"
    if "art" in s:
        return "article"
    return "paragraph"


def normalize_article(section: str) -> str:
    """Strikt aus echtem Legal-Header: '§ 38'->'38', 'Artikel 13'->'13', '[Artikel 17]'->'17',
    'Art. 13a'->'13a'. Kein Header (Datum/Seite/Titel) -> '' (keine falsche Fundstelle)."""
    m = _LEGAL_HEADER_RE.match(section or "")
    return (m.group(2) + (m.group(3) or "")) if m else ""


def normalize_paragraph(paragraph: str) -> str:
    """'(1)' -> '1', 'Abs. 2' -> '2', '' -> ''."""
    m = re.search(r"\d+", paragraph or "")
    return m.group(0) if m else ""


def extract_sub(text: str) -> str:
    """Best-effort feinste Granularitaet: 'lit. c' / 'Satz 2' / 'Nr. 3' (sonst '')."""
    m = _SUB_RE.search(text or "")
    return re.sub(r"\s+", " ", m.group(1)).strip() if m else ""


def is_recital(section: str, section_title: str) -> bool:
    blob = f"{section} {section_title}".lower()
    return any(k in blob for k in ("erwaegungsgrund", "erwägungsgrund", "recital"))


def format_article_label(
    regulation_label: str, citation_style: str, article: str, paragraph: str, sub: str
) -> str:
    """Druckbare Fundstelle: 'ProdHaftG § 1' bzw. 'Art. 13 Abs. 1 lit. c DSGVO'.

    `regulation_label` ist der Klarname (regulation_short, Originalschreibweise) — NICHT
    uppercasen, damit 'ProdHaftG'/'GeschGehG'/'MuSchG' korrekt erscheinen. Akronyme
    (BDSG/DSGVO/HGB) sind ohnehin schon gross."""
    code = (regulation_label or "").strip()
    if not article:
        return code
    abs_part = f" Abs. {paragraph}" if paragraph else ""
    sub_part = f" {sub}" if sub else ""
    if citation_style == "article":
        return f"Art. {article}{abs_part}{sub_part} {code}".strip()
    return f"{code} § {article}{abs_part}{sub_part}".strip()


def compute_chunk_hash(text: str) -> str:
    """sha256 des whitespace-normalisierten Chunk-Texts (Re-Link-/Ledger-Anker)."""
    norm = re.sub(r"\s+", " ", (text or "").strip())
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()


def deterministic_point_id(
    regulation_code: str, article: str, paragraph: str, chunk_index: int, document_version: str
) -> str:
    """Deterministische Qdrant-Point-ID (uuid5) fuer stabilen Re-Link + Alt/Neu-Koexistenz."""
    raw = f"{regulation_code}|{article}|{paragraph}|{chunk_index}|{document_version}"
    return str(uuid.uuid5(_ID_NS, raw))


def build_legal_fields(
    struct_meta: dict,
    regulation_code: str,
    chunk_text: str = "",
    citation_style: str = None,
    display_name: str = None,
    source_type: str = None,
) -> dict:
    """Roh-Chunker-Metadaten -> consumer-facing Spec-Felder (§2/§3).

    citation_style ist PRO REGULIERUNG (EU-VO/CH -> 'article', DE-§-Gesetz -> 'paragraph')
    und sollte explizit ueber die Ingest-Metadaten gesetzt werden; ohne Angabe wird er
    aus dem Section-String geraten (unzuverlaessig bei gemischten §/Art.-Referenzen).

    `display_name` (= regulation_short) ist der druckbare Klarname fuers Label; `regulation_code`
    bleibt der GROSS-Feldwert (Filter/Gruppierung). `source_type` steuert Urteils-Sonderfall."""
    sm = struct_meta or {}
    section = sm.get("section", "") or ""
    section_title = sm.get("section_title", "") or ""
    printable = (display_name or regulation_code or "").strip()
    if (source_type or "").strip().lower() in _RULING_TYPES:
        # Urteil: kein §/Art. (Querverweise verfaelschen). Label = Aktenzeichen.
        style = citation_style if citation_style in ("article", "paragraph") else ""
        article = ""
        label = printable
    else:
        style = (
            citation_style
            if citation_style in ("article", "paragraph")
            else detect_citation_style(section)
        )
        article = normalize_article(section)
        # Absatz/sub bewusst NICHT aus dem Chunk-Text raten: Querverweise ("Artikel 8 Absatz 1")
        # erzeugen FALSCHE Fundstellen. Zitat bleibt artikel-genau (zuverlaessig); feinere
        # Granularitaet ist ein spaeteres Refinement mit eigener, sicherer Extraktion.
        label = format_article_label(printable, style, article, "", "")
    return {
        "regulation_code": (regulation_code or "").upper(),
        "citation_style": style,
        "article": article,
        "paragraph": "",
        "sub": "",
        "is_recital": is_recital(section, section_title),
        "section_header": section_title,
        "article_label": label,
    }