transform_pipeline.py

#!/usr/bin/env python3
"""
Data Transformation Pipeline for Mini Data Warehouse
ETL processes for data cleansing, enrichment, and aggregation
"""

import pandas as pd
import psycopg2
import numpy as np
from datetime import datetime, timedelta
import logging
import json
from faker import Faker
import re

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class DataTransformationPipeline:
    def __init__(self, connection_params=None):
        if connection_params is None:
            self.connection_params = {
                'host': 'localhost',
                'port': '5432',
                'database': 'warehouse',
                'user': 'admin',
                'password': 'secret'
            }
        else:
            self.connection_params = connection_params
        
        self.fake = Faker()
        
    def connect_db(self):
        """Connect to PostgreSQL database"""
        try:
            conn = psycopg2.connect(**self.connection_params)
            return conn
        except psycopg2.Error as e:
            logger.error(f"Database connection failed: {e}")
            return None
    
    def create_staging_tables(self):
        """Create staging tables for transformation process"""
        conn = self.connect_db()
        if not conn:
            return False
        
        try:
            with conn.cursor() as cur:
                # Staging table for enriched customers
                cur.execute("""
                    DROP TABLE IF EXISTS staging_customers_enriched CASCADE;
                    CREATE TABLE staging_customers_enriched (
                        customer_id INTEGER,
                        first_name TEXT,
                        last_name TEXT,
                        email TEXT,
                        email_domain TEXT,
                        city TEXT,
                        country TEXT,
                        region TEXT,
                        continent TEXT,
                        customer_tier TEXT,
                        registration_date DATE,
                        days_since_registration INTEGER,
                        is_active BOOLEAN,
                        total_orders INTEGER,
                        total_spent NUMERIC(12,2),
                        avg_order_value NUMERIC(10,2),
                        last_order_date DATE,
                        days_since_last_order INTEGER
                    )
                """)
                
                # Staging table for product analytics
                cur.execute("""
                    DROP TABLE IF EXISTS staging_product_analytics CASCADE;
                    CREATE TABLE staging_product_analytics (
                        product_id INTEGER,
                        product_name TEXT,
                        product_name_clean TEXT,
                        category TEXT,
                        price NUMERIC(10,2),
                        price_tier TEXT,
                        price_percentile INTEGER,
                        total_sold INTEGER,
                        total_revenue NUMERIC(12,2),
                        avg_quantity_per_order NUMERIC(5,2),
                        popularity_rank INTEGER,
                        revenue_rank INTEGER,
                        margin_category TEXT,
                        reorder_rate NUMERIC(5,2),
                        seasonal_trend TEXT
                    )
                """)
                
                # Staging table for sales aggregates
                cur.execute("""
                    DROP TABLE IF EXISTS staging_sales_aggregates CASCADE;
                    CREATE TABLE staging_sales_aggregates (
                        date_key INTEGER,
                        order_date DATE,
                        year INTEGER,
                        quarter INTEGER,
                        month INTEGER,
                        day_of_week INTEGER,
                        is_weekend BOOLEAN,
                        is_holiday BOOLEAN,
                        daily_orders INTEGER,
                        daily_revenue NUMERIC(12,2),
                        daily_customers INTEGER,
                        avg_order_value NUMERIC(10,2),
                        new_customers INTEGER,
                        returning_customers INTEGER,
                        top_category TEXT,
                        top_category_revenue NUMERIC(10,2)
                    )
                """)
                
                conn.commit()
                logger.info("Staging tables created successfully")
                return True
                
        except psycopg2.Error as e:
            logger.error(f"Error creating staging tables: {e}")
            return False
        finally:
            conn.close()
    
    def clean_and_enrich_customers(self):
        """Clean and enrich customer data"""
        logger.info("Transforming customer data...")
        
        conn = self.connect_db()
        if not conn:
            return False
        
        try:
            # Get customer data
            customers_df = pd.read_sql("""
                SELECT c.*, 
                       COUNT(o.order_id) as total_orders,
                       COALESCE(SUM(o.total_amount), 0) as total_spent,
                       COALESCE(AVG(o.total_amount), 0) as avg_order_value,
                       MAX(o.order_date) as last_order_date
                FROM customers c
                LEFT JOIN orders o ON c.customer_id = o.customer_id
                GROUP BY c.customer_id, c.first_name, c.last_name, c.email, c.city, c.country
            """, conn)
            
            # Data cleaning and enrichment
            customers_df['email_domain'] = customers_df['email'].str.split('@').str[1]
            
            # Geographic enrichment
            region_mapping = {
                'United States': 'North America', 'Canada': 'North America', 'Mexico': 'North America',
                'United Kingdom': 'Europe', 'France': 'Europe', 'Germany': 'Europe', 'Spain': 'Europe', 
                'Italy': 'Europe', 'Netherlands': 'Europe', 'Poland': 'Europe', 'Sweden': 'Europe',
                'China': 'Asia', 'Japan': 'Asia', 'India': 'Asia', 'South Korea': 'Asia', 
                'Thailand': 'Asia', 'Singapore': 'Asia', 'Indonesia': 'Asia',
                'Brazil': 'South America', 'Argentina': 'South America', 'Chile': 'South America', 
                'Colombia': 'South America', 'Peru': 'South America',
                'Australia': 'Oceania', 'New Zealand': 'Oceania',
                'South Africa': 'Africa', 'Egypt': 'Africa', 'Nigeria': 'Africa', 'Kenya': 'Africa'
            }
            
            customers_df['region'] = customers_df['country'].map(region_mapping).fillna('Other')
            
            continent_mapping = {
                'North America': 'North America', 'Europe': 'Europe', 'Asia': 'Asia',
                'South America': 'South America', 'Oceania': 'Oceania', 'Africa': 'Africa', 'Other': 'Other'
            }
            customers_df['continent'] = customers_df['region'].map(continent_mapping)
            
            # Customer tier based on spending
            customers_df['customer_tier'] = pd.cut(
                customers_df['total_spent'], 
                bins=[-np.inf, 0, 500, 2000, np.inf], 
                labels=['Inactive', 'Bronze', 'Silver', 'Gold']
            )
            
            # Simulated registration dates (6 months to 2 years ago)
            customers_df['registration_date'] = customers_df['customer_id'].apply(
                lambda x: (datetime.now() - timedelta(days=np.random.randint(180, 730))).date()
            )
            
            customers_df['days_since_registration'] = customers_df['registration_date'].apply(
                lambda x: (datetime.now().date() - x).days
            )
            
            # Activity status
            customers_df['is_active'] = customers_df['total_orders'] > 0
            
            # Days since last order
            customers_df['last_order_date'] = pd.to_datetime(customers_df['last_order_date'])
            customers_df['days_since_last_order'] = (
                datetime.now() - customers_df['last_order_date']
            ).dt.days
            customers_df['days_since_last_order'] = customers_df['days_since_last_order'].fillna(9999).astype(int)
            
            # Insert into staging table
            with conn.cursor() as cur:
                cur.execute("DELETE FROM staging_customers_enriched")
                
                for _, row in customers_df.iterrows():
                    cur.execute("""
                        INSERT INTO staging_customers_enriched (
                            customer_id, first_name, last_name, email, email_domain, city, country,
                            region, continent, customer_tier, registration_date, days_since_registration,
                            is_active, total_orders, total_spent, avg_order_value, last_order_date, days_since_last_order
                        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    """, (
                        row['customer_id'], row['first_name'], row['last_name'], row['email'],
                        row['email_domain'], row['city'], row['country'], row['region'], row['continent'],
                        row['customer_tier'], row['registration_date'], row['days_since_registration'],
                        row['is_active'], row['total_orders'], row['total_spent'], row['avg_order_value'],
                        row['last_order_date'].date() if pd.notna(row['last_order_date']) else None,
                        row['days_since_last_order']
                    ))
                
                conn.commit()
                logger.info(f"Processed {len(customers_df)} customer records")
                return True
                
        except Exception as e:
            logger.error(f"Error in customer transformation: {e}")
            return False
        finally:
            conn.close()
    
    def analyze_product_performance(self):
        """Analyze and enrich product data"""
        logger.info("Analyzing product performance...")
        
        conn = self.connect_db()
        if not conn:
            return False
        
        try:
            # Get product performance data
            products_df = pd.read_sql("""
                SELECT p.*,
                       COALESCE(SUM(oi.quantity), 0) as total_sold,
                       COALESCE(SUM(oi.quantity * oi.unit_price), 0) as total_revenue,
                       COALESCE(AVG(oi.quantity), 0) as avg_quantity_per_order,
                       COUNT(DISTINCT oi.order_id) as unique_orders
                FROM products p
                LEFT JOIN order_items oi ON p.product_id = oi.product_id
                GROUP BY p.product_id, p.product_name, p.category, p.price
            """, conn)
            
            # Clean product names
            products_df['product_name_clean'] = products_df['product_name'].str.title().str.strip()
            
            # Price analysis
            products_df['price_tier'] = pd.cut(
                products_df['price'], 
                bins=[-np.inf, 25, 100, 500, np.inf], 
                labels=['Budget', 'Mid-Range', 'Premium', 'Luxury']
            )
            
            products_df['price_percentile'] = products_df['price'].rank(pct=True) * 100
            products_df['price_percentile'] = products_df['price_percentile'].round().astype(int)
            
            # Rankings
            products_df['popularity_rank'] = products_df['total_sold'].rank(method='dense', ascending=False)
            products_df['revenue_rank'] = products_df['total_revenue'].rank(method='dense', ascending=False)
            
            # Margin categories (simulated)
            margin_mapping = {
                'Electronics': 'Low', 'Furniture': 'Medium', 'Clothing': 'High',
                'Books': 'Medium', 'Sports': 'Medium'
            }
            products_df['margin_category'] = products_df['category'].map(margin_mapping)
            
            # Simulated reorder rate
            products_df['reorder_rate'] = np.random.uniform(0.1, 0.8, len(products_df))
            
            # Seasonal trends (simulated)
            seasonal_trends = ['Stable', 'Winter Peak', 'Summer Peak', 'Holiday Peak', 'Spring Peak']
            products_df['seasonal_trend'] = np.random.choice(seasonal_trends, len(products_df))
            
            # Insert into staging table
            with conn.cursor() as cur:
                cur.execute("DELETE FROM staging_product_analytics")
                
                for _, row in products_df.iterrows():
                    cur.execute("""
                        INSERT INTO staging_product_analytics (
                            product_id, product_name, product_name_clean, category, price, price_tier,
                            price_percentile, total_sold, total_revenue, avg_quantity_per_order,
                            popularity_rank, revenue_rank, margin_category, reorder_rate, seasonal_trend
                        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    """, (
                        row['product_id'], row['product_name'], row['product_name_clean'],
                        row['category'], row['price'], row['price_tier'], row['price_percentile'],
                        row['total_sold'], row['total_revenue'], row['avg_quantity_per_order'],
                        int(row['popularity_rank']), int(row['revenue_rank']), row['margin_category'],
                        row['reorder_rate'], row['seasonal_trend']
                    ))
                
                conn.commit()
                logger.info(f"Processed {len(products_df)} product records")
                return True
                
        except Exception as e:
            logger.error(f"Error in product analysis: {e}")
            return False
        finally:
            conn.close()
    
    def create_daily_aggregates(self):
        """Create daily sales aggregates"""
        logger.info("Creating daily sales aggregates...")
        
        conn = self.connect_db()
        if not conn:
            return False
        
        try:
            # Get daily sales data
            daily_sales_df = pd.read_sql("""
                SELECT 
                    o.order_date,
                    COUNT(DISTINCT o.order_id) as daily_orders,
                    SUM(o.total_amount) as daily_revenue,
                    COUNT(DISTINCT o.customer_id) as daily_customers,
                    AVG(o.total_amount) as avg_order_value
                FROM orders o
                GROUP BY o.order_date
                ORDER BY o.order_date
            """, conn)
            
            # Date enrichment
            daily_sales_df['order_date'] = pd.to_datetime(daily_sales_df['order_date'])
            daily_sales_df['date_key'] = daily_sales_df['order_date'].dt.strftime('%Y%m%d').astype(int)
            daily_sales_df['year'] = daily_sales_df['order_date'].dt.year
            daily_sales_df['quarter'] = daily_sales_df['order_date'].dt.quarter
            daily_sales_df['month'] = daily_sales_df['order_date'].dt.month
            daily_sales_df['day_of_week'] = daily_sales_df['order_date'].dt.dayofweek
            daily_sales_df['is_weekend'] = daily_sales_df['day_of_week'].isin([5, 6])
            
            # Simulate holiday detection
            daily_sales_df['is_holiday'] = False
            # Mark some days as holidays (simplified)
            holiday_dates = ['2024-01-01', '2024-07-04', '2024-12-25', '2024-11-28']
            for holiday in holiday_dates:
                daily_sales_df.loc[daily_sales_df['order_date'] == holiday, 'is_holiday'] = True
            
            # Get customer segmentation for each day
            for idx, row in daily_sales_df.iterrows():
                date = row['order_date'].date()
                
                # New vs returning customers (simplified logic)
                with conn.cursor() as cur:
                    cur.execute("""
                        SELECT COUNT(DISTINCT customer_id) 
                        FROM orders 
                        WHERE order_date = %s 
                        AND customer_id IN (
                            SELECT customer_id FROM orders WHERE order_date < %s
                        )
                    """, (date, date))
                    
                    returning = cur.fetchone()[0] or 0
                    new_customers = row['daily_customers'] - returning
                    
                    daily_sales_df.at[idx, 'new_customers'] = max(0, new_customers)
                    daily_sales_df.at[idx, 'returning_customers'] = returning
                    
                    # Top category for the day
                    cur.execute("""
                        SELECT p.category, SUM(oi.quantity * oi.unit_price) as revenue
                        FROM order_items oi
                        JOIN orders o ON oi.order_id = o.order_id
                        JOIN products p ON oi.product_id = p.product_id
                        WHERE o.order_date = %s
                        GROUP BY p.category
                        ORDER BY revenue DESC
                        LIMIT 1
                    """, (date,))
                    
                    top_cat_result = cur.fetchone()
                    if top_cat_result:
                        daily_sales_df.at[idx, 'top_category'] = top_cat_result[0]
                        daily_sales_df.at[idx, 'top_category_revenue'] = float(top_cat_result[1])
                    else:
                        daily_sales_df.at[idx, 'top_category'] = None
                        daily_sales_df.at[idx, 'top_category_revenue'] = 0
            
            # Insert into staging table
            with conn.cursor() as cur:
                cur.execute("DELETE FROM staging_sales_aggregates")
                
                for _, row in daily_sales_df.iterrows():
                    cur.execute("""
                        INSERT INTO staging_sales_aggregates (
                            date_key, order_date, year, quarter, month, day_of_week, is_weekend,
                            is_holiday, daily_orders, daily_revenue, daily_customers, avg_order_value,
                            new_customers, returning_customers, top_category, top_category_revenue
                        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    """, (
                        row['date_key'], row['order_date'].date(), row['year'], row['quarter'],
                        row['month'], row['day_of_week'], row['is_weekend'], row['is_holiday'],
                        row['daily_orders'], row['daily_revenue'], row['daily_customers'],
                        row['avg_order_value'], row['new_customers'], row['returning_customers'],
                        row['top_category'], row['top_category_revenue']
                    ))
                
                conn.commit()
                logger.info(f"Processed {len(daily_sales_df)} daily aggregate records")
                return True
                
        except Exception as e:
            logger.error(f"Error creating daily aggregates: {e}")
            return False
        finally:
            conn.close()
    
    def run_full_pipeline(self):
        """Execute the complete transformation pipeline"""
        logger.info("Starting full data transformation pipeline...")
        
        steps = [
            ("Creating staging tables", self.create_staging_tables),
            ("Enriching customer data", self.clean_and_enrich_customers),
            ("Analyzing product performance", self.analyze_product_performance),
            ("Creating daily aggregates", self.create_daily_aggregates)
        ]
        
        for step_name, step_function in steps:
            logger.info(f"Executing: {step_name}")
            if not step_function():
                logger.error(f"Pipeline failed at step: {step_name}")
                return False
        
        logger.info("Data transformation pipeline completed successfully")
        return True
    
    def generate_transformation_report(self):
        """Generate a report of transformation results"""
        conn = self.connect_db()
        if not conn:
            return {}
        
        try:
            report = {}
            
            with conn.cursor() as cur:
                # Customer enrichment stats
                cur.execute("SELECT COUNT(*) FROM staging_customers_enriched")
                report['enriched_customers'] = cur.fetchone()[0]
                
                cur.execute("SELECT customer_tier, COUNT(*) FROM staging_customers_enriched GROUP BY customer_tier")
                report['customer_tiers'] = dict(cur.fetchall())
                
                cur.execute("SELECT continent, COUNT(*) FROM staging_customers_enriched GROUP BY continent")
                report['customer_continents'] = dict(cur.fetchall())
                
                # Product analytics stats
                cur.execute("SELECT COUNT(*) FROM staging_product_analytics")
                report['analyzed_products'] = cur.fetchone()[0]
                
                cur.execute("SELECT price_tier, COUNT(*) FROM staging_product_analytics GROUP BY price_tier")
                report['product_price_tiers'] = dict(cur.fetchall())
                
                # Sales aggregates stats
                cur.execute("SELECT COUNT(*) FROM staging_sales_aggregates")
                report['daily_aggregates'] = cur.fetchone()[0]
                
                cur.execute("SELECT SUM(daily_revenue) FROM staging_sales_aggregates")
                report['total_analyzed_revenue'] = float(cur.fetchone()[0] or 0)
            
            return report
            
        except Exception as e:
            logger.error(f"Error generating transformation report: {e}")
            return {}
        finally:
            conn.close()

def main():
    """Main function to run transformation pipeline"""
    import argparse
    
    parser = argparse.ArgumentParser(description='Data transformation pipeline for Mini Data Warehouse')
    parser.add_argument('--report', action='store_true', help='Generate transformation report')
    
    args = parser.parse_args()
    
    pipeline = DataTransformationPipeline()
    
    success = pipeline.run_full_pipeline()
    
    if args.report or success:
        report = pipeline.generate_transformation_report()
        
        print("\n" + "="*50)
        print("DATA TRANSFORMATION PIPELINE REPORT")
        print("="*50)
        print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print()
        
        if success:
            print("✅ Pipeline Status: SUCCESS")
        else:
            print("❌ Pipeline Status: FAILED")
        
        print()
        print("TRANSFORMATION RESULTS:")
        print(f"  Enriched customers: {report.get('enriched_customers', 0):,}")
        print(f"  Analyzed products: {report.get('analyzed_products', 0):,}")
        print(f"  Daily aggregates: {report.get('daily_aggregates', 0):,}")
        print(f"  Total revenue analyzed: ${report.get('total_analyzed_revenue', 0):,.2f}")
        
        if 'customer_tiers' in report:
            print("\nCustomer Tiers:")
            for tier, count in report['customer_tiers'].items():
                print(f"  {tier}: {count:,}")
        
        if 'product_price_tiers' in report:
            print("\nProduct Price Tiers:")
            for tier, count in report['product_price_tiers'].items():
                print(f"  {tier}: {count:,}")
    
    return 0 if success else 1

if __name__ == "__main__":
    exit(main())