""" Quality Parameters Handler Manages CRUD operations for business-specific quality parameters """ import pymysql from pymysql.cursors import DictCursor import logging import csv import io import os from contextlib import contextmanager logger = logging.getLogger(__name__) class QualityParametersHandler: """Handle all quality parameter operations""" 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', 'mcube@admin123'), 'database': config.get('DB_NAME', 'voicebot_cluster'), 'charset': 'utf8mb4', 'cursorclass': DictCursor, 'autocommit': True } @contextmanager def get_connection(self): """Context manager for database connections""" conn = None try: conn = pymysql.connect(**self.db_config) yield conn except pymysql.Error as e: logger.error(f"Database connection error: {e}") raise finally: if conn: conn.close() def get_parameters(self, bid): """Get all quality parameters for a specific business""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT * FROM quality_parameters WHERE bid = %s ORDER BY parameter_group, parameter_name """ cursor.execute(query, (bid,)) results = cursor.fetchall() return results @staticmethod def _normalize_header(value): return ''.join(ch for ch in str(value or '').strip().lower() if ch.isalnum()) @staticmethod def _clean_cell(value): if value is None: return '' return str(value).strip() @staticmethod def _parse_bool(value): if isinstance(value, bool): return value text = str(value or '').strip().lower() return text in {'1', 'true', 'yes', 'y', 'applicable', 'enabled'} @staticmethod def _parse_int(value, default=None): 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, content): 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, *names, default=''): 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): 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') score = self._parse_int(self._row_get(normalized, 'Score', 'Max Score', 'max_score'), default=None) if not parameter_group or not parameter_name: return None, 'Parameter Group and Parameter Name are required' if score is None or score < 1: return None, 'Score must be a number greater than 0' na_condition = self._row_get( normalized, 'When is this check not applicable?', 'Not Applicable Condition', 'applicability_condition', ) return { 'parameter_group': parameter_group, 'parameter_name': parameter_name, 'parameter_type': self._row_get(normalized, 'Parameter Type', 'parameter_type', default='Required') or 'Required', 'check_description': self._row_get( normalized, 'What do you want to check on?', 'Check Description', 'check_description', ), 'detailed_description': self._row_get(normalized, 'Description', 'Detailed Description', 'detailed_description'), 'max_score': score, 'sample_utterances': self._row_get(normalized, 'Sample Utterances', 'sample_utterances'), 'applicability_condition': na_condition, 'auto_detect_na': self._parse_bool(self._row_get(normalized, 'Auto Detect N/A', 'auto_detect_na')), 'default_na': self._parse_bool(self._row_get(normalized, 'Default N/A', 'default_na')), 'applicable_once_per_customer': self._parse_bool(self._row_get(normalized, 'Applicable Once Per Customer', 'applicable_once_per_customer')), 'applicable_threshold': self._parse_int(self._row_get(normalized, 'Applicable Threshold', 'applicable_threshold')), 'is_fatal': self._parse_bool(self._row_get(normalized, 'Fatal', 'is_fatal')), 'enable_subjective_marking': self._parse_bool(self._row_get(normalized, 'Subjective Marking', 'enable_subjective_marking')), 'status': self._row_get(normalized, 'Status', 'status', default='Applicable') or 'Applicable', }, None def import_parameters_file(self, bid, filename, content): """Import quality parameters from a CSV/XLSX upload and 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') inserted = 0 updated = 0 skipped = 0 errors = [] 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 quality_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'] query = """ UPDATE quality_parameters SET parameter_type = %s, check_description = %s, detailed_description = %s, max_score = %s, sample_utterances = %s, applicability_condition = %s, auto_detect_na = %s, default_na = %s, applicable_once_per_customer = %s, applicable_threshold = %s, is_fatal = %s, enable_subjective_marking = %s, status = %s, updated_at = NOW() WHERE id = %s AND bid = %s """ cursor.execute(query, ( parameter_data.get('parameter_type'), parameter_data.get('check_description'), parameter_data.get('detailed_description'), parameter_data.get('max_score'), parameter_data.get('sample_utterances'), parameter_data.get('applicability_condition'), parameter_data.get('auto_detect_na', False), parameter_data.get('default_na', False), parameter_data.get('applicable_once_per_customer', False), parameter_data.get('applicable_threshold'), parameter_data.get('is_fatal', False), parameter_data.get('enable_subjective_marking', False), parameter_data.get('status', 'Applicable'), existing['id'], bid, )) updated += 1 continue query = """ INSERT INTO quality_parameters (bid, parameter_group, parameter_name, parameter_type, check_description, detailed_description, max_score, sample_utterances, applicability_condition, auto_detect_na, default_na, applicable_once_per_customer, applicable_threshold, is_fatal, enable_subjective_marking, status) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor.execute(query, ( bid, parameter_data.get('parameter_group'), parameter_data.get('parameter_name'), parameter_data.get('parameter_type'), parameter_data.get('check_description'), parameter_data.get('detailed_description'), parameter_data.get('max_score'), parameter_data.get('sample_utterances'), parameter_data.get('applicability_condition'), parameter_data.get('auto_detect_na', False), parameter_data.get('default_na', False), parameter_data.get('applicable_once_per_customer', False), parameter_data.get('applicable_threshold'), parameter_data.get('is_fatal', False), parameter_data.get('enable_subjective_marking', False), parameter_data.get('status', 'Applicable'), )) inserted += 1 conn.commit() 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 that the file uses comma or semicolon separators and includes Parameter Group, Parameter Name, and Score columns.' raise ValueError(detail) return result def ensure_template_table(self): """Create storage for reusable system parameter templates.""" with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS quality_parameter_templates ( id INT AUTO_INCREMENT PRIMARY KEY, template_key VARCHAR(100) NOT NULL DEFAULT 'default', parameter_group VARCHAR(255) NOT NULL, parameter_name VARCHAR(255) NOT NULL, parameter_type VARCHAR(100), check_description TEXT, detailed_description TEXT, max_score INT NOT NULL, sample_utterances TEXT, applicability_condition VARCHAR(255), auto_detect_na BOOLEAN DEFAULT FALSE, default_na BOOLEAN DEFAULT FALSE, applicable_once_per_customer BOOLEAN DEFAULT FALSE, applicable_threshold INT, is_fatal BOOLEAN DEFAULT FALSE, enable_subjective_marking BOOLEAN DEFAULT FALSE, status VARCHAR(50) DEFAULT 'Applicable', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_template_key (template_key), UNIQUE KEY unique_template_param (template_key, parameter_group, parameter_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """) def get_template_parameters(self, template_key='default', seed_from_bid=None): """Get reusable template parameters, optionally seeding from an existing BID once.""" self.ensure_template_table() template_key = str(template_key or 'default') with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM quality_parameter_templates WHERE template_key = %s ORDER BY parameter_group, parameter_name """, (template_key,), ) results = cursor.fetchall() if not results and seed_from_bid: source = self.get_parameters(str(seed_from_bid)) if source: for param in source: item = dict(param) item.pop('id', None) item.pop('bid', None) item.pop('created_at', None) item.pop('updated_at', None) self.save_template_parameter(template_key, item) return self.get_template_parameters(template_key) return results def save_template_parameter(self, template_key, parameter_data): """Save or update a reusable template parameter.""" self.ensure_template_table() template_key = str(template_key or 'default') with self.get_connection() as conn: cursor = conn.cursor() if 'id' in parameter_data and parameter_data['id']: query = """ UPDATE quality_parameter_templates SET parameter_group = %s, parameter_name = %s, parameter_type = %s, check_description = %s, detailed_description = %s, max_score = %s, sample_utterances = %s, applicability_condition = %s, auto_detect_na = %s, default_na = %s, applicable_once_per_customer = %s, applicable_threshold = %s, is_fatal = %s, enable_subjective_marking = %s, status = %s, updated_at = NOW() WHERE id = %s AND template_key = %s """ cursor.execute(query, ( parameter_data.get('parameter_group'), parameter_data.get('parameter_name'), parameter_data.get('parameter_type'), parameter_data.get('check_description'), parameter_data.get('detailed_description'), parameter_data.get('max_score'), parameter_data.get('sample_utterances'), parameter_data.get('applicability_condition'), parameter_data.get('auto_detect_na', False), parameter_data.get('default_na', False), parameter_data.get('applicable_once_per_customer', False), parameter_data.get('applicable_threshold'), parameter_data.get('is_fatal', False), parameter_data.get('enable_subjective_marking', False), parameter_data.get('status', 'Applicable'), parameter_data['id'], template_key, )) return parameter_data['id'] query = """ INSERT INTO quality_parameter_templates (template_key, parameter_group, parameter_name, parameter_type, check_description, detailed_description, max_score, sample_utterances, applicability_condition, auto_detect_na, default_na, applicable_once_per_customer, applicable_threshold, is_fatal, enable_subjective_marking, status) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor.execute(query, ( template_key, parameter_data.get('parameter_group'), parameter_data.get('parameter_name'), parameter_data.get('parameter_type'), parameter_data.get('check_description'), parameter_data.get('detailed_description'), parameter_data.get('max_score'), parameter_data.get('sample_utterances'), parameter_data.get('applicability_condition'), parameter_data.get('auto_detect_na', False), parameter_data.get('default_na', False), parameter_data.get('applicable_once_per_customer', False), parameter_data.get('applicable_threshold'), parameter_data.get('is_fatal', False), parameter_data.get('enable_subjective_marking', False), parameter_data.get('status', 'Applicable'), )) return cursor.lastrowid def delete_template_parameter(self, template_key, parameter_id): """Delete a reusable template parameter.""" self.ensure_template_table() with self.get_connection() as conn: cursor = conn.cursor() cursor.execute( """ DELETE FROM quality_parameter_templates WHERE template_key = %s AND id = %s """, (str(template_key or 'default'), parameter_id), ) return cursor.rowcount > 0 def get_parameter_by_id(self, bid, parameter_id): """Get a specific quality parameter by ID""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT * FROM quality_parameters WHERE bid = %s AND id = %s """ cursor.execute(query, (bid, parameter_id)) result = cursor.fetchone() return result def save_parameter(self, bid, parameter_data): """Save or update a quality parameter""" with self.get_connection() as conn: cursor = conn.cursor() # Check if parameter already exists (for update) if 'id' in parameter_data and parameter_data['id']: # Update existing parameter query = """ UPDATE quality_parameters SET parameter_group = %s, parameter_name = %s, parameter_type = %s, check_description = %s, detailed_description = %s, max_score = %s, sample_utterances = %s, applicability_condition = %s, auto_detect_na = %s, default_na = %s, applicable_once_per_customer = %s, applicable_threshold = %s, is_fatal = %s, enable_subjective_marking = %s, status = %s, updated_at = NOW() WHERE id = %s AND bid = %s """ cursor.execute(query, ( parameter_data.get('parameter_group'), parameter_data.get('parameter_name'), parameter_data.get('parameter_type'), parameter_data.get('check_description'), parameter_data.get('detailed_description'), parameter_data.get('max_score'), parameter_data.get('sample_utterances'), parameter_data.get('applicability_condition'), parameter_data.get('auto_detect_na', False), parameter_data.get('default_na', False), parameter_data.get('applicable_once_per_customer', False), parameter_data.get('applicable_threshold'), parameter_data.get('is_fatal', False), parameter_data.get('enable_subjective_marking', False), parameter_data.get('status', 'Applicable'), parameter_data['id'], bid )) return parameter_data['id'] else: # Insert new parameter query = """ INSERT INTO quality_parameters (bid, parameter_group, parameter_name, parameter_type, check_description, detailed_description, max_score, sample_utterances, applicability_condition, auto_detect_na, default_na, applicable_once_per_customer, applicable_threshold, is_fatal, enable_subjective_marking, status) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor.execute(query, ( bid, parameter_data.get('parameter_group'), parameter_data.get('parameter_name'), parameter_data.get('parameter_type'), parameter_data.get('check_description'), parameter_data.get('detailed_description'), parameter_data.get('max_score'), parameter_data.get('sample_utterances'), parameter_data.get('applicability_condition'), parameter_data.get('auto_detect_na', False), parameter_data.get('default_na', False), parameter_data.get('applicable_once_per_customer', False), parameter_data.get('applicable_threshold'), parameter_data.get('is_fatal', False), parameter_data.get('enable_subjective_marking', False), parameter_data.get('status', 'Applicable') )) return cursor.lastrowid def delete_parameter(self, bid, parameter_id): """Delete a quality parameter""" with self.get_connection() as conn: cursor = conn.cursor() query = """ DELETE FROM quality_parameters WHERE bid = %s AND id = %s """ cursor.execute(query, (bid, parameter_id)) return cursor.rowcount > 0 def calculate_total_possible_score(self, bid): """Calculate total possible score for all parameters""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT SUM(max_score) as total_score FROM quality_parameters WHERE bid = %s AND default_na = FALSE """ cursor.execute(query, (bid,)) result = cursor.fetchone() return result['total_score'] if result and result['total_score'] else 0 def get_parameters_by_group(self, bid, parameter_group): """Get all parameters for a specific group""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT * FROM quality_parameters WHERE bid = %s AND parameter_group = %s ORDER BY parameter_name """ cursor.execute(query, (bid, parameter_group)) results = cursor.fetchall() return results def get_parameter_groups(self, bid): """Get list of all parameter groups for a business""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT DISTINCT parameter_group FROM quality_parameters WHERE bid = %s ORDER BY parameter_group """ cursor.execute(query, (bid,)) results = cursor.fetchall() return [row['parameter_group'] for row in results]