| from __future__ import annotations |
|
|
| import json |
| import logging |
| import math |
| import os |
| import re |
| import httpx |
| import asyncio |
| from collections import Counter, defaultdict |
| from dataclasses import dataclass |
| from datetime import datetime, timezone |
| from functools import lru_cache |
| from pathlib import Path |
| from statistics import fmean |
| from typing import Iterable |
|
|
| import joblib |
| import pandas as pd |
| import spacy |
| from spacy.tokens import Doc, Span |
| from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer |
|
|
| from .schemas import ( |
| AnalysisMetadata, |
| AnalysisResponse, |
| ConversionScore, |
| HighlightRange, |
| PipelineStage, |
| ProductSentiment, |
| SentimentCounts, |
| SentimentSummary, |
| ) |
|
|
| logger = logging.getLogger(__name__) |
|
|
| SPACE_RX = re.compile(r"\s+") |
| EDGE_PUNCT_RX = re.compile(r"^[^A-Za-z0-9]+|[^A-Za-z0-9]+$") |
| ALPHA_RX = re.compile(r"[A-Za-z]") |
| DIGIT_RX = re.compile(r"\d") |
| BUDGET_RX = re.compile(r"(?i)^(?:rs\.?|inr|₹|\$|€|£)?\s*\d[\d,]*(?:\.\d+)?\s*(?:[kKmMlL]|lakh|lakhs|cr|crore|crores|bucks|ks)?$") |
| SENTENCE_SPLIT_RX = re.compile(r"(?<=[.!?])\s+") |
|
|
| GENERIC_ASPECT_TERMS = { |
| "app", |
| "apps", |
| "brand", |
| "brands", |
| "call", |
| "conversation", |
| "experience", |
| "feature", |
| "features", |
| "issue", |
| "issues", |
| "item", |
| "items", |
| "people", |
| "person", |
| "product", |
| "products", |
| "service", |
| "services", |
| "something", |
| "stuff", |
| "team", |
| "thing", |
| "things", |
| "today", |
| } |
| CLAUSE_BREAKERS = {"but", "however", "though", "although", "yet", "while"} |
| VALID_ENTITY_LABELS = {"PRODUCT", "BRAND", "BUDGET", "FEATURE", "ISSUE", "INTENT", "URGENCY", "DECISION_STAGE"} |
| GENERIC_FEATURES = {"good", "best", "nice", "great", "unknown"} |
| INVALID_ENTITY_TERMS = { |
| "bonus", |
| "cashback", |
| "consideration", |
| "deal", |
| "deals", |
| "decision", |
| "discount", |
| "discounts", |
| "daily use", |
| "emi", |
| "emi option", |
| "emi options", |
| "family", |
| "festive", |
| "festive offer", |
| "medium", |
| "no", |
| "offer", |
| "offers", |
| "photos", |
| "strong", |
| "weak", |
| "yes", |
| "maybe", |
| "none", |
| "na", |
| } |
| BRANDS = { |
| "apple", |
| "asus", |
| "blue star", |
| "bosch", |
| "daikin", |
| "dell", |
| "godrej", |
| "haier", |
| "hp", |
| "ifb", |
| "lenovo", |
| "lg", |
| "mi", |
| "oneplus", |
| "oppo", |
| "panasonic", |
| "samsung", |
| "sony", |
| "vivo", |
| "whirlpool", |
| "xiaomi", |
| } |
| PRODUCTS = { |
| "ac", |
| "air conditioner", |
| "camera", |
| "dishwasher", |
| "earbuds", |
| "headphones", |
| "laptop", |
| "microwave", |
| "mobile", |
| "phone", |
| "refrigerator", |
| "smartphone", |
| "smartwatch", |
| "tablet", |
| "tv", |
| "vacuum cleaner", |
| "washing machine", |
| "desktop", |
| } |
| PREFERENCES = {"budget", "durability", "energy efficiency", "performance", "reliability"} |
| LEAD_CATEGORICAL_COLS = [ |
| "product", |
| "preference", |
| "intent_strength", |
| "decision_stage", |
| "sentiment", |
| "hesitation", |
| "follow_up_needed", |
| "offer_given", |
| "emi_option", |
| "product_suggested", |
| ] |
| LEAD_NUMERIC_COLS = ["budget", "brand_count", "use_case_count", "word_count", "sentence_count"] |
| REPO_ROOT = Path(__file__).resolve().parents[2] |
| DEFAULT_LEAD_MODEL_PATH = REPO_ROOT / "data" / "processed" / "lead_scoring_model.joblib" |
|
|
| FEATURE_READER_SYSTEM_PROMPT = "You are an expert reader for an AI sales CRM. Extract structured sales data. Return STRICT JSON." |
|
|
|
|
| @dataclass(slots=True) |
| class AspectMention: |
| key: str |
| name: str |
| context: str |
| start_char: int |
| end_char: int |
| text: str |
| label: str = "ASPECT" |
|
|
|
|
| @dataclass(slots=True) |
| class ExtractionResult: |
| mentions: list[AspectMention] |
| provider: str |
|
|
|
|
| def _utc_timestamp() -> str: |
| return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") |
|
|
|
|
| @lru_cache(maxsize=1) |
| def load_nlp(): |
| for model_name in ("en_core_web_sm", "en_core_web_md"): |
| try: |
| logger.info("Loading spaCy model '%s'", model_name) |
| return spacy.load(model_name) |
| except OSError: |
| continue |
|
|
| logger.warning("No pretrained spaCy English model found. Falling back to a blank English pipeline.") |
| nlp = spacy.blank("en") |
| if "sentencizer" not in nlp.pipe_names: |
| nlp.add_pipe("sentencizer") |
| return nlp |
|
|
|
|
| @lru_cache(maxsize=1) |
| def load_vader() -> SentimentIntensityAnalyzer: |
| analyzer = SentimentIntensityAnalyzer() |
| analyzer.lexicon.update( |
| { |
| "crisp": 2.1, |
| "drain": -2.4, |
| "drains": -2.6, |
| "lag": -2.2, |
| "laggy": -2.6, |
| "overheat": -2.7, |
| "overheats": -2.8, |
| "premium": 1.8, |
| "responsive": 2.1, |
| "sharp": 2.0, |
| "sluggish": -2.4, |
| "smooth": 2.2, |
| "stable": 1.7, |
| } |
| ) |
| return analyzer |
|
|
|
|
| @lru_cache(maxsize=1) |
| def load_lead_model() -> dict | None: |
| model_path = Path(os.getenv("LEAD_SCORING_MODEL_PATH", str(DEFAULT_LEAD_MODEL_PATH))) |
| if not model_path.exists(): |
| logger.warning("Lead scoring model not found at %s", model_path) |
| return None |
| return joblib.load(model_path) |
|
|
|
|
| class AspectSentimentEngine: |
| def __init__(self) -> None: |
| self.nlp = load_nlp() |
| self.analyzer = load_vader() |
| self.spacy_model_name = getattr(self.nlp, "meta", {}).get("name") or "blank-en" |
| self.llama_api_key = os.getenv("LLAMA_API_KEY") or os.getenv("GROQ_API_KEY") |
| raw_url = os.getenv("LLAMA_API_URL", "https://api.groq.com/openai/v1/chat/completions") |
| self.llama_api_url = raw_url if raw_url.endswith("/chat/completions") else f"{raw_url.rstrip('/')}/chat/completions" |
| self.llama_model = os.getenv("LLAMA_MODEL", "llama-3.3-70b-versatile") |
| self.lead_model = load_lead_model() |
|
|
| @staticmethod |
| def normalize_text(text: str) -> str: |
| return SPACE_RX.sub(" ", text).strip() |
|
|
| def parse(self, text: str) -> Doc: |
| return self.nlp(self.normalize_text(text)) |
|
|
| @staticmethod |
| def _strip_edges(text: str) -> str: |
| return EDGE_PUNCT_RX.sub("", text).strip() |
|
|
| @staticmethod |
| def _valid_phrase(text: str) -> bool: |
| return bool(text and ALPHA_RX.search(text)) |
|
|
| @staticmethod |
| def _valid_entity_text(text: str) -> bool: |
| return bool(text and (ALPHA_RX.search(text) or DIGIT_RX.search(text))) |
|
|
| @staticmethod |
| def _compact_entity_name(text: str) -> str: |
| return SPACE_RX.sub(" ", text).strip().lower() |
|
|
| @staticmethod |
| def _is_budget_value(text: str) -> bool: |
| normalized = text.strip().lower().replace(" ", "") |
| return bool(BUDGET_RX.match(normalized)) |
|
|
| def _is_relevant(self, text: str) -> bool: |
| text_lower = text.lower() |
| keywords = {"buy", "price", "budget", "product", "purchase", "model", "cost", "review", "issue", "battery", "display"} |
| if any(word in text_lower for word in keywords): |
| return True |
| if any(p in text_lower for p in PRODUCTS): |
| return True |
| if any(b in text_lower for b in BRANDS): |
| return True |
| return False |
|
|
| def _classify_known_entity(self, name: str) -> str | None: |
| normalized = self._compact_entity_name(name) |
| if normalized in BRANDS: |
| return "BRAND" |
| if normalized in PRODUCTS: |
| return "PRODUCT" |
| if self._is_budget_value(normalized): |
| return "BUDGET" |
| return None |
|
|
| def _is_valid_extracted_entity(self, name: str, label: str) -> bool: |
| normalized = self._compact_entity_name(name) |
| if not self._valid_entity_text(normalized): |
| return False |
| if len(normalized) < 2: |
| return False |
| if label == "FEATURE" and normalized in GENERIC_FEATURES: |
| return False |
| if normalized in GENERIC_ASPECT_TERMS or normalized in INVALID_ENTITY_TERMS: |
| return False |
| if any(term in normalized.split() for term in INVALID_ENTITY_TERMS): |
| return False |
| if label not in VALID_ENTITY_LABELS: |
| return False |
| if label == "BUDGET": |
| return self._is_budget_value(normalized) |
| if label == "BRAND" and not any(b in normalized for b in BRANDS): |
| return False |
| if label == "PRODUCT" and not any(p in normalized for p in PRODUCTS): |
| return False |
| return len(normalized.split()) <= 3 |
|
|
| @staticmethod |
| def _binary_signal(value: str) -> int: |
| return 1 if value.lower().strip() in {"yes", "true", "1"} else 0 |
|
|
| @staticmethod |
| def _budget_number(value: str) -> float: |
| normalized = value.lower().replace(",", "").replace("rs", "").replace("inr", "").replace("₹", "").strip() |
| multiplier = 1.0 |
| if normalized.endswith("k"): |
| multiplier = 1000.0 |
| normalized = normalized[:-1] |
| elif normalized.endswith("lakh"): |
| multiplier = 100000.0 |
| normalized = normalized[:-4] |
| elif normalized.endswith("lakhs"): |
| multiplier = 100000.0 |
| normalized = normalized[:-5] |
| try: |
| return float(normalized.strip()) * multiplier |
| except ValueError: |
| return 0.0 |
|
|
| @staticmethod |
| def _normalize_model_category(value: str) -> str: |
| return value.lower().strip().replace(" ", "_") |
|
|
| def _lead_features_from_mentions( |
| self, |
| mentions: list[AspectMention], |
| products: list[ProductSentiment], |
| word_count: int, |
| sentence_count: int, |
| ) -> dict[str, int | float | str]: |
| by_label: dict[str, list[str]] = defaultdict(list) |
| for mention in mentions: |
| by_label[mention.label].append(self._compact_entity_name(mention.name)) |
|
|
| product = by_label.get("PRODUCT", ["unknown"])[0] |
| if product not in PRODUCTS: |
| product = "unknown" |
|
|
| preference = by_label.get("PREFERENCE", ["unknown"])[0] |
| if preference not in PREFERENCES: |
| preference = "unknown" |
|
|
| budget = 0.0 |
| if by_label.get("BUDGET"): |
| budget = self._budget_number(by_label["BUDGET"][0]) |
|
|
| sentiment = "neutral" |
| if products: |
| avg_score = fmean(product_item.score for product_item in products) |
| sentiment = self._label_for_score(avg_score) |
|
|
| return { |
| "id": 0, |
| "product": product, |
| "budget": budget, |
| "brand_count": len(set(by_label.get("BRAND", []))), |
| "use_case_count": len(set(by_label.get("USE_CASE", []))), |
| "preference": preference, |
| "intent_strength": by_label.get("INTENT", ["medium"])[0], |
| "decision_stage": by_label.get("DECISION_STAGE", ["consideration"])[0], |
| "sentiment": sentiment, |
| "hesitation": self._binary_signal(by_label.get("HESITATION", ["no"])[0]), |
| "follow_up_needed": self._binary_signal(by_label.get("FOLLOW_UP", ["no"])[0]), |
| "offer_given": self._binary_signal(by_label.get("OFFER", ["no"])[0]), |
| "emi_option": self._binary_signal(by_label.get("EMI", ["no"])[0]), |
| "product_suggested": self._binary_signal(by_label.get("PRODUCT_SUGGESTED", ["no"])[0]), |
| "word_count": word_count, |
| "sentence_count": sentence_count, |
| } |
|
|
| def predict_conversion( |
| self, |
| mentions: list[AspectMention], |
| products: list[ProductSentiment], |
| word_count: int, |
| sentence_count: int, |
| ) -> ConversionScore | None: |
| if not self.lead_model: |
| return None |
|
|
| features = self._lead_features_from_mentions(mentions, products, word_count, sentence_count) |
| payload = self.lead_model |
| feature_columns = payload["feature_columns"] |
| model = payload["model"] |
| scaler = payload.get("scaler") |
|
|
| frame = pd.DataFrame([features]) |
| frame["product"] = frame["product"].map(self._normalize_model_category) |
| frame["preference"] = frame["preference"].map(self._normalize_model_category) |
|
|
| encoded = pd.get_dummies(frame, columns=[col for col in LEAD_CATEGORICAL_COLS if col in frame], drop_first=False) |
| for column in feature_columns: |
| if column not in encoded: |
| encoded[column] = 0 |
| encoded = encoded[feature_columns] |
|
|
| if scaler: |
| numeric_columns = [column for column in LEAD_NUMERIC_COLS if column in encoded] |
| encoded[numeric_columns] = scaler.transform(encoded[numeric_columns]) |
|
|
| probability = float(model.predict_proba(encoded)[0][1]) |
| label = "hot" if probability >= 0.7 else "warm" if probability >= 0.4 else "cold" |
| confidence = round(abs(probability - 0.5) * 2, 2) |
| return ConversionScore( |
| probability=round(probability, 3), |
| label=label, |
| confidence=confidence, |
| features=features, |
| model=str(Path(os.getenv("LEAD_SCORING_MODEL_PATH", str(DEFAULT_LEAD_MODEL_PATH))).name), |
| ) |
|
|
| def _aspect_name_from_span(self, span: Span) -> tuple[str, str] | None: |
| if not span.text.strip(): |
| return None |
|
|
| display_parts: list[str] = [] |
| key_parts: list[str] = [] |
|
|
| for token in span: |
| if token.is_space or token.is_punct: |
| continue |
| if token.pos_ in {"DET", "PRON"}: |
| continue |
| if token.is_stop and token.pos_ not in {"NOUN", "PROPN"}: |
| continue |
| if token.pos_ in {"NOUN", "PROPN"} or token.dep_ == "compound" or not token.pos_: |
| display_parts.append(token.text) |
| lemma = token.lemma_.lower() if token.lemma_ not in {"", "-PRON-"} else token.text.lower() |
| key_parts.append(lemma) |
|
|
| if not display_parts: |
| fallback = [ |
| token.text |
| for token in span |
| if token.is_alpha and not token.is_stop and token.pos_ not in {"DET", "PRON"} |
| ] |
| if fallback: |
| display_parts = fallback[-2:] |
| key_parts = [part.lower() for part in display_parts] |
|
|
| name = self._strip_edges(" ".join(display_parts)).lower() |
| key = self._strip_edges(" ".join(key_parts)).lower() |
| if not self._valid_phrase(name): |
| return None |
| if key in GENERIC_ASPECT_TERMS or name in GENERIC_ASPECT_TERMS: |
| return None |
| if len(name.split()) > 4: |
| return None |
|
|
| return key, name |
|
|
| @staticmethod |
| def _trim_span(doc: Doc, start: int, end: int) -> Span: |
| while start < end and doc[start].is_space: |
| start += 1 |
| while end > start and doc[end - 1].is_space: |
| end -= 1 |
| return doc[start:end] |
|
|
| def _context_window(self, span: Span) -> str: |
| sent = span.sent |
| start = sent.start |
| end = sent.end |
|
|
| for index in range(span.start - 1, sent.start - 1, -1): |
| token = span.doc[index] |
| if token.text in {",", ";", ":"} or token.lower_ in CLAUSE_BREAKERS: |
| start = index + 1 |
| break |
| if token.lower_ in {"and", "or"}: |
| lookback_start = max(sent.start, index - 4) |
| if any(span.doc[left].pos_ in {"VERB", "AUX", "ADJ", "ADV"} for left in range(lookback_start, index)): |
| start = index + 1 |
| break |
|
|
| saw_predicate = False |
| for index in range(span.end, sent.end): |
| token = span.doc[index] |
| if token.text in {",", ";", ":"} or token.lower_ in CLAUSE_BREAKERS: |
| end = index |
| break |
| if token.pos_ in {"VERB", "AUX", "ADJ", "ADV"}: |
| saw_predicate = True |
| if token.lower_ in {"and", "or"} and saw_predicate: |
| next_index = index + 1 |
| while next_index < sent.end and span.doc[next_index].is_space: |
| next_index += 1 |
| if next_index < sent.end: |
| next_token = span.doc[next_index] |
| if next_token.pos_ in {"DET", "NOUN", "PROPN", "PRON"} or next_token.lower_ in {"the", "this", "that"}: |
| end = index |
| break |
|
|
| candidate = self._trim_span(span.doc, start, end) |
| context = self.normalize_text(candidate.text) |
|
|
| if len(context.split()) < 3: |
| local_start = max(sent.start, span.start - 6) |
| local_end = min(sent.end, span.end + 6) |
| context = self.normalize_text(self._trim_span(span.doc, local_start, local_end).text) |
|
|
| return context or self.normalize_text(sent.text) |
|
|
| @staticmethod |
| def _dedupe_mentions(mentions: Iterable[AspectMention]) -> list[AspectMention]: |
| seen: set[tuple[str, int, int]] = set() |
| deduped: list[AspectMention] = [] |
| for mention in mentions: |
| signature = (mention.key, mention.start_char, mention.end_char) |
| if signature in seen: |
| continue |
| seen.add(signature) |
| deduped.append(mention) |
| return deduped |
|
|
| def _build_feature_prompt(self, text: str) -> str: |
| return ( |
| f"{FEATURE_READER_SYSTEM_PROMPT}\n" |
| 'Format: {"features":[{"name":"value","label":"PRODUCT|BRAND|BUDGET|FEATURE|ISSUE|INTENT|URGENCY|DECISION_STAGE","context":"short evidence"}]}\n\n' |
| "Rules:\n" |
| "- Only extract relevant product/sales info.\n" |
| "- If a field is missing, skip it.\n" |
| "- Do NOT guess or hallucinate.\n" |
| "- Do NOT output generic words: yes, no, maybe, medium, consideration, decision, offer, discount, emi, none.\n" |
| "- If not product-related -> return empty JSON: {\"features\": []}\n" |
| "- Keep names short, ideally 1-3 words.\n\n" |
| "Definitions:\n" |
| "- INTENT: buying / exploring\n" |
| "- URGENCY: immediate / later\n" |
| "- DECISION_STAGE: early / mid / final\n\n" |
| "Example:\n" |
| 'Input: "I want a Samsung TV under 50000, planning to buy this week"\n' |
| "Output:\n" |
| '{"features": [' |
| '{"name":"tv", "label":"PRODUCT", "context":"Samsung TV"}, ' |
| '{"name":"samsung", "label":"BRAND", "context":"Samsung TV"}, ' |
| '{"name":"50000", "label":"BUDGET", "context":"under 50000"}, ' |
| '{"name":"buying", "label":"INTENT", "context":"planning to buy"}, ' |
| '{"name":"immediate", "label":"URGENCY", "context":"this week"}, ' |
| '{"name":"final", "label":"DECISION_STAGE", "context":"planning to buy"}' |
| ']}\n\n' |
| f"Input:\n{text}\n" |
| "Output:\n" |
| ) |
|
|
| @staticmethod |
| def _sentence_candidates(text: str) -> list[str]: |
| parts = [part.strip() for part in SENTENCE_SPLIT_RX.split(text) if part.strip()] |
| return parts or [text.strip()] |
|
|
| def _find_context_in_text(self, text: str, context: str, feature_name: str) -> str: |
| normalized_text = self.normalize_text(text) |
| normalized_context = self.normalize_text(context) |
| if normalized_context and normalized_context.lower() in normalized_text.lower(): |
| return normalized_context |
|
|
| candidates = self._sentence_candidates(normalized_text) |
| feature_lower = feature_name.lower() |
| for candidate in candidates: |
| if feature_lower in candidate.lower(): |
| return candidate |
| return normalized_context or normalized_text |
|
|
| def _locate_span(self, text: str, feature_name: str, context: str) -> tuple[int, int, str]: |
| normalized_text = self.normalize_text(text) |
| feature_lower = feature_name.lower().strip() |
| if not feature_lower: |
| return 0, 0, normalized_text |
|
|
| start = normalized_text.lower().find(feature_lower) |
| if start != -1: |
| return start, start + len(feature_name), self._find_context_in_text(normalized_text, context, feature_name) |
|
|
| words = feature_lower.split() |
| if words: |
| fallback = words[-1] |
| start = normalized_text.lower().find(fallback) |
| if start != -1: |
| return start, start + len(fallback), self._find_context_in_text(normalized_text, context, feature_name) |
|
|
| return 0, len(feature_name), self._find_context_in_text(normalized_text, context, feature_name) |
|
|
| def _extract_mentions_from_feature_items(self, text: str, items: list[dict[str, object]]) -> list[AspectMention]: |
| mentions: list[AspectMention] = [] |
| for item in items: |
| name = self._compact_entity_name(str(item.get("name", ""))) |
| label = self.normalize_text(str(item.get("label", "FEATURE"))).upper() |
| if label not in VALID_ENTITY_LABELS: |
| continue |
|
|
| if self._is_budget_value(name) and label != "BUDGET": |
| label = "BUDGET" |
|
|
| known_label = self._classify_known_entity(name) |
| if known_label: |
| label = known_label |
|
|
| if not self._is_valid_extracted_entity(name, label): |
| continue |
|
|
| key = f"{label}:{self._strip_edges(name).lower()}" |
| start_char, end_char, resolved_context = self._locate_span(text, name, str(item.get("context", ""))) |
| mentions.append( |
| AspectMention( |
| key=key, |
| name=name, |
| context=resolved_context, |
| start_char=start_char, |
| end_char=end_char, |
| text=text[start_char:end_char] if end_char > start_char else name, |
| label=label, |
| ) |
| ) |
|
|
| return self._dedupe_mentions(mentions) |
|
|
| async def _extract_mentions_with_llama_api(self, text: str) -> list[AspectMention]: |
| if not self.llama_api_key: |
| raise RuntimeError("Set LLAMA_API_KEY or GROQ_API_KEY before starting the backend.") |
|
|
| if not self._is_relevant(text): |
| return [] |
|
|
| payload = { |
| "model": self.llama_model, |
| "temperature": 0, |
| "response_format": {"type": "json_object"}, |
| "messages": [ |
| {"role": "system", "content": FEATURE_READER_SYSTEM_PROMPT}, |
| {"role": "user", "content": self._build_feature_prompt(text)}, |
| ], |
| } |
|
|
| headers = { |
| "Authorization": f"Bearer {self.llama_api_key}", |
| "Content-Type": "application/json", |
| "Accept": "application/json", |
| "User-Agent": "aspect-sentiment-client/1.0", |
| } |
|
|
| async with httpx.AsyncClient() as client: |
| try: |
| response = await client.post( |
| self.llama_api_url, json=payload, headers=headers, timeout=90.0 |
| ) |
| response.raise_for_status() |
| data = response.json() |
| |
| message = data.get("choices", [{}])[0].get("message", {}) |
| model_response = str(message.get("content", "")).strip() |
| if not model_response: |
| return [] |
| |
| parsed = json.loads(model_response) |
| features = parsed.get("features", []) |
| |
| items = [item for item in features if isinstance(item, dict)] |
| return self._extract_mentions_from_feature_items(text, items) |
| |
| except httpx.HTTPStatusError as exc: |
| logger.error("Groq API HTTP error %s: %s", exc.response.status_code, exc.response.text) |
| except httpx.RequestError as exc: |
| logger.error("Groq API connection error: %s", exc) |
| except (json.JSONDecodeError, ValueError) as exc: |
| logger.error("Error parsing Groq response: %s", exc) |
|
|
| return [] |
|
|
|
|
|
|
| def extract_mentions(self, doc: Doc) -> list[AspectMention]: |
| mentions: list[AspectMention] = [] |
| covered_tokens: set[int] = set() |
| has_dependencies = doc.has_annotation("DEP") |
|
|
| if has_dependencies: |
| for chunk in doc.noun_chunks: |
| aspect = self._aspect_name_from_span(chunk) |
| if not aspect: |
| continue |
| key, name = aspect |
| mentions.append( |
| AspectMention( |
| key=key, |
| name=name, |
| context=self._context_window(chunk), |
| start_char=chunk.start_char, |
| end_char=chunk.end_char, |
| text=chunk.text, |
| ) |
| ) |
| covered_tokens.update(range(chunk.start, chunk.end)) |
|
|
| for token in doc: |
| if token.i in covered_tokens: |
| continue |
| if token.is_space or token.is_punct or token.is_stop: |
| continue |
|
|
| if doc.has_annotation("POS"): |
| if token.pos_ not in {"NOUN", "PROPN"}: |
| continue |
| elif not token.is_alpha: |
| continue |
|
|
| span = doc[token.i : token.i + 1] |
| aspect = self._aspect_name_from_span(span) |
| if not aspect: |
| continue |
| key, name = aspect |
| mentions.append( |
| AspectMention( |
| key=key, |
| name=name, |
| context=self._context_window(span), |
| start_char=token.idx, |
| end_char=token.idx + len(token.text), |
| text=token.text, |
| ) |
| ) |
|
|
| return self._dedupe_mentions(mentions) |
|
|
| async def extract_mentions_with_provider(self, text: str) -> ExtractionResult: |
| normalized = self.normalize_text(text) |
| mentions = await self._extract_mentions_with_llama_api(normalized) |
| return ExtractionResult(mentions=mentions, provider=f"llama:{self.llama_model}") |
|
|
| @staticmethod |
| def _label_for_score(score: float) -> str: |
| if score > 0.05: |
| return "positive" |
| if score < -0.05: |
| return "negative" |
| return "neutral" |
|
|
| @staticmethod |
| def _confidence(scores: list[float], mention_count: int) -> float: |
| if not scores: |
| return 0.35 |
| strength = fmean(abs(score) for score in scores) |
| mention_bonus = min(0.16, max(0, mention_count - 1) * 0.04) |
| return round(min(0.99, 0.42 + strength * 0.46 + mention_bonus), 2) |
|
|
| def score_products(self, mentions: list[AspectMention]) -> list[ProductSentiment]: |
| grouped: dict[str, list[AspectMention]] = defaultdict(list) |
| for mention in mentions: |
| grouped[mention.key].append(mention) |
|
|
| products: list[ProductSentiment] = [] |
| for group in grouped.values(): |
| contexts = list(dict.fromkeys(mention.context for mention in group if mention.context)) |
| scores = [self.analyzer.polarity_scores(context)["compound"] for context in contexts] or [0.0] |
| score = round(float(fmean(scores)), 3) |
| label = self._label_for_score(score) |
|
|
| products.append( |
| ProductSentiment( |
| name=group[0].name, |
| entityType=group[0].label, |
| sentiment=label, |
| score=score, |
| confidence=self._confidence(scores, len(group)), |
| mentions=len(group), |
| context=contexts[0] if contexts else "", |
| contexts=contexts, |
| highlights=[ |
| HighlightRange( |
| product=group[0].name, |
| text=mention.text, |
| start=mention.start_char, |
| end=mention.end_char, |
| label=mention.label, |
| ) |
| for mention in group |
| ], |
| ) |
| ) |
|
|
| return sorted(products, key=lambda item: (-item.mentions, -abs(item.score), item.name)) |
|
|
| @staticmethod |
| def _percentages(counts: Counter[str], total: int) -> dict[str, int]: |
| if total == 0: |
| return {"positive": 0, "negative": 0, "neutral": 0} |
|
|
| raw = {label: counts.get(label, 0) * 100 / total for label in ("positive", "negative", "neutral")} |
| floors = {label: math.floor(value) for label, value in raw.items()} |
| remaining = 100 - sum(floors.values()) |
| order = sorted(raw, key=lambda label: raw[label] - floors[label], reverse=True) |
| for index in range(remaining): |
| floors[order[index % len(order)]] += 1 |
| return {label: int(value) for label, value in floors.items()} |
|
|
| def summarize(self, products: list[ProductSentiment]) -> SentimentSummary: |
| counts = Counter(product.sentiment for product in products) |
| total = len(products) |
| percentages = self._percentages(counts, total) |
| average_score = round(float(fmean(product.score for product in products)), 3) if products else 0.0 |
|
|
| top_count = max(counts.values(), default=0) |
| leaders = [label for label, count in counts.items() if count == top_count and count > 0] |
| dominant = leaders[0] if len(leaders) == 1 else "balanced" if leaders else "neutral" |
|
|
| return SentimentSummary( |
| positive=percentages["positive"], |
| negative=percentages["negative"], |
| neutral=percentages["neutral"], |
| counts=SentimentCounts( |
| positive=counts.get("positive", 0), |
| negative=counts.get("negative", 0), |
| neutral=counts.get("neutral", 0), |
| ), |
| dominant=dominant, |
| averageScore=average_score, |
| totalProducts=total, |
| ) |
|
|
| @staticmethod |
| def extraction_quality(mentions: list[AspectMention], products: list[ProductSentiment]) -> dict[str, int | float | str]: |
| label_counts = Counter(mention.label for mention in mentions) |
| return { |
| "mentionCount": len(mentions), |
| "uniqueEntityCount": len(products), |
| "labelCount": len(label_counts), |
| "avgConfidence": round(float(fmean(product.confidence for product in products)), 2) if products else 0.0, |
| **{f"{label.lower()}Count": count for label, count in sorted(label_counts.items())}, |
| } |
|
|
| async def analyze_text( |
| self, |
| text: str, |
| *, |
| source_name: str, |
| source_type: str, |
| language: str | None, |
| transcription_confidence: float | None, |
| whisper_model: str | None, |
| pipeline: list[PipelineStage], |
| processing_ms: int, |
| ) -> AnalysisResponse: |
| normalized = self.normalize_text(text) |
| doc = self.parse(normalized) |
| extraction_result = await self.extract_mentions_with_provider(normalized) |
| mentions = extraction_result.mentions |
| products = self.score_products(mentions) |
| highlights = [highlight for product in products for highlight in product.highlights] |
| sentence_count = sum(1 for _ in doc.sents) if normalized else 0 |
| word_count = sum(1 for token in doc if not token.is_space and not token.is_punct) |
| conversion_score = self.predict_conversion(mentions, products, word_count, sentence_count) |
|
|
| return AnalysisResponse( |
| transcript=normalized, |
| normalizedText=normalized, |
| products=products, |
| highlights=highlights, |
| summary=self.summarize(products), |
| conversionScore=conversion_score, |
| pipeline=pipeline, |
| metadata=AnalysisMetadata( |
| sourceType=source_type, |
| sourceName=source_name, |
| language=language, |
| processingMs=processing_ms, |
| extractionProvider=extraction_result.provider, |
| transcriptionConfidence=transcription_confidence, |
| whisperModel=whisper_model, |
| wordCount=word_count, |
| sentenceCount=sentence_count, |
| extractionQuality=self.extraction_quality(mentions, products), |
| createdAt=_utc_timestamp(), |
| ), |
| ) |
|
|