File size: 6,603 Bytes
e61cf07 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 | """Algorithmic rewrite quality gates (no word/phrase denylists)."""
from __future__ import annotations
import re
from collections import Counter
from app.pipeline.nlp import get_nlp
_WORD = re.compile(r"[A-Za-z][A-Za-z']*")
_MODAL = frozenset(
{"can", "could", "will", "would", "may", "might", "should", "must", "shall"}
)
def _alpha_tokens(text: str) -> list[str]:
return [m.group(0).lower() for m in _WORD.finditer(text or "")]
def closed_compounds_not_split(original: str, candidate: str) -> bool:
"""Reject when a solid source token is opened into two words (teamwork→team work)."""
source_tokens = _alpha_tokens(original)
cand = re.sub(r"\s+", " ", (candidate or "").lower())
cand_token_set = set(_alpha_tokens(candidate))
for token in source_tokens:
if len(token) < 6:
continue
if token in cand_token_set:
continue
for index in range(3, len(token) - 2):
left, right = token[:index], token[index:]
if f"{left} {right}" in cand:
return False
return True
def modal_head_pos_preserved(original: str, candidate: str) -> bool:
"""Keep modal/aux clause heads stable in POS and lexical relatedness."""
nlp = get_nlp()
if nlp is None:
return True
try:
source_doc = nlp(original or "")
cand_doc = nlp(candidate or "")
except Exception:
return True
def _modal_heads(doc) -> list[tuple[str, str, str]]:
heads: list[tuple[str, str, str]] = []
for token in doc:
lemma = token.lemma_.lower()
is_modal = lemma in _MODAL or (
token.pos_ == "AUX" and token.dep_ in {"aux", "auxpass"}
)
if not is_modal:
continue
head = token.head
if head.i == token.i:
continue
heads.append((lemma, head.pos_, head.lemma_.lower()))
return heads
def _verb_related(source_lemma: str, candidate_lemma: str) -> bool:
if source_lemma == candidate_lemma:
return True
try:
import wn
from app.config import ENGINE_WORDNET_LEXICON
wn.config.allow_multithreading = True
resource = wn.Wordnet(ENGINE_WORDNET_LEXICON)
synsets = list(resource.synsets(source_lemma, pos="v"))
except Exception:
# If lexicon is unavailable, keep POS-only behavior.
return True
for synset in synsets:
try:
words = {
(word.lemma() or "").replace("_", " ").strip().lower()
for word in synset.words()
}
except Exception:
continue
if candidate_lemma in words:
return True
return False
source_heads = _modal_heads(source_doc)
if not source_heads:
return True
cand_heads = _modal_heads(cand_doc)
cand_by_modal: dict[str, list[tuple[str, str]]] = {}
for modal, pos, lemma in cand_heads:
cand_by_modal.setdefault(modal, []).append((pos, lemma))
for modal, pos, lemma in source_heads:
options = cand_by_modal.get(modal)
if not options:
continue
matched = False
for cand_pos, cand_lemma in options:
if cand_pos != pos:
continue
if _verb_related(lemma, cand_lemma):
matched = True
break
if not matched:
return False
return True
def content_pos_balance_ok(original: str, candidate: str) -> bool:
"""Reject large shifts in content POS counts (noun/verb/adj)."""
nlp = get_nlp()
if nlp is None:
return True
try:
source_doc = nlp(original or "")
cand_doc = nlp(candidate or "")
except Exception:
return True
def _counts(doc) -> Counter:
return Counter(
token.pos_
for token in doc
if token.is_alpha and not token.is_stop and token.pos_ in {"NOUN", "VERB", "ADJ"}
)
source = _counts(source_doc)
cand = _counts(cand_doc)
for pos in ("NOUN", "VERB", "ADJ"):
if abs(source[pos] - cand[pos]) > 2:
return False
return True
def substitution_pos_stable(
sentence: str,
token_index: int,
replacement: str,
) -> bool:
"""Require the replaced span to keep the same coarse POS after re-parse."""
nlp = get_nlp()
if nlp is None:
return True
try:
doc = nlp(sentence)
except Exception:
return True
if token_index < 0 or token_index >= len(doc):
return False
token = doc[token_index]
updated = sentence[: token.idx] + replacement + sentence[token.idx + len(token.text) :]
try:
new_doc = nlp(updated)
except Exception:
return True
# Locate replacement by character offset.
target = None
for item in new_doc:
if item.idx == token.idx:
target = item
break
if target is None:
# Fallback: first overlapping alpha token near the old index.
for item in new_doc:
if item.is_alpha and abs(item.idx - token.idx) <= max(1, len(replacement)):
target = item
break
if target is None:
return False
return target.pos_ == token.pos_
def meaning_ok(original: str, candidate: str, *, min_sim: float = 0.72) -> bool | None:
"""MiniLM meaning check when available; None means backend unavailable."""
try:
from app.pipeline.minilm import score_candidate
scored = score_candidate(original, candidate)
except Exception:
return None
if scored is None:
return None
return float(scored) >= min_sim
def naturalness_reasons(original: str, candidate: str) -> list[str]:
"""Algorithmic naturalness failures (no phrase denylist)."""
reasons: list[str] = []
if not closed_compounds_not_split(original, candidate):
reasons.append("compound_split")
if not modal_head_pos_preserved(original, candidate):
reasons.append("modal_pos_shift")
if not content_pos_balance_ok(original, candidate):
reasons.append("pos_balance")
meaning = meaning_ok(original, candidate)
if meaning is False:
reasons.append("meaning_drop")
return reasons
|