Spaces:
Runtime error
Runtime error
File size: 4,004 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """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
|