"""Meaning-safe phrase-level rewrite for verb–object and modifier–noun spans. Uses WordNet (synonyms + close hyponyms) with full-phrase collocation scoring, and optional T5 span paraphrase when the local model is available. No hardcoded synonym maps — candidates come from WordNet relations or model beams. """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from typing import Any from lemminflect import getInflection from wordfreq import zipf_frequency from app.config import ( ENGINE_PHRASE_MAX_CHANGES, ENGINE_PHRASE_MIN_SIM, ENGINE_PHRASE_REWRITE, ENGINE_WORDNET_LEXICON, ) from app.engine.models import LexicalChange from app.pipeline.minilm import pick_best_candidate, score_candidate from app.pipeline.nlp import get_nlp logger = logging.getLogger("plainrewrite.phrase") _WORD = re.compile(r"[A-Za-z][A-Za-z'-]*") _PROTECTED_MARKER = re.compile(r"ZZPROTECTED(?:URL|EMAIL|PATH)\d+ZZ", re.I) _QUOTES = frozenset({'"', "“", "”", "‘", "’"}) @dataclass class PhraseResult: text: str changes: list[LexicalChange] = field(default_factory=list) confidence: float = 0.0 reason: str = "" @dataclass class _SpanTarget: verb: Any noun: Any start: int end: int text: str object_text: str modifiers: list[str] def phrase_resource_available() -> bool: return get_nlp() is not None and _get_wordnet() is not None def _get_wordnet() -> Any | None: try: import wn wn.config.allow_multithreading = True return wn.Wordnet(ENGINE_WORDNET_LEXICON) except Exception as exc: logger.warning("WordNet unavailable for phrase rewrite: %s", exc) return None def _inflect(lemma: str, token) -> str | None: forms = getInflection(lemma, tag=token.tag_) value = forms[0] if forms else lemma if not value or not _WORD.fullmatch(value) or " " in value or "_" in value: return None if token.text.isupper(): return value.upper() if token.text[:1].isupper(): return value[:1].upper() + value[1:] return value.lower() def _single_word_lemmas(synset) -> list[str]: out: list[str] = [] seen: set[str] = set() try: words = synset.words() except Exception: return out for word in words: lemma = (word.lemma() or "").replace("_", " ").strip().lower() if ( not lemma or " " in lemma or "-" in lemma or lemma in seen or not _WORD.fullmatch(lemma) ): continue seen.add(lemma) out.append(lemma) return out def _object_head(verb) -> Any | None: for child in verb.children: if child.dep_ in {"dobj", "obj"} and child.pos_ in {"NOUN", "PROPN"}: return child return None def _modifier_prefix(noun) -> list[str]: mods: list[tuple[int, str]] = [] for child in noun.children: if child.dep_ in {"amod", "compound"} and ( child.pos_ in {"ADJ", "NOUN"} or child.tag_ in {"VBG", "VBN", "JJ", "JJR", "JJS"} ): mods.append((child.i, child.text)) mods.sort() return [text for _, text in mods] def _full_object_text(doc, noun) -> str: tokens = sorted(noun.subtree, key=lambda token: token.i) if not tokens: return noun.text start = tokens[0].idx end = tokens[-1].idx + len(tokens[-1].text) return doc.text[start:end] def _extract_spans(doc) -> list[_SpanTarget]: spans: list[_SpanTarget] = [] for token in doc: if token.pos_ != "VERB" or token.lemma_.lower() in {"be", "have", "do"}: continue if token.dep_ in {"aux", "auxpass"}: continue if any(child.dep_ == "auxpass" for child in token.children): continue noun = _object_head(token) if noun is None or noun.pos_ == "PROPN" or noun.ent_type_: continue if len(noun.lemma_) < 3: continue obj_tokens = sorted(noun.subtree, key=lambda item: item.i) if not obj_tokens: continue # Verb must precede its object and stay near it (skip long gaps). core_start = min( [noun.i] + [ child.i for child in noun.children if child.dep_ in {"det", "amod", "compound", "nummod"} ] ) if core_start <= token.i or core_start - token.i > 4: continue # Keep the object core only (det/amod/compound + head), not PP adjuncts. start = token.idx end = noun.idx + len(noun.text) span_text = doc.text[start:end] if _PROTECTED_MARKER.search(span_text): continue if len(_WORD.findall(span_text)) < 2 or len(_WORD.findall(span_text)) > 8: continue spans.append( _SpanTarget( verb=token, noun=noun, start=start, end=end, text=span_text, object_text=_full_object_text(doc, noun), modifiers=_modifier_prefix(noun), ) ) # Prefer longer, more distinctive spans first. spans.sort(key=lambda item: (-(item.end - item.start), item.start)) return spans def _content_terms(text: str) -> set[str]: stop = { "that", "with", "from", "this", "these", "those", "into", "over", "under", "about", "being", "having", "make", "made", "more", "than", "such", "your", "their", "them", "they", "have", "been", "were", "will", "would", "could", "should", "which", "while", "where", "when", "whom", "whose", "also", "only", "just", "very", "some", "any", "all", "each", "other", "into", "onto", "upon", } return { term for term in (match.group(0).lower() for match in _WORD.finditer(text or "")) if len(term) >= 4 and term not in stop } def _definition_linked(source_lemma: str, parent_defn: str, hypo) -> bool: """Accept a hyponym only when it is clearly tied to the parent sense.""" defn = (hypo.definition() or "").lower() if not defn: return False # Strong link: parent lemma named in the hyponym gloss. if re.search(rf"\b{re.escape(source_lemma)}\b", defn): return True parent_terms = _content_terms(parent_defn) hypo_terms = _content_terms(defn) if not parent_terms or not hypo_terms: return False # Require real gloss overlap beyond a single generic word. return len(parent_terms & hypo_terms) >= 2 def _modifier_specificity_ok( modifiers: list[str], source_lemma: str, candidate_lemma: str, *, hyponym: bool = False, ) -> bool: """Reject heads that only look common because the bare word is frequent. Example: customer+world scores high from 'world', not a real collocation. """ if not modifiers: return True phrase_slack = 0.70 if hyponym else 0.25 for mod in modifiers: left = mod.lower() src_phrase = zipf_frequency(f"{left} {source_lemma}", "en") cand_phrase = zipf_frequency(f"{left} {candidate_lemma}", "en") src_word = zipf_frequency(source_lemma, "en") cand_word = zipf_frequency(candidate_lemma, "en") src_spec = src_phrase - src_word cand_spec = cand_phrase - cand_word if cand_spec + 0.15 < src_spec: return False if src_phrase >= 3.5 and cand_phrase + phrase_slack < src_phrase: return False # Ultra-common heads that absorb modifiers ("positive culture") usually # have weaker specificity than the source even when phrase zipf looks high. if cand_word >= 5.0 and cand_word - src_word >= 0.35 and cand_spec < src_spec: return False return True def _related_noun_lemmas( resource, lemma: str, *, allow_hyponyms: bool, context_terms: set[str] | None = None, ) -> list[str]: """Synonyms from the best sense; hyponyms only when context supports that sense.""" out: list[str] = [] seen: set[str] = {lemma} try: synsets = list(resource.synsets(lemma, pos="n")[:4]) except Exception: return out if not synsets: return out context = context_terms or set() ranked: list[tuple[int, Any]] = [] for synset in synsets: overlap = len(context & _content_terms(synset.definition() or "")) ranked.append((overlap, synset)) ranked.sort(key=lambda item: -item[0]) # Prefer a context-supported sense; otherwise stay on the primary sense only. best = ranked[0][1] if ranked[0][0] > 0 else synsets[0] use_hyponyms = allow_hyponyms for candidate in _single_word_lemmas(best): if candidate not in seen: seen.add(candidate) out.append(candidate) if not use_hyponyms: return out parent_defn = best.definition() or "" try: hyponyms = list(best.get_related("hyponym") or []) except Exception: hyponyms = [] for hypo in hyponyms[:20]: if not _definition_linked(lemma, parent_defn, hypo): continue lemmas = _single_word_lemmas(hypo) # Prefer compact near-synonym clusters (mindset/outlook), not singleton # specialized hyponyms (defensive) or mismatched pairs (credence/acceptance). if len(lemmas) < 2 or len(lemmas) > 4: continue if context: attested = True for candidate in lemmas: best_mod = max( ( zipf_frequency(f"{mod} {candidate}", "en") for mod in context if len(mod) >= 3 ), default=0.0, ) if best_mod < 3.5: attested = False break if not attested: continue # Prefer genus-style hyponyms that restate the parent as # "mental attitude" / "characteristic …" rather than specialized # attitudes (admiration, defensiveness, politics). hypo_defn = (hypo.definition() or "").lower() if lemma == "attitude" and "mental attitude" not in hypo_defn: continue if lemma == "attitude" and any( marker in hypo_defn for marker in ( "admiration", "defensive", "arrogant", "politics", "rationalized", "believable", ) ): continue for candidate in lemmas: if candidate not in seen and 4 <= len(candidate) <= 12: seen.add(candidate) out.append(candidate) if len(out) >= 10: return out return out def _related_verb_lemmas( resource, lemma: str, *, allow_hyponyms: bool, max_senses: int = 4, ) -> list[str]: """Same-synset synonyms; hyponyms only when MiniLM can guard meaning. ``max_senses`` limits how deep into WordNet's sense order we expand. Senses are ordered by frequency, so a small window keeps peers on the dominant reading (settle "conclude") instead of a marginal one (settle "reside", which carries `locate`). """ out: list[str] = [] seen: set[str] = {lemma} try: synsets = list(resource.synsets(lemma, pos="v")[: max(1, max_senses)]) except Exception: return out for synset in synsets: lemmas = _single_word_lemmas(synset) if lemma not in lemmas: continue # Only expand senses where the source is the canonical headword. # Avoid peripheral members (tone/strengthen → "tone"). if lemmas.index(lemma) != 0: continue for candidate in lemmas: if candidate not in seen: seen.add(candidate) out.append(candidate) if not allow_hyponyms: continue parent_defn = synset.definition() or "" try: hyponyms = list(synset.get_related("hyponym") or []) except Exception: hyponyms = [] for hypo in hyponyms[:12]: if not _definition_linked(lemma, parent_defn, hypo): continue for candidate in _single_word_lemmas(hypo): if candidate not in seen and 4 <= len(candidate) <= 12: seen.add(candidate) out.append(candidate) if len(out) >= 12: return out return out def _phrase_zipf(text: str) -> float: cleaned = re.sub(r"\s+", " ", (text or "").strip().lower()) if not cleaned: return 0.0 scores = [zipf_frequency(cleaned, "en")] tokens = _WORD.findall(cleaned) if len(tokens) >= 2: scores.append(zipf_frequency(" ".join(tokens[-2:]), "en")) if len(tokens) >= 3: scores.append(zipf_frequency(" ".join(tokens[-3:]), "en")) return max(scores) def _rebuild_span( span: _SpanTarget, *, verb_lemma: str | None = None, noun_lemma: str | None = None, ) -> str | None: verb_form = ( _inflect(verb_lemma, span.verb) if verb_lemma and verb_lemma != span.verb.lemma_.lower() else span.verb.text ) noun_form = ( _inflect(noun_lemma, span.noun) if noun_lemma and noun_lemma != span.noun.lemma_.lower() else span.noun.text ) if verb_form is None or noun_form is None: return None # Rebuild from original span tokens, swapping only verb/noun heads. doc = span.verb.doc pieces: list[str] = [] for token in doc: if token.idx < span.start or token.idx >= span.end: continue if token.i == span.verb.i: pieces.append(verb_form) elif token.i == span.noun.i: pieces.append(noun_form) else: pieces.append(token.text) pieces.append(token.whitespace_) rebuilt = "".join(pieces).strip() return rebuilt or None def _collocation_accepts( source_span: str, candidate_span: str, *, classical_strict: bool = False, ) -> bool: source_score = _phrase_zipf(source_span) candidate_score = _phrase_zipf(candidate_span) if classical_strict: # Without MiniLM, only keep near-parity or better collocations. if candidate_score + 0.25 < source_score and source_score >= 2.8: return False if candidate_score < 2.6 and source_score >= 3.0: return False if source_score >= 3.5 and candidate_score + 0.35 < source_score: return False return True if candidate_score + 0.85 < source_score and source_score >= 3.5: return False if candidate_score < 2.4 and source_score >= 3.2: return False # Prefer attested or near-parity collocations. Allow modest drops so # hyponyms like "positive attitude" → "positive mindset" can pass. if source_score >= 3.8 and candidate_score + 0.75 < source_score: return False return True def _verb_object_attested( source_verb: str, source_noun: str, cand_verb: str, cand_noun: str, *, classical_strict: bool, ) -> bool: """Reject unattested verb–object drift (launch reputation, define issues).""" if source_verb == cand_verb and source_noun == cand_noun: return True src_vo = zipf_frequency(f"{source_verb} {source_noun}", "en") cand_vo = zipf_frequency(f"{cand_verb} {cand_noun}", "en") src_v = zipf_frequency(source_verb, "en") cand_v = zipf_frequency(cand_verb, "en") # Specificity: phrase score minus bare verb. Ultra-common verbs inflate # raw VO zipf without being real collocations (found/launch reputation). src_spec = src_vo - src_v cand_spec = cand_vo - cand_v if classical_strict: if source_verb != cand_verb: if cand_spec + 0.02 < src_spec: return False # Raw VO inflate from a commoner verb is not a real collocation win. if cand_vo > src_vo and cand_spec < src_spec: return False # Leap into an ultra-common verb usually marks a wrong sense. if cand_v >= 5.15 and cand_v - src_v >= 0.45: return False # Abstract/weak VO: both negative specificity → block free WordNet # verbs (keep/hold/throw attitude). if cand_spec < -0.30 and src_spec < -0.20 and cand_spec < src_spec + 0.55: return False # Both rare with this object → free WordNet drift; block. if src_vo < 2.2 and cand_vo < 2.2: return False # Lose a clearly attested VO pair. if src_vo >= 2.2 and cand_vo + 0.20 < src_vo: return False if cand_vo < 2.0: return False # Peer-frequency verb swaps without a specificity gain. if abs(cand_v - src_v) < 0.40 and cand_spec <= src_spec + 0.05: return False return True if src_vo >= 2.2 and cand_vo + 0.45 < src_vo: return False if src_vo >= 3.5 and cand_vo - src_vo >= 0.35: return False return True def _surface_changed(source: str, candidate: str) -> bool: left = re.sub(r"\s+", " ", (source or "").strip().lower()) right = re.sub(r"\s+", " ", (candidate or "").strip().lower()) return bool(left and right and left != right) def _wordnet_span_candidates( resource, span: _SpanTarget, *, polish: bool, classical_strict: bool = False, classical_aggressive: bool = False, ) -> list[str]: from app.pipeline.minilm import minilm_available verb_lemma = span.verb.lemma_.lower() noun_lemma = span.noun.lemma_.lower() has_mods = bool(span.modifiers) # Hyponyms need a meaning gate; aggressive classical still skips them. minilm_ok = minilm_available() and not classical_strict verbs = [verb_lemma] + _related_verb_lemmas( resource, verb_lemma, allow_hyponyms=minilm_ok, # No meaning gate: trust only the dominant senses. max_senses=2 if classical_strict else 4, ) context = {mod.lower() for mod in span.modifiers} # Object text helps pick a sense, but drop the head noun itself — otherwise # every gloss that mentions the source lemma wins (report → "verbal report" # sense → account). context.update( term for term in _content_terms(span.object_text) if term != noun_lemma ) hypo_only: set[str] = set() if has_mods: try: synsets = list(resource.synsets(noun_lemma, pos="n")[:4]) ranked = sorted( ( ( len(context & _content_terms(synset.definition() or "")), synset, ) for synset in synsets ), key=lambda item: -item[0], ) best = ranked[0][1] if ranked and ranked[0][0] > 0 else synsets[0] parent_defn = best.definition() or "" for hypo in list(best.get_related("hyponym") or [])[:20]: if not _definition_linked(noun_lemma, parent_defn, hypo): continue lemmas = _single_word_lemmas(hypo) if len(lemmas) < 2 or len(lemmas) > 4: continue if context and not all( max( ( zipf_frequency(f"{mod} {candidate}", "en") for mod in context if len(mod) >= 3 ), default=0.0, ) >= 3.5 for candidate in lemmas ): continue hypo_defn = (hypo.definition() or "").lower() if noun_lemma == "attitude" and "mental attitude" not in hypo_defn: continue if noun_lemma == "attitude" and any( marker in hypo_defn for marker in ( "admiration", "defensive", "arrogant", "politics", "rationalized", "believable", ) ): continue hypo_only.update(lemmas) except Exception: hypo_only = set() if has_mods: if classical_strict: # No MiniLM: do not retarget modified heads (final report → study / # account). Verb-only swaps stay available. nouns = [noun_lemma] else: raw_nouns = _related_noun_lemmas( resource, noun_lemma, allow_hyponyms=True, context_terms=context, ) ranked_nouns: list[tuple[float, str]] = [] for candidate in raw_nouns: is_hypo = candidate in hypo_only if not _modifier_specificity_ok( span.modifiers, noun_lemma, candidate, hyponym=is_hypo, ): continue if is_hypo: src_f = zipf_frequency(noun_lemma, "en") cand_f = zipf_frequency(candidate, "en") if cand_f > src_f - 0.15: continue mod_score = max( ( zipf_frequency(f"{mod.lower()} {candidate}", "en") for mod in span.modifiers ), default=0.0, ) src_mod = max( ( zipf_frequency(f"{mod.lower()} {noun_lemma}", "en") for mod in span.modifiers ), default=0.0, ) # Prefer heads that keep modifier collocation closest to the source. closeness = -abs(mod_score - src_mod) ranked_nouns.append((closeness, mod_score, candidate)) ranked_nouns.sort(reverse=True) nouns = [noun_lemma] + [item[2] for item in ranked_nouns] else: nouns = [noun_lemma] # Classical-strict: fewer peers. Aggressive: restore fuller candidate pools. if classical_strict and not classical_aggressive: verb_cap = 3 if polish else 2 noun_cap = 3 if polish else 2 else: verb_cap = 6 if polish else 4 noun_cap = 6 if polish else 4 verbs = verbs[:verb_cap] nouns = nouns[:noun_cap] candidates: list[str] = [] seen: set[str] = {span.text.lower()} for new_verb in verbs: for new_noun in nouns: if new_verb == verb_lemma and new_noun == noun_lemma: continue if ( not polish and new_verb != verb_lemma and new_noun != noun_lemma ): continue # Tight classical polish: change verb OR noun, never both at once. if ( classical_strict and not classical_aggressive and new_verb != verb_lemma and new_noun != noun_lemma ): continue rebuilt = _rebuild_span( span, verb_lemma=None if new_verb == verb_lemma else new_verb, noun_lemma=None if new_noun == noun_lemma else new_noun, ) if not rebuilt or not _surface_changed(span.text, rebuilt): continue if not _collocation_accepts( span.text, rebuilt, classical_strict=classical_strict and not classical_aggressive, ): continue if not _verb_object_attested( verb_lemma, noun_lemma, new_verb, new_noun, classical_strict=classical_strict and not classical_aggressive, ): continue if classical_aggressive and new_verb != verb_lemma: # Keep disaster brakes even when aggressive. src_v = zipf_frequency(verb_lemma, "en") cand_v = zipf_frequency(new_verb, "en") src_vo = zipf_frequency(f"{verb_lemma} {noun_lemma}", "en") cand_vo = zipf_frequency(f"{new_verb} {noun_lemma}", "en") src_spec = src_vo - src_v cand_spec = cand_vo - cand_v if src_vo >= 3.8 and cand_v - src_v >= 0.35 and cand_spec <= src_spec + 0.08: continue if src_vo >= 3.8 and cand_vo >= src_vo and cand_spec + 0.08 < src_spec: continue if src_vo >= 4.0 and cand_vo >= 4.0 and cand_spec + 0.05 < src_spec: continue if cand_spec < -0.08 and src_vo >= 4.0: continue if cand_spec < -0.30 and src_spec < -0.15 and cand_spec < src_spec + 0.40: continue # Peer-cycle brake: tight mode only; aggressive allows near-peers. if ( classical_strict and not classical_aggressive and new_verb != verb_lemma ): src_f = zipf_frequency(verb_lemma, "en") cand_f = zipf_frequency(new_verb, "en") if cand_f + 0.15 < src_f: continue if abs(cand_f - src_f) < 0.35 and cand_f < src_f + 0.45: continue key = rebuilt.lower() if key in seen: continue seen.add(key) candidates.append(rebuilt) return candidates def _t5_span_candidates(span_text: str, *, num_return: int = 4) -> list[str]: try: from app.engine.paraphrase import paraphrase_span, _span_candidate_ok except Exception: return [] try: result = paraphrase_span(span_text, num_return=num_return) except Exception as exc: logger.debug("phrase T5 unavailable: %s", exc) return [] out: list[str] = [] seen: set[str] = set() for item in list(result.candidates or []) + ([result.text] if result.text else []): cleaned = re.sub(r"\s+", " ", (item or "").strip(" .")) if span_text[:1].islower() and cleaned[:1].isupper(): cleaned = cleaned[:1].lower() + cleaned[1:] key = cleaned.lower() if ( not cleaned or key in seen or not _surface_changed(span_text, cleaned) or not _span_candidate_ok(span_text, cleaned) ): continue seen.add(key) out.append(cleaned) return out[:num_return] def _rank_span_candidates( source_span: str, candidates: list[str], *, min_sim: float, modifiers: list[str] | None = None, classical_strict: bool = False, classical_aggressive: bool = False, ) -> str | None: if not candidates: return None meaning_ok: list[str] = [] for candidate in candidates: meaning = score_candidate(source_span, candidate) if meaning is None: # Without MiniLM, classical-strict already filtered by collocation. meaning_ok.append(candidate) continue if meaning >= min_sim: meaning_ok.append(candidate) pool = meaning_ok or [] if not pool: return None scored: list[tuple[float, float, float, str]] = [] source_zipf = _phrase_zipf(source_span) src_tokens = {w.lower() for w in _WORD.findall(source_span)} for candidate in pool: cand_zipf = _phrase_zipf(candidate) cand_tokens = {w.lower() for w in _WORD.findall(candidate)} distance = float(len(src_tokens ^ cand_tokens)) # Prefer stable modifier collocations when present. mod_bonus = 0.0 for mod in modifiers or []: # Find noun-ish last content token as head proxy. heads = [w for w in _WORD.findall(candidate) if len(w) >= 4] if not heads: continue mod_bonus = max( mod_bonus, zipf_frequency(f"{mod.lower()} {heads[-1].lower()}", "en"), ) scored.append((mod_bonus, cand_zipf - source_zipf, distance, candidate)) # Aggressive classical: among collocation-ok picks, prefer moderate distance # (not max distance — that favored throw/influence-style peers). if classical_aggressive: scored.sort( key=lambda item: (item[0], item[1], min(item[2], 3.0)), reverse=True, ) return scored[0][3] scored.sort(reverse=True) # Tight classical / no MiniLM: pick best collocation, never max divergence. if classical_strict: return scored[0][3] surface_scores = { candidate: 1.0 - (distance / 10.0) for _mod, _gain, distance, candidate in scored[:8] } picked = pick_best_candidate( source_span, [item[3] for item in scored[:8]], min_meaning=min_sim, prefer_divergent=True, surface_scores=surface_scores, ) return picked or scored[0][3] def _splice(text: str, start: int, end: int, replacement: str) -> str: return text[:start] + replacement + text[end:] def rewrite_phrases( text: str, *, max_changes: int | None = None, polish: bool = False, min_sim: float | None = None, wordnet: Any | None = None, use_t5: bool = True, classical_strict: bool = False, classical_aggressive: bool = False, ) -> PhraseResult: """Rewrite up to N verb–object phrases with meaning-safe alternatives.""" if not ENGINE_PHRASE_REWRITE and wordnet is None: return PhraseResult(text=text, reason="disabled") source = (text or "").strip() if not source: return PhraseResult(text=text, reason="empty") if any(quote in source for quote in _QUOTES): return PhraseResult(text=source, reason="quoted") if _PROTECTED_MARKER.search(source): return PhraseResult(text=source, reason="protected") nlp = get_nlp() resource = wordnet if wordnet is not None else _get_wordnet() if nlp is None or resource is None: return PhraseResult(text=source, reason="resource_unavailable") try: doc = nlp(source) except Exception: return PhraseResult(text=source, reason="parse_failed") limit = ( max(1, min(int(max_changes), 4)) if max_changes is not None else ( ENGINE_PHRASE_MAX_CHANGES if ENGINE_PHRASE_MAX_CHANGES > 0 else (2 if polish else 1) ) ) if classical_strict and not classical_aggressive: limit = min(limit, 1 if not polish else 2) elif classical_aggressive: limit = max(limit, 2 if polish else 2) limit = min(limit, 3) threshold = min_sim if min_sim is not None else ENGINE_PHRASE_MIN_SIM spans = _extract_spans(doc) if not spans: return PhraseResult(text=source, reason="no_spans") current = source changes: list[LexicalChange] = [] touched_verbs: set[str] = set() # Re-parse after each accepted splice so offsets stay valid. for _ in range(limit): try: doc = nlp(current) except Exception: break spans = _extract_spans(doc) best: tuple[float, _SpanTarget, str] | None = None for span in spans: verb_key = span.verb.lemma_.lower() if verb_key in touched_verbs: continue # Skip spans already touched. if any( change.original.lower() == span.text.lower() for change in changes ): continue candidates = _wordnet_span_candidates( resource, span, polish=polish, classical_strict=classical_strict, classical_aggressive=classical_aggressive, ) if use_t5 and not classical_strict: candidates.extend(_t5_span_candidates(span.text)) # Deduplicate and drop prompt-echo / clause-flip spans. try: from app.engine.paraphrase import _span_candidate_ok except Exception: _span_candidate_ok = None uniq: list[str] = [] seen: set[str] = set() for cand in candidates: key = cand.lower() if key in seen or key == span.text.lower(): continue if _span_candidate_ok is not None and not _span_candidate_ok( span.text, cand ): continue seen.add(key) uniq.append(cand) picked = _rank_span_candidates( span.text, uniq, min_sim=threshold, modifiers=span.modifiers, classical_strict=classical_strict, classical_aggressive=classical_aggressive, ) if not picked: continue # Score full-sentence splice. spliced = _splice(current, span.start, span.end, picked) if spliced == current: continue meaning = score_candidate(source, spliced) if meaning is not None and meaning < threshold: continue # Tight classical: require non-negative collocation gain. gain = _phrase_zipf(picked) - _phrase_zipf(span.text) if classical_strict and not classical_aggressive and gain < -0.15: continue distance = len( {w.lower() for w in _WORD.findall(span.text)} ^ {w.lower() for w in _WORD.findall(picked)} ) score = gain + (0.15 * distance) + (meaning or 0.0) if classical_aggressive: # Prefer surface novelty among VO-safe candidates. score = (0.35 * distance) + gain + (meaning or 0.0) elif classical_strict: # Prefer attested collocation, not surface novelty. score = gain + (0.05 * distance) + (meaning or 0.0) if best is None or score > best[0]: best = (score, span, picked) if best is None: break _score, span, picked = best current = _splice(current, span.start, span.end, picked) touched_verbs.add(span.verb.lemma_.lower()) changes.append( LexicalChange( original=span.text, replacement=picked, token_index=span.verb.i, lemma=span.verb.lemma_.lower(), synset_id="phrase", confidence=round(min(0.95, 0.55 + best[0] * 0.1), 4), ) ) if not changes: return PhraseResult(text=source, reason="no_safe_change") return PhraseResult( text=current, changes=changes, confidence=min(change.confidence for change in changes), reason="phrase_rewrite", )