""" Data Validation Module - Sales Dashboard Validates Google Sheet data integrity and alerts on issues """ import pandas as pd import re from datetime import datetime from collections import Counter class DataValidator: """Validates sales data and returns issues with severity levels""" def __init__(self, df, sheet_name="Sales Data"): self.df = df self.sheet_name = sheet_name self.issues = [] self.warnings = [] self.info = [] def validate_all(self): """Run all validation checks""" self.check_required_columns() self.check_date_formats() self.check_numeric_values() self.check_familia_consistency() self.check_missing_values() self.check_duplicates() self.check_outliers() self.check_comercial_names() return { 'critical': self.issues, 'warnings': self.warnings, 'info': self.info, 'total_rows': len(self.df), 'validated_at': datetime.now().strftime('%d/%m/%Y %H:%M:%S') } def check_required_columns(self): """Check if all required columns exist""" required = ['comercial', 'cliente', 'fatura', 'quant', 'mês', 'familia'] found_cols = [col.lower() for col in self.df.columns] missing = [] for req in required: if not any(req in col for col in found_cols): missing.append(req) if missing: self.issues.append({ 'type': 'MISSING_COLUMNS', 'severity': 'CRITICAL', 'message': f"Missing required columns: {', '.join(missing)}", 'affected_rows': None, 'recommendation': 'Add missing columns to the Google Sheet' }) def check_date_formats(self): """Validate date column formats""" mes_col = self._find_column('mês', 'mes') if not mes_col: return invalid_dates = [] unparsed_count = 0 for idx, value in enumerate(self.df[mes_col]): if pd.isna(value) or value == '' or str(value).strip() == '': continue s = str(value).strip() # Try to parse parsed = self._parse_date(s) if parsed is None: invalid_dates.append({ 'row': idx + 2, # +2 for header and 1-based index 'value': s }) unparsed_count += 1 if invalid_dates: sample = invalid_dates[:5] self.issues.append({ 'type': 'INVALID_DATES', 'severity': 'CRITICAL', 'message': f"{unparsed_count} rows with unparseable dates", 'affected_rows': [d['row'] for d in sample], 'samples': [d['value'] for d in sample], 'recommendation': 'Use format YYYY/MM or MM/YYYY. Examples: 2026/03 or 03/2026' }) def check_numeric_values(self): """Check for invalid numeric values in Faturação and Quantidade""" fat_col = self._find_column('fatura') quant_col = self._find_column('quant') for col_name, col in [('Faturação', fat_col), ('Quantidade', quant_col)]: if not col: continue invalid = [] for idx, value in enumerate(self.df[col]): if pd.isna(value) or value == '' or str(value).strip() == '': continue # Try to parse as numeric try: s = str(value).strip() # Remove common formatting s = s.replace('€', '').replace('$', '').replace(' ', '') s = s.replace('.', '').replace(',', '.') # European format float(s) except (ValueError, AttributeError): invalid.append({ 'row': idx + 2, 'value': str(value) }) if invalid: sample = invalid[:5] self.warnings.append({ 'type': f'INVALID_{col_name.upper()}', 'severity': 'WARNING', 'message': f"{len(invalid)} rows with invalid {col_name} values", 'affected_rows': [d['row'] for d in sample], 'samples': [d['value'] for d in sample], 'recommendation': f'Check {col_name} column for text/formula errors' }) def check_familia_consistency(self): """Check Familia column for spelling variations""" familia_col = self._find_column('familia', 'família') if not familia_col: return # Get all unique familia values familias = self.df[familia_col].dropna().astype(str).str.strip().str.lower() familia_counts = Counter(familias) # Check for urnas variations urna_variants = {k: v for k, v in familia_counts.items() if 'urn' in k and k not in ['urna', 'urnas']} if urna_variants: self.warnings.append({ 'type': 'FAMILIA_SPELLING', 'severity': 'WARNING', 'message': f"Found {len(urna_variants)} non-standard Urnas spellings", 'variants': list(urna_variants.keys()), 'counts': list(urna_variants.values()), 'recommendation': 'Standardize to exactly "Urnas" or "Urna" (case insensitive)' }) # List all familia values for reference top_familias = familia_counts.most_common(10) self.info.append({ 'type': 'FAMILIA_DISTRIBUTION', 'severity': 'INFO', 'message': f"Found {len(familia_counts)} unique familia values", 'top_values': [{'name': k, 'count': v} for k, v in top_familias], 'recommendation': 'Review for typos or inconsistent naming' }) def check_missing_values(self): """Check for missing values in critical columns""" critical_cols = { 'Comercial': self._find_column('comercial'), 'Cliente': self._find_column('cliente'), 'Faturação': self._find_column('fatura'), 'Quantidade': self._find_column('quant'), 'Mês': self._find_column('mês', 'mes') } # Numeric columns should only check for NaN, not empty strings numeric_cols = {'Faturação', 'Quantidade'} for name, col in critical_cols.items(): if not col: continue # For numeric columns, only check for NaN/None values if name in numeric_cols: total_missing = self.df[col].isna().sum() missing_rows = self.df[self.df[col].isna()].index.tolist() else: # For text columns, check both NaN and empty strings missing_count = self.df[col].isna().sum() empty_count = (self.df[col].astype(str).str.strip() == '').sum() total_missing = missing_count + empty_count missing_rows = self.df[self.df[col].isna() | (self.df[col].astype(str).str.strip() == '')].index.tolist() if total_missing > 0: sample_rows = [r + 2 for r in missing_rows[:10]] # +2 for header and 1-based severity = 'CRITICAL' if total_missing > 10 else 'WARNING' target = self.issues if severity == 'CRITICAL' else self.warnings target.append({ 'type': f'MISSING_{name.upper()}', 'severity': severity, 'message': f"{total_missing} rows missing {name}", 'affected_rows': sample_rows, 'percentage': round(total_missing / len(self.df) * 100, 2), 'recommendation': f'Fill in missing {name} values in sheet' }) def check_duplicates(self): """Check for potential duplicate entries""" # Check for exact duplicates duplicates = self.df[self.df.duplicated(keep='first')] if not duplicates.empty: dup_rows = [idx + 2 for idx in duplicates.index.tolist()[:10]] self.warnings.append({ 'type': 'DUPLICATE_ROWS', 'severity': 'WARNING', 'message': f"{len(duplicates)} potentially duplicate rows found", 'affected_rows': dup_rows, 'recommendation': 'Review and remove duplicate entries' }) def check_outliers(self): """Check for unusual values that might be data entry errors""" fat_col = self._find_column('fatura') quant_col = self._find_column('quant') if fat_col: numeric_vals = pd.to_numeric(self.df[fat_col], errors='coerce').dropna() if len(numeric_vals) > 0: mean = numeric_vals.mean() std = numeric_vals.std() # Flag values > 3 standard deviations from mean outliers = self.df[pd.to_numeric(self.df[fat_col], errors='coerce') > (mean + 3 * std)] if not outliers.empty and len(outliers) < len(self.df) * 0.01: # Less than 1% of data out_rows = [idx + 2 for idx in outliers.index.tolist()[:5]] out_vals = [pd.to_numeric(self.df.loc[idx, fat_col], errors='coerce') for idx in outliers.index.tolist()[:5]] self.info.append({ 'type': 'FATURACAO_OUTLIERS', 'severity': 'INFO', 'message': f"{len(outliers)} unusually high Faturação values detected", 'affected_rows': out_rows, 'values': [f"€{v:,.2f}" for v in out_vals if not pd.isna(v)], 'recommendation': 'Review these values to ensure they are correct' }) def check_comercial_names(self): """Check for consistent comercial names""" com_col = self._find_column('comercial') if not com_col: return comercials = self.df[com_col].dropna().astype(str).str.strip() comercial_counts = Counter(comercials) # Look for similar names (potential typos) names = list(comercial_counts.keys()) similar_pairs = [] for i, name1 in enumerate(names): for name2 in names[i+1:]: # Simple similarity check if self._similar_strings(name1.lower(), name2.lower()): similar_pairs.append((name1, name2, comercial_counts[name1], comercial_counts[name2])) if similar_pairs: self.warnings.append({ 'type': 'SIMILAR_COMERCIAL_NAMES', 'severity': 'WARNING', 'message': f"Found {len(similar_pairs)} pairs of similar comercial names", 'pairs': [{'name1': p[0], 'name2': p[1], 'count1': p[2], 'count2': p[3]} for p in similar_pairs], 'recommendation': 'Check if these are typos or different people' }) # List all comercials self.info.append({ 'type': 'COMERCIAL_LIST', 'severity': 'INFO', 'message': f"Found {len(comercial_counts)} unique comercial names", 'comercials': [{'name': k, 'sales_count': v} for k, v in comercial_counts.most_common()], 'recommendation': 'Verify all names are spelled correctly' }) def _find_column(self, *keywords): """Find column by keywords""" for col in self.df.columns: col_lower = col.lower() if all(k in col_lower for k in keywords): return col return None def _parse_date(self, value): """Try to parse date value""" if not value: return None s = str(value).strip().replace('-', '/').replace('.', '/') # Try YYYY/MM m = re.search(r"(\d{4})\D?(\d{1,2})", s) if m: year = m.group(1) month = m.group(2).zfill(2) if 1 <= int(month) <= 12: return (year, month) # Try MM/YYYY m = re.search(r"(\d{1,2})\D?(\d{4})", s) if m: month = m.group(1).zfill(2) year = m.group(2) if 1 <= int(month) <= 12: return (year, month) return None def _similar_strings(self, s1, s2, threshold=0.8): """Check if two strings are similar (potential typos)""" # Simple Levenshtein-like check if len(s1) < 3 or len(s2) < 3: return False # Check if one contains the other if s1 in s2 or s2 in s1: return True # Check character overlap common = set(s1) & set(s2) similarity = len(common) / max(len(set(s1)), len(set(s2))) return similarity > threshold def validate_objectives_sheet(df_obj): """Quick validation for objectives sheet""" issues = [] if df_obj is None or df_obj.empty: return [{ 'type': 'OBJECTIVES_MISSING', 'severity': 'WARNING', 'message': 'Objectives sheet is empty or not found', 'recommendation': 'Run /setup-objectives to generate objectives' }] # Check required columns required = ['Comercial', 'Cliente', 'Target_Valor', 'Target_Urnas'] missing = [col for col in required if col not in df_obj.columns] if missing: issues.append({ 'type': 'OBJECTIVES_COLUMNS', 'severity': 'CRITICAL', 'message': f"Objectives sheet missing columns: {', '.join(missing)}", 'recommendation': 'Regenerate objectives using /setup-objectives' }) return issues