"""Deterministic template families for structural reorder.""" from __future__ import annotations import re from functools import lru_cache from app.engine.models import SentenceSlots, TemplateCandidate from app.pipeline.nlp import get_nlp # Directional phrases are usually verb arguments ("went to school"), and # fronting an argument reads as marked ("To school, Ram went"). Locative # prepositions are adjuncts and front cleanly ("In the library, students read"). _DIRECTIONAL_PLACE = re.compile( r"^(to|into|onto|towards?|from|out of)\b", re.IGNORECASE ) def _place_fronts_cleanly(slots: SentenceSlots) -> bool: return bool(slots.place) and not _DIRECTIONAL_PLACE.match(slots.place.strip()) def rank_templates(slots: SentenceSlots) -> list[TemplateCandidate]: """Return confidence-ranked template candidates (never random).""" if not slots.subject or not slots.verb_phrase: return [] if slots.confidence < 0.45 or "low_confidence" in slots.reasons: return [] out: list[TemplateCandidate] = [] if slots.time and slots.manner and (slots.place or slots.object): out.append(TemplateCandidate("time_subj_manner_verb_place", 0.9)) if slots.time and slots.place: out.append(TemplateCandidate("time_subj_verb_place", 0.8)) if slots.time and slots.object: out.append(TemplateCandidate("time_subj_verb_object", 0.78)) if slots.time: out.append(TemplateCandidate("time_subj_verb_object_place", 0.72)) out.append(TemplateCandidate("subj_verb_object_time", 0.68)) out.append(TemplateCandidate("time_front", 0.64)) if slots.place and slots.subject and slots.verb_phrase: if _place_fronts_cleanly(slots): out.append(TemplateCandidate("place_front", 0.74)) if slots.manner and slots.place: out.append(TemplateCandidate("subj_manner_verb_place", 0.7)) if slots.manner and slots.object: out.append(TemplateCandidate("subj_manner_verb_rest", 0.66)) if slots.manner: out.append(TemplateCandidate("adv_subj_verb_object", 0.72)) out.append(TemplateCandidate("subj_manner_verb_rest", 0.65)) if slots.place: out.append(TemplateCandidate("subj_verb_place", 0.6)) if slots.object: out.append(TemplateCandidate("subj_verb_object", 0.58)) seen: set[str] = set() ranked: list[TemplateCandidate] = [] for c in sorted(out, key=lambda x: -x.confidence): if c.template_id in seen: continue seen.add(c.template_id) ranked.append(c) return ranked def _cap(text: str) -> str: t = (text or "").strip() if not t: return t return t[0].upper() + t[1:] def _cont(phrase: str, slots: SentenceSlots) -> str: """Lowercase a moved subject unless its parsed head is a proper noun.""" p = (phrase or "").strip() if not p: return p first = p.split()[0].strip("\"'") for ent in slots.entities: ent_first = ent.split()[0] if first == ent_first and ent_first[:1].isupper(): return p if p == slots.subject and not slots.subject_is_proper and first: return first.lower() + p[len(first) :] return p def _join_slots(*parts: str) -> str: bits = [p.strip() for p in parts if p and p.strip()] s = " ".join(bits) s = re.sub(r"\s+", " ", s).strip() s = re.sub(r"\s+([,.;:!?])", r"\1", s) return s def _terminal(text: str) -> str: if text.rstrip().endswith(("!", "?")): return text.rstrip()[-1] return "." @lru_cache(maxsize=2048) def _initial_subordinate_parts(text: str) -> tuple[str, str, str, bool] | None: """Return an initial parsed adverbial clause and its main clause.""" raw = (text or "").strip() if "," not in raw: return None nlp = get_nlp() if nlp is None: return None try: doc = nlp(raw) except Exception: return None comma = next((token for token in doc if token.text == ","), None) root = next((token for token in doc if token.dep_ == "ROOT"), None) if comma is None or root is None or root.i <= comma.i: return None clause = next( ( token for token in doc if token.dep_ == "advcl" and min(part.i for part in token.subtree) == 0 and max(part.i for part in token.subtree) < comma.i and any( part.dep_ == "mark" or part.pos_ == "SCONJ" for part in token.subtree ) ), None, ) if clause is None: return None end = _terminal(raw) core = raw[:-1].rstrip() if raw.endswith((".", "!", "?")) else raw subordinate, main = (part.strip() for part in core.split(",", 1)) if len(subordinate.split()) < 3 or len(main.split()) < 3: return None return subordinate, main, end, doc[0].dep_ == "mark" def can_swap_initial_subordinate(text: str) -> bool: return _initial_subordinate_parts(text) is not None def try_complex_clause_swap(text: str) -> str | None: """Move a parsed initial adverbial clause behind the main clause.""" parts = _initial_subordinate_parts(text) if parts is None: return None subordinate, main, end, needs_comma = parts continuation = subordinate[0].lower() + subordinate[1:] separator = ", " if needs_comma else " " return f"{_cap(main)}{separator}{continuation}{end}" # Connectives relate the sentence to the previous one; moving them to the tail # breaks the link the writer set up. _DISCOURSE_OPENERS = frozenset( { "however", "moreover", "therefore", "thus", "also", "instead", "nevertheless", "nonetheless", "meanwhile", "furthermore", "besides", "otherwise", "consequently", "similarly", "conversely", "finally", "first", "firstly", "second", "secondly", "third", "thirdly", "next", "then", "overall", "generally", "typically", "unfortunately", "fortunately", "importantly", "specifically", "notably", } ) @lru_cache(maxsize=2048) def try_unfront_opener(text: str) -> str | None: """Move a sentence-initial modifier phrase to the end. The inverse of the ``*_front`` family. Fronted openers are otherwise a dead end: the fronting templates regenerate the same sentence, so a text that already opens with "For the success of any business, ..." has no remaining structural move. """ raw = (text or "").strip() if "," not in raw: return None nlp = get_nlp() if nlp is None: return None try: doc = nlp(raw) except Exception: return None comma = next((token for token in doc if token.text == ","), None) root = next((token for token in doc if token.dep_ == "ROOT"), None) if comma is None or root is None or root.i <= comma.i: return None # A second comma usually means a list or an embedded aside; the opener is # then no longer a clean prefix to relocate. if any(token.text == "," for token in doc[comma.i + 1 :]): return None opener_head = next( ( token for token in doc if token.head.i == root.i and token.dep_ in {"prep", "advmod", "npadvmod", "nmod"} and min(part.i for part in token.subtree) == 0 and max(part.i for part in token.subtree) == comma.i - 1 ), None, ) if opener_head is None: return None if doc[0].lower_ in _DISCOURSE_OPENERS: return None # The subject must follow the comma, otherwise the "opener" is really part # of the subject and moving it would strand the verb. Copular sentences # carry a clausal subject ("providing good service is vital"). subject = next( ( child for child in root.children if child.dep_ in {"nsubj", "nsubjpass", "csubj"} ), None, ) if subject is None or subject.i < comma.i: return None end = _terminal(raw) core = raw[:-1].rstrip() if raw.endswith((".", "!", "?")) else raw opener, main = (part.strip() for part in core.split(",", 1)) if len(opener.split()) < 2 or len(main.split()) < 4: return None continuation = opener[0].lower() + opener[1:] if not doc[0].pos_ == "PROPN" else opener return f"{_cap(main)} {continuation}{end}" def _manner_before_verb(slots: SentenceSlots) -> bool: return not slots.verb_starts_with_aux def _object_without_place(slots: SentenceSlots) -> str: obj = (slots.object or "").strip() place = (slots.place or "").strip() if place and obj and place in obj: obj = obj.replace(place, "").strip(" ,") obj = re.sub(r"\s+", " ", obj).strip() return obj def fill_template(template_id: str, slots: SentenceSlots) -> str | None: """Build sentence from slots using the selected template (reorder only).""" s = slots end = _terminal(s.text) tid = template_id obj = _object_without_place(s) if tid == "place_front": if not s.place: return None # Copula/AUX: Subject + Verb + Manner + Object (not Subject + Manner + Verb) if _manner_before_verb(s): body = _join_slots( _cap(s.place) + ",", _cont(s.subject, s), s.manner, s.verb_phrase, obj, s.time, ) else: body = _join_slots( _cap(s.place) + ",", _cont(s.subject, s), s.verb_phrase, s.manner, obj, s.time, ) out = body + end if out.lower().rstrip(".!?") == s.text.lower().rstrip(".!?"): return None return out if tid == "time_subj_manner_verb_place": place_or_obj = s.place or obj if _manner_before_verb(s): body = _join_slots( _cap(s.time) + ",", _cont(s.subject, s), s.manner, s.verb_phrase, place_or_obj ) else: body = _join_slots( _cap(s.time) + ",", _cont(s.subject, s), s.verb_phrase, s.manner, place_or_obj ) return body + end if tid == "time_subj_verb_place": body = _join_slots(_cap(s.time) + ",", _cont(s.subject, s), s.verb_phrase, s.place) return body + end if tid == "time_subj_verb_object": body = _join_slots( _cap(s.time) + ",", _cont(s.subject, s), s.verb_phrase, obj, s.manner, s.place if s.place not in (obj or "") else "", ) return body + end if tid == "time_subj_verb_object_place": body = _join_slots( _cap(s.time) + ",", _cont(s.subject, s), s.verb_phrase, obj, s.place, s.manner, ) return body + end if tid == "subj_verb_object_time": body = _join_slots(_cap(s.subject), s.verb_phrase, obj, s.place, s.manner, s.time) return body + end if tid == "adv_subj_verb_object": body = _join_slots( _cap(s.manner) + ",", _cont(s.subject, s), s.verb_phrase, obj, s.place, s.time, ) return body + end if tid == "subj_manner_verb_place": if _manner_before_verb(s): body = _join_slots(_cap(s.subject), s.manner, s.verb_phrase, s.place) else: body = _join_slots(_cap(s.subject), s.verb_phrase, s.manner, s.place) return body + end if tid == "time_front": if s.time and re.search( rf"\b(for|over|within|during)\s+{re.escape(s.time)}\b", s.text, flags=re.I, ): return None if s.subject and s.verb_phrase: body = _join_slots( _cap(s.time) + ",", _cont(s.subject, s), s.verb_phrase, obj, s.place, s.manner, ) return body.rstrip(".!?") + end rest = s.text if s.time: rest = re.sub(rf"\b{re.escape(s.time)}\b", "", rest, count=1, flags=re.I) rest = re.sub(r"\s+", " ", rest).strip(" ,.") if not rest: return None if re.search(r"\b(for|over|within|during)\s+(every|each|\.|$)", rest, flags=re.I): return None body = _join_slots(_cap(s.time) + ",", _cont(rest, s)) return body.rstrip(".!?") + end if tid == "subj_manner_verb_rest": if _manner_before_verb(s): body = _join_slots(_cap(s.subject), s.manner, s.verb_phrase, obj, s.place, s.time) else: body = _join_slots(_cap(s.subject), s.verb_phrase, s.manner, obj, s.place, s.time) return body + end if tid == "subj_verb_place": body = _join_slots(_cap(s.subject), s.verb_phrase, s.place, s.time, s.manner) return body + end if tid == "subj_verb_object": body = _join_slots(_cap(s.subject), s.verb_phrase, obj, s.place, s.time, s.manner) return body + end return None def try_because_front(text: str) -> str | None: """Main … because Sub → Because Sub, main …""" raw = (text or "").strip() if not raw: return None end = "." m_end = re.search(r"[.!?]+$", raw) if m_end: end = m_end.group(0) raw = raw[: m_end.start()].strip() m = re.match(r"^(?P
.+?)\s+because\s+(?P.+)$", raw, flags=re.I) if not m: return None main = m.group("main").strip(" ,") sub = m.group("sub").strip(" ,") if len(main.split()) < 3 or len(sub.split()) < 2: return None if re.search(r"\bbecause\b", main, flags=re.I) or re.search( r"\bbecause\b", sub, flags=re.I ): return None body = ( f"Because {sub[0].lower() + sub[1:]}, " f"{main[0].lower() + main[1:]}" ) return re.sub(r"\s+", " ", body).strip() + end def _sentence_end(text: str) -> tuple[str, str]: raw = (text or "").strip() end = "." m_end = re.search(r"[.!?]+$", raw) if m_end: end = m_end.group(0) raw = raw[: m_end.start()].strip() return raw, end def try_for_purpose_front(text: str) -> str | None: """X is essential for Y → For Y, x is essential.""" raw, end = _sentence_end(text) match = re.match( r"^(?P.+?)\s+(?Pis|are|was|were)\s+" r"(?Pessential|important|critical|necessary|vital|crucial)\s+" r"for\s+(?P.+)$", raw, flags=re.I, ) if not match: return None head = match.group("head").strip(" ,") purpose = match.group("purpose").strip(" ,") if len(head.split()) < 3 or len(purpose.split()) < 2: return None if re.search(r"\bfor\b", head, flags=re.I): return None lowered = head[0].lower() + head[1:] if head[:1].isupper() else head body = ( f"For {purpose}, {lowered} {match.group('be')} {match.group('adj')}" ) return re.sub(r"\s+", " ", body).strip() + end def try_in_pp_front(text: str) -> str | None: """… play/take … in Y → In Y, …""" raw, end = _sentence_end(text) match = re.match( r"^(?P.+?)\s+" r"(?Pplay|plays|take|takes|have|has|hold|holds)\s+" r"(?P(?:an?\s+)?(?:important\s+|key\s+|major\s+)?(?:role|roles|part|parts|place))\s+" r"in\s+(?P.+)$", raw, flags=re.I, ) if not match: return None head = match.group("head").strip(" ,") place = match.group("place").strip(" ,") if len(head.split()) < 2 or len(place.split()) < 2: return None lowered = head[0].lower() + head[1:] if head[:1].isupper() else head body = ( f"In {place}, {lowered} {match.group('verb')} {match.group('object')}" ) return re.sub(r"\s+", " ", body).strip() + end def try_copula_np_invert(text: str) -> str | None: """Subject + is/are a/an/the NP → NP is/are subject.""" raw, end = _sentence_end(text) match = re.match( r"^(?P[A-Z][^,]{1,100}?)\s+" r"(?Pis|are|was|were)\s+" r"(?P(?:a|an|the)\s+[^,;:]+)$", raw, flags=0, ) if not match: return None subject = match.group("subj").strip(" ,") predicate = match.group("pred").strip(" ,") be = match.group("be") if len(subject.split()) < 1 or len(predicate.split()) < 2: return None # Avoid inverting long clausal subjects with internal finite verbs. if re.search( r"\b(that|which|who|when|where|because|while|although)\b", subject, flags=re.I, ): return None front = predicate[0].upper() + predicate[1:] lowered = subject[0].lower() + subject[1:] if subject[:1].isupper() else subject if re.match(r"^(a|an)\b", predicate, flags=re.I) and be in {"are", "were"}: be = "is" if be == "are" else "was" if re.match(r"^the\b", predicate, flags=re.I) and be in {"is", "was"}: # Keep agreement for pluralish subjects ending with s when obvious. if subject.lower().endswith("s") and not subject.lower().endswith( ("ss", "ness", "ics") ): be = "are" if be == "is" else "were" body = f"{front} {be} {lowered}" return re.sub(r"\s+", " ", body).strip() + end def try_when_clause_front(text: str) -> str | None: """Main … when Sub → When Sub, main …""" raw, end = _sentence_end(text) match = re.match( r"^(?P
.+?)\s+when\s+(?P.+)$", raw, flags=re.I, ) if not match: return None main = match.group("main").strip(" ,") sub = match.group("sub").strip(" ,") if len(main.split()) < 4 or len(sub.split()) < 3: return None if re.search(r"\bwhen\b", main, flags=re.I) or re.search( r"\bwhen\b", sub, flags=re.I ): return None body = ( f"When {sub[0].lower() + sub[1:]}, " f"{main[0].lower() + main[1:]}" ) return re.sub(r"\s+", " ", body).strip() + end def try_by_agent_front(text: str) -> str | None: """… are/is VERBen by AGENT → By AGENT, … are/is VERBen.""" raw, end = _sentence_end(text) match = re.match( r"^(?P.+?)\s+" r"(?Pis|are|was|were)\s+" r"(?P\w+ed|\w+en)\s+" r"by\s+(?P.+)$", raw, flags=re.I, ) if not match: return None head = match.group("head").strip(" ,") agent = match.group("agent").strip(" ,") if len(head.split()) < 3 or len(agent.split()) < 2: return None if re.search(r"\bby\b", head, flags=re.I): return None lowered = head[0].lower() + head[1:] if head[:1].isupper() else head body = ( f"By {agent}, {lowered} {match.group('be')} {match.group('verb')}" ) return re.sub(r"\s+", " ", body).strip() + end def try_in_both_front(text: str) -> str | None: """X is Y in both A and B → In both A and B, x is Y.""" raw, end = _sentence_end(text) match = re.match( r"^(?P.+?)\s+" r"(?Pis|are|was|were)\s+" r"(?P.+?)\s+" r"in both\s+(?P.+)$", raw, flags=re.I, ) if not match: return None head = match.group("head").strip(" ,") pred = match.group("pred").strip(" ,") scope = match.group("scope").strip(" ,") if len(head.split()) < 1 or len(pred.split()) < 2 or len(scope.split()) < 3: return None lowered = head[0].lower() + head[1:] if head[:1].isupper() else head body = ( f"In both {scope}, {lowered} {match.group('be')} {pred}" ) return re.sub(r"\s+", " ", body).strip() + end def try_such_as_front(text: str) -> str | None: """Label such as EXAMPLES can TAIL → EXAMPLES can TAIL as label.""" raw, end = _sentence_end(text) match = re.match( r"^(?P