sales_processor.py

import pandas as pd
import numpy as np
from datetime import datetime
import re
from config import SALES_FILE

class SalesProcessor:
    def __init__(self):
        self.sales_data = None
        self.load_sales_data()
    
    def load_sales_data(self):
        """Load sales data from CSV file"""
        try:
            self.sales_data = pd.read_csv(SALES_FILE)
            
            # Convert Date column to datetime
            self.sales_data['Date'] = pd.to_datetime(self.sales_data['Date'])
            
            # Add derived columns
            self.sales_data['Month'] = self.sales_data['Date'].dt.strftime('%Y-%m')
            self.sales_data['Year'] = self.sales_data['Date'].dt.year
            
            print(f"Loaded {len(self.sales_data)} sales records")
            
        except Exception as e:
            print(f"Error loading sales data: {e}")
            self.sales_data = None
    
    def process_sales_query(self, query):
        """Process natural language sales queries"""
        if self.sales_data is None:
            return "Sales data is not available."
        
        query = query.lower()
        
        # Monthly sales patterns
        if any(word in query for word in ['monthly', 'month', 'each month']):
            return self.get_monthly_sales()
        
        # Product-wise sales patterns
        elif any(word in query for word in ['product', 'which product', 'best product']):
            if any(word in query for word in ['highest', 'best', 'top', 'maximum']):
                return self.get_top_products()
            else:
                return self.get_product_sales()
        
        # Total sales patterns
        elif any(word in query for word in ['total', 'sum', 'overall']):
            return self.get_total_sales()
        
        # Specific product queries
        elif self.contains_product_name(query):
            product_name = self.extract_product_name(query)
            return self.get_specific_product_sales(product_name)
        
        # Date-specific queries
        elif self.contains_date(query):
            date_info = self.extract_date_info(query)
            return self.get_sales_by_date(date_info)
        
        # Average sales
        elif any(word in query for word in ['average', 'avg', 'mean']):
            return self.get_average_sales()
        
        # Trends
        elif any(word in query for word in ['trend', 'growth', 'increase', 'decrease']):
            return self.get_sales_trends()
        
        else:
            return self.get_general_sales_info()
    
    def get_monthly_sales(self):
        """Get sales summary by month"""
        monthly_sales = self.sales_data.groupby('Month')['Sales'].sum().sort_index()
        
        response = "📊 Monthly Sales Summary:\n\n"
        for month, sales in monthly_sales.items():
            response += f"{month}: ${sales:,.2f}\n"
        
        return response
    
    def get_product_sales(self):
        """Get sales summary by product"""
        product_sales = self.sales_data.groupby('Product')['Sales'].sum().sort_values(ascending=False)
        
        response = "📈 Product Sales Summary:\n\n"
        for product, sales in product_sales.items():
            response += f"{product}: ${sales:,.2f}\n"
        
        return response
    
    def get_top_products(self):
        """Get top-performing products"""
        product_sales = self.sales_data.groupby('Product')['Sales'].sum().sort_values(ascending=False)
        
        response = "🏆 Top Performing Products:\n\n"
        for i, (product, sales) in enumerate(product_sales.head().items(), 1):
            response += f"{i}. {product}: ${sales:,.2f}\n"
        
        return response
    
    def get_total_sales(self):
        """Get total sales across all products and periods"""
        total = self.sales_data['Sales'].sum()
        count = len(self.sales_data)
        
        response = f"💰 Total Sales Summary:\n\n"
        response += f"Total Sales: ${total:,.2f}\n"
        response += f"Number of Records: {count}\n"
        response += f"Average per Record: ${total/count:,.2f}\n"
        
        return response
    
    def get_specific_product_sales(self, product_name):
        """Get sales for a specific product"""
        product_data = self.sales_data[
            self.sales_data['Product'].str.contains(product_name, case=False)
        ]
        
        if product_data.empty:
            return f"No sales data found for product containing '{product_name}'"
        
        total_sales = product_data['Sales'].sum()
        monthly_sales = product_data.groupby('Month')['Sales'].sum()
        
        response = f"📊 Sales for '{product_name}':\n\n"
        response += f"Total Sales: ${total_sales:,.2f}\n"
        response += f"Number of Records: {len(product_data)}\n\n"
        response += "Monthly Breakdown:\n"
        
        for month, sales in monthly_sales.items():
            response += f"{month}: ${sales:,.2f}\n"
        
        return response
    
    def get_sales_by_date(self, date_info):
        """Get sales for specific date/period"""
        # This is a simplified implementation
        # You can expand this based on your needs
        return "Date-specific queries are being processed..."
    
    def get_average_sales(self):
        """Get average sales statistics"""
        avg_total = self.sales_data['Sales'].mean()
        avg_by_product = self.sales_data.groupby('Product')['Sales'].mean().sort_values(ascending=False)
        
        response = f"📊 Average Sales:\n\n"
        response += f"Overall Average: ${avg_total:,.2f}\n\n"
        response += "Average by Product:\n"
        
        for product, avg_sales in avg_by_product.items():
            response += f"{product}: ${avg_sales:,.2f}\n"
        
        return response
    
    def get_sales_trends(self):
        """Analyze sales trends"""
        monthly_sales = self.sales_data.groupby('Month')['Sales'].sum().sort_index()
        
        if len(monthly_sales) < 2:
            return "Not enough data to analyze trends."
        
        # Calculate month-over-month growth
        growth_rates = monthly_sales.pct_change().dropna()
        
        response = "📈 Sales Trends:\n\n"
        response += f"Best Month: {monthly_sales.idxmax()} (${monthly_sales.max():,.2f})\n"
        response += f"Worst Month: {monthly_sales.idxmin()} (${monthly_sales.min():,.2f})\n\n"
        
        if not growth_rates.empty:
            avg_growth = growth_rates.mean() * 100
            response += f"Average Monthly Growth: {avg_growth:+.1f}%\n"
        
        return response
    
    def get_general_sales_info(self):
        """Get general sales information"""
        total_sales = self.sales_data['Sales'].sum()
        unique_products = self.sales_data['Product'].nunique()
        date_range = f"{self.sales_data['Date'].min().strftime('%Y-%m-%d')} to {self.sales_data['Date'].max().strftime('%Y-%m-%d')}"
        
        response = "📊 Sales Data Overview:\n\n"
        response += f"Total Sales: ${total_sales:,.2f}\n"
        response += f"Number of Products: {unique_products}\n"
        response += f"Date Range: {date_range}\n"
        response += f"Total Records: {len(self.sales_data)}\n"
        
        return response
    
    def contains_product_name(self, query):
        """Check if query contains a product name"""
        if self.sales_data is None:
            return False
        
        products = self.sales_data['Product'].unique()
        for product in products:
            if product.lower() in query:
                return True
        return False
    
    def extract_product_name(self, query):
        """Extract product name from query"""
        if self.sales_data is None:
            return ""
        
        products = self.sales_data['Product'].unique()
        for product in products:
            if product.lower() in query:
                return product
        return ""
    
    def contains_date(self, query):
        """Check if query contains date information"""
        date_patterns = [
            r'\b\d{4}\b',  # Year
            r'\b(january|february|march|april|may|june|july|august|september|october|november|december)\b',
            r'\b(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)\b'
        ]
        
        for pattern in date_patterns:
            if re.search(pattern, query, re.IGNORECASE):
                return True
        return False
    
    def extract_date_info(self, query):
        """Extract date information from query"""
        # Simplified implementation
        return query