"""Deterministic long-sentence splitting for structural variety.""" from __future__ import annotations import re from functools import lru_cache from app.pipeline.nlp import get_nlp _WORD = re.compile(r"[A-Za-z0-9']+") _QUOTE = re.compile(r"[\"“”‘’]") # Prefer clause joins that usually mark a safe split boundary. _SEPARATORS = ( "; ", ", and ", ", but ", ", so ", ", which ", ", although ", ", because ", ", while ", ", whereas ", " although ", " because ", " whereas ", ) @lru_cache(maxsize=4096) def _is_independent_clause(text: str) -> bool: """True when the fragment can stand alone: a finite verb with a subject. Splitting a coordinated verb phrase ("…gain retention, and encourage X") otherwise strands a bare VP that reads as an imperative. """ candidate = (text or "").strip().rstrip(".!?") if not candidate: return False nlp = get_nlp() if nlp is None: # Without a parser, only trust splits on an explicit clause marker. return False try: doc = nlp(candidate) except Exception: return False root = next((token for token in doc if token.dep_ == "ROOT"), None) if root is None: return False if root.pos_ not in {"VERB", "AUX"}: # Copular clauses attach the subject to a predicate ROOT. return any(child.dep_ in {"nsubj", "nsubjpass"} for child in root.children) if root.tag_ in {"VB", "VBG", "VBN"} and not any( child.dep_ in {"aux", "auxpass"} for child in root.children ): return False return any(child.dep_ in {"nsubj", "nsubjpass"} for child in root.children) def _subtree_text(doc, indices: set[int]) -> str: ordered = sorted(indices) if not ordered: return "" return doc[ordered[0] : ordered[-1] + 1].text.strip() @lru_cache(maxsize=4096) def try_split_coordinated_verbs( text: str, *, min_words: int = 16, ) -> str | None: """Break a shared-subject verb list into two sentences. ``S can establish A, gain B, and encourage C`` becomes ``S can establish A and gain B. S can also encourage C``. The subject and auxiliary are repeated rather than dropped, which is what turns the tail into a real sentence instead of an imperative fragment. ``also`` carries the additive sense of the ``and`` that the split removes; no other wording is introduced. """ source = (text or "").strip() if not source or _QUOTE.search(source): return None if len(_WORD.findall(source)) < min_words: return None nlp = get_nlp() if nlp is None: return None try: doc = nlp(source) except Exception: return None root = next( (token for token in doc if token.dep_ == "ROOT" and token.pos_ in {"VERB", "AUX"}), None, ) if root is None: return None subject = next( (child for child in root.children if child.dep_ in {"nsubj", "nsubjpass"}), None, ) if subject is None: return None # spaCy chains a verb list rather than attaching each verb to the root: # establish -> conj gain -> conj encourage. conjuncts: list = [] frontier = root while True: nxt = next( ( child for child in frontier.children if child.dep_ == "conj" and child.pos_ in {"VERB", "AUX"} ), None, ) if nxt is None: break conjuncts.append(nxt) frontier = nxt if len(conjuncts) < 2: # A single conjunct is often a misparse of a relative clause; requiring a # real list keeps this off ambiguous sentences. return None last = conjuncts[-1] # "Customers appreciate organizations that respond, resolve, and treat …" # parses the relative-clause verbs as conjuncts of the matrix verb, so a # split would hand them the wrong subject. Ambiguous: leave it alone. if any( token.dep_ == "relcl" and token.i < last.i for token in doc ): return None # Every conjunct must lean on the shared subject; its own subject means the # clause is already independent and belongs to plain splitting. if any( grand.dep_ in {"nsubj", "nsubjpass"} for conj in conjuncts for grand in conj.children ): return None tail_indices = {token.i for token in last.subtree} if min(tail_indices) <= max(token.i for token in subject.subtree): return None if len(_WORD.findall(_subtree_text(doc, tail_indices))) < 3: return None head_indices = { token.i for token in doc if token.i not in tail_indices and not token.is_punct or token.text in {","} } head_indices = {index for index in head_indices if index not in tail_indices} head = _subtree_text(doc, head_indices) head = re.sub(r"[\s,]*\b(and|or)\s*$", "", head).strip() head = head.rstrip(" ,;") if len(_WORD.findall(head)) < 5: return None # Removing the final list item leaves the remaining two joined by a comma # ("establish a reputation, gain retention"), so restore the coordinator. if len(conjuncts) == 2: previous = doc[conjuncts[-2].i - 1] if previous.text == ",": offset = previous.idx - doc[min(head_indices)].idx if 0 <= offset < len(head) and head[offset] == ",": head = f"{head[:offset]} and{head[offset + 1:]}" auxiliaries = " ".join( child.text for child in sorted(root.children, key=lambda t: t.i) if child.dep_ in {"aux", "auxpass"} ) subject_text = _subtree_text(doc, {token.i for token in subject.subtree}) if not subject_text: return None subject_text = subject_text[:1].upper() + subject_text[1:] tail = _subtree_text(doc, tail_indices).lstrip(", ") parts = [subject_text, auxiliaries, "also", tail] second = " ".join(part for part in parts if part).strip() terminal = source[-1] if source.endswith(("!", "?")) else "." candidate = f"{head}{terminal} {second.rstrip('.!?')}{terminal}" if candidate.lower().rstrip(".!?") == source.lower().rstrip(".!?"): return None if not _is_independent_clause(head) or not _is_independent_clause(second): return None return candidate @lru_cache(maxsize=4096) def try_split_long_sentence( text: str, *, min_words: int = 16, ) -> str | None: """Split one long sentence into two on a clause join when safe. Returns a two-sentence string, or None when no safe split exists. """ source = (text or "").strip() if not source: return None words = _WORD.findall(source) if len(words) < min_words: return None if _QUOTE.search(source): return None for sep in _SEPARATORS: if sep not in source: continue left, right = source.split(sep, 1) left, right = left.strip(), right.strip() if len(_WORD.findall(left)) < 5 or len(_WORD.findall(right)) < 5: continue if not _is_independent_clause(left) or not _is_independent_clause(right): continue if right and right[0].islower(): right = right[0].upper() + right[1:] if not left.endswith((".", "!", "?")): left += "." if not right.endswith((".", "!", "?")): right += "." candidate = f"{left} {right}" if candidate.lower().rstrip(".!?") == source.lower().rstrip(".!?"): continue return candidate return None