"""Sentence rewrite generation from plans/templates.""" from __future__ import annotations import re from collections import Counter from app.engine.models import RewritePlan from app.engine.templates import ( fill_template, try_because_front, try_complex_clause_swap, ) from app.engine.voice import active_to_passive from app.pipeline.nlp import get_nlp def _content_tokens(text: str) -> list[str]: return [ w for w in re.findall(r"[a-zA-Z']+", (text or "").lower()) if len(w) >= 3 ] def _attributive_phrases_preserved(source: str, candidate: str) -> bool: """Keep adjective/noun compounds attached (for example 'daily tasks').""" nlp = get_nlp() if nlp is None: return True try: doc = nlp(source) except Exception: return True cand = (candidate or "").lower() for token in doc: if token.dep_ not in {"amod", "compound"}: continue if token.head.pos_ not in {"NOUN", "PROPN"}: continue if token.pos_ not in {"ADJ", "NOUN", "PROPN"}: continue left = f"{token.text} {token.head.text}".lower() right = f"{token.head.text} {token.text}".lower() if left not in cand and right not in cand: return False return True def reorder_quality_ok(source: str, candidate: str) -> bool: """Reject junk reorders that drop clauses or shuffle degree adverbs.""" src = (source or "").strip() cand = (candidate or "").strip() if not src or not cand: return False if cand.lower().rstrip(".!?") == src.lower().rstrip(".!?"): return False # Preserve short comma-delimited sentence openers (for example discourse # markers) unless a dedicated clause template handles them. opener = re.match(r"^\s*([^,]{1,24}),", src) if opener and len(opener.group(1).split()) <= 2: expected = opener.group(1).strip().lower() + "," if not cand.lower().startswith(expected): return False # Reject fronting an internal attributive word such as "daily" from # "planning daily tasks" into "Daily, planning tasks ...". fronted = re.match(r"^\s*([A-Za-z]+),\s+", cand) if fronted: word = fronted.group(1) if not re.match(rf"^{re.escape(word)}\b", src, flags=re.I): nlp = get_nlp() if nlp is not None: try: source_doc = nlp(src) token = next( ( item for item in source_doc if item.text.lower() == word.lower() ), None, ) if ( token is not None and token.dep_ in {"amod", "compound"} and token.head.pos_ in {"NOUN", "PROPN"} ): return False except Exception: pass if re.search( r"\b(more|most|less|least|very|really|quite)\s+" r"(makes?|make|is|are|was|were|has|have|had|does|do|did)\b", cand, flags=re.I, ): return False # Reject reorders that split degree+adverb pairs such as "more efficiently". for match in re.finditer( r"\b(more|most|less|least)\s+(efficiently|effectively|carefully|" r"quickly|slowly|easily|clearly|accurately)\b", src, flags=re.I, ): phrase = match.group(0).lower() if phrase not in cand.lower(): return False # Duration fronting that strands "for/over": "At least thirty minutes, … for every day" if re.search( r"^(at\s+least\s+)?\d*\s*(thirty|forty|fifty|\d+)\s+" r"(minutes?|hours?|seconds?)\s*,", cand, flags=re.I, ): return False if re.search(r"\b(for|over|within|during)\s+(every|each)\b", cand, flags=re.I): if re.search(r"\b(for|over|within|during)\s+\S+\s+(minutes?|hours?)\b", src, flags=re.I): return False if re.search(r",\s+(for|over|within|during)\s+\w+\s*$", cand, flags=re.I): return False src_toks = _content_tokens(src) cand_toks = _content_tokens(cand) if not src_toks: return False # Slot templates reorder existing words; additions, omissions, or # duplicate degree words indicate a bad parse. if Counter(src_toks) != Counter(cand_toks): return False if not _attributive_phrases_preserved(src, cand): return False # Soften keep ratio slightly for structural adverb fronts that drop discourse ADVs (also) keep = len(set(src_toks) & set(cand_toks)) / max(1, len(set(src_toks))) if keep < 0.80: return False if len(cand_toks) < int(len(src_toks) * 0.75) or len(cand_toks) > int( len(src_toks) * 1.35 ): return False return True def generate_from_plan( plan: RewritePlan, *, template_id: str | None = None, ) -> str | None: """Generate a rewritten sentence from a plan (slot reorder only).""" if not plan.safe: return None tid = template_id or plan.template_id raw = plan.slots.text if plan.slots else "" if tid == "because_front": return try_because_front(raw) if tid == "complex_clause_swap": return try_complex_clause_swap(raw) if tid == "active_to_passive": return active_to_passive(raw) if plan.slots is None: return None filled = fill_template(tid, plan.slots) if not filled: return None if filled.strip().lower().rstrip(".!?") == raw.lower().rstrip(".!?"): return None if not reorder_quality_ok(raw, filled): return None return filled.strip() def generate_candidates(plan: RewritePlan) -> list[tuple[str, str, float]]: """Try each ranked template; return (template_id, text, confidence).""" results: list[tuple[str, str, float]] = [] if not plan.safe: return results for cand in plan.candidates: text = generate_from_plan(plan, template_id=cand.template_id) if text: results.append((cand.template_id, text, cand.confidence)) return results