| """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
|
|
|
|
|
| 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
|
|
|
|
|
| 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
|
|
|
| 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
|
|
|
| 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
|
|
|
|
|
| if Counter(src_toks) != Counter(cand_toks):
|
| return False
|
| if not _attributive_phrases_preserved(src, cand):
|
| return False
|
|
|
| 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
|
|
|
