runtime/core.py

"""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