"""Per-BID customer sentiment configuration (keywords, rubric, detection mode).""" from __future__ import annotations import csv import io import os from contextlib import contextmanager from typing import Any, Dict, List, Optional, Tuple import pymysql from pymysql.cursors import DictCursor DEFAULT_SENTIMENT_CONFIG: Dict[str, Any] = { "detection_mode": "hybrid", "customer_only": True, "positive_keywords": ( "thank you, thanks, great, excellent, happy, satisfied, love, perfect, " "wonderful, appreciate, good experience, helpful, amazing, pleased" ), "negative_keywords": ( "angry, frustrated, disappointed, terrible, worst, unhappy, complaint, " "refund, cancel, bad service, not happy, upset, horrible, never again" ), "neutral_keywords": "okay, fine, maybe, not sure, average, alright", "prompt_rubric": ( "Classify overall customer mood from their words only (ignore agent tone). " "positive = satisfied, grateful, or enthusiastic; " "negative = frustrated, angry, or complaining; " "neutral = factual, mixed, or unclear." ), "keyword_positive_threshold": 2, "keyword_negative_threshold": 2, "max_keyword_hits": 5, "status": "Active", } class SentimentConfigHandler: def __init__(self, config): self.config = config self.db_config = { "host": config.get("DB_HOST", "127.0.0.1"), "port": config.get("DB_PORT", 3306), "user": config.get("DB_USER", "admin"), "password": config.get("DB_PASSWORD", ""), "database": config.get("DB_NAME", "voicebot_cluster"), "charset": "utf8mb4", "cursorclass": DictCursor, "autocommit": True, } @contextmanager def get_connection(self): conn = None try: conn = pymysql.connect(**self.db_config) yield conn finally: if conn: conn.close() def ensure_table(self) -> None: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( """ CREATE TABLE IF NOT EXISTS business_sentiment_config ( id INT AUTO_INCREMENT PRIMARY KEY, bid VARCHAR(50) NOT NULL, detection_mode VARCHAR(32) NOT NULL DEFAULT 'hybrid', customer_only TINYINT(1) NOT NULL DEFAULT 1, positive_keywords TEXT, negative_keywords TEXT, neutral_keywords TEXT, prompt_rubric TEXT, keyword_positive_threshold INT NOT NULL DEFAULT 2, keyword_negative_threshold INT NOT NULL DEFAULT 2, max_keyword_hits INT NOT NULL DEFAULT 5, status VARCHAR(50) DEFAULT 'Active', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY unique_bid (bid), INDEX idx_bid (bid) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) def get_config(self, bid: str) -> Optional[Dict[str, Any]]: self.ensure_table() bid = str(bid) with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM business_sentiment_config WHERE bid = %s LIMIT 1", (bid,), ) row = cursor.fetchone() if not row: return None row["customer_only"] = bool(row.get("customer_only")) return row def get_config_or_defaults(self, bid: str) -> Dict[str, Any]: existing = self.get_config(bid) if existing: return existing defaults = dict(DEFAULT_SENTIMENT_CONFIG) defaults["bid"] = str(bid) return defaults def seed_defaults_if_empty(self, bid: str) -> int: if self.get_config(bid): return 0 self.save_config(bid, DEFAULT_SENTIMENT_CONFIG) return 1 def save_config(self, bid: str, payload: Dict[str, Any]) -> int: self.ensure_table() bid = str(bid) mode = str(payload.get("detection_mode") or "hybrid").strip().lower() if mode not in {"hybrid", "llm", "keyword"}: mode = "hybrid" fields = { "detection_mode": mode, "customer_only": 1 if payload.get("customer_only", True) else 0, "positive_keywords": payload.get("positive_keywords") or "", "negative_keywords": payload.get("negative_keywords") or "", "neutral_keywords": payload.get("neutral_keywords") or "", "prompt_rubric": payload.get("prompt_rubric") or "", "keyword_positive_threshold": max( 1, int(payload.get("keyword_positive_threshold") or 2) ), "keyword_negative_threshold": max( 1, int(payload.get("keyword_negative_threshold") or 2) ), "max_keyword_hits": max(1, int(payload.get("max_keyword_hits") or 5)), "status": payload.get("status") or "Active", } with self.get_connection() as conn: cursor = conn.cursor() existing = self.get_config(bid) if existing and existing.get("id"): cursor.execute( """ UPDATE business_sentiment_config SET detection_mode=%s, customer_only=%s, positive_keywords=%s, negative_keywords=%s, neutral_keywords=%s, prompt_rubric=%s, keyword_positive_threshold=%s, keyword_negative_threshold=%s, max_keyword_hits=%s, status=%s, updated_at=NOW() WHERE id=%s AND bid=%s """, ( fields["detection_mode"], fields["customer_only"], fields["positive_keywords"], fields["negative_keywords"], fields["neutral_keywords"], fields["prompt_rubric"], fields["keyword_positive_threshold"], fields["keyword_negative_threshold"], fields["max_keyword_hits"], fields["status"], existing["id"], bid, ), ) return int(existing["id"]) cursor.execute( """ INSERT INTO business_sentiment_config (bid, detection_mode, customer_only, positive_keywords, negative_keywords, neutral_keywords, prompt_rubric, keyword_positive_threshold, keyword_negative_threshold, max_keyword_hits, status) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, ( bid, fields["detection_mode"], fields["customer_only"], fields["positive_keywords"], fields["negative_keywords"], fields["neutral_keywords"], fields["prompt_rubric"], fields["keyword_positive_threshold"], fields["keyword_negative_threshold"], fields["max_keyword_hits"], fields["status"], ), ) return int(cursor.lastrowid) @staticmethod def _normalize_header(value: str) -> str: return "".join(ch for ch in str(value or "").strip().lower() if ch.isalnum()) @staticmethod def _clean_cell(value: Any) -> str: if value is None: return "" return str(value).strip() @staticmethod def _parse_int(value: Any, default: Optional[int] = None) -> Optional[int]: text = str(value or "").strip() if not text: return default try: return int(float(text)) except (TypeError, ValueError): return default def _read_upload_rows(self, filename: str, content: bytes) -> List[Dict[str, Any]]: ext = os.path.splitext(str(filename or "").lower())[1] if ext == ".xlsx": try: from openpyxl import load_workbook except ImportError as exc: raise ValueError("openpyxl is required to upload XLSX files") from exc workbook = load_workbook(io.BytesIO(content), data_only=True) sheet = workbook.active values = list(sheet.iter_rows(values_only=True)) if not values: return [] headers = [self._clean_cell(cell) for cell in values[0]] rows = [] for row in values[1:]: if not row or not any(self._clean_cell(cell) for cell in row): continue rows.append( {headers[idx]: row[idx] if idx < len(row) else "" for idx in range(len(headers))} ) return rows if ext not in {".csv", ""}: raise ValueError("Only CSV and XLSX files are supported") try: text = content.decode("utf-8-sig") except UnicodeDecodeError: text = content.decode("latin-1") delimiter = "," sample = text[:8192] try: dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|") delimiter = dialect.delimiter except csv.Error: first_line = (text.splitlines() or [""])[0] if first_line.count(";") > first_line.count(","): delimiter = ";" elif first_line.count("\t") > first_line.count(","): delimiter = "\t" reader = csv.DictReader(io.StringIO(text), delimiter=delimiter) return [row for row in reader if row and any(self._clean_cell(v) for v in row.values())] def _row_get(self, normalized_row: Dict[str, str], *names: str, default: str = "") -> str: for name in names: key = self._normalize_header(name) if key in normalized_row: return normalized_row[key] return default def _row_to_config(self, row: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: normalized = {self._normalize_header(k): self._clean_cell(v) for k, v in row.items()} mode = self._row_get(normalized, "Detection Mode", "detection_mode", default="hybrid").lower() if mode not in {"hybrid", "llm", "keyword"}: mode = "hybrid" customer_only_raw = self._row_get( normalized, "Customer Only", "customer_only", default="yes" ).lower() customer_only = customer_only_raw not in {"no", "false", "0", "off"} return { "detection_mode": mode, "customer_only": customer_only, "positive_keywords": self._row_get( normalized, "Positive Keywords", "positive_keywords" ), "negative_keywords": self._row_get( normalized, "Negative Keywords", "negative_keywords" ), "neutral_keywords": self._row_get( normalized, "Neutral Keywords", "neutral_keywords" ), "prompt_rubric": self._row_get(normalized, "Prompt Rubric", "prompt_rubric", "Rubric"), "keyword_positive_threshold": max( 1, self._parse_int( self._row_get( normalized, "Positive Threshold", "keyword_positive_threshold" ), default=2, ) or 2, ), "keyword_negative_threshold": max( 1, self._parse_int( self._row_get( normalized, "Negative Threshold", "keyword_negative_threshold" ), default=2, ) or 2, ), "max_keyword_hits": max( 1, self._parse_int( self._row_get(normalized, "Max Keyword Hits", "max_keyword_hits"), default=5, ) or 5, ), "status": self._row_get(normalized, "Status", "status", default="Active") or "Active", }, None def import_config_file(self, bid: str, filename: str, content: bytes) -> Dict[str, Any]: bid = str(bid).strip() if not content: raise ValueError("Uploaded file is empty") rows = self._read_upload_rows(filename, content) if not rows: raise ValueError("No configuration rows found in uploaded file") config_data, error = self._row_to_config(rows[0]) if error: raise ValueError(error) config_id = self.save_config(bid, config_data) return {"updated": 1, "config_id": config_id}