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"}


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,
    }