| """Portable runtime foundation for the NLL deterministic editor.""" |
| from __future__ import annotations |
|
|
| import os |
| import re |
| import tempfile |
| from collections import Counter |
| from dataclasses import dataclass, field |
| from pathlib import Path |
| from typing import Any |
|
|
| import spacy |
|
|
|
|
| @dataclass |
| class Config: |
| batch: int = int(os.getenv("NLL_BATCH_SIZE", "32")) |
| out: str = os.getenv("NLL_OUTPUT_DIR", tempfile.gettempdir()) |
| stanza: bool = False |
| sts: bool = False |
| nli: bool = False |
| mlm: bool = False |
| lt: bool = os.getenv("NLL_ENABLE_LANGUAGETOOL", "false").lower() == "true" |
| online: bool = False |
| min_sts: float = 0.72 |
| min_nli: float = 0.45 |
|
|
| CFG = Config() |
| STATE: dict[str, Any] = {"warnings": [], "start": None} |
| DISCLAIMER = "The Writing Pattern Score describes observable structural and stylistic characteristics only." |
|
|
|
|
| def _load_spacy(): |
| try: |
| return spacy.load("en_core_web_sm") |
| except Exception as exc: |
| if os.getenv("NLL_REQUIRE_SPACY_MODEL", "true").lower() == "true": |
| raise RuntimeError("spaCy model en_core_web_sm is required for deployment.") from exc |
| pipe = spacy.blank("en") |
| if "sentencizer" not in pipe.pipe_names: |
| pipe.add_pipe("sentencizer") |
| return pipe |
|
|
| _NLP = _load_spacy() |
|
|
| def nlp(): |
| return _NLP |
|
|
|
|
| def sent_split(text): |
| value = str(text or "").strip() |
| if not value: |
| return [] |
| try: |
| doc = _NLP(value) |
| parts = [s.text.strip() for s in doc.sents if s.text.strip()] |
| if parts: |
| return parts |
| except Exception: |
| pass |
| return [x.strip() for x in re.split(r"(?<=[.!?])\s+", value) if x.strip()] |
|
|
|
|
| def sub(token): |
| try: |
| items = list(token.subtree) |
| if not items: |
| return token.text |
| start = min(x.i for x in items) |
| end = max(x.i for x in items) + 1 |
| return token.doc[start:end].text |
| except Exception: |
| return getattr(token, "text", str(token)) |
|
|
|
|
| @dataclass |
| class Candidate: |
| text: str |
| lane: str |
| family: str |
| reason: str |
| priority: float = .5 |
| score: float = 0.0 |
| checks: dict = field(default_factory=dict) |
|
|
| @dataclass |
| class Analysis: |
| text: str |
| families: list[str] = field(default_factory=list) |
| enumerations: list[list[str]] = field(default_factory=list) |
| notes: list[str] = field(default_factory=list) |
| metadata: dict = field(default_factory=dict) |
|
|
| @dataclass |
| class Profile: |
| name: str |
| keywords: set[str] = field(default_factory=set) |
| phrases: set[str] = field(default_factory=set) |
| immutable: set[str] = field(default_factory=set) |
|
|
| @dataclass |
| class Finding: |
| sentence: int |
| category: str |
| text: str |
| score: float = 0.0 |
| message: str = "" |
|
|
| @dataclass |
| class SentenceResult: |
| original: str |
| revised: str |
| status: str |
| families: list[str] |
| family: str | None |
| reason: str |
| alternatives: list[str] |
| checks: dict |
| diagnostics: dict |
|
|
| @property |
| def trace(self): |
| return self.diagnostics |
| @trace.setter |
| def trace(self, value): |
| self.diagnostics = value |
|
|
| @dataclass |
| class DocumentResult: |
| original: str |
| revised: str |
| workflow: str |
| mode: str |
| discipline: str |
| evidence: list[str] |
| pattern_score: float |
| band: str |
| findings: list |
| sentences: list[SentenceResult] |
| warnings: list[str] |
| stats: dict |
|
|
|
|
| PROFILES = { |
| "General Academic": Profile( |
| "General Academic", |
| {"study", "research", "analysis", "method", "results", "framework"}, |
| {"this study", "the results"}, |
| set(), |
| ), |
| "Technical": Profile( |
| "Technical", |
| {"system", "service", "handler", "configuration", "deployment", "model"}, |
| {"environment variable", "source code"}, |
| set(), |
| ), |
| "Business": Profile( |
| "Business", |
| {"supplier", "schedule", "invoice", "budget", "manager", "procurement"}, |
| {"implementation schedule", "cost ledger"}, |
| set(), |
| ), |
| "Contract": Profile( |
| "Contract", |
| {"contractor", "shall", "owner", "payment", "reimbursement", "eligible"}, |
| {"for purposes of", "approved work"}, |
| set(), |
| ), |
| } |
|
|
|
|
| def _extract_enumerations(text): |
| groups = [] |
| for match in re.finditer( |
| r"(?is)([^.;:]{1,120}?,\s*[^.;:]{1,120}?(?:,\s*[^.;:]{1,120}?)*,?\s+(?:and|or)\s+[^.;:]{1,120}?)(?=[.;:]|$)", |
| text, |
| ): |
| raw = match.group(1) |
| parts = [x.strip(" ,") for x in re.split(r",\s*|\s+(?:and|or)\s+", raw) if x.strip(" ,")] |
| if len(parts) >= 3: |
| groups.append(parts) |
| return groups |
|
|
|
|
| class Analyzer: |
| def analyze(self, sentence, profile, user_terms): |
| text = str(sentence or "").strip() |
| families = set() |
| enumerations = _extract_enumerations(text) |
| low = text.lower() |
| if enumerations: |
| families.update({"ENUMERATION", "COORDINATED_OBJECTS_OR_ENUMERATION"}) |
| if re.match(r"^(?:Applying|Using|Evaluating|Comparing|Assessing|Examining|Analyzing|Implementing|Testing|Measuring)\b", text, re.I): |
| families.add("GERUND_SUBJECT") |
| if re.search(r"\b(?:is|are|was|were|has been|have been|had been)\s+\w+(?:ed|en)\b", low): |
| families.add("PASSIVE") |
| try: |
| doc = _NLP(text) |
| dep = doc.has_annotation("DEP") |
| except Exception: |
| dep = False |
| return Analysis( |
| text=text, |
| families=sorted(families), |
| enumerations=enumerations, |
| metadata={"dependency_available": dep, "base_analyzer": "portable"}, |
| ) |
|
|
|
|
| class Scanner: |
| def scan(self, text): |
| sentences = sent_split(text) |
| findings = [] |
| for idx, sentence in enumerate(sentences): |
| words = sentence.split() |
| commas = sentence.count(",") |
| if len(words) >= 32 or commas >= 4: |
| findings.append(Finding(idx, "sentence complexity", sentence, .6, "Long or list-dense sentence may be complex.")) |
| score = round(min(100.0, sum((len(s.split()) / 35.0) + (s.count(",") * .45) for s in sentences)), 2) |
| band = "Low" if score < 20 else "Moderate" if score < 50 else "High" |
| features = {"sentences": len(sentences), "long_sentences": sum(len(s.split()) >= 32 for s in sentences)} |
| return findings, score, band, features |
|
|
|
|
| _RESOURCES: dict[str, Any] = {} |
|
|
| def optional(name): |
| if name in {"sts", "nli", "mlm"}: |
| return None |
| if name == "lt": |
| if not CFG.lt: |
| return None |
| if "lt" not in _RESOURCES: |
| try: |
| import language_tool_python |
| _RESOURCES["lt"] = language_tool_python.LanguageTool("en-US") |
| except Exception as exc: |
| STATE.setdefault("warnings", []).append(f"LanguageTool unavailable: {exc}") |
| _RESOURCES["lt"] = None |
| return _RESOURCES["lt"] |
| return None |
|
|
|
|
| def protect(text, hard_terms): |
| value = str(text or "") |
| found = [] |
| patterns = { |
| "url": r"https?://[^\s)\]]+", |
| "citation": r"\b[A-Z][A-Za-z-]+\s+et\s+al\.\s*\(\d{4}[a-z]?\)|\([A-Z][A-Za-z-]+(?:\s+et\s+al\.)?,?\s*\d{4}[a-z]?\)", |
| "quantity": r"(?<!\w)\d+(?:\.\d+)?%?(?!\w)", |
| "equation": r"\b[A-Za-z]\s*=\s*[^,.;]+", |
| "code": r"`[^`]+`", |
| } |
| for kind, pattern in patterns.items(): |
| for match in re.finditer(pattern, value): |
| found.append((kind, match.group(0))) |
| for term in set(hard_terms or set()): |
| if term and re.search(r"(?<!\w)" + re.escape(term) + r"(?!\w)", value, re.I): |
| found.append(("term", term)) |
| return found |
|
|
