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(), ), )