import pandas as pd
import glob
from datetime import datetime

# 1. Read all files (30 lines just for this)
files = sorted(glob.glob('data/sales_2026-*.csv'))
dfs = []
for f in files:
    try:
        df = pd.read_csv(f)
        dfs.append(df)
    except Exception as e:
        print(f"Error reading {f}: {e}")

# 2. Concatenate (another 50 lines)
result = pd.concat(dfs, ignore_index=True)

# 3. Clean dirty data (complex logic, 80 lines)
result = result[result['amount'] > 0]
result = result[result['region'].notna()]
result['date'] = pd.to_datetime(result['date'])

# 4. Align columns (different files have different column names!)
# ... lots of manual adaptation ...

# 5. Output (20 lines)
result.to_parquet('output/sales_agg.parquet')

import duckdb

con = duckdb.connect('etl.duckdb')

# Step 1: Multi-file read + clean + merge
# (auto schema inference, handles inconsistent columns gracefully)
con.execute("""
    CREATE OR REPLACE TABLE sales_agg AS
    SELECT 
        filename,
        region,
        SUM(amount) AS total_amount,
        COUNT(*) AS order_cnt
    FROM read_csv_auto('data/sales_*.csv', 
                       header=true, 
                       filename=true,
                       sample_size=-1)
    WHERE amount > 0
    GROUP BY 1, 2
""")

# Step 2: Output directly as Parquet (10x compression, ready for downstream)
con.execute("COPY sales_agg TO 'output/sales_agg.parquet' (FORMAT PARQUET)")
con.close()

import duckdb
from pathlib import Path

class SalesETLPipeline:
    def __init__(self, db_path='sales_analytics.duckdb'):
        self.con = duckdb.connect(db_path)
        self._setup_schema()
    
    def _setup_schema(self):
        """Initialize table structures"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS daily_sales (
                order_id VARCHAR,
                product_id VARCHAR,
                amount DECIMAL(10,2),
                region VARCHAR,
                sale_date DATE,
                source_file VARCHAR
            )
        """)
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS sales_summary (
                report_date DATE,
                region VARCHAR,
                total_amount DECIMAL(12,2),
                order_count BIGINT,
                avg_order DECIMAL(10,2),
                updated_at TIMESTAMP
            )
        """)
    
    def run_daily_etl(self, date_str=None):
        """Run daily ETL"""
        if date_str is None:
            date_str = Path('.').absolute().strftime('%Y-%m-%d')
        
        print(f"🔄 Processing sales data for {date_str}...")
        
        # Read today's sales CSV (auto-merge, auto-handle schema differences)
        csv_pattern = f'data/sales_{date_str}.csv'
        
        self.con.execute(f"""
            INSERT INTO daily_sales
            SELECT 
                order_id, product_id, amount, region, 
                '{date_str}'::DATE as sale_date,
                '{csv_pattern}' as source_file
            FROM read_csv_auto('{csv_pattern}', 
                              header=true, 
                              filename=true,
                              sample_size=-1)
            WHERE amount > 0
              AND order_id IS NOT NULL
        """)
        
        # Update summary table
        self.con.execute(f"""
            INSERT INTO sales_summary
            SELECT 
                '{date_str}'::DATE as report_date,
                region,
                SUM(amount) as total_amount,
                COUNT(*) as order_count,
                AVG(amount) as avg_order,
                CURRENT_TIMESTAMP as updated_at
            FROM daily_sales
            WHERE sale_date = '{date_str}'::DATE
            GROUP BY region
        """)
        
        # Output Parquet for downstream use
        self.con.execute("""
            COPY (SELECT * FROM sales_summary) 
            TO 'output/sales_summary.parquet' (FORMAT PARQUET)
        """)
        
        print(f"✅ ETL complete! Written to output/sales_summary.parquet")

if __name__ == '__main__':
    pipeline = SalesETLPipeline()
    pipeline.run_daily_etl('2026-09-02')

-- Only process new files (filenames contain dates)
FROM read_csv_auto('data/sales_2026-09-*.csv', 
                   header=true, filename=true)
WHERE filename >= 'data/sales_2026-09-01.csv'

-- Different files have different columns, DuckDB auto-aligns
SELECT * FROM read_csv_auto('data/business_*.csv',
                             header=true,
                             union_by_name=true,
                             null_padding=true)