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

# 一行代码读取所有 CSV 文件
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

# 自动检测分隔符、编码、列名
df = duckdb.sql("""
    SELECT * FROM read_csv_auto('sales/2026/*.csv')
    WHERE amount > 100
    ORDER BY amount DESC
    LIMIT 10
""").df()

import duckdb

# 使用 UNION BY NAME 自动对齐列
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

# 增量处理：只处理新增/变更的文件
today = datetime.now().strftime('%Y-%m-%d')
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')

# 创建或连接到目标数据库
con = duckdb.connect('sales_analytics.duckdb')

# 增量合并到目标表
con.execute(f"""
    INSERT INTO daily_sales
    SELECT * FROM read_csv_auto('sales/{today}/*.csv')
    WHERE order_date = '{today}'
    ON CONFLICT DO NOTHING
""")

# 更新汇总表
con.execute("""
    REFRESH MATERIALIZED VIEW sales_summary
""")

# 查询最新结果
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} 销售分析完成")
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):
        """初始化数据库 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):
        """运行每日 ETL"""
        if target_date is None:
            target_date = datetime.now().strftime('%Y-%m-%d')
        
        print(f"🔄 开始处理 {target_date} 的销售数据...")
        
        # 1. 读取并合并当日所有 CSV
        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"📄 读取了 {len(df)} 条订单记录")
        
        # 2. 数据清洗
        df = self._clean_data(df)
        
        # 3. 增量插入
        self._insert_orders(df, target_date)
        
        # 4. 生成汇总报表
        summary = self._generate_summary(target_date)
        
        # 5. 异常检测
        anomalies = self._detect_anomalies(target_date)
        
        if anomalies:
            print(f"⚠️ 发现 {len(anomalies)} 个异常订单")
            self._send_alert(anomalies)
        else:
            print("✅ 无异常订单")
        
        print(f"✅ ETL 完成！总计 {summary['total_orders'].sum()} 笔订单")
        return summary
    
    def _clean_data(self, df):
        """数据清洗"""
        # 填充缺失值
        df['region'] = df['region'].fillna('UNKNOWN')
        df['amount'] = df['amount'].fillna(0)
        
        # 过滤异常值（金额 > 100000）
        df = df[df['amount'] <= 100000]
        
        return df
    
    def _insert_orders(self, df, date_str):
        """增量插入订单数据"""
        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):
        """生成销售汇总"""
        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):
        """检测异常订单"""
        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):
        """发送告警（示例，实际可对接 Slack/邮件）"""
        print(f"🚨 告警：发现 {len(anomalies)} 个异常订单")
        for _, row in anomalies.head(5).iterrows():
            print(f"   - 订单 {row['order_id']}: ¥{row['amount']:.2f}")

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