File size: 4,074 Bytes
d1f3f31 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | from src.aspect_sentiment.semantic_mapper import mapper
from src.aspect_sentiment.normalized_features import NormalizedOutput
import re
def clean_budget(value: str) -> int:
"""Safely extract integer budget from string."""
digits = re.sub(r'[^\d]', '', str(value))
return int(digits) if digits else 0
def normalize_decision_stage(value: str) -> str:
name = value.replace(" ", "_").lower()
if name in {"converted", "ready_to_purchase", "near_purchase", "negotiation", "decision", "final"}:
return "decision"
if name in {"evaluating", "comparing_alternatives", "considering", "budget_discussion", "mid"}:
return "mid"
if name in {"purchase_delayed", "follow_up_required", "late"}:
return "late"
if name in {"exploring", "awareness", "curious", "early"}:
return "early"
return "mid"
def normalize_urgency(value: str) -> str:
name = value.replace(" ", "_").lower()
if any(term in name for term in ["today", "now", "immediate", "urgent", "high"]):
return "immediate"
if any(term in name for term in ["tomorrow", "week", "soon", "medium"]):
return "soon"
if any(term in name for term in ["month", "later", "low", "flexible"]):
return "flexible"
if any(term in name for term in ["no_rush", "browsing", "exploring"]):
return "no_rush"
return "flexible"
def normalize_price_sensitivity(value: str) -> str:
name = value.lower()
if any(term in name for term in ["high", "very", "expensive", "cheap", "discount", "low budget", "price sensitive"]):
return "high"
if any(term in name for term in ["low", "premium", "not important"]):
return "low"
return "medium"
def process_extractions(raw_features: list[dict]) -> NormalizedOutput:
"""
Takes the raw features from LLaMA 3 and converts them into the
normalized XGBoost categorical structure.
"""
output = NormalizedOutput()
for f in raw_features:
name = str(f.get("name", "")).strip().lower()
label = str(f.get("label", "")).upper()
if not name:
continue
# 1. Budget extraction
if label == "BUDGET" or any(c in name for c in ['$', '₹', 'rs', 'rupees']):
# Prevent GPU names like RTX 3050 from becoming a budget
gpu = mapper.extract_gpu(name)
if gpu:
# It's a GPU, not a budget!
output.use_cases.append("gaming") # Infer gaming
continue
val = clean_budget(name)
if val > 0:
output.budget = val
continue
# 2. Urgency
if label == "URGENCY" or label == "URGENCY_LEVEL":
output.urgency = normalize_urgency(name)
continue
# 2.5 Decision Stage
if label == "DECISION_STAGE":
output.decision_stage = normalize_decision_stage(name)
continue
if label in {"PRICE_SENSITIVITY", "OBJECTION_TYPE"}:
output.price_sensitivity = normalize_price_sensitivity(name)
continue
# 3. Semantic Mapping for Product, Brand, Use Case
mapped_tuples = mapper.map_term(name)
# If mapping engine found nothing, use raw label as fallback
if not mapped_tuples:
if label == "PRODUCT": output.products.append(name.replace(" ", "_"))
elif label == "BRAND": output.brands.append(name.replace(" ", "_"))
elif label == "USE_CASE": output.use_cases.append(name.replace(" ", "_"))
for val, mapped_label in mapped_tuples:
if mapped_label == "PRODUCT" and val not in output.products:
output.products.append(val)
elif mapped_label == "BRAND" and val not in output.brands:
output.brands.append(val)
elif mapped_label == "USE_CASE" and val not in output.use_cases:
output.use_cases.append(val)
return output
|