Upload 12 files

.gitattributes

@@ -1,35 +1,2 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.db filter=lfs diff=lfs merge=lfs -text
+lfs_tmp/sales_management.db filter=lfs diff=lfs merge=lfs -text

analytics.py

@@ -0,0 +1,192 @@
+import pandas as pd
+import numpy as np
+from datetime import datetime, timedelta
+
+class Analytics:
+    def __init__(self, db_manager):
+        self.db = db_manager
+    
+    def get_sales_summary(self):
+        """Get comprehensive sales summary statistics"""
+        try:
+            sales_df = self.db.get_dataframe('sales')
+            payments_df = self.db.get_dataframe('payments')
+            
+            if sales_df.empty:
+                return {
+                    'total_sales': 0,
+                    'total_payments': 0,
+                    'pending_amount': 0,
+                    'total_transactions': 0,
+                    'avg_sale_value': 0
+                }
+            
+            total_sales = sales_df['total_amount'].sum()
+            total_payments = payments_df['amount'].sum() if not payments_df.empty else 0
+            pending_amount = total_sales - total_payments
+            
+            return {
+                'total_sales': total_sales,
+                'total_payments': total_payments,
+                'pending_amount': pending_amount,
+                'total_transactions': len(sales_df),
+                'avg_sale_value': sales_df['total_amount'].mean()
+            }
+        except Exception as e:
+            return {
+                'total_sales': 0,
+                'total_payments': 0,
+                'pending_amount': 0,
+                'total_transactions': 0,
+                'avg_sale_value': 0
+            }
+    
+    def get_customer_analysis(self):
+        """Analyze customer data"""
+        try:
+            customers_df = self.db.get_dataframe('customers')
+            sales_df = self.db.get_dataframe('sales')
+            
+            if customers_df.empty:
+                return {
+                    'total_customers': 0,
+                    'village_distribution': {},
+                    'top_customers': {}
+                }
+            
+            # Customer distribution by village
+            village_stats = customers_df['village'].value_counts().head(10)
+            
+            # Top customers by spending
+            if not sales_df.empty:
+                customer_sales = sales_df.groupby('customer_id')['total_amount'].sum()
+                top_customers = customer_sales.nlargest(10)
+            else:
+                top_customers = pd.Series(dtype=float)
+            
+            return {
+                'total_customers': len(customers_df),
+                'village_distribution': village_stats.to_dict(),
+                'top_customers': top_customers.to_dict()
+            }
+        except Exception as e:
+            return {
+                'total_customers': 0,
+                'village_distribution': {},
+                'top_customers': {}
+            }
+    
+    def get_payment_analysis(self):
+        """Analyze payment data"""
+        try:
+            pending_payments = self.db.get_pending_payments()
+            payments_df = self.db.get_dataframe('payments')
+            
+            if pending_payments.empty:
+                return {
+                    'total_pending': 0,
+                    'customer_pending': {},
+                    'payment_methods': {}
+                }
+            
+            # Group by customer
+            customer_pending = pending_payments.groupby('customer_id')['pending_amount'].sum()
+            
+            # Payment method distribution
+            if not payments_df.empty:
+                payment_methods = payments_df['payment_method'].value_counts()
+            else:
+                payment_methods = pd.Series(dtype=object)
+            
+            return {
+                'total_pending': pending_payments['pending_amount'].sum(),
+                'customer_pending': customer_pending.to_dict(),
+                'payment_methods': payment_methods.to_dict()
+            }
+        except Exception as e:
+            return {
+                'total_pending': 0,
+                'customer_pending': {},
+                'payment_methods': {}
+            }
+    
+    def get_demo_conversion_rates(self):
+        """Calculate demo conversion rates"""
+        try:
+            demos_df = self.db.get_demo_conversions()
+            
+            if demos_df.empty:
+                return {
+                    'total_demos': 0,
+                    'converted_demos': 0,
+                    'conversion_rate': 0
+                }
+            
+            total_demos = len(demos_df)
+            converted_demos = len(demos_df[demos_df['conversion_status'] == 'Converted'])
+            conversion_rate = (converted_demos / total_demos) * 100 if total_demos > 0 else 0
+            
+            return {
+                'total_demos': total_demos,
+                'converted_demos': converted_demos,
+                'conversion_rate': conversion_rate
+            }
+        except Exception as e:
+            return {
+                'total_demos': 0,
+                'converted_demos': 0,
+                'conversion_rate': 0
+            }
+    
+    def get_sales_trend(self):
+        """Get sales trend data for charts"""
+        try:
+            sales_df = self.db.get_dataframe('sales')
+            
+            if sales_df.empty:
+                return pd.DataFrame()
+            
+            # Convert sale_date to datetime if it's not
+            sales_df['sale_date'] = pd.to_datetime(sales_df['sale_date'])
+            
+            # Group by date
+            daily_sales = sales_df.groupby('sale_date')['total_amount'].sum().reset_index()
+            daily_sales = daily_sales.sort_values('sale_date')
+            
+            return daily_sales
+        except Exception as e:
+            return pd.DataFrame()
+    
+    def get_payment_distribution(self):
+        """Get payment distribution for charts"""
+        try:
+            payments_df = self.db.get_dataframe('payments')
+            
+            if payments_df.empty:
+                return pd.DataFrame()
+            
+            payment_dist = payments_df.groupby('payment_method')['amount'].sum().reset_index()
+            return payment_dist
+        except Exception as e:
+            return pd.DataFrame()
+    
+    def get_product_performance(self):
+        """Get product performance data"""
+        try:
+            sale_items_df = self.db.get_dataframe('sale_items', '''
+            SELECT si.*, p.product_name 
+            FROM sale_items si 
+            JOIN products p ON si.product_id = p.product_id
+            ''')
+            
+            if sale_items_df.empty:
+                return pd.DataFrame()
+            
+            product_perf = sale_items_df.groupby('product_name').agg({
+                'quantity': 'sum',
+                'amount': 'sum'
+            }).reset_index()
+            
+            return product_perf
+        except Exception as e:
+            return pd.DataFrame()

automation.py

@@ -0,0 +1,90 @@
+# enhanced_automation.py
+import schedule
+import time
+import smtplib
+from email.mime.text import MimeText
+from email.mime.multipart import MimeMultipart
+from datetime import datetime, timedelta
+from analytics import Analytics
+import os
+
+class AutomationManager:
+    def __init__(self, db_manager, whatsapp_manager):
+        self.db = db_manager
+        self.whatsapp = whatsapp_manager
+    
+    def daily_payment_reminders(self):
+        """Send payment reminders for overdue payments"""
+        overdue_payments = self.db.get_pending_payments()
+        
+        for _, payment in overdue_payments.iterrows():
+            if payment['pending_amount'] > 0:
+                customer = self.db.get_dataframe('customers', 
+                    f"SELECT * FROM customers WHERE customer_id = {payment['customer_id']}")
+                
+                if not customer.empty:
+                    customer_data = customer.iloc[0]
+                    message = f"""Hello {customer_data['name']},
+
+This is a friendly reminder that your payment of ₹{payment['pending_amount']:,.2f} for invoice {payment['invoice_no']} is overdue.
+
+Please make the payment at your earliest convenience.
+
+Thank you,
+Sales Team"""
+                    
+                    self.whatsapp.send_message(customer_data['mobile'], message)
+    
+    def demo_followups(self):
+        """Send follow-up messages for demos"""
+        upcoming_followups = self.db.get_dataframe('demos', '''
+        SELECT d.*, c.name as customer_name, c.mobile, p.product_name
+        FROM demos d
+        JOIN customers c ON d.customer_id = c.customer_id
+        JOIN products p ON d.product_id = p.product_id
+        WHERE d.follow_up_date = date('now')
+        AND d.conversion_status = 'Not Converted'
+        ''')
+        
+        for _, demo in upcoming_followups.iterrows():
+            message = f"""Hello {demo['customer_name']},
+
+Following up on your demo of {demo['product_name']} on {demo['demo_date']}. 
+
+How was your experience? Would you like to place an order or need another demo?
+
+Best regards,
+Sales Team"""
+            
+            self.whatsapp.send_message(demo['mobile'], message)
+    
+    def weekly_performance_report(self):
+        """Generate and send weekly performance report"""
+        analytics = Analytics(self.db)
+        
+        sales_summary = analytics.get_sales_summary()
+        demo_stats = analytics.get_demo_conversion_rates()
+        payment_analysis = analytics.get_payment_analysis()
+        
+        report = f"""
+        📊 WEEKLY PERFORMANCE REPORT
+        ----------------------------
+        Total Sales: ₹{sales_summary.get('total_sales', 0):,.2f}
+        Pending Payments: ₹{sales_summary.get('pending_amount', 0):,.2f}
+        Demo Conversion Rate: {demo_stats.get('conversion_rate', 0):.1f}%
+        Total Customers: {analytics.get_customer_analysis().get('total_customers', 0)}
+        
+        Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M')}
+        """
+        
+        # You can extend this to email the report
+        self._save_report(report)
+        return report
+    
+    def _save_report(self, report):
+        """Save report to file"""
+        filename = f"reports/weekly_report_{datetime.now().strftime('%Y%m%d')}.txt"
+        os.makedirs('reports', exist_ok=True)
+        
+        with open(filename, 'w') as f:
+            f.write(report)

data_processor.py

@@ -0,0 +1,710 @@
+import pandas as pd
+import numpy as np
+import os
+import re
+from datetime import datetime
+import logging
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class DataProcessor:
+    def __init__(self, db_manager):
+        self.db = db_manager
+        self.product_mapping = self._create_product_mapping()
+    
+    def _create_product_mapping(self):
+        """Create product mapping from database"""
+        try:
+            products_df = self.db.get_dataframe('products')
+            return {row['product_name'].upper(): row['product_id'] for _, row in products_df.iterrows()}
+        except Exception as e:
+            logger.error(f"Error creating product mapping: {e}")
+            return {}
+    
+    def process_excel_file(self, file_path):
+        """Enhanced file processing with all data types"""
+        try:
+            file_name = os.path.basename(file_path)
+            print(f"🚀 Processing file: {file_name}")
+            
+            excel_file = pd.ExcelFile(file_path)
+            processed_sheets = 0
+            
+            for sheet_name in excel_file.sheet_names:
+                df = pd.read_excel(file_path, sheet_name=sheet_name)
+                df_clean = self._clean_dataframe(df)
+                
+                print(f"\n📊 Sheet: {sheet_name}")
+                print(f"   Columns: {df_clean.columns.tolist()}")
+                
+                # Check all types with priority
+                is_payment = self._is_payment_sheet(df_clean)
+                is_sales = self._is_sales_sheet(df_clean)
+                is_customer = self._is_customer_sheet(df_clean)
+                is_distributor = self._is_distributor_sheet(df_clean)
+                
+                print(f"   Detection - Payment: {is_payment}, Sales: {is_sales}, Customer: {is_customer}, Distributor: {is_distributor}")
+                
+                processed = False
+                if is_payment:
+                    processed = self.process_payment_sheet(df_clean, file_name, sheet_name)
+                elif is_sales:
+                    processed = self.process_sales_sheet(df_clean, file_name, sheet_name)
+                elif is_distributor:
+                    processed = self.process_distributor_sheet(df_clean, file_name, sheet_name)
+                elif is_customer:
+                    processed = self.process_customer_sheet(df_clean, file_name, sheet_name)
+                
+                if processed:
+                    processed_sheets += 1
+                    print(f"   ✅ Successfully processed as detected type")
+                else:
+                    print(f"   ❌ Failed to process")
+            
+            print(f"\n🎉 File processing complete: {processed_sheets}/{len(excel_file.sheet_names)} sheets processed")
+            return processed_sheets > 0
+            
+        except Exception as e:
+            print(f"💥 Error processing file {file_path}: {e}")
+            return False
+    
+    def _clean_dataframe(self, df):
+        """Clean and prepare dataframe for processing"""
+        # Remove completely empty rows and columns
+        df = df.dropna(how='all').dropna(axis=1, how='all')
+        
+        # Reset index
+        df = df.reset_index(drop=True)
+        
+        # Convert column names to string and clean them
+        df.columns = [str(col).strip().upper() for col in df.columns]
+        
+        return df
+    
+    def _is_sales_sheet(self, df):
+        """Check if sheet contains sales data"""
+        required_columns = ['INVOICE', 'CUSTOMER', 'PRODUCT', 'QUANTITY', 'AMOUNT']
+        existing_columns = [col for col in df.columns if any(req in col for req in required_columns)]
+        return len(existing_columns) >= 3
+    
+    def _is_customer_sheet(self, df):
+        """Check if sheet contains customer data"""
+        required_columns = ['CUSTOMER', 'NAME', 'MOBILE', 'VILLAGE']
+        existing_columns = [col for col in df.columns if any(req in col for req in required_columns)]
+        return len(existing_columns) >= 2
+    
+    def _is_distributor_sheet(self, df):
+        """Check if sheet contains distributor data"""
+        required_columns = ['DISTRIBUTOR', 'MANTRI', 'SABHASAD']
+        existing_columns = [col for col in df.columns if any(req in col for req in required_columns)]
+        return len(existing_columns) >= 2
+    
+    def process_sales_sheet(self, df, file_name, sheet_name):
+        """Process sales data from sheet"""
+        try:
+            processed_rows = 0
+            
+            for index, row in df.iterrows():
+                try:
+                    # Skip header rows and empty rows
+                    if self._is_header_row(row) or pd.isna(row.iloc[0]):
+                        continue
+                    
+                    # Extract sales data (adjust column indices based on your Excel structure)
+                    invoice_no = str(row.iloc[0]) if len(row) > 0 else f"INV_{datetime.now().strftime('%Y%m%d%H%M%S')}_{index}"
+                    customer_name = str(row.iloc[1]) if len(row) > 1 else "Unknown Customer"
+                    product_name = str(row.iloc[2]) if len(row) > 2 else "Unknown Product"
+                    quantity = self._safe_float(row.iloc[3]) if len(row) > 3 else 0
+                    amount = self._safe_float(row.iloc[4]) if len(row) > 4 else 0
+                    
+                    # Get or create customer
+                    customer_id = self._get_or_create_customer(customer_name, "", "", "", "")
+                    
+                    # Get product ID
+                    product_id = self._get_product_id(product_name)
+                    
+                    if customer_id and product_id and quantity > 0:
+                        # Create sale
+                        sale_date = datetime.now().date()
+                        sale_items = [{
+                            'product_id': product_id,
+                            'quantity': quantity,
+                            'rate': amount / quantity if quantity > 0 else 0
+                        }]
+                        
+                        self.db.add_sale(invoice_no, customer_id, sale_date, sale_items)
+                        processed_rows += 1
+                        
+                except Exception as e:
+                    logger.warning(f"Error processing row {index} in sales sheet: {e}")
+                    continue
+            
+            logger.info(f"Processed {processed_rows} sales from {sheet_name}")
+            return processed_rows > 0
+            
+        except Exception as e:
+            logger.error(f"Error processing sales sheet: {e}")
+            return False
+    
+    def process_customer_sheet(self, df, file_name, sheet_name):
+        """Process customer data from sheet with duplicate handling"""
+        try:
+            processed_rows = 0
+            duplicate_rows = 0
+            error_rows = 0
+            
+            print(f"🔄 Processing customer sheet: {sheet_name} with {len(df)} rows")
+            
+            for index, row in df.iterrows():
+                try:
+                    # Skip header rows and empty rows
+                    if self._is_header_row(row) or pd.isna(row.iloc[0]):
+                        continue
+                    
+                    # Extract customer data
+                    customer_code = str(row.iloc[0]) if len(row) > 0 and pd.notna(row.iloc[0]) else None
+                    name = str(row.iloc[1]) if len(row) > 1 and pd.notna(row.iloc[1]) else "Unknown"
+                    mobile = str(row.iloc[2]) if len(row) > 2 and pd.notna(row.iloc[2]) else ""
+                    
+                    # Extract location - adjust indices based on your Excel structure
+                    village = str(row.iloc[3]) if len(row) > 3 and pd.notna(row.iloc[3]) else ""
+                    taluka = str(row.iloc[4]) if len(row) > 4 and pd.notna(row.iloc[4]) else ""
+                    district = str(row.iloc[5]) if len(row) > 5 and pd.notna(row.iloc[5]) else ""
+                    
+                    # If village is combined with name, split them
+                    if not village and "(" in name:
+                        name_parts = name.split("(")
+                        if len(name_parts) > 1:
+                            name = name_parts[0].strip()
+                            village = name_parts[1].replace(")", "").strip()
+                    
+                    # Skip if no name
+                    if not name or name == "Unknown":
+                        continue
+                    
+                    # Add customer to database (method now handles duplicates)
+                    customer_id = self.db.add_customer(name, mobile, village, taluka, district, customer_code)
+                    
+                    if customer_id and customer_id != -1:
+                        processed_rows += 1
+                        if processed_rows % 50 == 0:  # Progress update
+                            print(f"📊 Processed {processed_rows} customers...")
+                    else:
+                        duplicate_rows += 1
+                        
+                except Exception as e:
+                    error_rows += 1
+                    if error_rows <= 5:  # Only log first few errors
+                        print(f"❌ Error in row {index}: {e}")
+                    continue
+            
+            print(f"🎉 Customer processing complete: {processed_rows} added, {duplicate_rows} duplicates, {error_rows} errors")
+            return processed_rows > 0
+            
+        except Exception as e:
+            print(f"💥 Error processing customer sheet: {e}")
+            return False
+    
+    def process_distributor_sheet(self, df, file_name, sheet_name):
+        """Process distributor data from sheet"""
+        try:
+            processed_rows = 0
+            
+            # Clean the dataframe - convert column names to consistent format
+            df.columns = [str(col).strip().upper() for col in df.columns]
+            print(f"DEBUG: Processing distributor sheet with columns: {df.columns.tolist()}")
+            
+            for index, row in df.iterrows():
+                try:
+                    # Skip header rows and empty rows
+                    if self._is_header_row(row) or pd.isna(row.iloc[0]):
+                        print(f"DEBUG: Skipping row {index} - header or empty")
+                        continue
+                    
+                    print(f"DEBUG: Processing row {index}")
+                    
+                    # Extract distributor data based on YOUR ACTUAL COLUMNS
+                    # Map your Excel columns to database fields
+                    name = self._extract_distributor_name(row)  # We'll use Village + Taluka as name
+                    village = self._safe_get(row, 'Village', 1)
+                    taluka = self._safe_get(row, 'Taluka', 2) 
+                    district = self._safe_get(row, 'District', 3)
+                    mantri_name = self._safe_get(row, 'Mantri_Name', 4)
+                    mantri_mobile = self._safe_get(row, 'Mantri_Mobile', 5)
+                    sabhasad_count = self._safe_get_int(row, 'Sabhasad', 6)
+                    contact_in_group = self._safe_get_int(row, 'Contact_In_Group', 7)
+                    
+                    print(f"DEBUG: Extracted - Village: {village}, Taluka: {taluka}, Mantri: {mantri_name}")
+                    
+                    # Validate we have essential data
+                    if not village or not taluka:
+                        print(f"DEBUG: Skipping - missing village or taluka")
+                        continue
+                    
+                    # Create distributor name from village + taluka
+                    if not name:
+                        name = f"{village} - {taluka}"
+                    
+                    # Add distributor to database with ALL fields
+                    self.db.execute_query('''
+                    INSERT OR REPLACE INTO distributors 
+                    (name, village, taluka, district, mantri_name, mantri_mobile, sabhasad_count, contact_in_group)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                    ''', (name, village, taluka, district, mantri_name, mantri_mobile, sabhasad_count, contact_in_group))
+                    
+                    processed_rows += 1
+                    print(f"DEBUG: Successfully added distributor: {name}")
+                    
+                except Exception as e:
+                    logger.warning(f"Error processing row {index} in distributor sheet: {e}")
+                    continue
+            
+            logger.info(f"Processed {processed_rows} distributors from {sheet_name}")
+            return processed_rows > 0
+            
+        except Exception as e:
+            logger.error(f"Error processing distributor sheet: {e}")
+            return False
+
+    def _extract_distributor_name(self, row):
+        """Extract distributor name from village and taluka"""
+        village = self._safe_get(row, 'Village', 1)
+        taluka = self._safe_get(row, 'Taluka', 2)
+        
+        if village and taluka:
+            return f"{village} - {taluka}"
+        elif village:
+            return village
+        elif taluka:
+            return taluka
+        else:
+            return "Unknown Distributor"
+
+    def _safe_get(self, row, column_name, default_index):
+        """Safely get value from row by column name or index"""
+        try:
+            # Try by column name first
+            if column_name in row.index:
+                value = row[column_name]
+                if pd.isna(value):
+                    return ""
+                return str(value).strip()
+            
+            # Fallback to index
+            if len(row) > default_index:
+                value = row.iloc[default_index]
+                if pd.isna(value):
+                    return ""
+                return str(value).strip()
+            
+            return ""
+        except Exception:
+            return ""
+
+    def _safe_get_int(self, row, column_name, default_index):
+        """Safely get integer value from row"""
+        try:
+            str_value = self._safe_get(row, column_name, default_index)
+            if str_value and str_value.strip():
+                return int(float(str_value))  # Handle both int and float strings
+            return 0
+        except (ValueError, TypeError):
+            return 0
+    
+    def _is_header_row(self, row):
+        """Check if row is a header row - updated for your data"""
+        if len(row) == 0:
+            return True
+            
+        first_value = str(row.iloc[0]) if pd.notna(row.iloc[0]) else ""
+        first_value_upper = first_value.upper()
+        
+        # Header indicators for YOUR data
+        header_indicators = [
+            'DATE', 'VILLAGE', 'TALUKA', 'DISTRICT', 'MANTRI', 
+            'SABHASAD', 'CONTACT', 'TOTAL', 'SR', 'NO', 'NAME'
+        ]
+        
+        # If first value contains any header indicator, it's likely a header
+        return any(indicator in first_value_upper for indicator in header_indicators)
+        
+    def _safe_float(self, value):
+        """Safely convert value to float"""
+        try:
+            if pd.isna(value):
+                return 0.0
+            return float(value)
+        except (ValueError, TypeError):
+            return 0.0
+    
+    def _get_or_create_customer(self, name, mobile, village, taluka, district):
+        """Get existing customer or create new one"""
+        try:
+            # Check if customer exists
+            result = self.db.execute_query(
+                'SELECT customer_id FROM customers WHERE name = ? AND mobile = ?', 
+                (name, mobile)
+            )
+            
+            if result:
+                return result[0][0]
+            else:
+                # Create new customer
+                customer_code = f"CUST_{datetime.now().strftime('%Y%m%d%H%M%S')}"
+                self.db.add_customer(name, mobile, village, taluka, district, customer_code)
+                
+                # Get the new customer ID
+                result = self.db.execute_query(
+                    'SELECT customer_id FROM customers WHERE customer_code = ?', 
+                    (customer_code,)
+                )
+                return result[0][0] if result else None
+                
+        except Exception as e:
+            logger.error(f"Error getting/creating customer: {e}")
+            return None
+    
+    def _get_product_id(self, product_name):
+        """Get product ID from product name"""
+        clean_name = product_name.upper().strip()
+        return self.product_mapping.get(clean_name, None)
+    
+    def _extract_location_from_name(self, name):
+        """Extract village and taluka from customer name"""
+        name_upper = name.upper()
+        
+        locations = {
+            'AMIYAD': ('Amiyad', ''),
+            'AMVAD': ('Amvad', ''),
+            'ANKALAV': ('', 'Ankalav'),
+            'PETLAD': ('', 'Petlad'),
+            'BORSAD': ('', 'Borsad'),
+            'VADODARA': ('', 'Vadodara'),
+            'ANAND': ('', 'Anand'),
+            'NADIAD': ('', 'Nadiad')
+        }
+        
+        village, taluka = "", ""
+        for location, (v, t) in locations.items():
+            if location in name_upper:
+                if v: 
+                    village = v
+                if t: 
+                    taluka = t
+                break
+        
+        return village, taluka
+    
+
+
+    # Add to DataProcessor class in data_processor.py
+
+    def _is_sales_sheet(self, df):
+        """Enhanced sales sheet detection with better logging"""
+        columns_lower = [str(col).lower() for col in df.columns]
+        
+        print(f"\n🔍 ENHANCED SALES DETECTION:")
+        print(f"   All columns: {columns_lower}")
+        
+        sales_indicators = [
+            'invoice', 'sale', 'amount', 'product', 'quantity', 'rate', 
+            'total', 'price', 'bill', 'payment', 'item', 'qty'
+        ]
+        
+        found_indicators = []
+        for indicator in sales_indicators:
+            matching_cols = [col for col in columns_lower if indicator in col]
+            if matching_cols:
+                found_indicators.append((indicator, matching_cols))
+        
+        print(f"   Found sales indicators: {found_indicators}")
+        
+        score = len(found_indicators)
+        print(f"   Sales detection score: {score}")
+        
+        return score >= 2
+
+    def process_sales_sheet(self, df, file_name, sheet_name):
+        """Enhanced sales data processing with better logging"""
+        try:
+            processed_rows = 0
+            print(f"🔄 Processing sales sheet: {sheet_name} with {len(df)} rows")
+            
+            for index, row in df.iterrows():
+                try:
+                    # Skip header rows and empty rows
+                    if self._is_header_row(row) or pd.isna(row.iloc[0]):
+                        continue
+                    
+                    print(f"🔧 Processing row {index}")
+                    
+                    # Extract sales data with flexible column mapping
+                    invoice_no = self._extract_sales_value(row, 'invoice', 0, f"INV_{datetime.now().strftime('%Y%m%d%H%M%S')}_{index}")
+                    customer_name = self._extract_sales_value(row, 'customer', 1, "Unknown Customer")
+                    product_name = self._extract_sales_value(row, 'product', 2, "Unknown Product")
+                    quantity = self._safe_float(self._extract_sales_value(row, 'quantity', 3, 0))
+                    amount = self._safe_float(self._extract_sales_value(row, 'amount', 4, 0))
+                    
+                    print(f"   Extracted - Invoice: '{invoice_no}', Customer: '{customer_name}', Product: '{product_name}', Qty: {quantity}, Amount: {amount}")
+                    
+                    # Validate essential data
+                    if not customer_name or customer_name == "Unknown Customer":
+                        print(f"   ⚠️ Skipping - invalid customer name")
+                        continue
+                    
+                    if quantity <= 0:
+                        print(f"   ⚠️ Skipping - invalid quantity: {quantity}")
+                        continue
+                    
+                    if amount <= 0:
+                        print(f"   ⚠️ Skipping - invalid amount: {amount}")
+                        continue
+                    
+                    # Get or create customer
+                    customer_id = self._get_or_create_customer(customer_name, "", "", "", "")
+                    if not customer_id:
+                        print(f"   ⚠️ Skipping - could not get/create customer")
+                        continue
+                    
+                    # Get product ID
+                    product_id = self._get_product_id(product_name)
+                    if not product_id:
+                        print(f"   ⚠️ Skipping - product not found: '{product_name}'")
+                        print(f"   Available products: {list(self.product_mapping.keys())}")
+                        continue
+                    
+                    # Calculate rate
+                    rate = amount / quantity if quantity > 0 else 0
+                    
+                    # Create sale items
+                    sale_date = datetime.now().date()
+                    sale_items = [{
+                        'product_id': product_id,
+                        'quantity': quantity,
+                        'rate': rate
+                    }]
+                    
+                    # Generate proper invoice number
+                    if not invoice_no or invoice_no.startswith('INV_'):
+                        invoice_no = self.db.generate_invoice_number()
+                    
+                    print(f"   Creating sale - Customer ID: {customer_id}, Product ID: {product_id}")
+                    
+                    # Add sale to database
+                    sale_id = self.db.add_sale(invoice_no, customer_id, sale_date, sale_items)
+                    
+                    if sale_id and sale_id > 0:
+                        processed_rows += 1
+                        print(f"   ✅ Successfully created sale ID: {sale_id}")
+                    else:
+                        print(f"   ❌ Failed to create sale")
+                        
+                except Exception as e:
+                    print(f"   ❌ Error in row {index}: {e}")
+                    import traceback
+                    traceback.print_exc()
+                    continue
+            
+            print(f"🎉 Processed {processed_rows} sales from {sheet_name}")
+            return processed_rows > 0
+            
+        except Exception as e:
+            print(f"💥 Error processing sales sheet: {e}")
+            import traceback
+            traceback.print_exc()
+            return False
+    def _extract_sales_value(self, row, field_name, default_index, default_value):
+        """Extract sales values with flexible column matching"""
+        # Try to find column by name
+        for col_name in row.index:
+            if field_name in str(col_name).lower():
+                value = row[col_name]
+                if pd.notna(value):
+                    return str(value).strip()
+        
+        # Fallback to index
+        if len(row) > default_index:
+            value = row.iloc[default_index]
+            if pd.notna(value):
+                return str(value).strip()
+        
+        return default_value
+
+    # Add to DataProcessor class
+
+    def _is_payment_sheet(self, df):
+        """Detect payment sheets"""
+        columns_lower = [str(col).lower() for col in df.columns]
+        
+        payment_indicators = [
+            'payment', 'paid', 'amount', 'invoice', 'date', 'method',
+            'cash', 'gpay', 'cheque', 'bank', 'rrn', 'reference'
+        ]
+        
+        score = sum(1 for indicator in payment_indicators 
+                    if any(indicator in col for col in columns_lower))
+        
+        print(f"🔍 Payment detection - Score: {score}, Columns: {columns_lower}")
+        return score >= 2
+
+    def process_payment_sheet(self, df, file_name, sheet_name):
+        """Process payment data from sheet"""
+        try:
+            processed_rows = 0
+            print(f"🔄 Processing payment sheet: {sheet_name}")
+            
+            for index, row in df.iterrows():
+                try:
+                    if self._is_header_row(row) or pd.isna(row.iloc[0]):
+                        continue
+                    
+                    # Extract payment data
+                    invoice_no = self._extract_sales_value(row, 'invoice', 0, "")
+                    amount = self._safe_float(self._extract_sales_value(row, 'amount', 1, 0))
+                    payment_date = self._extract_sales_value(row, 'date', 2, datetime.now().date())
+                    payment_method = self._extract_sales_value(row, 'method', 3, "Cash")
+                    
+                    if invoice_no and amount > 0:
+                        # Find sale by invoice number
+                        sale_result = self.db.execute_query(
+                            'SELECT sale_id FROM sales WHERE invoice_no = ?',
+                            (invoice_no,),
+                            log_action=False
+                        )
+                        
+                        if sale_result:
+                            sale_id = sale_result[0][0]
+                            
+                            # Add payment
+                            self.db.execute_query('''
+                            INSERT INTO payments (sale_id, payment_date, payment_method, amount)
+                            VALUES (?, ?, ?, ?)
+                            ''', (sale_id, payment_date, payment_method, amount))
+                            
+                            processed_rows += 1
+                            print(f"✅ Processed payment for invoice {invoice_no}")
+                    
+                except Exception as e:
+                    print(f"❌ Error processing payment row {index}: {e}")
+                    continue
+            
+            print(f"🎉 Processed {processed_rows} payments from {sheet_name}")
+            return processed_rows > 0
+            
+        except Exception as e:
+            print(f"💥 Error processing payment sheet: {e}")
+            return False
+
+
+    def _is_customer_sheet(self, df):
+        """Check if sheet contains customer data - IMPROVED"""
+        columns_lower = [str(col).lower() for col in df.columns]
+        
+        customer_indicators = [
+            'customer', 'name', 'mobile', 'phone', 'village', 'taluka', 
+            'district', 'code', 'contact'
+        ]
+        
+        score = sum(1 for indicator in customer_indicators 
+                if any(indicator in col for col in columns_lower))
+        
+        print(f"🔍 Customer sheet detection - Score: {score}, Columns: {columns_lower}")
+        return score >= 2
+
+    def _is_distributor_sheet(self, df):
+        """Enhanced distributor sheet detection with better logging"""
+        columns_lower = [str(col).lower() for col in df.columns]
+        
+        print(f"\n🔍 ENHANCED DISTRIBUTOR DETECTION:")
+        print(f"   All columns: {columns_lower}")
+        
+        distributor_indicators = [
+            'distributor', 'mantri', 'sabhasad', 'contact_in_group',
+            'village', 'taluka', 'district', 'leader', 'team', 'sabh'
+        ]
+        
+        found_indicators = []
+        for indicator in distributor_indicators:
+            matching_cols = [col for col in columns_lower if indicator in col]
+            if matching_cols:
+                found_indicators.append((indicator, matching_cols))
+        
+        print(f"   Found indicators: {found_indicators}")
+        
+        score = len(found_indicators)
+        print(f"   Detection score: {score}")
+        
+        # More flexible detection - lower threshold
+        return score >= 1  # Even if we find just one indicator, try processing
+
+    def process_single_sheet(self, df, sheet_name, file_name):
+        """Process a single sheet with detailed logging"""
+        print(f"🔄 Processing sheet: {sheet_name} from {file_name}")
+        
+        if self._is_sales_sheet(df):
+            print("✅ Detected as SALES sheet")
+            return self.process_sales_sheet(df, file_name, sheet_name)
+        elif self._is_customer_sheet(df):
+            print("✅ Detected as CUSTOMER sheet") 
+            return self.process_customer_sheet(df, file_name, sheet_name)
+        elif self._is_distributor_sheet(df):
+            print("✅ Detected as DISTRIBUTOR sheet")
+            return self.process_distributor_sheet(df, file_name, sheet_name)
+        else:
+            print("❓ Unknown sheet type - trying customer processing as fallback")
+            return self.process_customer_sheet(df, file_name, sheet_name)
+        
+    def process_excel_file(self, file_path):
+        """Enhanced file processing with all data types"""
+        try:
+            file_name = os.path.basename(file_path)
+            print(f"🚀 Processing file: {file_name}")
+            
+            excel_file = pd.ExcelFile(file_path)
+            processed_sheets = 0
+            
+            for sheet_name in excel_file.sheet_names:
+                df = pd.read_excel(file_path, sheet_name=sheet_name)
+                df_clean = self._clean_dataframe(df)
+                
+                print(f"\n📊 Sheet: {sheet_name}")
+                print(f"   Columns: {df_clean.columns.tolist()}")
+                
+                # Check all types with priority
+                is_payment = self._is_payment_sheet(df_clean)
+                is_sales = self._is_sales_sheet(df_clean)
+                is_customer = self._is_customer_sheet(df_clean)
+                is_distributor = self._is_distributor_sheet(df_clean)
+                
+                print(f"   Detection - Payment: {is_payment}, Sales: {is_sales}, Customer: {is_customer}, Distributor: {is_distributor}")
+                
+                processed = False
+                if is_payment:
+                    print("   💳 Processing as PAYMENT sheet")
+                    processed = self.process_payment_sheet(df_clean, file_name, sheet_name)
+                elif is_sales:
+                    print("   💰 Processing as SALES sheet")
+                    processed = self.process_sales_sheet(df_clean, file_name, sheet_name)
+                elif is_distributor:
+                    print("   🤝 Processing as DISTRIBUTOR sheet")
+                    processed = self.process_distributor_sheet(df_clean, file_name, sheet_name)
+                elif is_customer:
+                    print("   👥 Processing as CUSTOMER sheet")
+                    processed = self.process_customer_sheet(df_clean, file_name, sheet_name)
+                else:
+                    print("   ❓ Unknown sheet type")
+                
+                if processed:
+                    processed_sheets += 1
+                    print(f"   ✅ Successfully processed")
+                else:
+                    print(f"   ❌ Failed to process")
+            
+            print(f"\n🎉 File processing complete: {processed_sheets}/{len(excel_file.sheet_names)} sheets processed")
+            return processed_sheets > 0
+            
+        except Exception as e:
+            print(f"💥 Error processing file {file_path}: {e}")
+            return False

database.py

@@ -0,0 +1,1106 @@
+import sqlite3
+import pandas as pd
+import os
+import logging
+from datetime import datetime, timedelta
+from typing import List, Dict, Any, Optional
+import random 
+
+# Set up logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class DatabaseManager:
+    def __init__(self, db_path="sales_management.db"):
+        self.db_path = db_path
+        self._is_logging = False  # Prevent recursion
+        self.init_database()
+
+    def get_connection(self):
+        """Get database connection with error handling"""
+        try:
+            conn = sqlite3.connect(self.db_path)
+            conn.row_factory = sqlite3.Row  # This enables column access by name
+            return conn
+        except sqlite3.Error as e:
+            logger.error(f"Database connection error: {e}")
+            raise
+
+    def init_database(self):
+        """Initialize database with all tables and relationships"""
+        conn = self.get_connection()
+
+        try:
+            # Customers table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS customers (
+                customer_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                customer_code TEXT UNIQUE,
+                name TEXT NOT NULL,
+                mobile TEXT,
+                village TEXT,
+                taluka TEXT,
+                district TEXT,
+                status TEXT DEFAULT 'Active',
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """)
+
+            # Distributors table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS distributors (
+                distributor_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                name TEXT NOT NULL,
+                village TEXT,
+                taluka TEXT,
+                district TEXT,
+                mantri_name TEXT,
+                mantri_mobile TEXT,
+                sabhasad_count INTEGER DEFAULT 0,
+                contact_in_group INTEGER DEFAULT 0,
+                status TEXT DEFAULT 'Active',
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """)
+
+            # Products table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS products (
+                product_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                product_name TEXT UNIQUE NOT NULL,
+                packing_type TEXT,
+                capacity_ltr REAL,
+                category TEXT,
+                standard_rate REAL,
+                is_active INTEGER DEFAULT 1,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """)
+
+            # Sales table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS sales (
+                sale_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                invoice_no TEXT UNIQUE NOT NULL,
+                customer_id INTEGER,
+                sale_date DATE,
+                total_amount REAL DEFAULT 0,
+                total_liters REAL DEFAULT 0,
+                payment_status TEXT DEFAULT 'Pending',
+                notes TEXT,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ON DELETE SET NULL
+            )
+            """)
+
+            # Sale items table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS sale_items (
+                item_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                sale_id INTEGER,
+                product_id INTEGER,
+                quantity INTEGER,
+                rate REAL,
+                amount REAL,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (sale_id) REFERENCES sales (sale_id) ON DELETE CASCADE,
+                FOREIGN KEY (product_id) REFERENCES products (product_id) ON DELETE SET NULL
+            )
+            """)
+
+            # Payments table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS payments (
+                payment_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                sale_id INTEGER,
+                payment_date DATE,
+                payment_method TEXT,
+                amount REAL,
+                rrn TEXT,
+                reference TEXT,
+                status TEXT DEFAULT 'Completed',
+                notes TEXT,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (sale_id) REFERENCES sales (sale_id) ON DELETE CASCADE
+            )
+            """)
+
+            # Demos table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS demos (
+                demo_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                customer_id INTEGER,
+                distributor_id INTEGER,
+                demo_date DATE,
+                demo_time TIME,
+                product_id INTEGER,
+                quantity_provided INTEGER,
+                follow_up_date DATE,
+                conversion_status TEXT DEFAULT 'Not Converted',
+                notes TEXT,
+                demo_location TEXT,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ON DELETE SET NULL,
+                FOREIGN KEY (distributor_id) REFERENCES distributors (distributor_id) ON DELETE SET NULL,
+                FOREIGN KEY (product_id) REFERENCES products (product_id) ON DELETE SET NULL
+            )
+            """)
+
+            # WhatsApp logs table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS whatsapp_logs (
+                log_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                customer_id INTEGER,
+                distributor_id INTEGER,
+                message_type TEXT,
+                message_content TEXT,
+                status TEXT,
+                sent_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                response TEXT,
+                FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ON DELETE SET NULL,
+                FOREIGN KEY (distributor_id) REFERENCES distributors (distributor_id) ON DELETE SET NULL
+            )
+            """)
+
+            # Follow-ups table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS follow_ups (
+                follow_up_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                customer_id INTEGER,
+                distributor_id INTEGER,
+                demo_id INTEGER,
+                follow_up_date DATE,
+                follow_up_type TEXT,
+                notes TEXT,
+                status TEXT DEFAULT 'Pending',
+                next_follow_up_date DATE,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ON DELETE SET NULL,
+                FOREIGN KEY (distributor_id) REFERENCES distributors (distributor_id) ON DELETE SET NULL,
+                FOREIGN KEY (demo_id) REFERENCES demos (demo_id) ON DELETE SET NULL
+            )
+            """)
+
+            # System logs table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS system_logs (
+                log_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                log_type TEXT,
+                log_message TEXT,
+                table_name TEXT,
+                record_id INTEGER,
+                action TEXT,
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                user_info TEXT
+            )
+            """)
+
+            # Rollback logs table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS rollback_logs (
+                rollback_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                table_name TEXT,
+                record_id INTEGER,
+                old_data TEXT,
+                new_data TEXT,
+                action TEXT,
+                rollback_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                rolled_back_by TEXT
+            )
+            """)
+
+            # Offers table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS offers (
+                offer_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                offer_name TEXT NOT NULL,
+                offer_description TEXT,
+                product_id INTEGER,
+                discount_percentage REAL,
+                discount_amount REAL,
+                start_date DATE,
+                end_date DATE,
+                status TEXT DEFAULT 'Active',
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (product_id) REFERENCES products (product_id) ON DELETE SET NULL
+            )
+            """)
+
+            # Demo teams table
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS demo_teams (
+                team_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                team_name TEXT NOT NULL,
+                team_leader TEXT,
+                team_members TEXT,
+                assigned_villages TEXT,
+                status TEXT DEFAULT 'Active',
+                created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """)
+
+            conn.commit()
+            logger.info("Database tables initialized successfully")
+
+        except sqlite3.Error as e:
+            logger.error(f"Error initializing database: {e}")
+            raise
+        finally:
+            conn.close()
+
+        self.initialize_default_data()
+        self.migrate_database()
+        self.create_indexes()
+
+    def create_indexes(self):
+        """Create indexes for better performance"""
+        conn = self.get_connection()
+
+        try:
+            # Create indexes for frequently queried columns
+            indexes = [
+                "CREATE INDEX IF NOT EXISTS idx_customers_village ON customers(village)",
+                "CREATE INDEX IF NOT EXISTS idx_customers_mobile ON customers(mobile)",
+                "CREATE INDEX IF NOT EXISTS idx_sales_customer_id ON sales(customer_id)",
+                "CREATE INDEX IF NOT EXISTS idx_sales_date ON sales(sale_date)",
+                "CREATE INDEX IF NOT EXISTS idx_sales_invoice ON sales(invoice_no)",
+                "CREATE INDEX IF NOT EXISTS idx_payments_sale_id ON payments(sale_id)",
+                "CREATE INDEX IF NOT EXISTS idx_demos_customer_id ON demos(customer_id)",
+                "CREATE INDEX IF NOT EXISTS idx_demos_date ON demos(demo_date)",
+                "CREATE INDEX IF NOT EXISTS idx_sale_items_sale_id ON sale_items(sale_id)",
+                "CREATE INDEX IF NOT EXISTS idx_follow_ups_date ON follow_ups(follow_up_date)",
+                "CREATE INDEX IF NOT EXISTS idx_whatsapp_customer_id ON whatsapp_logs(customer_id)",
+            ]
+
+            for index_sql in indexes:
+                conn.execute(index_sql)
+
+            conn.commit()
+            logger.info("Database indexes created successfully")
+
+        except sqlite3.Error as e:
+            logger.error(f"Error creating indexes: {e}")
+        finally:
+            conn.close()
+
+    def migrate_database(self):
+        """Migrate existing database to add missing columns"""
+        conn = self.get_connection()
+        try:
+            cursor = conn.cursor()
+
+            # Check if demo_time column exists
+            cursor.execute("PRAGMA table_info(demos)")
+            columns = [column[1] for column in cursor.fetchall()]
+
+            # Add demo_time column if it doesn't exist
+            if "demo_time" not in columns:
+                cursor.execute("ALTER TABLE demos ADD COLUMN demo_time TIME")
+                logger.info("Added demo_time column to demos table")
+
+            # Add demo_location column if it doesn't exist
+            if "demo_location" not in columns:
+                cursor.execute("ALTER TABLE demos ADD COLUMN demo_location TEXT")
+                logger.info("Added demo_location column to demos table")
+
+            conn.commit()
+            logger.info("Database migration completed successfully")
+
+        except sqlite3.Error as e:
+            logger.error(f"Error during database migration: {e}")
+            conn.rollback()
+        finally:
+            conn.close()
+
+    def initialize_default_data(self):
+        """Initialize with default products and demo teams"""
+        default_products = [
+            ("1 LTR PLASTIC JAR", "PLASTIC_JAR", 1.0, "Regular", 95),
+            ("2 LTR PLASTIC JAR", "PLASTIC_JAR", 2.0, "Regular", 185),
+            ("5 LTR PLASTIC JAR", "PLASTIC_JAR", 5.0, "Regular", 460),
+            ("5 LTR STEEL BARNI", "STEEL_BARNI", 5.0, "Premium", 680),
+            ("10 LTR STEEL BARNI", "STEEL_BARNI", 10.0, "Premium", 1300),
+            ("20 LTR STEEL BARNI", "STEEL_BARNI", 20.0, "Premium", 2950),
+            ("20 LTR PLASTIC CAN", "PLASTIC_CAN", 20.0, "Regular", 2400),
+            ("1 LTR PET BOTTLE", "PET_BOTTLE", 1.0, "Regular", 85),
+        ]
+
+        default_teams = [
+            (
+                "Team A - North Region",
+                "Rajesh Kumar",
+                "Mohan, Suresh, Priya",
+                "Amiyad, Amvad, Ankalav",
+            ),
+            (
+                "Team B - South Region",
+                "Sunil Patel",
+                "Anita, Vijay, Deepak",
+                "Petlad, Borsad, Vadodara",
+            ),
+        ]
+
+        conn = self.get_connection()
+        try:
+            # Insert default products
+            for product in default_products:
+                conn.execute(
+                    """
+                INSERT OR IGNORE INTO products (product_name, packing_type, capacity_ltr, category, standard_rate)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                    product,
+                )
+
+            # Insert default demo teams
+            for team in default_teams:
+                conn.execute(
+                    """
+                INSERT OR IGNORE INTO demo_teams (team_name, team_leader, team_members, assigned_villages)
+                VALUES (?, ?, ?, ?)
+                """,
+                    team,
+                )
+
+            conn.commit()
+            logger.info("Default data initialized successfully")
+
+        except sqlite3.Error as e:
+            logger.error(f"Error initializing default data: {e}")
+        finally:
+            conn.close()
+
+    def _execute_query_internal(self, query: str, params: tuple = None) -> List[tuple]:
+        """Internal method to execute SQL query without logging"""
+        conn = self.get_connection()
+        try:
+            cursor = conn.cursor()
+            if params:
+                cursor.execute(query, params)
+            else:
+                cursor.execute(query)
+
+            # Only try to fetch results for SELECT queries
+            if query.strip().upper().startswith("SELECT"):
+                result = cursor.fetchall()
+            elif query.strip().upper().startswith("INSERT"):
+                # For INSERT queries, return the lastrowid as a single-row result
+                result = [(cursor.lastrowid,)]
+            else:
+                result = []
+
+            conn.commit()
+            return result
+
+        except sqlite3.Error as e:
+            logger.error(f"Database query error: {e}")
+            conn.rollback()
+            raise
+        finally:
+            conn.close()
+
+    def execute_query(
+        self, query: str, params: tuple = None, log_action: bool = True
+    ) -> List[tuple]:
+        """Execute a SQL query with comprehensive error handling"""
+        try:
+            result = self._execute_query_internal(query, params)
+
+            # Log the query execution (but avoid recursion)
+            if log_action and not self._is_logging:
+                try:
+                    self._is_logging = True
+                    self._execute_query_internal(
+                        """
+                    INSERT INTO system_logs (log_type, log_message, table_name, record_id, action)
+                    VALUES (?, ?, ?, ?, ?)
+                    """,
+                        (
+                            "QUERY_EXECUTION",
+                            f"Executed query: {query[:100]}...",
+                            None,
+                            None,
+                            "EXECUTE",
+                        ),
+                    )
+                except Exception as e:
+                    logger.error(f"Error logging system action: {e}")
+                finally:
+                    self._is_logging = False
+
+            return result
+        except Exception as e:
+            logger.error(f"Error in execute_query: {e}")
+            return []  # Return empty list instead of raising exception
+
+    def get_dataframe(
+        self, table_name: str = None, query: str = None, params: tuple = None
+    ) -> pd.DataFrame:
+        """Get table data as DataFrame with flexible query support"""
+        conn = self.get_connection()
+        try:
+            if query:
+                df = pd.read_sql_query(query, conn, params=params)
+            else:
+                df = pd.read_sql_query(f"SELECT * FROM {table_name}", conn)
+            return df
+        except Exception as e:
+            logger.error(
+                f"Error getting DataFrame for {table_name if table_name else 'query'}: {e}"
+            )
+            # Return empty DataFrame with proper structure
+            return pd.DataFrame()
+        finally:
+            conn.close()
+
+    def add_customer(
+        self,
+        name: str,
+        mobile: str = "",
+        village: str = "",
+        taluka: str = "",
+        district: str = "",
+        customer_code: str = None,
+    ) -> int:
+        """Add a new customer with duplicate handling"""
+
+        # Generate customer code if not provided
+        if not customer_code:
+            customer_code = f"CUST{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(100, 999)}"
+
+        try:
+            # Check if customer already exists (by mobile or similar name+village)
+            existing_customer = self.execute_query(
+                "SELECT customer_id FROM customers WHERE mobile = ? OR (name = ? AND village = ?)",
+                (mobile, name, village),
+                log_action=False,
+            )
+
+            if existing_customer:
+                # Customer already exists, return existing ID
+                return existing_customer[0][0]
+
+            # If customer_code already exists, generate a new one
+            max_attempts = 5
+            for attempt in range(max_attempts):
+                try:
+                    result = self.execute_query(
+                        """
+                    INSERT INTO customers (customer_code, name, mobile, village, taluka, district)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                    """,
+                        (customer_code, name, mobile, village, taluka, district),
+                        log_action=False,
+                    )
+                    break
+                except sqlite3.IntegrityError as e:
+                    if (
+                        "UNIQUE constraint failed: customers.customer_code" in str(e)
+                        and attempt < max_attempts - 1
+                    ):
+                        # Generate new unique customer code
+                        customer_code = f"CUST{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(1000, 9999)}"
+                        continue
+                    else:
+                        raise e
+
+            # Get the inserted customer_id
+            customer_id = self.execute_query(
+                "SELECT last_insert_rowid()", log_action=False
+            )[0][0]
+
+            self.log_system_action(
+                "CUSTOMER_ADD",
+                f"Added customer: {name}",
+                "customers",
+                customer_id,
+                "INSERT",
+            )
+
+            return customer_id
+        except Exception as e:
+            logger.error(f"Error adding customer: {e}")
+            # Return a fallback - this won't be in database but prevents crashes
+            return -1
+
+    def add_distributor(
+        self,
+        name: str,
+        village: str = "",
+        taluka: str = "",
+        district: str = "",
+        mantri_name: str = "",
+        mantri_mobile: str = "",
+        sabhasad_count: int = 0,
+        contact_in_group: int = 0,
+        status: str = "Active",
+    ) -> int:
+        """Add a new distributor with duplicate handling"""
+
+        try:
+            # Check if distributor already exists
+            existing_distributor = self.execute_query(
+                "SELECT distributor_id FROM distributors WHERE name = ? AND village = ? AND taluka = ?",
+                (name, village, taluka),
+                log_action=False,
+            )
+
+            if existing_distributor:
+                # Distributor already exists, return existing ID
+                return existing_distributor[0][0]
+
+            # Insert new distributor
+            self.execute_query(
+                """
+            INSERT INTO distributors (name, village, taluka, district, mantri_name, mantri_mobile,
+                                    sabhasad_count, contact_in_group, status)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+                (
+                    name,
+                    village,
+                    taluka,
+                    district,
+                    mantri_name,
+                    mantri_mobile,
+                    sabhasad_count,
+                    contact_in_group,
+                    status,
+                ),
+                log_action=False,
+            )
+
+            # Get the inserted distributor_id
+            distributor_id = self.execute_query(
+                "SELECT last_insert_rowid()", log_action=False
+            )[0][0]
+
+            self.log_system_action(
+                "DISTRIBUTOR_ADD",
+                f"Added distributor: {name}",
+                "distributors",
+                distributor_id,
+                "INSERT",
+            )
+
+            return distributor_id
+
+        except Exception as e:
+            logger.error(f"Error adding distributor: {e}")
+            return -1
+
+    def get_distributor_by_location(self, village: str, taluka: str) -> Optional[Dict]:
+        """Get distributor by village and taluka"""
+        try:
+            result = self.execute_query(
+                "SELECT * FROM distributors WHERE village = ? AND taluka = ?",
+                (village, taluka),
+                log_action=False,
+            )
+            if result:
+                return dict(result[0])
+            return None
+        except Exception as e:
+            logger.error(f"Error getting distributor by location: {e}")
+            return None
+
+    def distributor_exists(self, name: str, village: str, taluka: str) -> bool:
+        """Check if distributor already exists"""
+        try:
+            result = self.execute_query(
+                "SELECT distributor_id FROM distributors WHERE name = ? AND village = ? AND taluka = ?",
+                (name, village, taluka),
+                log_action=False,
+            )
+            return len(result) > 0
+        except Exception as e:
+            logger.error(f"Error checking distributor existence: {e}")
+            return False
+
+    # In your DatabaseManager class in database.py, replace the generate_invoice_number method:
+
+    def generate_invoice_number(self):
+        """Generate automatic invoice number in format: INVCLmmyyserial"""
+        try:
+            # Get current date components
+            now = datetime.now()
+            month = now.strftime("%m")  # Two-digit month
+            year = now.strftime("%y")  # Two-digit year
+
+            # Get the last invoice number for this month-year
+            result = self.execute_query(
+                "SELECT invoice_no FROM sales WHERE invoice_no LIKE ? ORDER BY sale_id DESC LIMIT 1",
+                (f"INVCL{month}{year}%",),
+                log_action=False,
+            )
+
+            if result:
+                last_invoice = result[0][0]
+                # Extract serial number and increment
+                try:
+                    # Format: INVCLmmyyXXX
+                    serial_part = last_invoice[8:]  # Get part after INVCLmmyy
+                    last_serial = int(serial_part)
+                    new_serial = last_serial + 1
+                except ValueError:
+                    new_serial = 1
+            else:
+                # First invoice of the month-year
+                new_serial = 1
+
+            # Format: INVCL + month(2) + year(2) + serial(3 digits)
+            return f"INVCL{month}{year}{new_serial:03d}"
+
+        except Exception as e:
+            logger.error(f"Error generating invoice number: {e}")
+            # Fallback: timestamp-based
+            return f"INVCL{int(datetime.now().timestamp())}"
+
+    # Or if you want a more flexible version with configurable prefix:
+    def generate_invoice_number(self, prefix="INVCL"):
+        """Generate automatic invoice number in format: PREFIXmmyyserial"""
+        try:
+            now = datetime.now()
+            month = now.strftime("%m")
+            year = now.strftime("%y")
+
+            result = self.execute_query(
+                "SELECT invoice_no FROM sales WHERE invoice_no LIKE ? ORDER BY sale_id DESC LIMIT 1",
+                (f"{prefix}{month}{year}%",),
+                log_action=False,
+            )
+
+            if result:
+                last_invoice = result[0][0]
+                try:
+                    # Remove prefix and date part, get serial
+                    serial_part = last_invoice[
+                        len(prefix) + 4 :
+                    ]  # prefix + 4 digits (mmyy)
+                    last_serial = int(serial_part)
+                    new_serial = last_serial + 1
+                except ValueError:
+                    new_serial = 1
+            else:
+                new_serial = 1
+
+            return f"{prefix}{month}{year}{new_serial:03d}"
+
+        except Exception as e:
+            logger.error(f"Error generating invoice number: {e}")
+            return f"{prefix}{int(datetime.now().timestamp())}"
+
+    # Add to your DatabaseManager class in database.py
+
+    def add_sale(
+        self,
+        invoice_no: str,
+        customer_id: int,
+        sale_date,
+        items: List[Dict],
+        payments: List[Dict] = None,
+        notes: str = "",
+    ) -> int:
+        """Add a new sale with items and optional payments - ENHANCED"""
+        conn = self.get_connection()
+        try:
+            cursor = conn.cursor()
+
+            # Calculate total amount and liters
+            total_amount = sum(item["quantity"] * item["rate"] for item in items)
+            total_liters = sum(item.get("liters", 0) for item in items)
+
+            print(
+                f"🔧 DEBUG: Creating sale - Invoice: {invoice_no}, Customer: {customer_id}, Total: {total_amount}"
+            )  # DEBUG
+
+            # Add sale record
+            cursor.execute(
+                """
+            INSERT INTO sales (invoice_no, customer_id, sale_date, total_amount, total_liters, notes)
+            VALUES (?, ?, ?, ?, ?, ?)
+            """,
+                (invoice_no, customer_id, sale_date, total_amount, total_liters, notes),
+            )
+
+            # Get the sale ID
+            sale_id = cursor.lastrowid
+            print(f"🔧 DEBUG: Sale created with ID: {sale_id}")  # DEBUG
+
+            # Add sale items
+            for item in items:
+                amount = item["quantity"] * item["rate"]
+                print(
+                    f"🔧 DEBUG: Adding item - Product: {item['product_id']}, Qty: {item['quantity']}, Rate: {item['rate']}"
+                )  # DEBUG
+
+                cursor.execute(
+                    """
+                INSERT INTO sale_items (sale_id, product_id, quantity, rate, amount)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                    (
+                        sale_id,
+                        item["product_id"],
+                        item["quantity"],
+                        item["rate"],
+                        amount,
+                    ),
+                )
+
+            # Add payments if provided
+            if payments:
+                for payment in payments:
+                    cursor.execute(
+                        """
+                    INSERT INTO payments (sale_id, payment_date, payment_method, amount, rrn, reference)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                    """,
+                        (
+                            sale_id,
+                            payment["payment_date"],
+                            payment["method"],
+                            payment["amount"],
+                            payment.get("rrn", ""),
+                            payment.get("reference", ""),
+                        ),
+                    )
+
+            conn.commit()
+
+            # Update payment status
+            self._update_payment_status(sale_id)
+
+            print(f"🔧 DEBUG: Sale {sale_id} completed successfully")  # DEBUG
+            return sale_id
+
+        except Exception as e:
+            conn.rollback()
+            logger.error(f"Error adding sale: {e}")
+            print(f"❌ ERROR in add_sale: {e}")  # DEBUG
+            raise
+        finally:
+            conn.close()
+
+    def _update_payment_status(self, sale_id: int):
+        """Update payment status for a sale"""
+        conn = self.get_connection()
+        try:
+            # Get total paid amount
+            cursor = conn.cursor()
+            cursor.execute(
+                "SELECT COALESCE(SUM(amount), 0) FROM payments WHERE sale_id = ?",
+                (sale_id,),
+            )
+            total_paid = cursor.fetchone()[0]
+
+            # Get sale total
+            cursor.execute(
+                "SELECT total_amount FROM sales WHERE sale_id = ?", (sale_id,)
+            )
+            sale_total = cursor.fetchone()[0]
+
+            # Determine payment status
+            if total_paid >= sale_total:
+                status = "Paid"
+            elif total_paid > 0:
+                status = "Partial"
+            else:
+                status = "Pending"
+
+            # Update status
+            cursor.execute(
+                "UPDATE sales SET payment_status = ? WHERE sale_id = ?",
+                (status, sale_id),
+            )
+            conn.commit()
+
+        except Exception as e:
+            logger.error(f"Error updating payment status: {e}")
+        finally:
+            conn.close()
+
+    def get_pending_payments(self) -> pd.DataFrame:
+        """Get all pending payments with customer details"""
+        return self.get_dataframe(
+            "sales",
+            """
+        SELECT s.sale_id, s.invoice_no, s.sale_date, c.name as customer_name,
+               c.mobile, c.village, s.total_amount,
+               (s.total_amount - COALESCE(SUM(p.amount), 0)) as pending_amount,
+               COALESCE(SUM(p.amount), 0) as paid_amount
+        FROM sales s
+        LEFT JOIN customers c ON s.customer_id = c.customer_id
+        LEFT JOIN payments p ON s.sale_id = p.sale_id
+        WHERE s.payment_status IN ('Pending', 'Partial')
+        GROUP BY s.sale_id
+        HAVING pending_amount > 0
+        ORDER BY s.sale_date DESC
+        """,
+        )
+
+    def get_demo_conversions(self) -> pd.DataFrame:
+        """Get demo conversion statistics with details"""
+        return self.get_dataframe(
+            "demos",
+            """
+        SELECT d.*, c.name as customer_name, p.product_name,
+               dist.name as distributor_name, c.village, c.taluka,
+               CASE WHEN d.conversion_status = 'Converted' THEN 1 ELSE 0 END as converted
+        FROM demos d
+        LEFT JOIN customers c ON d.customer_id = c.customer_id
+        LEFT JOIN products p ON d.product_id = p.product_id
+        LEFT JOIN distributors dist ON d.distributor_id = dist.distributor_id
+        ORDER BY d.demo_date DESC
+        """,
+        )
+
+    def get_sales_analytics(self, start_date: str = None, end_date: str = None) -> Dict:
+        """Get comprehensive sales analytics"""
+        if not start_date:
+            start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
+        if not end_date:
+            end_date = datetime.now().strftime("%Y-%m-%d")
+
+        query = """
+        SELECT
+            COUNT(*) as total_sales,
+            SUM(total_amount) as total_revenue,
+            AVG(total_amount) as avg_sale_value,
+            COUNT(DISTINCT customer_id) as unique_customers,
+            SUM(CASE WHEN payment_status = 'Paid' THEN 1 ELSE 0 END) as completed_payments,
+            SUM(CASE WHEN payment_status IN ('Pending', 'Partial') THEN 1 ELSE 0 END) as pending_payments
+        FROM sales
+        WHERE sale_date BETWEEN ? AND ?
+        """
+
+        result = self.execute_query(query, (start_date, end_date), log_action=False)
+
+        if result:
+            row = result[0]
+            return {
+                "total_sales": row[0] or 0,
+                "total_revenue": row[1] or 0,
+                "avg_sale_value": row[2] or 0,
+                "unique_customers": row[3] or 0,
+                "completed_payments": row[4] or 0,
+                "pending_payments": row[5] or 0,
+            }
+        return {}
+
+    def log_system_action(
+        self,
+        log_type: str,
+        message: str,
+        table_name: str = None,
+        record_id: int = None,
+        action: str = None,
+    ):
+        """Log system actions for audit trail - without recursion"""
+        if self._is_logging:
+            return  # Prevent recursion
+
+        try:
+            self._is_logging = True
+            self._execute_query_internal(
+                """
+            INSERT INTO system_logs (log_type, log_message, table_name, record_id, action)
+            VALUES (?, ?, ?, ?, ?)
+            """,
+                (log_type, message, table_name, record_id, action),
+            )
+        except Exception as e:
+            logger.error(f"Error logging system action: {e}")
+        finally:
+            self._is_logging = False
+
+    def create_rollback_point(
+        self, table_name: str, record_id: int, old_data: str, new_data: str, action: str
+    ):
+        """Create a rollback point for data changes"""
+        try:
+            self.execute_query(
+                """
+            INSERT INTO rollback_logs (table_name, record_id, old_data, new_data, action)
+            VALUES (?, ?, ?, ?, ?)
+            """,
+                (table_name, record_id, old_data, new_data, action),
+                log_action=False,
+            )
+        except Exception as e:
+            logger.error(f"Error creating rollback point: {e}")
+
+    def get_recent_activity(self, limit: int = 10) -> pd.DataFrame:
+        """Get recent system activity"""
+        return self.get_dataframe(
+            "system_logs",
+            f"""
+        SELECT log_type, log_message, table_name, record_id, action, created_date
+        FROM system_logs
+        ORDER BY created_date DESC
+        LIMIT {limit}
+        """,
+        )
+
+    def backup_database(self, backup_path: str = None):
+        """Create a database backup"""
+        if not backup_path:
+            backup_path = f"backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.db"
+
+        try:
+            conn = self.get_connection()
+            backup_conn = sqlite3.connect(backup_path)
+
+            with backup_conn:
+                conn.backup(backup_conn)
+
+            conn.close()
+            backup_conn.close()
+
+            logger.info(f"Database backup created: {backup_path}")
+            return backup_path
+
+        except Exception as e:
+            logger.error(f"Error creating database backup: {e}")
+            return None
+
+    def get_village_wise_sales(self) -> pd.DataFrame:
+        """Get sales data grouped by village"""
+        return self.get_dataframe(
+            "sales",
+            """
+        SELECT c.village, COUNT(s.sale_id) as total_sales,
+               SUM(s.total_amount) as total_revenue,
+               AVG(s.total_amount) as avg_sale_value,
+               COUNT(DISTINCT s.customer_id) as unique_customers
+        FROM sales s
+        JOIN customers c ON s.customer_id = c.customer_id
+        WHERE c.village IS NOT NULL AND c.village != ''
+        GROUP BY c.village
+        ORDER BY total_revenue DESC
+        """,
+        )
+
+    def get_product_performance(self) -> pd.DataFrame:
+        """Get product performance analytics"""
+        return self.get_dataframe(
+            "sale_items",
+            """
+        SELECT p.product_name, COUNT(si.item_id) as times_sold,
+               SUM(si.quantity) as total_quantity,
+               SUM(si.amount) as total_revenue,
+               AVG(si.rate) as avg_rate
+        FROM sale_items si
+        JOIN products p ON si.product_id = p.product_id
+        GROUP BY p.product_id, p.product_name
+        ORDER BY total_revenue DESC
+        """,
+        )
+
+    def get_upcoming_follow_ups(self) -> pd.DataFrame:
+        """Get upcoming follow-ups"""
+        return self.get_dataframe(
+            "follow_ups",
+            """
+        SELECT f.*, c.name as customer_name, c.mobile,
+               d.name as distributor_name, dm.demo_date
+        FROM follow_ups f
+        LEFT JOIN customers c ON f.customer_id = c.customer_id
+        LEFT JOIN distributors d ON f.distributor_id = d.distributor_id
+        LEFT JOIN demos dm ON f.demo_id = dm.demo_id
+        WHERE f.follow_up_date >= date('now')
+        AND f.status = 'Pending'
+        ORDER BY f.follow_up_date ASC
+        LIMIT 20
+        """,
+        )
+
+    def get_whatsapp_logs(self, customer_id: int = None) -> pd.DataFrame:
+        """Get WhatsApp communication logs"""
+        if customer_id:
+            return self.get_dataframe(
+                "whatsapp_logs",
+                """
+            SELECT w.*, c.name as customer_name, c.mobile
+            FROM whatsapp_logs w
+            LEFT JOIN customers c ON w.customer_id = c.customer_id
+            WHERE w.customer_id = ?
+            ORDER BY w.sent_date DESC
+            """,
+                (customer_id,),
+            )
+        else:
+            return self.get_dataframe(
+                "whatsapp_logs",
+                """
+            SELECT w.*, c.name as customer_name, c.mobile
+            FROM whatsapp_logs w
+            LEFT JOIN customers c ON w.customer_id = c.customer_id
+            ORDER BY w.sent_date DESC
+            LIMIT 50
+            """,
+            )
+
+    def cleanup_old_data(self, days: int = 365):
+        """Clean up old data (logs, etc.) older than specified days"""
+        try:
+            cutoff_date = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
+
+            # Clean system logs
+            self.execute_query(
+                "DELETE FROM system_logs WHERE created_date < ?",
+                (cutoff_date,),
+                log_action=False,
+            )
+
+            # Clean rollback logs
+            self.execute_query(
+                "DELETE FROM rollback_logs WHERE rollback_date < ?",
+                (cutoff_date,),
+                log_action=False,
+            )
+
+            logger.info(f"Cleaned up data older than {days} days")
+
+        except Exception as e:
+            logger.error(f"Error cleaning up old data: {e}")
+
+
+# Utility function to check database health
+def check_database_health(db_path: str = "sales_management.db") -> Dict:
+    """Check database health and statistics"""
+    try:
+        db = DatabaseManager(db_path)
+
+        # Get table counts
+        tables = ["customers", "sales", "distributors", "demos", "payments", "products"]
+        counts = {}
+
+        for table in tables:
+            result = db.execute_query(f"SELECT COUNT(*) FROM {table}", log_action=False)
+            counts[table] = result[0][0] if result else 0
+
+        # Get database size
+        db_size = os.path.getsize(db_path) if os.path.exists(db_path) else 0
+
+        return {
+            "status": "healthy",
+            "table_counts": counts,
+            "database_size_mb": round(db_size / (1024 * 1024), 2),
+            "last_backup": "N/A",  # You can implement backup tracking
+            "integrity_check": "passed",  # You can add actual integrity checks
+        }
+
+    except Exception as e:
+        return {
+            "status": "error",
+            "error": str(e),
+            "table_counts": {},
+            "database_size_mb": 0,
+            "integrity_check": "failed",
+        }

main.py

@@ -0,0 +1,230 @@
+import streamlit as st
+import os
+import sys
+
+# Add the current directory to Python path
+sys.path.append(os.path.dirname(__file__))
+
+# MUST BE FIRST - Page configuration
+st.set_page_config(
+    page_title="Sales Management System",
+    page_icon="📊",
+    layout="wide",
+    initial_sidebar_state="expanded"
+)
+
+# Import utilities - Create these if they don't exist
+try:
+    from utils.styling import apply_custom_css
+    from utils.helpers import init_session_state, check_module_availability
+    from components.database_status import show_database_status
+except ImportError:
+    # Create basic fallbacks if utils don't exist
+    def apply_custom_css():
+        st.markdown("""
+        <style>
+        .main-header { color: #1f77b4; }
+        .section-header { color: #2e86ab; margin-top: 2rem; }
+        </style>
+        """, unsafe_allow_html=True)
+    
+    def init_session_state():
+        if 'db' not in st.session_state:
+            st.session_state.db = None
+        if 'data_processor' not in st.session_state:
+            st.session_state.data_processor = None
+        if 'analytics' not in st.session_state:
+            st.session_state.analytics = None
+        if 'whatsapp_manager' not in st.session_state:
+            st.session_state.whatsapp_manager = None
+    
+    def check_module_availability():
+        try:
+            import pandas, plotly, sqlite3
+            MODULES_AVAILABLE = True
+        except ImportError:
+            MODULES_AVAILABLE = False
+        
+        try:
+            import pywhatkit
+            WHATSAPP_AVAILABLE = True
+        except ImportError:
+            WHATSAPP_AVAILABLE = False
+            
+        return MODULES_AVAILABLE, WHATSAPP_AVAILABLE
+    
+    def show_database_status(db):
+        if db:
+            try:
+                health = db.execute_query("SELECT COUNT(*) FROM sqlite_master", log_action=False)
+                st.sidebar.success("✅ Database Connected")
+            except:
+                st.sidebar.error("❌ Database Error")
+        else:
+            st.sidebar.warning("⚠️ Database Not Initialized")
+
+# Apply custom CSS
+apply_custom_css()
+
+# Initialize session state
+init_session_state()
+
+# Check module availability
+MODULES_AVAILABLE, WHATSAPP_AVAILABLE = check_module_availability()
+
+# Initialize components with error handling
+if MODULES_AVAILABLE:
+    try:
+        from database import DatabaseManager
+        from data_processor import DataProcessor
+        from analytics import Analytics
+        
+        if st.session_state.db is None:
+            st.session_state.db = DatabaseManager()
+            st.success("✅ Database initialized successfully!")
+        
+        if st.session_state.data_processor is None:
+            st.session_state.data_processor = DataProcessor(st.session_state.db)
+        
+        if st.session_state.analytics is None:
+            st.session_state.analytics = Analytics(st.session_state.db)
+        
+        if WHATSAPP_AVAILABLE and st.session_state.whatsapp_manager is None:
+            try:
+                from whatsapp_manager import WhatsAppManager
+                st.session_state.whatsapp_manager = WhatsAppManager(st.session_state.db)
+                st.success("✅ WhatsApp Manager initialized!")
+            except Exception as e:
+                st.warning(f"⚠️ WhatsApp Manager not available: {e}")
+                st.session_state.whatsapp_manager = None
+                
+    except Exception as e:
+        st.error(f"❌ Application initialization failed: {e}")
+        st.info("Please check that all required files are in the correct location.")
+
+# Assign to local variables for easier access
+db = st.session_state.db
+data_processor = st.session_state.data_processor
+analytics = st.session_state.analytics
+whatsapp_manager = st.session_state.whatsapp_manager
+
+# Add this in your main content area (before page routing)
+st.sidebar.markdown("""
+<div style='text-align: center; margin: 20px 0; margin-top:-10px'>
+    <img src='https://tse4.mm.bing.net/th/id/OIP.bvMgrnyDHrBdq_MmZeP8XgHaHa?rs=1&pid=ImgDetMain&o=7&rm=3' 
+         style='width: 200px; height: auto; margin-bottom: 10px;'>
+</div>
+""", unsafe_allow_html=True)
+st.sidebar.markdown("<h2 style='text-align: center;'>🚀 Sales Management</h2>", unsafe_allow_html=True)
+
+page = st.sidebar.radio("Navigation", [
+    "📊 System Dashboard", "👥 Customers", "💰 Sales", "💳 Payments", 
+    "🎯 Demos", "🤝 Distributors", "🔍 File Viewer", "📤 Data Import", "📊 Power BI Dashboard", "📈 Reports"
+], index=0)
+
+# app.py (add this)
+from sales_data_processor import SalesDataProcessor
+
+# Initialize in your main app
+if 'data_processor' not in st.session_state:
+    st.session_state.data_processor = SalesDataProcessor(db)
+    
+def show_basic_dashboard(db, analytics):
+    st.title("📊 Sales Dashboard")
+    
+    if db and analytics:
+        try:
+            sales_summary = analytics.get_sales_summary()
+            
+            col1, col2, col3, col4 = st.columns(4)
+            with col1:
+                st.metric("Total Sales", f"₹{sales_summary.get('total_sales', 0):,.0f}")
+            with col2:
+                st.metric("Pending Payments", f"₹{sales_summary.get('pending_amount', 0):,.0f}")
+            with col3:
+                st.metric("Total Transactions", sales_summary.get('total_transactions', 0))
+            with col4:
+                st.metric("Avg Sale", f"₹{sales_summary.get('avg_sale_value', 0):,.0f}")
+                
+        except Exception as e:
+            st.error(f"Error loading dashboard data: {e}")
+    else:
+        st.warning("Database or analytics not available")
+# Page routing with error handling
+try:
+    if page == "📊 System Dashboard":
+        try:
+            from pages.system_dashboard import create_dashboard
+            create_dashboard(db, analytics)
+        except ImportError:
+            st.error("Dashboard page not available. Creating basic dashboard...")
+            show_basic_dashboard(db, analytics)        
+    
+    elif page == "👥 Customers":
+        try:
+            from pages.customers import show_customers_page
+            show_customers_page(db, whatsapp_manager)
+        except ImportError:
+            st.error("Customers page not available")
+    
+    elif page == "💰 Sales":
+        try:
+            from pages.sales import show_sales_page
+            show_sales_page(db, whatsapp_manager)
+        except ImportError:
+            st.error("Sales page not available")
+    
+    elif page == "🎯 Demos":
+        from pages.demos import show_demos_page
+        show_demos_page(db, whatsapp_manager)
+        
+    elif page == "💳 Payments":
+        from pages.payments import show_payments_page
+        show_payments_page(db, whatsapp_manager)
+    
+    elif page == "🤝 Distributors":
+        try:
+            from pages.distributors import show_distributors_page
+            show_distributors_page(db, whatsapp_manager)
+        except ImportError:
+            st.error("Distributors page not available")
+    
+    elif page == "🔍 File Viewer":
+        try:
+            from pages.file_viewer import show_file_viewer_page
+            show_file_viewer_page(db, data_processor)
+        except ImportError:
+            st.error("File Viewer page not available")
+    
+    elif page == "📤 Data Import":
+        try:
+            from pages.data_import import show_data_import_page
+            show_data_import_page(db, data_processor)
+        except ImportError:
+            st.error("Data Import page not available")
+    elif page == "📊 Power BI Dashboard":
+        try:
+            from pages.dashboard import create_dashboard
+            create_dashboard(db, analytics)
+        except ImportError:
+            st.error("Dashboard page not available. Creating basic dashboard...")
+            show_basic_dashboard(db, analytics)
+    
+    elif page == "📈 Reports":
+        try:
+            from pages.reports import show_reports_page
+            show_reports_page(db,whatsapp_manager)
+        except ImportError:
+            st.error("Reports page not available")
+
+except Exception as e:
+    st.error(f"Application error: {e}")
+    st.info("Please check the console for more details.")
+
+
+
+
+st.sidebar.markdown("---")
+st.sidebar.info("🚀 Sales Management System v2.0")
+
+# Basic dashboard fallback

requirements.txt

@@ -0,0 +1,8 @@
+streamlit==1.28.0
+pandas==2.1.0
+plotly==5.15.0
+openpyxl==3.1.2
+pywhatkit==5.4
+schedule==1.2.0
+deep-translator==1.11.4
+numpy==1.24.0

sales_data_processor.py

@@ -0,0 +1,502 @@
+# sales_data_processor.py
+import pandas as pd
+import re
+from datetime import datetime
+import numpy as np
+import sqlite3
+import os
+import streamlit as st
+
+class SalesDataProcessor:
+    def __init__(self, db):
+        self.db = db
+        self.setup_product_mapping()
+        self.setup_location_mapping()
+        self.setup_database_tables()
+    
+    def setup_database_tables(self):
+        """Initialize database tables if they don't exist"""
+        conn = self.db.get_connection()
+        cursor = conn.cursor()
+        
+        # Create sales table
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS sales (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                source_sheet TEXT,
+                sr_no TEXT,
+                customer_name TEXT,
+                village TEXT,
+                taluka TEXT,
+                district TEXT,
+                invoice_no TEXT UNIQUE,
+                reference TEXT,
+                dispatch_date TEXT,
+                product_type TEXT,
+                quantity INTEGER,
+                rate_per_unit REAL,
+                amount REAL,
+                final_amount REAL,
+                total_liters REAL,
+                payment_date TEXT,
+                gpay_amount REAL,
+                cash_amount REAL,
+                cheque_amount REAL,
+                rrn_number TEXT,
+                sold_by TEXT,
+                sale_type TEXT,
+                payment_status TEXT,
+                payment_method TEXT,
+                processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                source_file TEXT
+            )
+        ''')
+        
+        # Create customers table (aggregated from sales)
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS customers (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                customer_name TEXT,
+                village TEXT,
+                taluka TEXT,
+                district TEXT,
+                total_purchases REAL DEFAULT 0,
+                total_orders INTEGER DEFAULT 0,
+                last_order_date TEXT,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        ''')
+        
+        conn.commit()
+        conn.close()
+    
+    def setup_product_mapping(self):
+        """Standard product mapping for all packaging types"""
+        self.PRODUCT_MAPPING = {
+            '1 LTR PLASTIC JAR': '1L_PLASTIC_JAR',
+            '2 LTR PLASTIC JAR': '2L_PLASTIC_JAR', 
+            '5 LTR PLASTIC JAR': '5L_PLASTIC_JAR',
+            '10 LTR PLASTIC JAR': '10L_PLASTIC_JAR',
+            '5 LTR STEEL BARNI': '5L_STEEL_BARNI',
+            '10 LTR STEEL BARNI': '10L_STEEL_BARNI',
+            '20 LTR STEEL BARNI': '20L_STEEL_BARNI',
+            '20 LTR PLASTIC CAN': '20L_PLASTIC_CAN',
+            '1 LTR PET BOTTLE': '1L_PET_BOTTLE',
+            '20 LTR CARBO': '20L_CARBO'
+        }
+    
+    def setup_location_mapping(self):
+        """Gujarati location name standardization"""
+        self.GUJARATI_LOCALITIES = {
+            'રામપુરા': 'RAMPURA',
+            'શેખડી': 'SHEKHADI', 
+            'સિંહોલ': 'SINHOL',
+            'વનાદરા': 'VANADARA',
+            'માવલી': 'MAVLI',
+            'સિમરડા': 'SIMRADA',
+            'બિલપડ': 'BILPAD',
+            'વઘોડિયા': 'VAGHODIA',
+            'સાકરિયા': 'SAKARIYA'
+        }
+    
+    def safe_float(self, value):
+        """Safely convert to float, handle errors"""
+        if pd.isna(value) or value in ['', 'NOT_AVAILABLE', None, '_']:
+            return 0.0
+        try:
+            return float(value)
+        except (ValueError, TypeError):
+            return 0.0
+    
+    def safe_int(self, value):
+        """Safely convert to integer"""
+        return int(self.safe_float(value))
+    
+    def parse_date(self, date_str):
+        """Handle all date formats intelligently"""
+        if pd.isna(date_str) or date_str in ['', 'NOT_AVAILABLE', None, '_']:
+            return 'NOT_AVAILABLE'
+        
+        if isinstance(date_str, (int, float)):
+            try:
+                return (datetime(1899, 12, 30) + pd.Timedelta(days=date_str)).strftime('%Y-%m-%d')
+            except:
+                return 'INVALID_DATE'
+        
+        date_str = str(date_str).strip()
+        
+        date_formats = [
+            '%Y-%m-%d %H:%M:%S', 
+            '%d/%m/%Y', 
+            '%Y-%m-%d', 
+            '%d-%m-%Y',
+            '%d/%m/%Y %H:%M:%S'
+        ]
+        
+        for fmt in date_formats:
+            try:
+                return datetime.strptime(date_str, fmt).strftime('%Y-%m-%d')
+            except ValueError:
+                continue
+        
+        return 'INVALID_DATE'
+    
+    def clean_name(self, name):
+        """Handle names, duplicates, variations"""
+        if pd.isna(name) or name in ['', '-', '_', None]:
+            return 'NOT_AVAILABLE'
+        name = ' '.join(str(name).strip().split())
+        return name
+    
+    def standardize_location(self, location):
+        """Handle Gujarati location names"""
+        if pd.isna(location) or location in ['', 'NOT_AVAILABLE', None]:
+            return 'NOT_AVAILABLE'
+        
+        location_str = str(location).strip()
+        
+        if isinstance(location_str, str):
+            for guj_name, eng_name in self.GUJARATI_LOCALITIES.items():
+                if guj_name in location_str:
+                    return eng_name
+        
+        return location_str.upper()
+    
+    def standardize_product(self, product_name):
+        """Convert any product name to standard format"""
+        if pd.isna(product_name) or product_name in ['', 'NOT_AVAILABLE', None]:
+            return 'UNKNOWN_PRODUCT'
+        
+        product_str = str(product_name).strip()
+        product_upper = product_str.upper()
+        
+        for key, value in self.PRODUCT_MAPPING.items():
+            if key in product_upper:
+                return value
+        
+        # Fuzzy matching
+        if '1 LTR' in product_upper or '1L' in product_upper:
+            if 'PLASTIC' in product_upper or 'JAR' in product_upper:
+                return '1L_PLASTIC_JAR'
+            elif 'PET' in product_upper or 'BOTTLE' in product_upper:
+                return '1L_PET_BOTTLE'
+        elif '2 LTR' in product_upper or '2L' in product_upper:
+            return '2L_PLASTIC_JAR'
+        elif '5 LTR' in product_upper or '5L' in product_upper:
+            if 'STEEL' in product_upper or 'BARNI' in product_upper:
+                return '5L_STEEL_BARNI'
+            else:
+                return '5L_PLASTIC_JAR'
+        elif '10 LTR' in product_upper or '10L' in product_upper:
+            if 'STEEL' in product_upper or 'BARNI' in product_upper:
+                return '10L_STEEL_BARNI'
+            else:
+                return '10L_PLASTIC_JAR'
+        elif '20 LTR' in product_upper or '20L' in product_upper:
+            if 'STEEL' in product_upper or 'BARNI' in product_upper:
+                return '20L_STEEL_BARNI'
+            elif 'PLASTIC' in product_upper or 'CAN' in product_upper:
+                return '20L_PLASTIC_CAN'
+            elif 'CARBO' in product_upper:
+                return '20L_CARBO'
+        
+        return f"UNKNOWN_{product_upper.replace(' ', '_')}"
+    
+    def detect_sale_type(self, row):
+        """Detect if it's demo sale (single unit) or bulk sale"""
+        quantity = self.safe_int(row.get('QTN', 0))
+        reference = str(row.get('REF.', '')).upper()
+        
+        if reference == 'DEMO' or quantity == 1:
+            return 'DEMO_SALE'
+        else:
+            return 'BULK_SALE'
+    
+    def calculate_payment_status(self, row):
+        """Determine payment status intelligently"""
+        final_amt = self.safe_float(row.get('FINAL AMT', 0))
+        gpay = self.safe_float(row.get('G-PAY', 0))
+        cash = self.safe_float(row.get('CASH', 0))
+        cheque = self.safe_float(row.get('CHQ', 0))
+        
+        paid_amt = gpay + cash + cheque
+        
+        if paid_amt >= final_amt:
+            return 'PAID'
+        elif paid_amt > 0:
+            return 'PARTIAL_PAID'
+        elif self.parse_date(row.get('PAYMENT DATE')) not in ['NOT_AVAILABLE', 'INVALID_DATE']:
+            return 'PENDING'
+        else:
+            return 'UNPAID'
+    
+    def detect_payment_method(self, row):
+        """Intelligently detect payment method"""
+        gpay = self.safe_float(row.get('G-PAY', 0))
+        cash = self.safe_float(row.get('CASH', 0))
+        cheque = self.safe_float(row.get('CHQ', 0))
+        
+        if gpay > 0:
+            return 'GPAY'
+        elif cash > 0:
+            return 'CASH'
+        elif cheque > 0:
+            return 'CHEQUE'
+        else:
+            return 'NOT_PAID'
+    
+    def process_dataframe(self, df, sheet_name, source_file):
+        """Process entire dataframe and standardize all records"""
+        standardized_records = []
+        
+        for idx, row in df.iterrows():
+            if (pd.isna(row.get('NAME', '')) and 
+                pd.isna(row.get('PACKING', '')) and 
+                pd.isna(row.get('INV NO', ''))):
+                continue
+            
+            try:
+                standardized_record = self.standardize_record(row, sheet_name, source_file)
+                standardized_records.append(standardized_record)
+            except Exception as e:
+                st.error(f"⚠️ Error processing row {idx}: {e}")
+                continue
+        
+        return standardized_records
+    
+    def standardize_record(self, row, sheet_name, source_file):
+        """Standardize a single record"""
+        record = {
+            'source_sheet': sheet_name,
+            'sr_no': self.clean_name(row.get('SR NO.', 'NOT_AVAILABLE')),
+            'customer_name': self.clean_name(row.get('NAME', 'NOT_AVAILABLE')),
+            'village': self.standardize_location(row.get('VILLAGE', 'NOT_AVAILABLE')),
+            'taluka': self.standardize_location(row.get('TALUKA', 'NOT_AVAILABLE')),
+            'district': self.standardize_location(row.get('DISTRICT', 'NOT_AVAILABLE')),
+            'invoice_no': self.clean_name(row.get('INV NO', 'NOT_AVAILABLE')),
+            'reference': self.clean_name(row.get('REF.', 'NOT_AVAILABLE')),
+            'dispatch_date': self.parse_date(row.get('DISPATCH DATE')),
+            'product_type': self.standardize_product(row.get('PACKING', 'NOT_AVAILABLE')),
+            'quantity': self.safe_int(row.get('QTN', 0)),
+            'rate_per_unit': self.safe_float(row.get('RATE', 0)),
+            'amount': self.safe_float(row.get('AMT', 0)),
+            'final_amount': self.safe_float(row.get('FINAL AMT', 0)),
+            'total_liters': self.safe_float(row.get('TOTAL LTR', 0)),
+            'payment_date': self.parse_date(row.get('PAYMENT DATE')),
+            'gpay_amount': self.safe_float(row.get('G-PAY', 0)),
+            'cash_amount': self.safe_float(row.get('CASH', 0)),
+            'cheque_amount': self.safe_float(row.get('CHQ', 0)),
+            'rrn_number': self.clean_name(row.get('RRN', 'NOT_AVAILABLE')),
+            'sold_by': self.clean_name(row.get('BY', 'NOT_AVAILABLE')),
+            'sale_type': self.detect_sale_type(row),
+            'payment_status': self.calculate_payment_status(row),
+            'payment_method': self.detect_payment_method(row),
+            'source_file': os.path.basename(source_file)
+        }
+        
+        # Auto-calculate missing amounts
+        if record['amount'] == 0 and record['quantity'] > 0 and record['rate_per_unit'] > 0:
+            record['amount'] = record['quantity'] * record['rate_per_unit']
+        
+        if record['final_amount'] == 0 and record['amount'] > 0:
+            record['final_amount'] = record['amount']
+        
+        return record
+    
+    def insert_into_database(self, records):
+        """Insert processed records into database"""
+        conn = self.db.get_connection()
+        cursor = conn.cursor()
+        
+        inserted_count = 0
+        updated_count = 0
+        
+        for record in records:
+            try:
+                # Check if invoice already exists
+                cursor.execute('SELECT id FROM sales WHERE invoice_no = ?', (record['invoice_no'],))
+                existing = cursor.fetchone()
+                
+                if existing:
+                    # Update existing record
+                    update_query = '''
+                        UPDATE sales SET 
+                        source_sheet=?, sr_no=?, customer_name=?, village=?, taluka=?, district=?,
+                        reference=?, dispatch_date=?, product_type=?, quantity=?, rate_per_unit=?,
+                        amount=?, final_amount=?, total_liters=?, payment_date=?, gpay_amount=?,
+                        cash_amount=?, cheque_amount=?, rrn_number=?, sold_by=?, sale_type=?,
+                        payment_status=?, payment_method=?, source_file=?
+                        WHERE invoice_no=?
+                    '''
+                    cursor.execute(update_query, (
+                        record['source_sheet'], record['sr_no'], record['customer_name'],
+                        record['village'], record['taluka'], record['district'],
+                        record['reference'], record['dispatch_date'], record['product_type'],
+                        record['quantity'], record['rate_per_unit'], record['amount'],
+                        record['final_amount'], record['total_liters'], record['payment_date'],
+                        record['gpay_amount'], record['cash_amount'], record['cheque_amount'],
+                        record['rrn_number'], record['sold_by'], record['sale_type'],
+                        record['payment_status'], record['payment_method'], record['source_file'],
+                        record['invoice_no']
+                    ))
+                    updated_count += 1
+                else:
+                    # Insert new record
+                    insert_query = '''
+                        INSERT INTO sales (
+                            source_sheet, sr_no, customer_name, village, taluka, district,
+                            invoice_no, reference, dispatch_date, product_type, quantity,
+                            rate_per_unit, amount, final_amount, total_liters, payment_date,
+                            gpay_amount, cash_amount, cheque_amount, rrn_number, sold_by,
+                            sale_type, payment_status, payment_method, source_file
+                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    '''
+                    cursor.execute(insert_query, (
+                        record['source_sheet'], record['sr_no'], record['customer_name'],
+                        record['village'], record['taluka'], record['district'],
+                        record['invoice_no'], record['reference'], record['dispatch_date'],
+                        record['product_type'], record['quantity'], record['rate_per_unit'],
+                        record['amount'], record['final_amount'], record['total_liters'],
+                        record['payment_date'], record['gpay_amount'], record['cash_amount'],
+                        record['cheque_amount'], record['rrn_number'], record['sold_by'],
+                        record['sale_type'], record['payment_status'], record['payment_method'],
+                        record['source_file']
+                    ))
+                    inserted_count += 1
+                    
+            except Exception as e:
+                st.error(f"❌ Database error for invoice {record['invoice_no']}: {e}")
+                continue
+        
+        conn.commit()
+        
+        # Update customers table
+        self.update_customers_table(conn)
+        
+        conn.close()
+        
+        return inserted_count, updated_count
+    
+    def update_customers_table(self, conn):
+        """Update customers table from sales data"""
+        cursor = conn.cursor()
+        
+        # Clear and rebuild customers table
+        cursor.execute('DELETE FROM customers')
+        
+        # Insert aggregated customer data
+        cursor.execute('''
+            INSERT INTO customers (customer_name, village, taluka, district, total_purchases, total_orders, last_order_date)
+            SELECT 
+                customer_name,
+                village,
+                taluka,
+                district,
+                SUM(final_amount) as total_purchases,
+                COUNT(*) as total_orders,
+                MAX(dispatch_date) as last_order_date
+            FROM sales 
+            WHERE customer_name != 'NOT_AVAILABLE'
+            GROUP BY customer_name, village, taluka, district
+        ''')
+        
+        conn.commit()
+    
+    def process_excel_file(self, file_path):
+        """Main method to process Excel file - called from Streamlit"""
+        try:
+            st.info(f"🔄 Processing: {os.path.basename(file_path)}")
+            
+            # Read the Excel file
+            xl = pd.ExcelFile(file_path)
+            
+            # Process each sheet
+            all_records = []
+            
+            for sheet_name in xl.sheet_names:
+                with st.spinner(f"Processing sheet: {sheet_name}..."):
+                    # Read sheet
+                    df = pd.read_excel(file_path, sheet_name=sheet_name)
+                    
+                    # Standardize data
+                    standardized_records = self.process_dataframe(df, sheet_name, file_path)
+                    all_records.extend(standardized_records)
+            
+            if not all_records:
+                st.warning("⚠️ No valid records found in the file")
+                return False
+            
+            # Insert into database
+            with st.spinner("Inserting into database..."):
+                inserted, updated = self.insert_into_database(all_records)
+            
+            # Show results
+            if inserted > 0 or updated > 0:
+                st.success(f"✅ Processed {len(all_records)} records from {os.path.basename(file_path)}")
+                st.success(f"📊 New: {inserted}, Updated: {updated}")
+                
+                # Show quick summary
+                self.show_import_summary(all_records)
+                return True
+            else:
+                st.warning("⚠️ No records were inserted or updated")
+                return False
+            
+        except Exception as e:
+            st.error(f"❌ Error processing file: {e}")
+            return False
+    
+    def show_import_summary(self, records):
+        """Show summary of imported data"""
+        if not records:
+            return
+        
+        df = pd.DataFrame(records)
+        
+        col1, col2, col3, col4 = st.columns(4)
+        
+        with col1:
+            st.metric("Total Records", len(records))
+        with col2:
+            demo_sales = len(df[df['sale_type'] == 'DEMO_SALE'])
+            st.metric("Demo Sales", demo_sales)
+        with col3:
+            bulk_sales = len(df[df['sale_type'] == 'BULK_SALE'])
+            st.metric("Bulk Sales", bulk_sales)
+        with col4:
+            total_amount = df['final_amount'].sum()
+            st.metric("Total Amount", f"₹{total_amount:,.2f}")
+        
+        # Show top products
+        st.subheader("📦 Products Imported")
+        product_summary = df['product_type'].value_counts().head(5)
+        for product, count in product_summary.items():
+            st.write(f"- {product}: {count} records")
+    
+    def get_import_stats(self):
+        """Get import statistics for dashboard"""
+        conn = self.db.get_connection()
+        
+        try:
+            # Total records
+            total_records = pd.read_sql('SELECT COUNT(*) as count FROM sales', conn)['count'].iloc[0]
+            
+            # Files processed
+            files_processed = pd.read_sql('SELECT COUNT(DISTINCT source_file) as count FROM sales', conn)['count'].iloc[0]
+            
+            # Recent imports
+            recent_imports = pd.read_sql('''
+                SELECT source_file, COUNT(*) as records, MAX(processed_at) as last_import
+                FROM sales 
+                GROUP BY source_file 
+                ORDER BY last_import DESC 
+                LIMIT 5
+            ''', conn)
+            
+            return {
+                'total_records': total_records,
+                'files_processed': files_processed,
+                'recent_imports': recent_imports.to_dict('records')
+            }
+        finally:
+            conn.close()

sales_management.db

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6ae6dc8f1fb2158a1a11b75081b08290d985cba6ea96bf8943761f3cb0e4dc0d
+size 729088

sales_manager.py

@@ -0,0 +1,92 @@
+# enhanced_sales_manager.py
+from datetime import datetime, timedelta
+import pandas as pd
+import sqlite3
+
+class EnhancedSalesManager:
+    def __init__(self, db_manager, data_processor):
+        self.db_manager = db_manager
+        self.data_processor = data_processor
+    
+    def batch_import_data(self, directory_path):
+        """Import all data from a directory"""
+        return self.data_processor.process_directory(directory_path)
+    
+    def get_customer_insights(self):
+        """Get comprehensive customer insights"""
+        conn = self.db_manager.get_connection()
+        
+        # Top customers by spending
+        top_customers = pd.read_sql('''
+        SELECT c.name, c.village, SUM(s.total_amount) as total_spent, COUNT(s.sale_id) as purchases
+        FROM customers c
+        JOIN sales s ON c.customer_id = s.customer_id
+        GROUP BY c.customer_id
+        ORDER BY total_spent DESC
+        LIMIT 20
+        ''', conn)
+        
+        # Village performance
+        village_performance = pd.read_sql('''
+        SELECT village, COUNT(DISTINCT customer_id) as customers, 
+               SUM(total_amount) as total_sales, AVG(total_amount) as avg_sale
+        FROM sales s
+        JOIN customers c ON s.customer_id = c.customer_id
+        GROUP BY village
+        ORDER BY total_sales DESC
+        ''', conn)
+        
+        return {
+            'top_customers': top_customers,
+            'village_performance': village_performance
+        }
+    
+    def generate_comprehensive_report(self, start_date=None, end_date=None):
+        """Generate detailed business intelligence report"""
+        if not start_date:
+            start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
+        if not end_date:
+            end_date = datetime.now().strftime('%Y-%m-%d')
+        
+        conn = self.db_manager.get_connection()
+        
+        reports = {}
+        
+        # Sales trends
+        reports['sales_trends'] = pd.read_sql(f'''
+        SELECT DATE(sale_date) as date, SUM(total_amount) as daily_sales,
+            SUM(total_liters) as daily_liters, COUNT(*) as transactions
+        FROM sales
+        WHERE sale_date BETWEEN '{start_date}' AND '{end_date}'
+        GROUP BY DATE(sale_date)
+        ORDER BY date
+        ''', conn)
+        
+        # Product performance
+        reports['product_performance'] = pd.read_sql(f'''
+        SELECT p.packing_type, p.capacity_ltr, SUM(si.quantity) as total_quantity,
+            SUM(si.amount) as total_revenue, COUNT(DISTINCT s.sale_id) as transactions
+        FROM sale_items si
+        JOIN products p ON si.product_id = p.product_id
+        JOIN sales s ON si.sale_id = s.sale_id
+        WHERE s.sale_date BETWEEN '{start_date}' AND '{end_date}'
+        GROUP BY p.product_id
+        ORDER BY total_revenue DESC
+        ''', conn)
+        
+        # Payment analysis
+        reports['payment_analysis'] = pd.read_sql(f'''
+        SELECT 
+            CASE 
+                WHEN payment_date IS NOT NULL THEN 'Paid'
+                ELSE 'Pending'
+            END as payment_status,
+            COUNT(*) as transactions,
+            SUM(total_amount) as amount,
+            AVG(total_amount) as avg_amount
+        FROM sales
+        WHERE sale_date BETWEEN '{start_date}' AND '{end_date}'
+        GROUP BY payment_status
+        ''', conn)
+        
+        return reports

whatsapp_manager.py

@@ -0,0 +1,402 @@
+# whatsapp_manager.py
+import pywhatkit
+import logging
+from datetime import datetime, timedelta
+import time
+import pandas as pd
+import os
+import streamlit as st
+
+class WhatsAppManager:
+    def __init__(self, db_manager):
+        self.db = db_manager
+        self.logger = logging.getLogger(__name__)
+    
+    def send_message(self, phone_number, message, image_path=None):
+        """Send WhatsApp message with error handling"""
+        try:
+            # Clean phone number
+            phone_number = self._clean_phone_number(phone_number)
+            if not phone_number:
+                self._log_message(phone_number, message, "failed", "Invalid phone number")
+                return False
+            
+            st.info(f"📱 Preparing to send WhatsApp message to {phone_number}")
+            
+            # Schedule message (sends in 2 minutes)
+            send_time = datetime.now() + timedelta(minutes=2)
+            
+            try:
+                if image_path and os.path.exists(image_path):
+                    pywhatkit.sendwhats_image(
+                        phone_number, 
+                        image_path, 
+                        message,
+                        wait_time=20,
+                        tab_close=True
+                    )
+                else:
+                    pywhatkit.sendwhatmsg(
+                        phone_number,
+                        message,
+                        send_time.hour,
+                        send_time.minute,
+                        wait_time=20,
+                        tab_close=True
+                    )
+                
+                # Log the message
+                self._log_message(phone_number, message, "sent")
+                st.success(f"✅ Message sent successfully to {phone_number}")
+                return True
+                
+            except Exception as e:
+                error_msg = f"PyWhatKit error: {str(e)}"
+                st.error(f"❌ {error_msg}")
+                self._log_message(phone_number, message, "failed", error_msg)
+                return False
+            
+        except Exception as e:
+            error_msg = f"Unexpected error: {str(e)}"
+            st.error(f"❌ {error_msg}")
+            self._log_message(phone_number, message, "failed", error_msg)
+            return False
+    
+    def send_bulk_messages(self, customer_ids, message_template):
+        """Send messages to multiple customers"""
+        results = []
+        total_customers = len(customer_ids)
+        
+        progress_bar = st.progress(0)
+        status_text = st.empty()
+        
+        for i, customer_id in enumerate(customer_ids):
+            try:
+                # Update progress
+                progress = (i + 1) / total_customers
+                progress_bar.progress(progress)
+                status_text.text(f"Processing {i+1}/{total_customers} customers...")
+                
+                customer = self.db.get_dataframe('customers', 
+                    f"SELECT * FROM customers WHERE customer_id = {customer_id}")
+                
+                if not customer.empty:
+                    customer_data = customer.iloc[0]
+                    phone = customer_data['mobile']
+                    
+                    if phone and pd.notna(phone) and str(phone).strip():
+                        personalized_msg = self._personalize_message(message_template, customer_data)
+                        success = self.send_message(phone, personalized_msg)
+                        
+                        results.append({
+                            'customer_id': customer_id,
+                            'customer_name': customer_data['name'],
+                            'phone': phone,
+                            'status': 'sent' if success else 'failed',
+                            'message': personalized_msg[:50] + "..." if len(personalized_msg) > 50 else personalized_msg
+                        })
+                    else:
+                        results.append({
+                            'customer_id': customer_id,
+                            'customer_name': customer_data['name'],
+                            'phone': phone,
+                            'status': 'failed',
+                            'message': 'No phone number available'
+                        })
+                else:
+                    results.append({
+                        'customer_id': customer_id,
+                        'customer_name': 'Unknown',
+                        'phone': 'N/A',
+                        'status': 'failed',
+                        'message': 'Customer not found'
+                    })
+                    
+                # Small delay to avoid rate limiting
+                time.sleep(2)
+                
+            except Exception as e:
+                results.append({
+                    'customer_id': customer_id,
+                    'customer_name': 'Error',
+                    'phone': 'N/A',
+                    'status': 'failed',
+                    'message': f'Error: {str(e)}'
+                })
+        
+        progress_bar.empty()
+        status_text.empty()
+        
+        return results
+    
+    def send_distributor_message(self, distributor_id, message):
+        """Send message to distributor"""
+        try:
+            distributor = self.db.get_dataframe('distributors',
+                f"SELECT * FROM distributors WHERE distributor_id = {distributor_id}")
+            
+            if not distributor.empty:
+                distributor_data = distributor.iloc[0]
+                phone = distributor_data['mantri_mobile']
+                
+                if phone and pd.notna(phone) and str(phone).strip():
+                    personalized_msg = self._personalize_distributor_message(message, distributor_data)
+                    return self.send_message(phone, personalized_msg)
+                else:
+                    st.warning(f"No mobile number found for distributor: {distributor_data['name']}")
+                    return False
+            else:
+                st.error("Distributor not found")
+                return False
+                
+        except Exception as e:
+            st.error(f"Error sending distributor message: {e}")
+            return False
+    
+    def _personalize_distributor_message(self, template, distributor_data):
+        """Personalize message for distributor"""
+        message = template
+        message = message.replace('{name}', distributor_data.get('mantri_name', 'Distributor'))
+        message = message.replace('{distributor_name}', distributor_data.get('name', ''))
+        message = message.replace('{village}', distributor_data.get('village', ''))
+        message = message.replace('{taluka}', distributor_data.get('taluka', ''))
+        message = message.replace('{sabhasad_count}', str(distributor_data.get('sabhasad_count', 0)))
+        return message
+    
+    def _clean_phone_number(self, phone):
+        """Clean and validate phone number"""
+        if not phone or pd.isna(phone):
+            return None
+        
+        # Convert to string and remove spaces, hyphens, etc.
+        phone_str = str(phone).strip()
+        clean_phone = ''.join(filter(str.isdigit, phone_str))
+        
+        # Validate length
+        if len(clean_phone) < 10:
+            return None
+        
+        # Add country code if missing (assuming India)
+        if len(clean_phone) == 10:
+            clean_phone = '91' + clean_phone
+        elif len(clean_phone) == 11 and clean_phone.startswith('0'):
+            clean_phone = '91' + clean_phone[1:]
+        elif len(clean_phone) == 12 and clean_phone.startswith('91'):
+            # Already correct format
+            pass
+        else:
+            # If longer than 12 digits, take last 12
+            if len(clean_phone) > 12:
+                clean_phone = clean_phone[-12:]
+        
+        return '+' + clean_phone
+    
+    def _personalize_message(self, template, customer_data):
+        """Personalize message with customer data"""
+        message = template
+        message = message.replace('{name}', customer_data.get('name', 'Customer'))
+        message = message.replace('{village}', customer_data.get('village', ''))
+        message = message.replace('{taluka}', customer_data.get('taluka', ''))
+        message = message.replace('{district}', customer_data.get('district', ''))
+        
+        # Add current date
+        current_date = datetime.now().strftime('%d-%m-%Y')
+        message = message.replace('{date}', current_date)
+        
+        return message
+    
+    def _log_message(self, phone, message, status, error=None):
+        """Log WhatsApp message in database"""
+        try:
+            # Ensure whatsapp_logs table exists
+            self.db.execute_query('''
+            CREATE TABLE IF NOT EXISTS whatsapp_logs (
+                log_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                customer_id INTEGER,
+                distributor_id INTEGER,
+                phone_number TEXT,
+                message_content TEXT,
+                message_type TEXT,
+                status TEXT,
+                error_message TEXT,
+                sent_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (customer_id) REFERENCES customers (customer_id) ON DELETE SET NULL,
+                FOREIGN KEY (distributor_id) REFERENCES distributors (distributor_id) ON DELETE SET NULL
+            )
+            ''', log_action=False)
+            
+            # Find customer by phone (last 10 digits)
+            customer_result = self.db.execute_query(
+                'SELECT customer_id FROM customers WHERE mobile LIKE ?',
+                (f'%{phone[-10:]}%',),
+                log_action=False
+            )
+            
+            customer_id = customer_result[0][0] if customer_result else None
+            
+            # Find distributor by phone
+            distributor_result = self.db.execute_query(
+                'SELECT distributor_id FROM distributors WHERE mantri_mobile LIKE ?',
+                (f'%{phone[-10:]}%',),
+                log_action=False
+            )
+            
+            distributor_id = distributor_result[0][0] if distributor_result else None
+            
+            # Determine message type
+            message_type = 'customer' if customer_id else 'distributor' if distributor_id else 'general'
+            
+            self.db.execute_query('''
+            INSERT INTO whatsapp_logs (customer_id, distributor_id, phone_number, message_content, message_type, status, error_message)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            ''', (customer_id, distributor_id, phone, message, message_type, status, error), log_action=False)
+            
+        except Exception as e:
+            self.logger.error(f"Failed to log message: {e}")
+    
+    def get_message_stats(self):
+        """Get messaging statistics"""
+        try:
+            stats = self.db.execute_query('''
+            SELECT 
+                COUNT(*) as total_messages,
+                SUM(CASE WHEN status = 'sent' THEN 1 ELSE 0 END) as sent_messages,
+                SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed_messages,
+                MAX(sent_date) as last_message_date
+            FROM whatsapp_logs
+            ''', log_action=False)
+            
+            if stats:
+                return {
+                    'total_messages': stats[0][0] or 0,
+                    'sent_messages': stats[0][1] or 0,
+                    'failed_messages': stats[0][2] or 0,
+                    'last_message_date': stats[0][3]
+                }
+            return {}
+            
+        except Exception as e:
+            self.logger.error(f"Error getting message stats: {e}")
+            return {}
+    
+    def get_recent_messages(self, limit=10):
+        """Get recent WhatsApp messages"""
+        try:
+            messages = self.db.get_dataframe('whatsapp_logs', f'''
+            SELECT wl.*, 
+                   c.name as customer_name,
+                   d.name as distributor_name
+            FROM whatsapp_logs wl
+            LEFT JOIN customers c ON wl.customer_id = c.customer_id
+            LEFT JOIN distributors d ON wl.distributor_id = d.distributor_id
+            ORDER BY wl.sent_date DESC
+            LIMIT {limit}
+            ''')
+            return messages
+        except Exception as e:
+            self.logger.error(f"Error getting recent messages: {e}")
+            return pd.DataFrame()
+    
+    def send_welcome_message(self, phone_number, name, role="customer"):
+        """Send welcome message to new customers/distributors"""
+        if role == "distributor":
+            message = f"""Welcome {name}! 🎉
+
+Thank you for joining our distributor network! 
+
+We're excited to have you on board and look forward to working together to grow your business.
+
+Our team will contact you shortly to discuss:
+• Training schedule
+• Product information  
+• Sales strategies
+• Support systems
+
+For any immediate queries, feel free to contact us.
+
+Best regards,
+Sales Team"""
+        else:
+            message = f"""Welcome {name}! 🎉
+
+Thank you for choosing us! 
+
+We're delighted to have you as our customer and look forward to serving you with the best products and service.
+
+Feel free to reach out for any queries or support.
+
+Best regards,
+Sales Team"""
+        
+        return self.send_message(phone_number, message)
+    
+    def send_payment_reminder(self, customer_id, invoice_no, pending_amount):
+        """Send payment reminder to customer"""
+        try:
+            customer = self.db.get_dataframe('customers',
+                f"SELECT * FROM customers WHERE customer_id = {customer_id}")
+            
+            if not customer.empty:
+                customer_data = customer.iloc[0]
+                phone = customer_data['mobile']
+                
+                if phone and pd.notna(phone) and str(phone).strip():
+                    message = f"""Hello {customer_data['name']},
+
+This is a friendly reminder regarding your pending payment.
+
+Invoice: {invoice_no}
+Pending Amount: ₹{pending_amount:,.2f}
+
+Please make the payment at your earliest convenience.
+
+Thank you for your cooperation!
+
+Best regards,
+Sales Team"""
+                    
+                    return self.send_message(phone, message)
+                else:
+                    st.warning(f"No mobile number found for customer: {customer_data['name']}")
+                    return False
+            else:
+                st.error("Customer not found")
+                return False
+                
+        except Exception as e:
+            st.error(f"Error sending payment reminder: {e}")
+            return False
+
+# Utility function to check WhatsApp availability
+def check_whatsapp_availability():
+    """Check if WhatsApp features are available"""
+    try:
+        import pywhatkit
+        return True
+    except ImportError:
+        return False
+
+# Example usage and test function
+def test_whatsapp_manager(db):
+    """Test WhatsApp manager functionality"""
+    st.subheader("🧪 WhatsApp Manager Test")
+    
+    if st.button("Test WhatsApp Connection"):
+        try:
+            manager = WhatsAppManager(db)
+            
+            # Test phone number cleaning
+            test_numbers = ["9876543210", "09876543210", "919876543210"]
+            for num in test_numbers:
+                cleaned = manager._clean_phone_number(num)
+                st.write(f"Original: {num} → Cleaned: {cleaned}")
+            
+            # Test message personalization
+            test_customer = {'name': 'John Doe', 'village': 'Test Village'}
+            personalized = manager._personalize_message("Hello {name} from {village}!", test_customer)
+            st.write(f"Personalized message: {personalized}")
+            
+            st.success("✅ WhatsApp manager test completed successfully!")
+            
+        except Exception as e:
+            st.error(f"❌ WhatsApp manager test failed: {e}")