| |
|
| | 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()
|
| |
|
| |
|
| | 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
|
| | )
|
| | ''')
|
| |
|
| |
|
| | 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
|
| |
|
| |
|
| | 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)
|
| | }
|
| |
|
| |
|
| | 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:
|
| |
|
| | cursor.execute('SELECT id FROM sales WHERE invoice_no = ?', (record['invoice_no'],))
|
| | existing = cursor.fetchone()
|
| |
|
| | if existing:
|
| |
|
| | 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_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()
|
| |
|
| |
|
| | 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()
|
| |
|
| |
|
| | cursor.execute('DELETE FROM customers')
|
| |
|
| |
|
| | 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)}")
|
| |
|
| |
|
| | xl = pd.ExcelFile(file_path)
|
| |
|
| |
|
| | all_records = []
|
| |
|
| | for sheet_name in xl.sheet_names:
|
| | with st.spinner(f"Processing sheet: {sheet_name}..."):
|
| |
|
| | df = pd.read_excel(file_path, sheet_name=sheet_name)
|
| |
|
| |
|
| | 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
|
| |
|
| |
|
| | with st.spinner("Inserting into database..."):
|
| | inserted, updated = self.insert_into_database(all_records)
|
| |
|
| |
|
| | 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}")
|
| |
|
| |
|
| | 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}")
|
| |
|
| |
|
| | 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 = pd.read_sql('SELECT COUNT(*) as count FROM sales', conn)['count'].iloc[0]
|
| |
|
| |
|
| | files_processed = pd.read_sql('SELECT COUNT(DISTINCT source_file) as count FROM sales', conn)['count'].iloc[0]
|
| |
|
| |
|
| | 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() |
