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")