app/engine/safety/__init__.py

"""Semantic safety checks and fallback decisions."""

from __future__ import annotations

import re
from collections.abc import Iterable
from dataclasses import dataclass
from difflib import SequenceMatcher

from app.engine.quality import naturalness_reasons
from app.pipeline.candidate_validator import validate_candidate
from app.pipeline.meaning_safety import polarity_safe
from app.pipeline.nlp import get_nlp

_URL = re.compile(r"https?://[^\s<>\"']+|www\.[^\s<>\"']+", re.I)
_EMAIL = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
_NUMBER = re.compile(r"\b\d[\d,]*(?:\.\d+)?%?\b")


@dataclass
class SafetyResult:
    ok: bool
    confidence: float
    reasons: list[str]
    surface_sim: float = 0.0
    meaning: float = 0.0


def _entity_tokens(
    text: str,
    protected_entities: Iterable[str] | None = None,
) -> set[str]:
    """Return parsed named entities plus protected network identifiers."""
    toks = {entity for entity in (protected_entities or ()) if entity}
    if protected_entities is None:
        nlp = get_nlp()
        if nlp is not None:
            try:
                doc = nlp(text or "")
                toks.update(ent.text for ent in doc.ents)
                toks.update(token.text for token in doc if token.pos_ == "PROPN")
            except Exception:
                pass
    for m in _URL.finditer(text or ""):
        toks.add(m.group(0))
    for m in _EMAIL.finditer(text or ""):
        toks.add(m.group(0))
    return toks


def _numbers(text: str) -> set[str]:
    return {m.group(0).replace(",", "") for m in _NUMBER.finditer(text or "")}


def _tense_aux_ok(
    original: str,
    candidate: str,
    protected_auxiliaries: Iterable[str] | None = None,
) -> bool:
    """Reject if dependency-parsed auxiliary markers disappear."""
    auxiliaries = list(protected_auxiliaries or ())
    if protected_auxiliaries is None:
        nlp = get_nlp()
        if nlp is not None:
            try:
                original_lemmas = {
                    token.lemma_.lower()
                    for token in nlp(original or "")
                    if token.pos_ == "AUX"
                }
                candidate_lemmas = {
                    token.lemma_.lower()
                    for token in nlp(candidate or "")
                    if token.pos_ == "AUX"
                }
                return original_lemmas.issubset(candidate_lemmas)
            except Exception:
                auxiliaries = []
    if not auxiliaries:
        return True
    candidate_tokens = set(
        re.findall(r"[a-zA-Z']+", (candidate or "").lower())
    )
    return all(aux.lower() in candidate_tokens for aux in auxiliaries)


def check_safety(
    original: str,
    candidate: str,
    *,
    min_meaning: float = 0.80,
    min_confidence: float = 0.55,
    use_minilm: bool = False,
    protected_entities: Iterable[str] | None = None,
    protected_auxiliaries: Iterable[str] | None = None,
    structural_validation: bool = True,
    hard_invariants_only: bool = False,
) -> SafetyResult:
    """Lightweight similarity/safety gate between original and rewrite."""
    reasons: list[str] = []
    o = (original or "").strip()
    c = (candidate or "").strip()
    if not o or not c:
        return SafetyResult(False, 0.0, ["empty"])

    if not polarity_safe(o, c):
        reasons.append("negation")

    reasons.extend(naturalness_reasons(o, c))

    if hard_invariants_only:
        # Forced rule fallbacks may add cleft auxiliaries; keep only hard facts.
        o_ents = _entity_tokens(o, protected_entities)
        for ent in o_ents:
            if ent not in c and ent.lower() not in c.lower():
                reasons.append(f"entity:{ent}")
                break
        o_nums, c_nums = _numbers(o), _numbers(c)
        if o_nums and not o_nums.issubset(c_nums):
            reasons.append("numbers")
        surface_sim = SequenceMatcher(None, o.lower(), c.lower()).ratio()
        return SafetyResult(
            ok=not reasons,
            confidence=0.55 if not reasons else 0.2,
            reasons=reasons,
            surface_sim=surface_sim,
            meaning=surface_sim,
        )

    # Broken duration fronting / stranded preposition
    if re.search(
        r"^(at\s+least\s+)?[\w\s]*\b(minutes?|hours?|seconds?)\s*,",
        c,
        flags=re.I,
    ) and re.search(r"\bfor\b", o, flags=re.I):
        reasons.append("duration_front")
    if re.search(r"\b(for|over|within)\s+(every|each)\b", c, flags=re.I) and re.search(
        r"\b(for|over|within)\s+.+\b(minutes?|hours?)\b", o, flags=re.I
    ):
        reasons.append("stranded_prep")

    o_ents = _entity_tokens(o, protected_entities)
    for ent in o_ents:
        if ent not in c and ent.lower() not in c.lower():
            reasons.append(f"entity:{ent}")
            break

    o_nums, c_nums = _numbers(o), _numbers(c)
    if o_nums and not o_nums.issubset(c_nums):
        reasons.append("numbers")

    if not _tense_aux_ok(o, c, protected_auxiliaries):
        reasons.append("tense")

    if structural_validation:
        # Structural reorder may be near-copy in surface ratio; relax max_surface
        vr = validate_candidate(
            o,
            c,
            min_meaning=min_meaning if use_minilm else 0.0,
            max_surface=0.995,
            min_surface=0.20,
        )
        # Filter validator reasons that fight structural reorder
        ignore = {"too_similar", "identical"}
        for r in vr.reasons:
            if r in ignore:
                continue
            if r.startswith("meaning:") and not use_minilm:
                continue
            if r not in reasons:
                reasons.append(r)
        surface_sim = vr.surface_sim
        meaning = vr.meaning
    else:
        # Grammar repair may legitimately change inflection or spelling. Recheck
        # only hard invariants while still reporting a lightweight similarity.
        surface_sim = SequenceMatcher(None, o.lower(), c.lower()).ratio()
        meaning = surface_sim

    # Optional MiniLM meaning score when enabled
    if use_minilm:
        try:
            from app.pipeline.minilm import score_candidate

            scored = score_candidate(o, c)
            if scored is not None:
                meaning = float(scored)
                if meaning < min_meaning:
                    reasons.append(f"meaning:{meaning:.2f}")
        except Exception:
            pass

    confidence = max(
        0.0,
        min(1.0, (meaning + (1.0 - abs(surface_sim - 0.7))) / 2),
    )
    if reasons:
        confidence = min(confidence, 0.4)

    ok = not reasons and confidence >= min_confidence * 0.5
    # If only soft issues, still allow when polarity+entities ok
    hard = {
        r
        for r in reasons
        if r
        in {
            "negation",
            "numbers",
            "tense",
            "polarity",
            "entity_inject",
            "invention",
            "broken",
            "duration_front",
            "stranded_prep",
            "compound_split",
            "modal_pos_shift",
            "pos_balance",
            "meaning_drop",
        }
        or r.startswith("entity:")
        or r.startswith("meaning:")
    }
    if hard:
        ok = False
    elif reasons and surface_sim >= 0.35:
        # Soft validator noise on reorders — accept if content preserved
        ok = True
        confidence = max(confidence, 0.6)

    return SafetyResult(
        ok=ok,
        confidence=confidence,
        reasons=reasons,
        surface_sim=surface_sim,
        meaning=float(meaning or 0.0),
    )