"""CRUD for per-business sales propensity scoring parameters.""" from __future__ import annotations import csv import io import logging import os from contextlib import contextmanager from typing import Any, Dict, List, Optional, Tuple import pymysql from pymysql.cursors import DictCursor logger = logging.getLogger(__name__) DEFAULT_PROPENSITY_PARAMETERS: List[Dict[str, Any]] = [ { "parameter_group": "Intent", "parameter_name": "Buying intent strength", "max_score": 25, "check_description": "Customer shows clear interest in purchasing or moving forward.", "detection_mode": "hybrid", "positive_keywords": "ready to buy,book visit,send quotation,interested in,want to purchase,finalize", "negative_keywords": "not interested,wrong number,just browsing,no need", "positive_points_per_hit": 8, "negative_points_per_hit": 10, }, { "parameter_group": "Urgency", "parameter_name": "Timeline and urgency", "max_score": 20, "check_description": "Customer indicates a timeframe or urgency to decide.", "detection_mode": "keyword", "positive_keywords": "this week,this month,urgent,asap,soon,immediately,within", "negative_keywords": "later,not sure when,no hurry,next year", "positive_points_per_hit": 7, "negative_points_per_hit": 8, }, { "parameter_group": "Commercial", "parameter_name": "Budget readiness", "max_score": 20, "check_description": "Budget, pricing, or payment willingness is discussed positively.", "detection_mode": "keyword", "positive_keywords": "budget,afford,price is fine,payment plan,loan approved,within budget", "negative_keywords": "too expensive,cannot afford,out of budget,overpriced", "positive_points_per_hit": 7, "negative_points_per_hit": 9, }, { "parameter_group": "Risk", "parameter_name": "Objection handling", "max_score": 15, "check_description": "Objections are raised and addressed constructively.", "detection_mode": "hybrid", "positive_keywords": "that makes sense,understood,resolved,clear now", "negative_keywords": "still concerned,not convinced,need to think,compare with", "positive_points_per_hit": 5, "negative_points_per_hit": 6, }, { "parameter_group": "Momentum", "parameter_name": "Next step commitment", "max_score": 20, "check_description": "A concrete next step or follow-up is agreed.", "detection_mode": "keyword", "positive_keywords": "schedule,callback tomorrow,send details,visit site,meeting,whatsapp me", "negative_keywords": "will call you,maybe later,not now", "positive_points_per_hit": 8, "negative_points_per_hit": 8, }, ] class PropensityParametersHandler: 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 sales_propensity_parameters ( id INT AUTO_INCREMENT PRIMARY KEY, bid VARCHAR(50) NOT NULL, parameter_group VARCHAR(255) NOT NULL, parameter_name VARCHAR(255) NOT NULL, check_description TEXT, detailed_description TEXT, max_score INT NOT NULL DEFAULT 10, detection_mode VARCHAR(32) NOT NULL DEFAULT 'keyword', positive_keywords TEXT, negative_keywords TEXT, positive_points_per_hit INT NOT NULL DEFAULT 5, negative_points_per_hit INT NOT NULL DEFAULT 5, max_keyword_hits INT NOT NULL DEFAULT 3, status VARCHAR(50) DEFAULT 'Applicable', sort_order INT NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_bid (bid), UNIQUE KEY unique_propensity_param (bid, parameter_group, parameter_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) def get_parameters(self, bid: str, active_only: bool = True) -> List[Dict[str, Any]]: self.ensure_table() bid = str(bid) with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT * FROM sales_propensity_parameters WHERE bid = %s """ params: List[Any] = [bid] if active_only: query += " AND status = 'Applicable'" query += " ORDER BY sort_order ASC, parameter_group ASC, parameter_name ASC" cursor.execute(query, params) return cursor.fetchall() or [] def seed_defaults_if_empty(self, bid: str) -> int: existing = self.get_parameters(bid, active_only=False) if existing: return 0 inserted = 0 for idx, param in enumerate(DEFAULT_PROPENSITY_PARAMETERS): payload = dict(param) payload["sort_order"] = idx self.save_parameter(bid, payload) inserted += 1 return inserted def save_parameter(self, bid: str, parameter_data: Dict[str, Any]) -> int: self.ensure_table() bid = str(bid) with self.get_connection() as conn: cursor = conn.cursor() fields = { "parameter_group": parameter_data.get("parameter_group"), "parameter_name": parameter_data.get("parameter_name"), "check_description": parameter_data.get("check_description"), "detailed_description": parameter_data.get("detailed_description"), "max_score": max(1, int(parameter_data.get("max_score") or 10)), "detection_mode": str(parameter_data.get("detection_mode") or "keyword").strip().lower(), "positive_keywords": parameter_data.get("positive_keywords") or "", "negative_keywords": parameter_data.get("negative_keywords") or "", "positive_points_per_hit": max(1, int(parameter_data.get("positive_points_per_hit") or 5)), "negative_points_per_hit": max(1, int(parameter_data.get("negative_points_per_hit") or 5)), "max_keyword_hits": max(1, int(parameter_data.get("max_keyword_hits") or 3)), "status": parameter_data.get("status") or "Applicable", "sort_order": int(parameter_data.get("sort_order") or 0), } if parameter_data.get("id"): cursor.execute( """ UPDATE sales_propensity_parameters SET parameter_group=%s, parameter_name=%s, check_description=%s, detailed_description=%s, max_score=%s, detection_mode=%s, positive_keywords=%s, negative_keywords=%s, positive_points_per_hit=%s, negative_points_per_hit=%s, max_keyword_hits=%s, status=%s, sort_order=%s, updated_at=NOW() WHERE id=%s AND bid=%s """, ( fields["parameter_group"], fields["parameter_name"], fields["check_description"], fields["detailed_description"], fields["max_score"], fields["detection_mode"], fields["positive_keywords"], fields["negative_keywords"], fields["positive_points_per_hit"], fields["negative_points_per_hit"], fields["max_keyword_hits"], fields["status"], fields["sort_order"], parameter_data["id"], bid, ), ) return int(parameter_data["id"]) cursor.execute( """ INSERT INTO sales_propensity_parameters (bid, parameter_group, parameter_name, check_description, detailed_description, max_score, detection_mode, positive_keywords, negative_keywords, positive_points_per_hit, negative_points_per_hit, max_keyword_hits, status, sort_order) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, ( bid, fields["parameter_group"], fields["parameter_name"], fields["check_description"], fields["detailed_description"], fields["max_score"], fields["detection_mode"], fields["positive_keywords"], fields["negative_keywords"], fields["positive_points_per_hit"], fields["negative_points_per_hit"], fields["max_keyword_hits"], fields["status"], fields["sort_order"], ), ) return int(cursor.lastrowid) def delete_parameter(self, bid: str, parameter_id: int) -> bool: self.ensure_table() with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( "DELETE FROM sales_propensity_parameters WHERE id=%s AND bid=%s", (int(parameter_id), str(bid)), ) return cursor.rowcount > 0 def calculate_total_possible_score(self, bid: str) -> int: params = self.get_parameters(bid) return sum(int(p.get("max_score") or 0) for p in params) @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] semicolons = first_line.count(";") commas = first_line.count(",") tabs = first_line.count("\t") if semicolons > commas and semicolons > 0: delimiter = ";" elif tabs > commas and tabs > 0: delimiter = "\t" reader = csv.DictReader(io.StringIO(text), delimiter=delimiter) return [row for row in reader if row and any(self._clean_cell(value) for value 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_parameter(self, row: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: normalized = {self._normalize_header(key): self._clean_cell(value) for key, value in row.items()} parameter_group = self._row_get(normalized, "Parameter Group", "parameter_group", "Group") parameter_name = self._row_get(normalized, "Parameter Name", "parameter_name", "Name") max_score = self._parse_int(self._row_get(normalized, "Max Score", "max_score", "Score"), default=None) if not parameter_group or not parameter_name: return None, "Parameter Group and Parameter Name are required" if max_score is None or max_score < 1: return None, "Max Score must be a number greater than 0" detection_mode = self._row_get(normalized, "Detection Mode", "detection_mode", default="keyword").lower() if detection_mode not in {"keyword", "hybrid"}: detection_mode = "keyword" return { "parameter_group": parameter_group, "parameter_name": parameter_name, "check_description": self._row_get( normalized, "Check Description", "check_description", "Description" ), "detailed_description": self._row_get( normalized, "Detailed Description", "detailed_description" ), "max_score": max_score, "detection_mode": detection_mode, "positive_keywords": self._row_get( normalized, "Positive Keywords", "positive_keywords" ), "negative_keywords": self._row_get( normalized, "Negative Keywords", "negative_keywords" ), "positive_points_per_hit": max( 1, self._parse_int( self._row_get(normalized, "Positive Points Per Hit", "positive_points_per_hit"), default=5, ) or 5, ), "negative_points_per_hit": max( 1, self._parse_int( self._row_get(normalized, "Negative Points Per Hit", "negative_points_per_hit"), default=5, ) or 5, ), "max_keyword_hits": max( 1, self._parse_int( self._row_get(normalized, "Max Keyword Hits", "max_keyword_hits"), default=3, ) or 3, ), "status": self._row_get(normalized, "Status", "status", default="Applicable") or "Applicable", }, None def import_parameters_file(self, bid: str, filename: str, content: bytes) -> Dict[str, Any]: """Import propensity parameters from CSV/XLSX; upsert by group + name.""" 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 parameter rows found in uploaded file") self.ensure_table() inserted = 0 updated = 0 skipped = 0 errors: List[Dict[str, Any]] = [] with self.get_connection() as conn: cursor = conn.cursor() for index, row in enumerate(rows, start=2): parameter_data, error = self._row_to_parameter(row) if error: skipped += 1 errors.append({"row": index, "error": error}) continue cursor.execute( """ SELECT id FROM sales_propensity_parameters WHERE bid = %s AND parameter_group = %s AND parameter_name = %s LIMIT 1 """, ( bid, parameter_data["parameter_group"], parameter_data["parameter_name"], ), ) existing = cursor.fetchone() if existing: parameter_data["id"] = existing["id"] self.save_parameter(bid, parameter_data) updated += 1 else: self.save_parameter(bid, parameter_data) inserted += 1 result = { "inserted": inserted, "updated": updated, "skipped": skipped, "errors": errors[:20], } if inserted == 0 and updated == 0: detail = "No rows were imported." if skipped: detail += f" {skipped} row(s) had errors." if errors: detail += f' Row {errors[0]["row"]}: {errors[0]["error"]}' elif rows: detail += " Check column headers (Parameter Group, Parameter Name, Max Score) and file separators." raise ValueError(detail) return result