""" nlp_analysis.py ----------------- Fully local, offline text analysis. No external API calls. Uses: * lexicon matching (see lexicons.py) for sentiment / emotion / domain markers * simple linguistic-complexity statistics * scikit-learn's CountVectorizer purely as an internal tokenizer helper (no network / pretrained weights required) Everything here produces small, explainable numeric features. These features feed into risk_engine.RiskEngine, which is the layer that turns them into depression / anxiety / stress / wellness scores. """ import re from dataclasses import dataclass, field from typing import Dict, List from . import lexicons _WORD_RE = re.compile(r"[a-zA-Z']+") _SENTENCE_RE = re.compile(r"[.!?]+") def _tokenize(text: str) -> List[str]: return [w.lower() for w in _WORD_RE.findall(text)] def _contains_phrase(text_lower: str, phrase: str) -> bool: return phrase in text_lower @dataclass class TextFeatures: word_count: int sentence_count: int avg_sentence_length: float avg_word_length: float type_token_ratio: float # lexical diversity sentiment_score: float # -1 (very negative) to +1 (very positive) positive_hits: List[str] = field(default_factory=list) negative_hits: List[str] = field(default_factory=list) depression_markers: List[str] = field(default_factory=list) anxiety_markers: List[str] = field(default_factory=list) stress_markers: List[str] = field(default_factory=list) crisis_markers: List[str] = field(default_factory=list) absolutist_count: int = 0 negation_count: int = 0 first_person_ratio: float = 0.0 # first-person singular / total words def to_dict(self) -> Dict: return { "word_count": self.word_count, "sentence_count": self.sentence_count, "avg_sentence_length": self.avg_sentence_length, "avg_word_length": self.avg_word_length, "type_token_ratio": self.type_token_ratio, "sentiment_score": self.sentiment_score, "positive_hits": self.positive_hits, "negative_hits": self.negative_hits, "depression_markers": self.depression_markers, "anxiety_markers": self.anxiety_markers, "stress_markers": self.stress_markers, "crisis_markers": self.crisis_markers, "absolutist_count": self.absolutist_count, "negation_count": self.negation_count, "first_person_ratio": self.first_person_ratio, } class TextAnalyzer: """Stateless analyzer for a single message. Aggregate across a session in ConversationManager / RiskEngine.""" def analyze(self, text: str) -> TextFeatures: text = text or "" text_lower = text.lower() tokens = _tokenize(text) word_count = len(tokens) sentences = [s for s in _SENTENCE_RE.split(text) if s.strip()] sentence_count = max(1, len(sentences)) avg_sentence_length = word_count / sentence_count if sentence_count else 0.0 avg_word_length = ( sum(len(w) for w in tokens) / word_count if word_count else 0.0 ) unique_tokens = set(tokens) type_token_ratio = len(unique_tokens) / word_count if word_count else 0.0 pos_hits = sorted(unique_tokens & lexicons.POSITIVE_WORDS) neg_hits = sorted(unique_tokens & lexicons.NEGATIVE_WORDS) sentiment_score = 0.0 if pos_hits or neg_hits: sentiment_score = (len(pos_hits) - len(neg_hits)) / max( 1, len(pos_hits) + len(neg_hits) ) dep_hits = sorted( phrase for phrase in lexicons.DEPRESSION_MARKERS if _contains_phrase(text_lower, phrase) ) anx_hits = sorted( phrase for phrase in lexicons.ANXIETY_MARKERS if _contains_phrase(text_lower, phrase) ) stress_hits = sorted( phrase for phrase in lexicons.STRESS_MARKERS if _contains_phrase(text_lower, phrase) ) crisis_hits = sorted( phrase for phrase in lexicons.CRISIS_MARKERS if _contains_phrase(text_lower, phrase) ) absolutist_count = sum(1 for t in tokens if t in lexicons.ABSOLUTIST_WORDS) negation_count = sum(1 for t in tokens if t in lexicons.NEGATION_WORDS) first_person_count = sum(1 for t in tokens if t in lexicons.FIRST_PERSON_SINGULAR) first_person_ratio = first_person_count / word_count if word_count else 0.0 return TextFeatures( word_count=word_count, sentence_count=sentence_count, avg_sentence_length=round(avg_sentence_length, 2), avg_word_length=round(avg_word_length, 2), type_token_ratio=round(type_token_ratio, 3), sentiment_score=round(sentiment_score, 3), positive_hits=pos_hits, negative_hits=neg_hits, depression_markers=dep_hits, anxiety_markers=anx_hits, stress_markers=stress_hits, crisis_markers=crisis_hits, absolutist_count=absolutist_count, negation_count=negation_count, first_person_ratio=round(first_person_ratio, 3), ) def aggregate(self, feature_list: List[TextFeatures]) -> Dict: """Combine per-message features across a whole session.""" if not feature_list: return { "message_count": 0, "avg_word_count": 0.0, "avg_sentiment": 0.0, "total_depression_markers": 0, "total_anxiety_markers": 0, "total_stress_markers": 0, "total_crisis_markers": 0, "avg_first_person_ratio": 0.0, "avg_absolutist_count": 0.0, "avg_type_token_ratio": 0.0, } n = len(feature_list) return { "message_count": n, "avg_word_count": round(sum(f.word_count for f in feature_list) / n, 2), "avg_sentiment": round(sum(f.sentiment_score for f in feature_list) / n, 3), "total_depression_markers": sum(len(f.depression_markers) for f in feature_list), "total_anxiety_markers": sum(len(f.anxiety_markers) for f in feature_list), "total_stress_markers": sum(len(f.stress_markers) for f in feature_list), "total_crisis_markers": sum(len(f.crisis_markers) for f in feature_list), "avg_first_person_ratio": round(sum(f.first_person_ratio for f in feature_list) / n, 3), "avg_absolutist_count": round(sum(f.absolutist_count for f in feature_list) / n, 3), "avg_type_token_ratio": round(sum(f.type_token_ratio for f in feature_list) / n, 3), }