"""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), )