Spaces:

omgy
/

pcsales

Sleeping

pcsales / database.py
Upload 12 files
943deb6 verified 12 months ago
41.6 kB
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",
        }