Repository Public
1"""Corpus-level style fidelity: stylometric feature vectors, distributional distances between the set of2reconstructions and the set of originals (vs the set of rewrites), and a calibrated original-vs-rewrite probe3whose score on reconstructions is reported as exploratory evidence only (never as proof of origin)."""4from __future__ import annotations56import math7import re8from collections import Counter910import numpy as np1112from htr.parse.protect import HEDGE_RE, NEGATION_RE, PLACEHOLDER_RE13from htr.utils.text import split_sentences1415_WORD = re.compile(r"[A-Za-z][A-Za-z'’-]*")16_DISCOURSE = re.compile(r"\b(?:however|moreover|furthermore|therefore|thus|hence|nevertheless|consequently|in addition|in contrast|on the other hand|for example|for instance|in particular|notably|importantly|overall|finally|first|second|third|accordingly|whereas|although|while)\b", re.I)17_PASSIVE = re.compile(r"\b(?:is|are|was|were|be|been|being)\s+(?:\w+ly\s+)?\w+(?:ed|en)\b", re.I)18_NOMINAL = re.compile(r"\b\w+(?:tion|sion|ment|ness|ity|ance|ence)s?\b", re.I)19_FIRST_PERSON = re.compile(r"\b(?:we|our|us|I|my)\b")20_SUBORD = re.compile(r"\b(?:that|which|who|whom|whose|because|since|although|though|whereas|if|unless|until|when|while|after|before|as)\b", re.I)21_LATINATE = re.compile(r"\b\w{10,}\b")22FEATURES = ["mean_sent_len", "sd_sent_len", "ttr", "mtld", "mean_word_len", "long_word_rate", "comma_rate", "semicolon_rate", "paren_rate", "discourse_rate", "passive_rate", "nominal_rate", "first_person_rate", "subord_rate", "hedge_rate", "negation_rate", "digit_rate", "flesch"]232425def _syllables(w: str) -> int:26 w = w.lower()27 groups = re.findall(r"[aeiouy]+", w)28 n = len(groups)29 if w.endswith("e") and n > 1 and not w.endswith(("le", "ee")):30 n -= 131 return max(1, n)323334def mtld(tokens: list[str], ttr_threshold: float = 0.72) -> float:35 def _pass(toks):36 factors, types, count = 0.0, set(), 037 for t in toks:38 count += 139 types.add(t)40 if len(types) / count <= ttr_threshold:41 factors += 142 types, count = set(), 043 if count:44 factors += (1 - len(types) / count) / (1 - ttr_threshold) if count else 045 return len(toks) / factors if factors else float(len(toks))46 if len(tokens) < 10:47 return float(len(set(tokens)))48 return (_pass(tokens) + _pass(tokens[::-1])) / 2495051def features(text: str) -> dict[str, float]:52 text_np = PLACEHOLDER_RE.sub(" ", text)53 sents = split_sentences(text_np) or [text_np]54 words = [w.lower() for w in _WORD.findall(text_np)]55 n_w = max(1, len(words))56 sl = [len(_WORD.findall(s)) for s in sents]57 syl = sum(_syllables(w) for w in words)58 return {59 "mean_sent_len": float(np.mean(sl)), "sd_sent_len": float(np.std(sl)), "ttr": len(set(words)) / n_w, "mtld": mtld(words),60 "mean_word_len": float(np.mean([len(w) for w in words])) if words else 0.0, "long_word_rate": len(_LATINATE.findall(text_np)) / n_w,61 "comma_rate": text_np.count(",") / n_w, "semicolon_rate": text_np.count(";") / n_w, "paren_rate": text_np.count("(") / n_w,62 "discourse_rate": len(_DISCOURSE.findall(text_np)) / n_w, "passive_rate": len(_PASSIVE.findall(text_np)) / max(1, len(sents)),63 "nominal_rate": len(_NOMINAL.findall(text_np)) / n_w, "first_person_rate": len(_FIRST_PERSON.findall(text_np)) / n_w,64 "subord_rate": len(_SUBORD.findall(text_np)) / n_w, "hedge_rate": len(HEDGE_RE.findall(text_np)) / n_w, "negation_rate": len(NEGATION_RE.findall(text_np)) / n_w,65 "digit_rate": len(re.findall(r"\d", text_np)) / max(1, len(text_np)), "flesch": 206.835 - 1.015 * (n_w / max(1, len(sents))) - 84.6 * (syl / n_w),66 }676869def feature_matrix(texts: list[str]) -> np.ndarray:70 return np.array([[features(t)[f] for f in FEATURES] for t in texts], dtype=float)717273def wasserstein_1d(a: np.ndarray, b: np.ndarray) -> float:74 a, b = np.sort(a), np.sort(b)75 q = np.linspace(0, 1, 201)76 return float(np.mean(np.abs(np.quantile(a, q) - np.quantile(b, q))))777879def distribution_distances(X_hyp: np.ndarray, X_ref: np.ndarray, X_scale: np.ndarray | None = None) -> dict[str, float]:80 """Per-feature 1-D Wasserstein distance (standardised by the pooled spread, floored so that near-constant81 features cannot dominate) + mean over features."""82 pool = X_scale if X_scale is not None else np.vstack([X_hyp, X_ref])83 scale = np.maximum(pool.std(0), 0.05 * (np.abs(pool.mean(0)) + 1e-9))84 scale = np.maximum(scale, 1e-6)85 d = {f"W1_{f}": wasserstein_1d(X_hyp[:, i] / scale[i], X_ref[:, i] / scale[i]) for i, f in enumerate(FEATURES)}86 d["W1_mean"] = float(np.mean(list(d.values())))87 return d888990def style_probe(X_orig: np.ndarray, X_rewrite: np.ndarray, X_eval: np.ndarray, seed: int = 0) -> dict[str, float]:91 """Logistic-regression probe (original=1 vs rewrite=0) trained on standardised stylometric features with a held-out92 split for calibration/AUC, then applied to reconstructions: mean P(original) and the fraction classified as93 original. Exploratory: a probe score is evidence about *feature distributions*, not about who wrote a text."""94 rng = np.random.default_rng(seed)95 X = np.vstack([X_orig, X_rewrite])96 y = np.r_[np.ones(len(X_orig)), np.zeros(len(X_rewrite))]97 mu, sd = X.mean(0), X.std(0) + 1e-998 Xs = (X - mu) / sd99 idx = rng.permutation(len(X))100 n_tr = int(0.7 * len(X))101 tr, te = idx[:n_tr], idx[n_tr:]102 w = np.zeros(X.shape[1])103 b = 0.0104 lr, l2 = 0.1, 1e-3105 for _ in range(500): # plain gradient descent on logistic loss106 z = Xs[tr] @ w + b107 p = 1 / (1 + np.exp(-z))108 g = p - y[tr]109 w -= lr * (Xs[tr].T @ g / len(tr) + l2 * w)110 b -= lr * g.mean()111 def predict(Xn):112 return 1 / (1 + np.exp(-((Xn - mu) / sd @ w + b)))113 p_te = predict(X[te])114 auc = _auc(y[te], p_te)115 acc = float(((p_te > 0.5) == y[te]).mean())116 p_eval = predict(X_eval)117 return {"probe_auc_orig_vs_rewrite": auc, "probe_acc_heldout": acc, "eval_mean_p_original": float(p_eval.mean()), "eval_frac_classified_original": float((p_eval > 0.5).mean()), "probe_weights": {f: float(wi) for f, wi in zip(FEATURES, w)}}118119120def _auc(y: np.ndarray, s: np.ndarray) -> float:121 pos, neg = s[y == 1], s[y == 0]122 if len(pos) == 0 or len(neg) == 0:123 return float("nan")124 order = np.argsort(np.r_[pos, neg])125 ranks = np.empty(len(order))126 ranks[order] = np.arange(1, len(order) + 1)127 return float((ranks[: len(pos)].sum() - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg)))128