text_analytics.py

# WriterAgent - AI Writing Assistant for LibreOffice
# Copyright (c) 2026 KeithCu (modifications and relicensing)
#
# SPDX-License-Identifier: GPL-3.0-or-later
"""Trusted venv text analytics compute — runs in user venv worker."""

from __future__ import annotations

from collections import Counter
from functools import lru_cache
import importlib.metadata
import importlib.util
import re
from typing import Any, cast

_LANG_MODELS: dict[str, tuple[str, ...]] = {
    "en": ("en_core_web_sm", "en_core_web_md", "en_core_web_lg"),
    "de": ("de_core_news_sm", "de_core_news_md", "de_core_news_lg"),
    "fr": ("fr_core_news_sm", "fr_core_news_md", "fr_core_news_lg"),
    "es": ("es_core_news_sm", "es_core_news_md", "es_core_news_lg"),
    "it": ("it_core_news_sm", "it_core_news_md", "it_core_news_lg"),
    "pt": ("pt_core_news_sm", "pt_core_news_md", "pt_core_news_lg"),
    "ru": ("ru_core_news_sm", "ru_core_news_md", "ru_core_news_lg"),
    "zh": ("zh_core_web_sm", "zh_core_web_md", "zh_core_web_lg"),
    "ja": ("ja_core_news_sm", "ja_core_news_md", "ja_core_news_lg"),
    "nl": ("nl_core_news_sm", "nl_core_news_md", "nl_core_news_lg"),
    "ko": ("ko_core_news_sm", "ko_core_news_md", "ko_core_news_lg"),
    "pl": ("pl_core_news_sm", "pl_core_news_md", "pl_core_news_lg"),
    "ca": ("ca_core_news_sm", "ca_core_news_md", "ca_core_news_lg"),
    "da": ("da_core_news_sm", "da_core_news_md", "da_core_news_lg"),
    "el": ("el_core_news_sm", "el_core_news_md", "el_core_news_lg"),
    "fi": ("fi_core_news_sm", "fi_core_news_md", "fi_core_news_lg"),
    "hr": ("hr_core_news_sm", "hr_core_news_md", "hr_core_news_lg"),
    "lt": ("lt_core_news_sm", "lt_core_news_md", "lt_core_news_lg"),
    "mk": ("mk_core_news_sm",),
    "nb": ("nb_core_news_sm", "nb_core_news_md", "nb_core_news_lg"),
    "ro": ("ro_core_news_sm", "ro_core_news_md", "ro_core_news_lg"),
    "sl": ("sl_core_news_sm", "sl_core_news_md", "sl_core_news_lg"),
    "sv": ("sv_core_news_sm", "sv_core_news_md", "sv_core_news_lg"),
    "uk": ("uk_core_news_sm", "uk_core_news_md", "uk_core_news_lg"),
}

_READABILITY_KEYS = frozenset({
    "flesch_reading_ease", "flesch_kincaid_grade", "smog",
    "gunning_fog", "automated_readability_index", "coleman_liau_index",
    "lix", "rix",
})

_DESCRIPTIVE_KEYS = frozenset({
    "proportion_unique_tokens",
    "token_length_mean", "token_length_median", "token_length_std",
    "sentence_length_mean", "sentence_length_median", "sentence_length_std",
    "syllables_per_token_mean", "syllables_per_token_median", "syllables_per_token_std",
})

_INSTALL_PACKAGES = ("spacy", "textdescriptives", "transformers", "torch", "scikit-learn")
_MAX_ANALYZED_CHARS = 1_000_000  # spaCy default nlp.max_length


def _lang_prefix(lang: Any) -> str | None:
    if not isinstance(lang, str):
        return None
    prefix = lang.lower().strip()[:2]
    return prefix or None


@lru_cache(maxsize=1)
def _sentiment_pipeline(model: str) -> Any:
    """Load or retrieve the cached sentiment pipeline for *model*.

    The worker is warm and handles one text call at a time; store only
    the last successful pipeline; do not store failures.
    """
    # Cache pipeline for model string across calls within warm worker.
    from transformers import pipeline

    return pipeline("sentiment-analysis", model=model)


def _model_candidates(lang: Any) -> list[str]:
    # xx_sent_ud_sm is tried once; xx_ent_wiki_sm stays last (NER-only).
    code = _lang_prefix(lang)
    return [*_LANG_MODELS.get(code, ()), "xx_sent_ud_sm", "xx_ent_wiki_sm"]


def _load_nlp(lang: str | None = None) -> Any:
    """Load the best available spaCy pipeline for the requested language.

    Strategy for high quality + multilingual support:
    - Prefer a language-specific model when we know the language (better accuracy).
    - Fall back to strong multilingual models that include senter/tagger/parser
      (required for good readability and linguistic metrics via textdescriptives).
    - `xx_sent_ud_sm` is an excellent multilingual base for 100+ languages.
    """
    import spacy

    candidates = _model_candidates(lang)

    last_err: Exception | None = None
    for model_name in candidates:
        try:
            nlp = spacy.load(model_name)
            # Add textdescriptives components for readability and stats.
            # In textdescriptives >=2 the factories are namespaced (e.g. textdescriptives/readability).
            # Adding the specific ones we need ensures the extensions are registered.
            try:
                import textdescriptives as td  # noqa: F401
                for comp in ("textdescriptives/descriptive_stats", "textdescriptives/readability"):
                    if comp not in nlp.pipe_names:
                        nlp.add_pipe(comp)
            except Exception:
                # textdescriptives not installed — we can still do entities + chunks.
                pass  # nosec B110

            return nlp
        except Exception as e:
            last_err = e

    raise RuntimeError(
        f"Could not load any suitable spaCy model. Tried: {candidates}. "
        f"Last error: {last_err}. "
        "Install a model with `python -m spacy download xx_sent_ud_sm` (or a language-specific one)."
    ) from last_err


def _get_lang_from_context(context: dict[str, Any] | None) -> str | None:
    if not context:
        return None
    # Common places we might receive language info from the host.
    for key in ("lang", "language", "locale", "doc_lang"):
        val = context.get(key)
        if isinstance(val, str) and val.strip():
            return val.strip()
    return None


# ---------------------------------------------------------------------------
# Core high-quality analyses
# ---------------------------------------------------------------------------

def _nlp_result(text: str, lang: str | None, context: dict[str, Any] | None, extract: Any) -> dict[str, Any]:
    """Run spaCy once and attach meta. Narrow helpers share this so they do not call nlp(text) themselves."""
    if lang is None:
        lang = _get_lang_from_context(context)
    nlp = _load_nlp(lang)
    # Cap input at _MAX_ANALYZED_CHARS to bound work and stay below spaCy's max_length.
    truncated: dict[str, Any] | None = None
    if len(text) > _MAX_ANALYZED_CHARS:
        truncated = {
            "analyzed_chars": _MAX_ANALYZED_CHARS,
            "total_chars": len(text),
        }
        text = text[:_MAX_ANALYZED_CHARS]
    doc = nlp(text)
    res = extract(doc)
    res["meta"] = _extract_meta(nlp, doc, lang)
    if truncated:
        res["truncated"] = truncated
    return {"status": "ok", "result": res}


def _extract_meta(nlp: Any, doc: Any, lang: str | None) -> dict[str, Any]:
    # Prefer doc.lang_ before falling back to requested lang or 'unknown'.
    return {
        "model": nlp.meta.get("name") or getattr(nlp, "path", None) or "unknown",
        "lang": getattr(doc, "lang_", None) or (lang or "unknown"),
        "has_textdescriptives": any(p.startswith("textdescriptives/") for p in nlp.pipe_names),
    }


def _sentence_count(doc: Any) -> int:
    # Only read doc.sents when SENT_START is annotated (xx_ent_wiki_sm omits it).
    if doc.has_annotation("SENT_START"):
        return len(list(doc.sents))
    return 0


def _extract_readability(doc: Any) -> tuple[dict[str, Any], dict[str, Any]]:
    """Return (readability_dict, descriptive_stats_dict) from textdescriptives (or basic fallback).

    textdescriptives >=2 returns a flat dict (after the components have run on the doc).
    We split it into the groups the rest of the code expects.
    """
    # Share _READABILITY_KEYS and _DESCRIPTIVE_KEYS across helpers.
    try:
        import textdescriptives as td

        m = cast("Any", td.extract_dict(doc))
        if isinstance(m, list):
            m = m[0]
        rd = {k: v for k, v in m.items() if k in _READABILITY_KEYS}
        ds = {k: v for k, v in m.items() if k in _DESCRIPTIVE_KEYS or k.startswith("n_")}
        # Return (readability_dict, descriptive_stats_dict) without unused empty extra groups.
        return rd, ds
    except Exception:
        # textdescriptives not available or failed to extract — fall back to basic spaCy stats.
        return {}, {"n_tokens": len(doc), "n_sentences": _sentence_count(doc)}


def _extract_entities(doc: Any) -> list[dict[str, Any]]:
    return [
        {"text": ent.text, "label": ent.label_, "start_char": ent.start_char, "end_char": ent.end_char}
        for ent in doc.ents
    ]


def _extract_key_phrases(doc: Any) -> list[dict[str, Any]]:
    # Guard noun_chunks with has_annotation("DEP") for NER-only models.
    if not doc.has_annotation("DEP"):
        return []
    try:
        chunks = list(doc.noun_chunks)
    except (NotImplementedError, ValueError):
        return []

    seen: set[str] = set()
    key_phrases = []
    for chunk in chunks:
        lemma = " ".join([t.lemma_ if t.lemma_ != "-PRON-" else t.text.lower() for t in chunk])
        key = lemma.lower().strip()
        if key and key not in seen and len(key) > 1:
            seen.add(key)
            key_phrases.append({"text": chunk.text, "lemma": lemma, "root": chunk.root.text if chunk.root else None})
    return key_phrases[:25]


def _as_sections(text: str | list[str]) -> tuple[list[str], bool]:
    """Normalize a string or list of section texts to (non-empty sections, is_multi)."""
    if isinstance(text, list):
        return [str(t).strip() for t in text if str(t).strip()], True
    return ([str(text).strip()] if str(text or "").strip() else []), False


def _extract_topics(text: str | list[str], n_topics: Any = 4) -> dict[str, Any]:
    """Topic modeling via TF-IDF + NMF (from scikit-learn).

    Accepts either a single string (whole document) or list[str] of section texts.
    When given sections, also returns per-section dominant topic assignments.

    This is intentionally simple and dependency-light on the scientific stack.
    No spaCy required for this helper.
    """
    try:
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.decomposition import NMF
    except ImportError:
        # Signal clearly; callers (dialog, insert, scripts) will surface a helpful message.
        return {
            "topics": [],
            "error": "MISSING_PACKAGE",
            "install_hint": "scikit-learn (part of the Data Analysis / EDA stack)",
            "install": "uv pip install scikit-learn",
        }

    docs, is_multi_section = _as_sections(text)

    if not docs:
        return {"topics": [], "note": "no text"}

    # Filter short sections, cap n_topics by filtered count, and cap components by feature count.
    filtered = [(idx, d) for idx, d in enumerate(docs) if len(d.split()) >= 8]
    if not filtered:
        filtered = [(0, docs[0])]

    try:
        # Route n_topics coercion errors into the topic_model_failed envelope.
        n_topics = max(1, min(int(n_topics or 4), len(filtered), 8))

        # Multilingual-friendly: no hardcoded English stop_words (users can have mixed docs).
        # ngram up to 2 helps with short technical phrases.
        vectorizer = TfidfVectorizer(
            max_features=1200,
            ngram_range=(1, 2),
            min_df=1,
            strip_accents="unicode",
        )
        X = vectorizer.fit_transform([d for _, d in filtered])
        if X.shape[1] < 2:
            return {"topics": [], "note": "insufficient distinct terms"}

        n_components = min(n_topics, X.shape[1])
        nmf = NMF(n_components=n_components, random_state=42, init="nndsvd", max_iter=300, tol=1e-4)
        W = nmf.fit_transform(X)  # (n_docs, n_topics) weights
        H = nmf.components_       # (n_topics, n_terms)

        feature_names = vectorizer.get_feature_names_out()

        topics = []
        for i in range(n_components):
            top_idx = H[i].argsort()[-7:][::-1]
            terms = [str(feature_names[j]) for j in top_idx]
            total = float(W[:, i].sum() + 1e-9)
            topics.append({
                "id": i,
                "terms": terms,
                "weight": round(total / (W.sum() + 1e-9), 3),
            })

        result: dict[str, Any] = {"topics": topics}

        if is_multi_section:
            assignments = []
            for row_idx, row in enumerate(W):
                dom = int(row.argmax())
                strength = float(row[dom])
                assignments.append({
                    "section_index": filtered[row_idx][0],
                    "dominant_topic": dom,
                    "strength": round(strength, 3),
                })
            result["assignments"] = assignments
            result["n_sections"] = len(docs)

        return result
    except Exception as exc:
        # Don't crash the whole analysis; surface what happened.
        return {"topics": [], "error": f"topic_model_failed: {exc}"}


_STAR_RE = re.compile(r"^([1-5])\s*stars?$")
_LABEL_N_RE = re.compile(r"^label_(\d+)$")


def _polarity_from_class_index(index: int, num_labels: int) -> str:
    if num_labels < 2 or not (0 <= index < num_labels):
        return "neutral"
    midpoint = (num_labels - 1) / 2
    if index < midpoint:
        return "negative"
    if index > midpoint:
        return "positive"
    return "neutral"


def _normalize_sentiment_label(label: str, num_labels: int | None = None) -> str:
    cleaned = str(label or "").lower().strip()
    if cleaned in ("pos", "positive", "1", "5"):
        return "positive"
    if cleaned in ("neg", "negative", "0"):
        return "negative"

    star_match = _STAR_RE.match(cleaned)
    if star_match:
        star = int(star_match.group(1))
        return _polarity_from_class_index(star - 1, 5)

    label_match = _LABEL_N_RE.match(cleaned)
    if label_match and isinstance(num_labels, int):
        return _polarity_from_class_index(int(label_match.group(1)), num_labels)
    return "neutral"


def _pipeline_num_labels(clf: Any) -> int | None:
    model = getattr(clf, "model", None)
    config = getattr(model, "config", None)
    num_labels = getattr(config, "num_labels", None)
    if isinstance(num_labels, int) and num_labels > 1:
        return num_labels
    id2label = getattr(config, "id2label", None)
    if isinstance(id2label, dict) and len(id2label) > 1:
        return len(id2label)
    return None


def _extract_sentiment(text: str | list[str], params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Sentiment analysis using transformers + a strong multilingual model (XLM-RoBERTa based).

    Default model provides good cross-lingual performance across many languages.
    This replaced the previous spacytextblob implementation (which had limited multilingual support).

    Returns overall score/label + per-section results when a list of sections is passed
    (reusing the same section extraction as topics for "by section" analysis on Writer docs).

    Config override via text_analytics_sentiment_model (JSON setting) for future flexibility.
    Only the transformers engine is implemented.
    """
    params = params or {}
    sections, is_multi = _as_sections(text)

    if not sections:
        return {"overall": {"score": 0.0, "label": "neutral"}, "note": "no text"}

    model = params.get("model") or "cardiffnlp/twitter-xlm-roberta-base-sentiment"

    try:
        import transformers
    except Exception as exc:
        return {
            "overall": {"score": 0.0, "label": "neutral"},
            "error": "MISSING_PACKAGE",
            "install_hint": "transformers (with CPU torch) for multilingual sentiment",
            "install": "uv pip install transformers torch --index-url https://download.pytorch.org/whl/cpu",
            "detail": f"Could not import transformers: {str(exc)}",
        }

    try:
        clf = _sentiment_pipeline(model)
    except Exception as exc:
        return {
            "overall": {"score": 0.0, "label": "neutral"},
            "error": "SENTIMENT_MODEL_LOAD_FAILED",
            "install_hint": f"Failed to load model '{model}'. This may be a download, cache, network, compatibility, or missing tokenizer library issue (e.g. sentencepiece). Not just the base package.",
            "install": "uv pip install transformers torch sentencepiece  (and ensure network access for first model download)",
            "detail": str(exc),
        }

    num_labels = _pipeline_num_labels(clf)
    per_section: list[dict[str, Any]] = []
    scores: list[float] = []

    for i, sec in enumerate(sections):
        # Truncate conservatively for transformer limits; sections from headings are usually reasonable.
        text_for_clf = sec[:512]
        res = clf(text_for_clf)[0]
        # HF returns label like 'positive'/'negative'/'neutral' and score 0-1.
        raw_label = str(res.get("label", "neutral"))
        sc = float(res.get("score", 0.0))
        label = _normalize_sentiment_label(raw_label, num_labels)
        # Normalize LABEL_n / star ratings with num_labels to a signed polarity score.
        signed = sc if label == "positive" else (-sc if label == "negative" else 0.0)
        scores.append(signed)
        per_section.append({
            "section_index": i,
            "score": round(signed, 3),
            "label": label,
        })

    if is_multi and scores:
        avg_score = sum(scores) / len(scores)
        # Derive overall label from avg.
        if avg_score > 0.15:
            overall_label = "positive"
        elif avg_score < -0.15:
            overall_label = "negative"
        else:
            overall_label = "neutral"
        overall = {
            "score": round(avg_score, 3),
            "label": overall_label,
        }
    else:
        # Single-string sentiment puts section_index only in per_section, not overall.
        overall = {"score": per_section[0]["score"], "label": per_section[0]["label"]}

    res: dict[str, Any] = {"overall": overall}
    if is_multi:
        res["per_section"] = per_section
        res["n_sections"] = len(per_section)
    return res


def analyze_text(text: str, *, lang: str | None = None, context: dict[str, Any] | None = None) -> dict[str, Any]:
    """Run a rich multilingual analysis on a single text using spaCy + textdescriptives.

    Returns a dict with:
      - readability (via textdescriptives when available)
      - descriptive_stats
      - entities
      - key_phrases (from noun chunks)
      - linguistic_profile (pos proportions, complexity signals)
      - meta (model used, language, etc.)
    """
    if not text or not text.strip():
        return {"status": "ok", "result": {}, "note": "empty text"}

    def _full(doc: Any) -> dict[str, Any]:
        rd, ds = _extract_readability(doc)
        # --- Linguistic profile (always useful) ---
        # Skip empty pos_ tags (some models leave token.pos_ empty).
        pos_counts = Counter(
            token.pos_ for token in doc
            if token.pos_ and not (token.is_punct or token.is_space)
        )
        total = sum(pos_counts.values()) or 1
        pos_props = {k: round(v / total, 4) for k, v in pos_counts.items()}
        return {
            "readability": rd,
            "descriptive_stats": ds,
            "entities": _extract_entities(doc),
            "key_phrases": _extract_key_phrases(doc),
            "linguistic_profile": {
                "pos_proportions": pos_props,
                "n_tokens": len(doc),
                "n_sents": _sentence_count(doc),
            },
        }

    return _nlp_result(text, lang, context, _full)


def _probe_package(package: str) -> tuple[bool, str | None]:
    """Presence and version from package metadata; never imports the package."""
    try:
        return True, importlib.metadata.version(package)
    except importlib.metadata.PackageNotFoundError:
        return False, None


def check_diagnostics(lang: Any = None) -> dict[str, Any]:
    """Perform self-diagnostics of the packages needed by Text Analytics features.

    This runs *inside the worker* using whatever python the extension resolved
    (based on Settings → Python → python_venv_path, or falling back to LO's python).

    Reports on:
      - spaCy + models (for entities, key phrases, basic stats, language)
      - textdescriptives (for real Readability scores)
      - transformers + torch (for the Sentiment feature using a good multilingual model)
      - scikit-learn (for topics)

    The output includes the python executable being used and the exact install command.
    """
    # Always capture the python the worker is actually using.
    import sys
    python_exe = sys.executable

    # _lang_prefix treats non-string lang as None to avoid AttributeError.
    lang = _lang_prefix(lang)
    recommended_model = _LANG_MODELS.get(lang, ("xx_sent_ud_sm",))[0]

    # Full command needed for everything the dialog supports.
    text_analytics_install = (
        "uv pip install " + " ".join(_INSTALL_PACKAGES)
        + f" && python -m spacy download {recommended_model}"
    )

    try:
        import spacy
        has_td = importlib.util.find_spec("textdescriptives") is not None
        models = list(spacy.util.get_installed_models())

        # Transformers + torch (for Sentiment). Probed independently: torch used
        # to sit inside the transformers import, so a missing transformers
        # reported torch missing too.
        has_transformers, transformers_version = _probe_package("transformers")
        has_torch, torch_version = _probe_package("torch")

        # scikit-learn (for topics). Metadata name, not the "sklearn" import name.
        has_sklearn, sklearn_version = _probe_package("scikit-learn")

        return {
            "status": "ok",
            "python_used_by_worker": python_exe,
            "spacy_version": spacy.__version__,
            "has_textdescriptives": has_td,
            "models": models,
            "has_transformers": has_transformers,
            "transformers_version": transformers_version,
            "has_torch": has_torch,
            "torch_version": torch_version,
            "has_sklearn": has_sklearn,
            "sklearn_version": sklearn_version,
            "text_analytics_install": text_analytics_install,
        }
    except Exception as e:
        return {
            "status": "error",
            "python_used_by_worker": python_exe,
            "message": str(e),
            "text_analytics_install": text_analytics_install,
        }


def _text_analytics_error(helper: str, message: str, result: dict[str, Any]) -> dict[str, Any]:
    return {
        "status": "error",
        "code": "TEXT_ANALYTICS_ERROR",
        "helper": helper,
        "message": message,
        "result": result,
    }


def dispatch_text_analytics(
    helper: str,
    text: Any,
    params: dict[str, Any],
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Execute text analytics helper or alias on *text*."""
    lang = params.get("lang") or (context or {}).get("lang")
    if helper in ("diagnostics", "check"):
        return {"status": "ok", "helper": helper, "result": check_diagnostics(lang=lang)}

    # Preserve original list form for section-aware helpers (topics, sentiment).
    # Only join for helpers that expect a single string.
    is_list = isinstance(text, list)
    flat = "\n\n".join(str(t) for t in text if t) if is_list else str(text or "")
    if not flat.strip():
        return {"status": "ok", "result": {}, "note": "no text provided"}

    # For the minimal dialog we mainly use "full" or specific helpers.
    if helper in ("full", "analyze", "all"):
        return analyze_text(flat, lang=lang, context=context)

    # Narrow helpers: run nlp(text) once and extract only the requested subset,
    # skipping the unneeded NER / noun-chunk / textdescriptives work.
    if helper in ("readability", "stats", "descriptive"):
        def _readability_body(doc: Any) -> dict[str, Any]:
            rd, ds = _extract_readability(doc)
            return {"readability": rd, "descriptive_stats": ds}
        return _nlp_result(flat, lang, context, _readability_body)

    if helper in ("entities", "ner"):
        return _nlp_result(flat, lang, context, lambda doc: {"entities": _extract_entities(doc)})

    if helper in ("key_phrases", "keyphrases", "chunks"):
        return _nlp_result(flat, lang, context, lambda doc: {"key_phrases": _extract_key_phrases(doc)})

    if helper in ("topics", "topic", "topic_model"):
        # Use preserved original (may still be list) for per-section support.
        topic_data = _extract_topics(text, n_topics=params.get("n_topics", 4))
        # Normalize into the same result envelope the rest of the system expects.
        if topic_data.get("error") == "MISSING_PACKAGE":
            return _text_analytics_error(helper, "install scikit-learn for topics", topic_data)
        # Report actual topic count in meta["n_topics"] and section count in meta["n_sections"].
        topics = topic_data.get("topics", [])
        meta: dict[str, Any] = {"n_topics": len(topics)}
        if isinstance(topic_data.get("n_sections"), int):
            meta["n_sections"] = topic_data["n_sections"]

        res_dict: dict[str, Any] = {
            "topics": topics,
            "assignments": topic_data.get("assignments"),
            "meta": meta,
        }
        res_dict.update({k: topic_data[k] for k in ("error", "note") if k in topic_data})

        return {"status": "ok", "result": res_dict}

    if helper in ("sentiment", "polarity"):
        # Pass params so model can be overridden via JSON config (see WriterAgentConfig).
        sent_data = _extract_sentiment(text, params=params)
        if sent_data.get("error"):
            # A missing model used to be status ok, so the runner reported success.
            return _text_analytics_error(helper, str(sent_data.get("error")), sent_data)
        res = {
            "sentiment": sent_data.get("overall", {}),
            "meta": {"n_sections": sent_data.get("n_sections")}
        }
        if "per_section" in sent_data:
            res["per_section"] = sent_data["per_section"]
        return {"status": "ok", "result": res}

    return {
        "status": "error",
        "code": "UNKNOWN_HELPER",
        "helper": str(helper),
        "message": f"Unknown text analytics helper {helper!r}",
    }


def run_text_analytics(
    spec: dict[str, Any] | str,
    text: str | list[str] | None = None,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Dispatcher used by the trusted worker stub (and by Run Python Script templates).

    `spec` can be a string like "full", "readability", "entities", "key_phrases"
    or a dict {"helper": "...", "params": {...}}.

    `text` is the document text (or list of section texts).
    """
    from plugin.scripting.venv.trusted_dispatch import run_domain

    return run_domain("text", spec, text, context=context)