"""Normalize messy inspector notes into bullet lines for RAG + generation.""" from __future__ import annotations import re from dataclasses import dataclass from app.chunking.splitter import count_tokens _ENCODING = None def _encoding(): global _ENCODING if _ENCODING is None: import tiktoken _ENCODING = tiktoken.get_encoding("cl100k_base") return _ENCODING def explode_note_lines(bullets: list[str]) -> list[str]: """Split dense note blobs into separate lines (semicolons, pipes, long sentences).""" out: list[str] = [] for raw in bullets: text = str(raw or "").strip() if not text: continue if len(text) <= 320 and text.count(";") + text.count("|") == 0: out.append(text) continue chunks = re.split(r"[;\n|]+", text) if len(chunks) == 1 and len(text) > 320: chunks = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9£])", text) for piece in chunks: p = piece.strip() if len(p) >= 2: out.append(p) return out def clean_and_clamp_bullets(bullets: list[str], *, max_items: int) -> list[str]: """Trim, de-dupe (exact), and cap bullet count.""" if not bullets: return [] out: list[str] = [] seen: set[str] = set() for raw in explode_note_lines(bullets): t = str(raw or "").strip() if not t: continue t = re.sub(r"^[-•*]+\s*", "", t).strip() if not t: continue key = t.casefold() if key in seen: continue seen.add(key) out.append(t) if len(out) >= max_items: break return out @dataclass(frozen=True, slots=True) class BulletClampReport: """Audit record describing what ``clean_and_clamp_bullets_with_report`` dropped. Used by ``_generate_section_text`` (standard path) and ``run_inspector_tool_loop`` (agentic path) so we can surface the warning in section metadata instead of silently losing content. Without this the user reports "data from messy notes didn't show up in the final report", but the cause is the silent clamp here — not the LLM. """ input_count: int cleaned_count: int duplicate_dropped: int overflow_dropped: int overflow_examples: tuple[str, ...] max_items: int @property def total_dropped(self) -> int: return self.duplicate_dropped + self.overflow_dropped @property def has_clamps(self) -> bool: return self.total_dropped > 0 def to_dict(self) -> dict: return { "input_count": self.input_count, "cleaned_count": self.cleaned_count, "duplicate_dropped": self.duplicate_dropped, "overflow_dropped": self.overflow_dropped, "overflow_examples": list(self.overflow_examples), "max_items": self.max_items, } def clean_and_clamp_bullets_with_report( bullets: list[str], *, max_items: int, overflow_example_limit: int = 6 ) -> tuple[list[str], BulletClampReport]: """Same behaviour as ``clean_and_clamp_bullets`` plus an audit report. We return the dropped overflow examples (first few that didn't fit) so the section metadata can show the user *which* observations were silently truncated. This is what powers the bullet-clamp warning surfaced on the report API response. """ out: list[str] = [] seen: set[str] = set() duplicate_dropped = 0 overflow_examples: list[str] = [] exploded = explode_note_lines(bullets or []) input_count = len(exploded) overflow_count = 0 for raw in exploded: t = str(raw or "").strip() if not t: continue t = re.sub(r"^[-•*]+\s*", "", t).strip() if not t: continue key = t.casefold() if key in seen: duplicate_dropped += 1 continue if len(out) >= max_items: overflow_count += 1 if len(overflow_examples) < overflow_example_limit: overflow_examples.append(t) continue seen.add(key) out.append(t) report = BulletClampReport( input_count=input_count, cleaned_count=len(out), duplicate_dropped=duplicate_dropped, overflow_dropped=overflow_count, overflow_examples=tuple(overflow_examples), max_items=max_items, ) return out, report def format_bullets_for_prompt(bullets: list[str], max_tokens: int) -> str: """Format bullets for the LLM user prompt within a token budget. Notes are pre-processed (typo normalisation + unverified-term flagging) at this boundary so the LLM never sees a raw transcription artefact, while the grounding pass continues to compare against the untouched bullet list. """ if not bullets: return "(none)" try: from app.config import settings from app.generator.term_glossary import preprocess_bullets if getattr(settings, "enable_unverified_term_flagging", True): bullets = preprocess_bullets(bullets) except Exception: # noqa: BLE001 — never block prompt assembly on preprocessing pass if max_tokens < 40: max_tokens = 40 parts: list[str] = [] used = 0 for b in bullets: line = f"- {b}" need = count_tokens(line) if parts and used + need > max_tokens: remaining = len(bullets) - len(parts) if remaining > 0: parts.append(f"- … ({remaining} further note line(s) still used for retrieval)") break if need > max_tokens: enc = _encoding().encode(line)[:max_tokens] parts.append(_encoding().decode(enc)) break parts.append(line) used += need return "\n".join(parts)