import pandas as pd
import glob

files = glob.glob('sales/2026/*.csv')
dfs = []
for f in files:
    df = pd.read_csv(f)
    dfs.append(df)
result = pd.concat(dfs)

import duckdb

# One line of code to read all CSV files
result = duckdb.sql("""
    SELECT 
        department,
        COUNT(*) as order_count,
        SUM(amount) as total_sales,
        AVG(amount) as avg_order,
        MAX(order_date) as last_order
    FROM 'sales/2026/*.csv'
    GROUP BY department
    ORDER BY total_sales DESC
""").df()

print(result)

import duckdb

# Auto-detect delimiter, encoding, column names
df = duckdb.sql("""
    SELECT * FROM read_csv_auto('sales/2026/*.csv')
    WHERE amount > 100
    ORDER BY amount DESC
    LIMIT 10
""").df()

import duckdb

# Use UNION BY NAME to auto-align columns
df = duckdb.sql("""
    SELECT * FROM (
        SELECT * FROM 'sales/region_a/*.csv'
        UNION BY NAME
        SELECT * FROM 'sales/region_b/*.csv'
        UNION BY NAME
        SELECT * FROM 'sales/region_c/*.csv'
    )
""").df()

import duckdb
from datetime import datetime, timedelta

# Incremental processing: only process new/changed files
today = datetime.now().strftime('%Y-%m-%d')

# Create or connect to target database
con = duckdb.connect('sales_analytics.duckdb')

# Incremental merge into target table
con.execute(f"""
    INSERT INTO daily_sales
    SELECT * FROM read_csv_auto('sales/{today}/*.csv')
    WHERE order_date = '{today}'
    ON CONFLICT DO NOTHING
""")

# Query latest results
result = con.execute("""
    SELECT 
        department,
        SUM(amount) as daily_sales,
        COUNT(*) as orders
    FROM daily_sales
    WHERE order_date = '{today}'
    GROUP BY department
""").df()

print(f"📊 {today} Sales Analysis Complete")
print(result)

import duckdb
import pandas as pd
from datetime import datetime, timedelta
import os

class SalesETLPipeline:
    def __init__(self, sales_dir='sales_data'):
        self.sales_dir = sales_dir
        self.con = duckdb.connect('sales_analytics.duckdb')
        self._init_schema()
    
    def _init_schema(self):
        """Initialize database schema"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS daily_orders (
                order_id VARCHAR,
                store_id VARCHAR,
                product_id VARCHAR,
                amount DECIMAL(10,2),
                order_date DATE,
                region VARCHAR
            )
        """)
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS sales_summary (
                report_date DATE,
                store_id VARCHAR,
                total_orders BIGINT,
                total_amount DECIMAL(12,2),
                avg_order DECIMAL(10,2),
                updated_at TIMESTAMP
            )
        """)
    
    def run_daily_etl(self, target_date=None):
        """Run daily ETL"""
        if target_date is None:
            target_date = datetime.now().strftime('%Y-%m-%d')
        
        print(f"🔄 Starting ETL for {target_date}...")
        
        # 1. Read and merge all CSV files for the day
        csv_pattern = f"{self.sales_dir}/{target_date}/*.csv"
        df = self.con.sql(f"""
            SELECT * FROM read_csv_auto('{csv_pattern}', 
                hive_partitioning=true,
                union_by_name=true)
        """).df()
        
        print(f"📄 Read {len(df)} order records")
        
        # 2. Data cleaning
        df = self._clean_data(df)
        
        # 3. Incremental insert
        self._insert_orders(df, target_date)
        
        # 4. Generate summary report
        summary = self._generate_summary(target_date)
        
        # 5. Anomaly detection
        anomalies = self._detect_anomalies(target_date)
        
        if anomalies:
            print(f"⚠️ Found {len(anomalies)} anomalous orders")
            self._send_alert(anomalies)
        else:
            print("✅ No anomalous orders")
        
        print(f"✅ ETL Complete! Total {summary['total_orders'].sum()} orders")
        return summary
    
    def _clean_data(self, df):
        """Data cleaning"""
        df['region'] = df['region'].fillna('UNKNOWN')
        df['amount'] = df['amount'].fillna(0)
        df = df[df['amount'] <= 100000]
        return df
    
    def _insert_orders(self, df, date_str):
        """Incremental order insertion"""
        for _, row in df.iterrows():
            self.con.execute("""
                INSERT OR IGNORE INTO daily_orders
                (order_id, store_id, product_id, amount, order_date, region)
                VALUES (?, ?, ?, ?, ?, ?)
            """, [
                row['order_id'], row['store_id'], row['product_id'],
                row['amount'], date_str, row['region']
            ])
    
    def _generate_summary(self, date_str):
        """Generate sales summary"""
        summary = self.con.sql(f"""
            SELECT 
                store_id,
                COUNT(*) as total_orders,
                SUM(amount) as total_amount,
                AVG(amount) as avg_order
            FROM daily_orders
            WHERE order_date = '{date_str}'
            GROUP BY store_id
        """).df()
        
        summary['updated_at'] = datetime.now()
        self.con.execute("DELETE FROM sales_summary WHERE report_date = ?", [date_str])
        
        for _, row in summary.iterrows():
            self.con.execute("""
                INSERT INTO sales_summary
                (report_date, store_id, total_orders, total_amount, avg_order, updated_at)
                VALUES (?, ?, ?, ?, ?, ?)
            """, [
                row['report_date'], row['store_id'], row['total_orders'],
                row['total_amount'], row['avg_order'], row['updated_at']
            ])
        
        return summary
    
    def _detect_anomalies(self, date_str):
        """Detect anomalous orders"""
        anomalies = self.con.sql(f"""
            SELECT * FROM daily_orders
            WHERE order_date = '{date_str}'
            AND amount > (
                SELECT avg(amount) * 3 
                FROM daily_orders 
                WHERE order_date = '{date_str}'
            )
        """).df()
        return anomalies
    
    def _send_alert(self, anomalies):
        """Send alert (example, can integrate with Slack/email)"""
        print(f"🚨 Alert: {len(anomalies)} anomalous orders found")
        for _, row in anomalies.head(5).iterrows():
            print(f"   - Order {row['order_id']}: ${row['amount']:.2f}")

# Run ETL
if __name__ == '__main__':
    pipeline = SalesETLPipeline()
    pipeline.run_daily_etl()