import duckdb
import json

# 创建 DuckDB 内存数据库（零配置，用完即销毁）
con = duckdb.connect(':memory:')

# ── 淘宝 CSV（自动推断 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')
""")

# ── 拼多多 CSV（列名不同，需要映射）──
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  -- 拼多多不导出买家ID
    FROM read_csv_auto('data/pdd_sales_2026-08-23.csv')
""")

# ── 抖音 JSON（嵌套结构自动展开）──
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')
""")

# 三步走：UNION ALL 合并 → 过滤无效订单 → 计算利润率
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  -- 过滤测试订单
    AND order_date >= '2026-01-01'  -- 只保留今年的数据
""")

# 验证数据质量
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)

# 品类销售排行（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'  -- 最近7天
    GROUP BY 1
    ORDER BY 2 DESC
    LIMIT 10
""").df()

# 平台对比分析
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()

# 漏斗分析：从浏览到购买（模拟数据）
con.sql("""
    WITH funnel AS (
        SELECT 
            'product_view' AS stage, COUNT(*) AS users UNION ALL
            SELECT 'add_to_cart', COUNT(DISTINCT buyer_id) FROM unified_sales WHERE revenue > 0 UNION ALL
            SELECT 'purchase', COUNT(DISTINCT buyer_id) FROM unified_sales WHERE revenue > 100
    )
    SELECT stage, users, 
           ROUND(1.0 * users / MAX(users) OVER() * 100, 2) AS conversion_pct
    FROM funnel
""").df()

#!/usr/bin/env python3
"""电商销售数据每日自动化分析"""
import duckdb
from datetime import datetime, timedelta

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

# 1. 自动读取所有平台的最新文件（glob 模式）
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. 生成日报（输出为 Parquet，供 BI 工具读取）
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. 发送结果到 Telegram / 飞书
result = con.sql("SELECT * FROM daily_sales LIMIT 5").df()
print(f"✅ 昨日销售分析完成，共 {len(result)} 条记录")