import duckdb
import json

# Create an in-memory DuckDB database (zero config,销毁 on exit)
con = duckdb.connect(':memory:')

# ── Taobao CSV (auto-infer schema) ──
con.sql("""
    CREATE TABLE taobao_orders AS
    SELECT 
        订单号 AS order_id,
        商品名称 AS product_name,
        CAST(实付金额 AS DOUBLE) AS revenue,
        STRFTIME(STRPTIME(支付时间, '%Y-%m-%d %H:%M:%S'), '%Y-%m-%d') AS order_date,
        买家ID AS buyer_id
    FROM read_csv_auto('data/taobao_sales_2026-08-23.csv')
""")

# ── Pinduoduo CSV (different column names, need mapping) ──
con.sql("""
    CREATE TABLE pdd_orders AS
    SELECT 
        订单编号 AS order_id,
        商品名 AS product_name,
        CAST(实际支付金额 AS DOUBLE) AS revenue,
        STRFTIME(STRPTIME(下单时间, '%Y-%m-%d %H:%M:%S'), '%Y-%m-%d') AS order_date,
        NULL AS buyer_id  -- Pinduoduo doesn't export buyer IDs
    FROM read_csv_auto('data/pdd_sales_2026-08-23.csv')
""")

# ── Douyin JSON (nested structure auto-flattened) ──
con.sql("""
    CREATE TABLE dy_orders AS
    SELECT 
        data:order_id AS order_id,
        data:product_name AS product_name,
        CAST(data:pay_amount AS DOUBLE) AS revenue,
        STRFTIME(STRPTIME(data:create_time, '%Y-%m-%dT%H:%M:%SZ'), '%Y-%m-%d') AS order_date,
        data:buyer_id AS buyer_id
    FROM read_json_auto('data/douyin_sales_2026-08-23.json')
""")

# Three steps: UNION ALL merge → filter invalid orders → calculate metrics
con.sql("""
    CREATE VIEW unified_sales AS
    SELECT 
        'taobao' AS platform,
        order_id, product_name, revenue, order_date, buyer_id
    FROM taobao_orders
    UNION ALL
    SELECT 'pdd', order_id, product_name, revenue, order_date, buyer_id
    FROM pdd_orders
    UNION ALL
    SELECT 'dy', order_id, product_name, revenue, order_date, buyer_id
    FROM dy_orders
    WHERE revenue > 0  -- Filter test orders
    AND order_date >= '2026-01-01'  -- Only keep this year's data
""")

# Data quality check
quality_report = con.sql("""
    SELECT 
        platform,
        COUNT(*) AS total_orders,
        SUM(revenue) AS total_revenue,
        AVG(revenue) AS avg_order_value,
        COUNT(DISTINCT buyer_id) AS unique_buyers
    FROM unified_sales
    GROUP BY platform
    ORDER BY total_revenue DESC
""").df()

print(quality_report)

# Category sales ranking (Top 10)
con.sql("""
    SELECT 
        product_name,
        SUM(revenue) AS total_sales,
        COUNT(*) AS order_count,
        SUM(revenue) / COUNT(*) AS avg_price
    FROM unified_sales
    WHERE order_date >= DATE '2026-08-17'  -- Last 7 days
    GROUP BY 1
    ORDER BY 2 DESC
    LIMIT 10
""").df()

# Platform comparison analysis
con.sql("""
    SELECT 
        platform,
        order_date,
        SUM(revenue) AS daily_revenue,
        COUNT(*) AS daily_orders,
        SUM(revenue) / NULLIF(COUNT(*), 0) AS avg_order_value
    FROM unified_sales
    WHERE order_date >= DATE '2026-08-01'
    GROUP BY 1, 2
    ORDER BY 2, 3 DESC
""").df()

#!/usr/bin/env python3
"""Daily E-Commerce Sales Data Automation"""
import duckdb
from datetime import datetime, timedelta

con = duckdb.connect(':memory:')

# 1. Auto-read all platform files (glob pattern)
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')

con.sql(f"""
    CREATE VIEW daily_sales AS
    SELECT 'taobao' AS platform, * FROM read_csv_auto('data/taobao_*{yesterday}*.csv')
    UNION ALL
    SELECT 'pdd', * FROM read_csv_auto('data/pdd_*{yesterday}*.csv')
    UNION ALL
    SELECT 'dy', * FROM read_json_auto('data/douyin_*{yesterday}*.json')
    WHERE revenue > 0
""")

# 2. Generate daily report (output as Parquet for BI tools)
con.sql(f"""
    COPY (
        SELECT 
            platform,
            order_date,
            SUM(revenue) AS daily_revenue,
            COUNT(*) AS order_count
        FROM daily_sales
        WHERE order_date = '{yesterday}'
        GROUP BY 1, 2
    ) TO 'output/daily_report_{yesterday}.parquet' (FORMAT PARQUET)
""")

# 3. Send results to Telegram / Feishu
result = con.sql("SELECT * FROM daily_sales LIMIT 5").df()
print(f"✅ Yesterday's sales analysis complete: {len(result)} records")