app/services/survey_level_classifier.py

"""Corpus-informed RICS survey product tier inference (Levels 1 / 2 / 3).

Uses the same local exemplar folders as the knowledge base (``knowledge_base_dirs``):
typically ``Behrang RICS Documents`` and ``RAW Context``. Files are scanned to build
lightweight lexical + section-heading profiles per inferred gold tier, then unknown
uploads are scored against those profiles.

Deterministic (no LLM). Degrades gracefully when no corpus files exist.
"""

from __future__ import annotations

import json
import logging
import math
import re
import time
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from app.config import settings
from app.services.knowledge_base import iter_kb_files
from app.templates.registry import ALL_VALID_SECTION_CODES, section_order_for_survey

logger = logging.getLogger(__name__)

_TOKEN_RE = re.compile(r"[a-z0-9']{3,}", re.I)
# RICS section codes appear in two distinct contexts in real surveys:
#
#   1. Compound codes (E1, F2, K5, …) — only Level 2 / 3 surveys use these and
#      they are universally diagnostic on their own. Match them anywhere in the
#      text (case-insensitive).
#
#   2. Plain-letter codes (A, B, …, L) — appear in Level 1 reports as section
#      headings. Matching them with a bare ``\b[A-L]\b`` is a disaster because
#      the English article "a" and pronoun "I" both match, biasing every prose
#      paragraph toward Level 1 (the only level whose section_order is pure
#      single letters). We restrict plain-letter matches to *heading position*:
#      line-start, optionally bracketed, followed by heading punctuation
#      (".", ":", ")", em/en dash, hyphen). This catches "A. Introduction" /
#      "E — Outside the property" without firing on "a crack" or "I noted".
_COMPOUND_SECTION_RE = re.compile(r"\b([A-L])(\d{1,2})\b", re.I)
_HEADING_LETTER_RE = re.compile(
    r"(?m)^\s*\(?([A-L])\)?\s*(?:[.:\)\u2013\u2014\-])",
)
# Below this threshold the section-code signal is too sparse to choose a tier
# reliably — the compactness penalty starts producing arbitrary winners. Real
# RICS reports easily exceed this; messy field-notes typically don't, which is
# the correct signal to suppress section_scores entirely for those uploads.
_SECTION_CODE_MIN_EVIDENCE = 3


# ── Phrase tiers ───────────────────────────────────────────────────────────────
# Phrases are separated into three weight tiers so the scorer can strongly favour
# documents that carry a distinctive title / product name.
#
# ANCHOR (weight 25) — near-exclusive product identifiers that only appear in
#   one RICS survey tier.  A single anchor hit dominates corpus + section scoring
#   and raises minimum confidence to 0.55 (hard-block territory).
#
# STRONG (weight 8) — highly distinctive but not completely exclusive (e.g. the
#   phrase might appear in a competitor document but rarely in another RICS tier).
#
# NORMAL (weight 2) — useful supporting evidence; not decisive on its own.
#
# Rule of thumb: if you're unsure, make it STRONG not ANCHOR.

_LEVEL_ANCHOR_PHRASES: dict[int, tuple[str, ...]] = {
    1: (
        "condition report",
        "rics condition report",
        "home survey – level 1",
        "home survey - level 1",
        "home survey level 1",
        "rics home survey level 1",
    ),
    2: (
        "homebuyer report",
        "home buyer report",
        "rics homebuyer",
        "rics home buyer",
        "homebuyer survey",
        "home buyer survey",
        "home survey – level 2",
        "home survey - level 2",
        "home survey level 2",
        "rics home survey level 2",
        "rics homebuyer report",
    ),
    3: (
        "building survey",
        "full building survey",
        "full structural survey",
        "rics building survey",
        "home survey – level 3",
        "home survey - level 3",
        "home survey level 3",
        "rics home survey level 3",
    ),
}

_LEVEL_STRONG_PHRASES: dict[int, tuple[str, ...]] = {
    1: (
        "level 1 condition",
        "traffic light rating",  # "traffic light" alone is too generic; keep the full phrase
        "condition rating 1",    # Level 1 uses condition ratings 1/2/3 — only "rating 1" alone
        "energy matters",
        # NOTE: "condition 3" removed — RICS condition ratings (1/2/3) appear in ALL survey levels.
        #   A Level 3 surveyor writing "condition 3 — severe cracking" would falsely score L1.
        # NOTE: "description of the property" removed — generic heading used at all levels.
        # NOTE: "traffic light" removed — phrase on its own appears in non-L1 contexts too.
    ),
    2: (
        "level 2 survey",
        "valuation not provided",
        "repair and maintenance",
        "purchase negotiation",
    ),
    3: (
        "level 3 survey",
        "fabric and structure",
        "defects analysis",
        "technical risks",
        "costed repairs",
        "further investigations",
        "schedule of condition",
        "specialist investigation",
        "structural movement",
        "further investigation required",
        "invasive investigation",
    ),
}

# Keep _LEVEL_PHRASES for backward-compat references in tests.
_LEVEL_PHRASES: dict[int, tuple[str, ...]] = {
    lvl: _LEVEL_ANCHOR_PHRASES[lvl] + _LEVEL_STRONG_PHRASES[lvl]
    for lvl in (1, 2, 3)
}

_PATH_HINTS: list[tuple[re.Pattern[str], int]] = [
    (re.compile(r"level[\s_-]*1|l1\b|condition[\s_-]*report", re.I), 1),
    (re.compile(r"level[\s_-]*2|l2\b|homebuyer|home[\s_-]*buyer", re.I), 2),
    (re.compile(r"level[\s_-]*3|l3\b|building[\s_-]*survey", re.I), 3),
]


def _cache_path() -> Path:
    return settings.cache_dir / "survey_level_corpus_profiles.json"


def _tokenize(text: str) -> list[str]:
    return [m.group(0).lower() for m in _TOKEN_RE.finditer(text.lower())]


def _extract_section_codes(text: str) -> set[str]:
    """Detect RICS section codes (E1, F2, …) in ``text`` without false positives.

    See the regex docstring above for the rationale. Returns the set of valid
    section codes only — codes outside ``ALL_VALID_SECTION_CODES`` (e.g. random
    letter+digit pairs like ``"A1"`` from "Annex 1") are filtered out.
    """
    found: set[str] = set()
    for m in _COMPOUND_SECTION_RE.finditer(text):
        code = m.group(1).upper() + m.group(2)
        if code in ALL_VALID_SECTION_CODES:
            found.add(code)
    for m in _HEADING_LETTER_RE.finditer(text):
        letter = m.group(1).upper()
        if letter in ALL_VALID_SECTION_CODES:
            found.add(letter)
    return found


def _infer_corpus_label(path: Path, text_lower: str) -> int | None:
    """Infer gold tier for an exemplar file from path + opening text."""
    joined = f"{path.as_posix().lower()} {text_lower[:12000]}"
    for pat, lvl in _PATH_HINTS:
        if pat.search(joined):
            return lvl
    # Text-only fallbacks (first chunk of document)
    best: int | None = None
    best_hits = 0
    for lvl, phrases in _LEVEL_PHRASES.items():
        hits = sum(1 for p in phrases if p in text_lower[:20000])
        if hits > best_hits:
            best_hits = hits
            best = lvl
    if best is not None and best_hits >= 2:
        return best
    return None


def _read_sample_text(fp: Path, max_chars: int = 120_000) -> str:
    """Load plain text from .docx / .pdf using the same loaders as ingestion."""
    try:
        from app.ingest.pipeline import load_raw_documents

        docs = load_raw_documents(fp)
        parts = [d.page_content for d in docs if getattr(d, "page_content", None)]
        raw = "\n".join(parts).strip()
        if len(raw) > max_chars:
            return raw[:max_chars]
        return raw
    except Exception as exc:  # noqa: BLE001
        logger.debug("survey_level corpus: skip %s: %s", fp, exc)
        return ""


@dataclass(frozen=True, slots=True)
class CorpusProfiles:
    created_at_unix: int
    files_used: int
    # level -> { "df": {token: count}, "n_docs": int }
    levels: dict[int, dict[str, Any]]


def _read_cache() -> CorpusProfiles | None:
    p = _cache_path()
    if not p.is_file():
        return None
    try:
        raw = json.loads(p.read_text(encoding="utf-8"))
        created = int(raw.get("created_at_unix") or 0)
        ttl = int(settings.survey_level_corpus_cache_seconds)
        if created and (int(time.time()) - created) > ttl:
            return None
        levels: dict[int, dict[str, Any]] = {}
        for k, v in (raw.get("levels") or {}).items():
            lvl = int(k)
            levels[lvl] = {
                "df": {str(t): int(c) for t, c in (v.get("df") or {}).items()},
                "n_docs": int(v.get("n_docs") or 0),
            }
        return CorpusProfiles(
            created_at_unix=created,
            files_used=int(raw.get("files_used") or 0),
            levels=levels,
        )
    except Exception:  # noqa: BLE001
        return None


def _write_cache(profiles: CorpusProfiles) -> None:
    try:
        settings.cache_dir.mkdir(parents=True, exist_ok=True)
        payload = {
            "created_at_unix": profiles.created_at_unix,
            "files_used": profiles.files_used,
            "levels": {str(k): {"df": v["df"], "n_docs": v["n_docs"]} for k, v in profiles.levels.items()},
        }
        _cache_path().write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
    except Exception:  # noqa: BLE001
        logger.debug("Could not write survey level corpus cache", exc_info=True)


def build_or_load_corpus_profiles(*, force_refresh: bool = False) -> CorpusProfiles:
    """Scan KB dirs, infer labelled exemplars, aggregate DF per level; cache on disk."""
    if not force_refresh:
        cached = _read_cache()
        if cached is not None:
            return cached

    files = iter_kb_files()[:400]
    per_level_docs: dict[int, list[str]] = {1: [], 2: [], 3: []}
    used = 0
    for fp in files:
        if fp.suffix.lower() not in (".pdf", ".docx"):
            continue
        text = _read_sample_text(fp, max_chars=80_000)
        if len(text) < 400:
            continue
        low = text.lower()
        label = _infer_corpus_label(fp, low)
        if label is None:
            continue
        per_level_docs[label].append(low)
        used += 1

    levels: dict[int, dict[str, Any]] = {}
    for lvl in (1, 2, 3):
        docs = per_level_docs.get(lvl) or []
        df: Counter[str] = Counter()
        for low in docs:
            df.update(_tokenize(low))
        # Trim to top terms per level for smaller cache
        top = dict(df.most_common(4000))
        levels[lvl] = {"df": top, "n_docs": len(docs)}

    prof = CorpusProfiles(created_at_unix=int(time.time()), files_used=used, levels=levels)
    _write_cache(prof)
    logger.info("survey_level corpus: built profiles from %d labelled exemplar files", used)
    return prof


@dataclass(frozen=True, slots=True)
class SurveyLevelCandidate:
    survey_level: int
    score: float


@dataclass(frozen=True, slots=True)
class SurveyLevelClassification:
    document_id: str | None
    filename: str
    predicted_survey_level: int
    confidence: float
    candidates: list[SurveyLevelCandidate]
    rationale: str


def _score_against_corpus(text_lower: str, profiles: CorpusProfiles) -> dict[int, float]:
    toks = _tokenize(text_lower)
    if not toks:
        return {1: 0.0, 2: 0.0, 3: 0.0}
    doc_tf: Counter[str] = Counter(toks)
    doc_norm = math.sqrt(sum(c * c for c in doc_tf.values())) or 1.0
    scores: dict[int, float] = {}
    for lvl in (1, 2, 3):
        df = profiles.levels.get(lvl, {}).get("df") or {}
        if not df:
            scores[lvl] = 0.0
            continue
        dot = 0.0
        for w, c in doc_tf.items():
            if w in df:
                dot += c * math.log(1 + float(df[w]))
        scores[lvl] = dot / doc_norm
    return scores


def _phrase_scores(text_lower: str) -> tuple[dict[int, float], int | None]:
    """Return (score_per_level, anchor_level_or_None).

    anchor_level is set to the first level for which ANY anchor phrase was
    found in the document.  A single anchor phrase is so distinctive that it
    should dominate the prediction and push confidence into hard-block territory
    — the caller must honour this.
    """
    out = {1: 0.0, 2: 0.0, 3: 0.0}
    anchor_level: int | None = None

    for lvl in (1, 2, 3):
        for p in _LEVEL_ANCHOR_PHRASES.get(lvl, ()):
            if p in text_lower:
                out[lvl] += 25.0
                if anchor_level is None:
                    anchor_level = lvl
        for p in _LEVEL_STRONG_PHRASES.get(lvl, ()):
            if p in text_lower:
                out[lvl] += 8.0

    return out, anchor_level


def _section_scores(codes: set[str]) -> dict[int, float]:
    """Score the document's section-code set against each survey level's expected codes.

    Problem with raw Jaccard: since L3 ⊇ L2 in terms of section codes, every code
    a L2 document contains is also valid for L3. Jaccard penalises L3 (larger union)
    but not enough to make the gap decisive.

    We use a "compact-coverage" formula that rewards how well the level's code set
    *explains* the document's codes (recall) AND penalises the level for having far
    more codes than the document observed (over-specification penalty):

        recall   = inter / |doc_codes|   (fraction of doc codes the level covers)
        compact  = inter / |level_codes| (fraction of level codes the doc uses)
        score    = 15.0 * (recall * compact) ** 0.5   (geometric mean)

    For a perfect L2 doc with 20 codes:
        vs L2 (27 codes): recall=1.0, compact=20/27=0.74  → score=15*√0.74≈12.9
        vs L3 (45 codes): recall=1.0, compact=20/45=0.44  → score=15*√0.44≈9.9

    The L2 advantage grows as the doc uses codes that are NOT exclusive to L3, which
    is exactly the case for HomeBuyer reports.
    """
    out = {1: 0.0, 2: 0.0, 3: 0.0}
    # Sparse evidence is worse than no evidence: with 1–2 codes the compactness
    # penalty arbitrarily favours whichever level happens to have a smaller
    # section_order set, which is L1 by construction. Suppress the signal until
    # we see enough codes for the geometric mean to mean something.
    if len(codes) < _SECTION_CODE_MIN_EVIDENCE:
        return out
    n_doc = len(codes)
    for lvl in (1, 2, 3):
        order = set(section_order_for_survey(lvl))
        inter = len(codes & order)
        if inter == 0:
            out[lvl] = 0.0
            continue
        recall = inter / n_doc
        compact = inter / (len(order) or 1)
        out[lvl] = 15.0 * math.sqrt(recall * compact)
    return out


def classify_text(
    text: str,
    *,
    filename: str = "",
    document_id: str | None = None,
    profiles: CorpusProfiles | None = None,
) -> SurveyLevelClassification:
    """Return predicted RICS tier 1–3 plus confidence and rationale."""
    prof = profiles or build_or_load_corpus_profiles()
    low = text.lower()
    codes = _extract_section_codes(text)

    corpus_part = _score_against_corpus(low, prof)
    phrase_part, anchor_level = _phrase_scores(low)
    sec_part = _section_scores(codes)

    total: dict[int, float] = {}
    for lvl in (1, 2, 3):
        n_docs = int(prof.levels.get(lvl, {}).get("n_docs") or 0)
        corpus_weight = 0.25 if n_docs == 0 else 1.0
        total[lvl] = corpus_weight * corpus_part[lvl] + phrase_part[lvl] + sec_part[lvl]

    ranked = sorted(total.items(), key=lambda x: x[1], reverse=True)
    best_lvl, best_s = ranked[0]
    second_s = ranked[1][1] if len(ranked) > 1 else 0.0

    # If best score is zero every level tied at 0 — no usable signal at all.
    # Return unknown (level 0) immediately rather than silently picking Level 1
    # due to dict-insertion-order tie-breaking, which produced systematic false
    # positives on messy field-notes that lacked formal RICS language.
    if best_s == 0.0:
        return SurveyLevelClassification(
            document_id=document_id,
            filename=filename,
            predicted_survey_level=0,
            confidence=0.0,
            candidates=[SurveyLevelCandidate(survey_level=lvl, score=0.0) for lvl in (3, 2, 1)],
            rationale=(
                "No recognisable RICS-level phrases or section codes were found. "
                "Tier unknown — the upload will be treated as matching any level."
            ),
        )

    denom = best_s + 1e-6
    raw_conf = max(0.0, min(1.0, (best_s - second_s) / denom)) if best_s > 0 else 0.2
    confidence = raw_conf

    # Anchor boost: a single hit on an anchor phrase (e.g. "homebuyer report",
    # "building survey") is so distinctive that it should dominate corpus noise
    # and section-code ambiguity. If the anchor agrees with best_lvl, lift
    # confidence into hard-block territory. If the anchor *disagrees* with the
    # scoring winner (rare), override the prediction to the anchor level because
    # the product-name title is the most reliable single signal we have.
    anchor_rationale = ""
    if anchor_level is not None:
        if anchor_level == best_lvl:
            confidence = max(confidence, 0.55)
            anchor_rationale = (
                f"Anchor phrase for Level {anchor_level} found in document "
                f"(e.g. 'homebuyer report' or 'building survey' — product-name title match). "
                f"Confidence lifted to {confidence:.0%}."
            )
        else:
            # Anchor disagrees with corpus winner → anchor wins, but cap confidence
            # at 0.50 (we don't want to be overconfident when corpus and phrase diverge).
            best_lvl = anchor_level
            confidence = 0.50
            # Re-sort so ranked[0] reflects the override for rationale display.
            ranked = sorted(total.items(), key=lambda x: (x[0] != anchor_level, -x[1]))
            anchor_rationale = (
                f"Anchor phrase for Level {anchor_level} found — overrides corpus scoring winner. "
                f"Product-name title is the most reliable single signal."
            )
        best_s = total.get(best_lvl, 0.0)

    any_corpus = any(int(prof.levels.get(L, {}).get("n_docs") or 0) > 0 for L in (1, 2, 3))

    if not any_corpus:
        # Without a labelled corpus the phrase + section signals still work; just cap
        # confidence when there were NO anchor hits (anchors are self-sufficient).
        if anchor_level is None and best_s < 1.0:
            confidence = min(confidence, 0.25)
        rationale_parts = [
            "No labelled exemplar corpus was detected under knowledge_base_dirs.",
            f"Tier prediction is based on document wording and section-code patterns (favours Level {best_lvl}).",
            "Run POST /documents/survey-level/corpus-refresh with exemplar PDFs to improve accuracy.",
        ]
    else:
        rationale_parts = [
            f"Lexical cues favour Level {best_lvl}.",
            f"Section-heading overlap with Level {best_lvl} pack scored highest among tiers.",
        ]

    if anchor_rationale:
        rationale_parts.append(anchor_rationale)
    if codes:
        rationale_parts.append(f"Detected {len(codes)} RICS-style section code(s) in the sample.")

    rationale = " ".join(rationale_parts)
    candidates = [SurveyLevelCandidate(survey_level=lvl, score=round(float(sc), 4)) for lvl, sc in ranked]

    return SurveyLevelClassification(
        document_id=document_id,
        filename=filename,
        predicted_survey_level=int(best_lvl),
        confidence=round(float(confidence), 3),
        candidates=candidates,
        rationale=rationale,
    )


def classify_file_path(
    path: Path,
    *,
    filename: str,
    document_id: str | None = None,
    profiles: CorpusProfiles | None = None,
) -> SurveyLevelClassification:
    text = _read_sample_text(path, max_chars=120_000)
    if len(text) < 200:
        # Try path-hint inference before giving up — a filename like "level1_report.pdf"
        # is still a reliable signal even when the file body is unreadable.
        path_hint: int | None = None
        joined_hint = f"{path.as_posix().lower()} {text.lower()}"
        for pat, lvl in _PATH_HINTS:
            if pat.search(joined_hint):
                path_hint = lvl
                break
        predicted_from_hint = path_hint if path_hint is not None else 0
        # If no path hint, do NOT pin to Level 3 — return a truly unknown signal so the
        # caller (guardrail) skips tier enforcement rather than silently passing any level.
        if predicted_from_hint:
            return SurveyLevelClassification(
                document_id=document_id,
                filename=filename,
                predicted_survey_level=predicted_from_hint,
                confidence=0.30,
                candidates=[
                    SurveyLevelCandidate(survey_level=predicted_from_hint, score=1.0),
                    SurveyLevelCandidate(survey_level=2 if predicted_from_hint != 2 else 1, score=0.0),
                    SurveyLevelCandidate(survey_level=3 if predicted_from_hint != 3 else 1, score=0.0),
                ],
                rationale=(
                    f"File body too short to classify; filename/path suggests Level {predicted_from_hint}."
                ),
            )
        return SurveyLevelClassification(
            document_id=document_id,
            filename=filename,
            predicted_survey_level=0,
            confidence=0.0,
            candidates=[
                SurveyLevelCandidate(survey_level=3, score=0.0),
                SurveyLevelCandidate(survey_level=2, score=0.0),
                SurveyLevelCandidate(survey_level=1, score=0.0),
            ],
            rationale="Could not read enough text from the file; tier unknown — set manually.",
        )
    return classify_text(text, filename=filename, document_id=document_id, profiles=profiles)