148 lines
4.3 KiB
Python
148 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import dataclass
|
|
from typing import Iterable
|
|
|
|
import numpy as np
|
|
from langdetect import DetectorFactory, detect
|
|
from sklearn.cluster import KMeans
|
|
from sklearn.feature_extraction.text import TfidfVectorizer
|
|
|
|
|
|
DetectorFactory.seed = 0
|
|
|
|
SENTENCE_SPLIT = re.compile(r"(?<=[.!?])\s+|\n+")
|
|
WORD_RE = re.compile(r"[A-Za-zÀ-ÿ0-9']+")
|
|
|
|
POSITIVE_WORDS = {
|
|
"support",
|
|
"approve",
|
|
"agree",
|
|
"welcome",
|
|
"benefit",
|
|
"safe",
|
|
"fair",
|
|
"help",
|
|
"yes",
|
|
}
|
|
|
|
NEGATIVE_WORDS = {
|
|
"oppose",
|
|
"reject",
|
|
"harm",
|
|
"unsafe",
|
|
"unfair",
|
|
"no",
|
|
"risk",
|
|
"delay",
|
|
"concern",
|
|
"object",
|
|
}
|
|
|
|
|
|
def detect_language(text: str) -> str:
|
|
sample = text.strip()
|
|
if len(sample) < 10:
|
|
return "und"
|
|
try:
|
|
return detect(sample)
|
|
except Exception:
|
|
return "und"
|
|
|
|
|
|
def tokenize(text: str) -> list[str]:
|
|
return [token.lower() for token in WORD_RE.findall(text)]
|
|
|
|
|
|
def split_sentences(text: str) -> list[str]:
|
|
parts = [part.strip() for part in SENTENCE_SPLIT.split(text) if part.strip()]
|
|
return parts or ([text.strip()] if text.strip() else [])
|
|
|
|
|
|
def sentence_centrality_summary(texts: Iterable[str], max_sentences: int = 3) -> list[str]:
|
|
sentences: list[str] = []
|
|
for text in texts:
|
|
sentences.extend(split_sentences(text))
|
|
|
|
unique_sentences: list[str] = []
|
|
seen = set()
|
|
for sentence in sentences:
|
|
key = sentence.lower()
|
|
if key not in seen:
|
|
seen.add(key)
|
|
unique_sentences.append(sentence)
|
|
|
|
if not unique_sentences:
|
|
return []
|
|
if len(unique_sentences) <= max_sentences:
|
|
return unique_sentences
|
|
|
|
vectorizer = TfidfVectorizer(stop_words="english")
|
|
matrix = vectorizer.fit_transform(unique_sentences)
|
|
similarity = (matrix * matrix.T).toarray()
|
|
scores = similarity.sum(axis=1)
|
|
ranked = sorted(range(len(unique_sentences)), key=lambda idx: (-scores[idx], idx))[:max_sentences]
|
|
return [unique_sentences[idx] for idx in sorted(ranked)]
|
|
|
|
|
|
def summarize_multilingual_comments(comments: list[dict], max_sentences: int = 3) -> dict[str, list[str]]:
|
|
by_language: dict[str, list[str]] = defaultdict(list)
|
|
for comment in comments:
|
|
language = comment.get("language") or detect_language(comment.get("text", ""))
|
|
by_language[language].append(comment.get("text", ""))
|
|
|
|
return {
|
|
language: sentence_centrality_summary(texts, max_sentences=max_sentences)
|
|
for language, texts in sorted(by_language.items())
|
|
}
|
|
|
|
|
|
def cluster_texts(texts: list[str]) -> list[dict]:
|
|
if not texts:
|
|
return []
|
|
if len(texts) == 1:
|
|
return [{"cluster": 0, "items": [texts[0]], "top_terms": tokenize(texts[0])[:5]}]
|
|
|
|
vectorizer = TfidfVectorizer(stop_words="english")
|
|
matrix = vectorizer.fit_transform(texts)
|
|
n_clusters = max(1, min(int(math.sqrt(len(texts))) or 1, 4))
|
|
if n_clusters == 1:
|
|
return [{"cluster": 0, "items": texts, "top_terms": _top_terms(matrix, vectorizer)}]
|
|
|
|
model = KMeans(n_clusters=n_clusters, n_init=10, random_state=0)
|
|
labels = model.fit_predict(matrix)
|
|
clusters: dict[int, list[str]] = defaultdict(list)
|
|
for label, text in zip(labels, texts):
|
|
clusters[int(label)].append(text)
|
|
|
|
return [
|
|
{"cluster": cluster_id, "items": items, "top_terms": _top_terms(matrix[[i for i, label in enumerate(labels) if label == cluster_id]], vectorizer)}
|
|
for cluster_id, items in sorted(clusters.items())
|
|
]
|
|
|
|
|
|
def _top_terms(matrix, vectorizer: TfidfVectorizer, limit: int = 5) -> list[str]:
|
|
if matrix.shape[0] == 0:
|
|
return []
|
|
averaged = np.asarray(matrix.mean(axis=0)).ravel()
|
|
terms = np.array(vectorizer.get_feature_names_out())
|
|
top_indices = averaged.argsort()[::-1][:limit]
|
|
return [str(terms[index]) for index in top_indices if averaged[index] > 0]
|
|
|
|
|
|
def sentiment_score(text: str) -> float:
|
|
tokens = tokenize(text)
|
|
if not tokens:
|
|
return 0.0
|
|
positive = sum(1 for token in tokens if token in POSITIVE_WORDS)
|
|
negative = sum(1 for token in tokens if token in NEGATIVE_WORDS)
|
|
return (positive - negative) / max(1, len(tokens))
|
|
|
|
|
|
def laplace_noise(scale: float, seed: int | None = None) -> float:
|
|
rng = np.random.default_rng(seed)
|
|
return float(rng.laplace(0.0, scale))
|