backend/core/notes_routing.py

"""Multi-stage RICS L3 note routing: explicit prefix → element label → keywords → semantic fallback.

Priority 1 — explicit section tokens at line start (``G1,``, ``D9:``, ``D4 -``).
Priority 1b — surveyor element labels (``Main roof:``, ``Rainwater fittings:``).
Priority 2 — high-confidence domain keyword tiers (full canonical lexicon).
Priority 3 — general keyword patterns, then embedding/RAG when enabled.
"""

from __future__ import annotations

import re
from collections import defaultdict

from backend.core.notes_element_labels import ELEMENT_LABEL_ROUTES
from backend.core.notes_high_confidence_lexicon import compile_high_confidence_patterns
from backend.core.rics_canonical_l3 import valid_leaf_section_ids

UNASSIGNED = "UNASSIGNED"

# Priority 1: canonical leaf id at line start, optional separator, remainder is body.
_EXPLICIT_SECTION_PREFIX = re.compile(
    r"^\s*(?:section\s+)?([A-N]\d{1,2})\s*(?:[:,;\-–—\.]\s*|\s+)(.+)$",
    re.IGNORECASE | re.DOTALL,
)

_ELEMENT_LABEL_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
    (
        section_id,
        re.compile(rf"^\s*({label})\s*:\s*(.+)$", re.IGNORECASE | re.DOTALL),
    )
    for section_id, label in ELEMENT_LABEL_ROUTES
]

# Priority 2: full canonical lexicon (order-sensitive; see notes_high_confidence_lexicon.py).
_HIGH_CONFIDENCE_PATTERNS = compile_high_confidence_patterns()

RAG_CLASSIFICATION_GUIDANCE = (
    "You are a strict RICS Level 3 surveyor assistant. Allocate the following note "
    "to exactly one of the 54 canonical sub-sections. If the text describes a building "
    "element (like gutters), do not assign it to the surrounding element (like main walls); "
    "assign it to its exact dedicated section (D3)."
)


def parse_explicit_section_prefix(line: str) -> tuple[str, str] | None:
    """Return ``(section_id, body)`` when the line starts with a canonical leaf token."""
    text = (line or "").strip()
    if len(text) < 3:
        return None
    match = _EXPLICIT_SECTION_PREFIX.match(text)
    if not match:
        return None
    section_id = match.group(1).upper()
    body = match.group(2).strip()
    allowed = valid_leaf_section_ids()
    if section_id in allowed and body:
        return section_id, body
    return None


def parse_element_label_prefix(line: str) -> tuple[str, str] | None:
    """Return ``(section_id, full_line)`` when the line opens with a known element label."""
    text = (line or "").strip()
    if len(text) < 3 or ":" not in text:
        return None
    allowed = valid_leaf_section_ids()
    for section_id, pattern in _ELEMENT_LABEL_PATTERNS:
        match = pattern.match(text)
        if not match:
            continue
        body = match.group(2).strip()
        if body and section_id.upper() in allowed:
            return section_id.upper(), text
    return None


def classify_high_confidence_keywords(text: str) -> tuple[str, float]:
    """Priority-2 keyword tier; returns ``UNASSIGNED`` when no rule fires."""
    line = (text or "").strip()
    if len(line) < 3:
        return UNASSIGNED, 0.0
    valid = valid_leaf_section_ids()
    for section_id, pattern in _HIGH_CONFIDENCE_PATTERNS:
        if pattern.search(line):
            sid = section_id.upper()
            if sid in valid:
                return sid, 1.0
    return UNASSIGNED, 0.0


def route_lines_to_l3_sections(lines: list[str]) -> tuple[dict[str, list[str]], list[str]]:
    """Route note lines into L3 section buckets using the full cascade."""
    from backend.core.notes_keyword_router import classify_note_cascade

    routed: dict[str, list[str]] = defaultdict(list)
    unmatched: list[str] = []
    for raw in lines:
        text = (raw or "").strip()
        if len(text) < 3:
            continue
        section_id, _, observation = classify_note_cascade(text)
        if section_id == UNASSIGNED:
            unmatched.append(observation)
        else:
            routed[section_id].append(observation)
    return dict(routed), unmatched