-- 在 DuckDB 中直接生成 100 万行测试数据
SET memory_limit = '4GB';

-- 创建 100 万行销售数据
CREATE TABLE sales AS
SELECT 
    row_number() OVER () AS order_id,
    '2025-01-01'::DATE + (random() * 365)::INTEGER AS order_date,
    'CUST_' || lpad((random() * 5000)::INTEGER::VARCHAR, 5, '0') AS customer_id,
    (CASE (random() * 4)::INTEGER
        WHEN 0 THEN '电子产品'
        WHEN 1 THEN '服装'
        WHEN 2 THEN '食品饮料'
        WHEN 3 THEN '家居用品'
        ELSE '图书'
    END) AS product_category,
    'Product_' || lpad((random() * 200)::INTEGER::VARCHAR, 3, '0') AS product_name,
    (random() * 10 + 1)::INTEGER AS quantity,
    round((random() * 500 + 10)::NUMERIC, 2) AS unit_price,
    round((quantity * unit_price)::NUMERIC, 2) AS total_amount,
    (CASE (random() * 5)::INTEGER
        WHEN 0 THEN '华北'
        WHEN 1 THEN '华东'
        WHEN 2 THEN '华南'
        WHEN 3 THEN '西南'
        ELSE '华中'
    END) AS region,
    (CASE (random() * 3)::INTEGER
        WHEN 0 THEN '支付宝'
        WHEN 1 THEN '微信支付'
        WHEN 2 THEN '银行卡'
        ELSE '货到付款'
    END) AS payment_method
FROM range(1000000);

-- 验证数据量
SELECT count(*) AS total_rows FROM sales;
-- 输出: 1000000

-- 导出为 CSV
COPY sales TO '/tmp/sales_2026.csv' (FORMAT CSV, HEADER true);

-- 直接查询 CSV 文件，零导入
SELECT 
    region,
    count(*) AS order_count,
    round(sum(total_amount)) AS total_revenue,
    round(avg(total_amount), 2) AS avg_order_value
FROM read_csv('/tmp/sales_2026.csv', 
    header = true, 
    columns = {
        'order_id': 'INTEGER',
        'order_date': 'DATE',
        'customer_id': 'VARCHAR',
        'product_category': 'VARCHAR',
        'product_name': 'VARCHAR',
        'quantity': 'INTEGER',
        'unit_price': 'DECIMAL(10,2)',
        'total_amount': 'DECIMAL(10,2)',
        'region': 'VARCHAR',
        'payment_method': 'VARCHAR'
    })
WHERE order_date >= '2025-06-01'
GROUP BY region
ORDER BY total_revenue DESC;

-- 导入到 DuckDB 内部（加速后续查询）
CREATE TABLE sales_imported AS 
SELECT * FROM read_csv('/tmp/sales_2026.csv', header = true);
-- 导入耗时 ~1.2 秒

-- 1. 月度销售趋势
SELECT 
    strftime(order_date, '%Y-%m') AS month,
    count(*) AS orders,
    count(DISTINCT customer_id) AS unique_customers,
    round(sum(total_amount)) AS revenue,
    round(avg(total_amount), 2) AS avg_order
FROM sales_imported
GROUP BY month
ORDER BY month;

-- 2. 产品类目销售排行
SELECT 
    product_category,
    count(*) AS orders,
    round(sum(total_amount)) AS revenue,
    round(avg(total_amount), 2) AS avg_order_value,
    sum(quantity) AS total_units
FROM sales_imported
GROUP BY product_category
ORDER BY revenue DESC;

-- 3. 区域×月份交叉分析（立方体查询）
SELECT 
    region,
    strftime(order_date, '%Y-%m') AS month,
    count(*) AS orders,
    round(sum(total_amount)) AS revenue
FROM sales_imported
GROUP BY CUBE(region, month)
ORDER BY region, month;

-- 4. TOP 10 高价值客户
SELECT 
    customer_id,
    count(*) AS order_count,
    round(sum(total_amount)) AS total_spent,
    round(avg(total_amount), 2) AS avg_order_value,
    max(order_date) AS last_order_date
FROM sales_imported
GROUP BY customer_id
ORDER BY total_spent DESC
LIMIT 10;

-- 1. 滚动 30 天移动平均销售额
SELECT 
    order_date,
    round(sum(total_amount)) AS daily_revenue,
    round(avg(sum(total_amount)) OVER (
        ORDER BY order_date 
        ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
    ), 2) AS moving_avg_30d
FROM sales_imported
GROUP BY order_date
ORDER BY order_date;

-- 2. 同环比分析
WITH monthly AS (
    SELECT 
        strftime(order_date, '%Y-%m') AS month,
        round(sum(total_amount)) AS revenue
    FROM sales_imported
    GROUP BY month
)
SELECT 
    month,
    revenue,
    lag(revenue) OVER (ORDER BY month) AS prev_month,
    round((revenue - lag(revenue) OVER (ORDER BY month)) / 
          lag(revenue) OVER (ORDER BY month) * 100, 2) AS mom_change_pct
FROM monthly
ORDER BY month;

-- 3. 每个客户的首次/末次购买分析
SELECT 
    customer_id,
    min(order_date) AS first_purchase,
    max(order_date) AS last_purchase,
    count(*) AS total_orders,
    datediff('day', min(order_date), max(order_date)) AS customer_lifetime_days,
    round(sum(total_amount)) AS lifetime_value
FROM sales_imported
GROUP BY customer_id
HAVING count(*) >= 5
ORDER BY lifetime_value DESC;

-- 导出聚合结果
COPY (
    SELECT 
        product_category,
        region,
        strftime(order_date, '%Y-%m') AS month,
        count(*) AS orders,
        round(sum(total_amount)) AS revenue
    FROM sales_imported
    GROUP BY product_category, region, month
    ORDER BY revenue DESC
) TO '/tmp/sales_report.csv' (FORMAT CSV, HEADER true);

-- 导出为 Parquet（压缩率更高，适合长期存储）
COPY sales_imported TO '/tmp/sales_2026.parquet' (FORMAT PARQUET);
-- CSV: 85 MB → Parquet: 12 MB（压缩率 7:1）

-- 导出 JSON
COPY (
    SELECT region, round(sum(total_amount)) AS revenue
    FROM sales_imported GROUP BY region
) TO '/tmp/region_summary.json' (FORMAT JSON);

-- 场景：100 个按天分区的 CSV 文件
-- 文件命名: sales_2025-01-01.csv ... sales_2025-04-10.csv

-- 一行代码合并所有文件并聚合
SELECT 
    strftime(order_date, '%Y-%m') AS month,
    product_category,
    count(*) AS orders,
    round(sum(total_amount)) AS revenue
FROM read_csv('/data/sales_*.csv', 
    header = true, 
    union_by_name = true  -- 自动匹配列名
)
WHERE order_date >= '2025-01-01'
GROUP BY month, product_category
ORDER BY month, revenue DESC;

-- 合并所有文件到一张大表（用于后续分析）
CREATE TABLE all_sales AS
SELECT * FROM read_csv('/data/sales_*.csv', 
    header = true, 
    union_by_name = true);

-- DuckDB 默认使用全部可用内存，但你可以限制
SET memory_limit = '1GB';  -- 限制最多使用 1GB
SET temp_directory = '/tmp/duckdb_tmp';  -- 内存溢出时写入磁盘

-- 差的做法：读所有列再筛选
SELECT region, sum(total_amount)
FROM read_csv('/data/huge_file.csv', header = true)
WHERE region = '华东';

-- 好的做法：投影下推（只读需要的列）
SELECT region, sum(total_amount)
FROM read_csv('/data/huge_file.csv', 
    header = true,
    columns = {'region': 'VARCHAR', 'total_amount': 'DECIMAL(10,2)'}
)
WHERE region = '华东';

-- CSV → Parquet（一次性投资，长期受益）
COPY tbl TO '/data/optimized.parquet' (FORMAT PARQUET);

-- 查询 Parquet 比 CSV 快 3-10 倍
-- 因为 DuckDB 可以直接读取列式元数据，跳过不相关列
SELECT region, count(*) 
FROM read_parquet('/data/optimized.parquet')
GROUP BY region;

-- Parquet 还支持谓词下推
-- 下面的查询只读取涉及的数据块
SELECT * FROM read_parquet('/data/optimized.parquet')
WHERE order_date BETWEEN '2025-06-01' AND '2025-06-30';

-- 方法：使用 WHERE 子句分批
CREATE TABLE result AS
SELECT * FROM read_csv_auto('/data/huge_dataset.csv')
WHERE order_date >= '2025-01-01' 
  AND order_date < '2025-04-01';  -- 只加载一个季度

-- 对每个季度分别处理，然后用 UNION ALL 合并
CREATE TABLE yearly_result AS
SELECT * FROM read_csv_auto('/data/huge_dataset.csv')
WHERE order_date >= '2025-01-01' AND order_date < '2025-04-01'
UNION ALL
SELECT * FROM read_csv_auto('/data/huge_dataset.csv')
WHERE order_date >= '2025-04-01' AND order_date < '2025-07-01'
UNION ALL
SELECT * FROM read_csv_auto('/data/huge_dataset.csv')
WHERE order_date >= '2025-07-01' AND order_date < '2025-10-01'
UNION ALL
SELECT * FROM read_csv_auto('/data/huge_dataset.csv')
WHERE order_date >= '2025-10-01' AND order_date < '2026-01-01';

-- 创建视图而不是复制数据
CREATE VIEW sales_view AS
SELECT * FROM read_csv('/data/sales_*.csv', header = true, union_by_name = true);

-- 查询视图 → DuckDB 在查询时实时读取文件
SELECT region, sum(total_amount) 
FROM sales_view
GROUP BY region;
-- 零存储开销，但每次查询都会重新读取文件

#!/usr/bin/env python3
"""
weekly_sales_report.py - DuckDB 驱动的自动化周报生成器
运行方式: python3 weekly_sales_report.py
"""
import duckdb
import datetime

today = datetime.date.today()
monday = today - datetime.timedelta(days=today.weekday())
last_monday = monday - datetime.timedelta(days=7)

# 连接 DuckDB（内存模式）
conn = duckdb.connect()

# 安装扩展
conn.execute("INSTALL httpfs; LOAD httpfs;")

print(f"📊 生成销售周报: {last_monday} → {monday}")

# 读取本周数据并聚合
result = conn.execute(f"""
    SELECT 
        product_category,
        region,
        count(*) AS orders,
        count(DISTINCT customer_id) AS customers,
        round(sum(total_amount), 2) AS revenue,
        round(avg(total_amount), 2) AS avg_order_value
    FROM read_csv('/data/sales_*.csv', header = true, union_by_name = true)
    WHERE order_date >= '{last_monday}' 
      AND order_date < '{monday}'
    GROUP BY product_category, region
    ORDER BY revenue DESC
""").fetchdf()

# 导出为 Excel 兼容格式
result.to_csv('/tmp/weekly_report.csv', index=False)
print(f"✅ 周报已生成: /tmp/weekly_report.csv")
print(f"📈 共 {len(result)} 行数据")

# 地区排名
print("\n🏆 地区销售排名:")
region_stats = conn.execute(f"""
    SELECT region, round(sum(total_amount), 2) AS revenue
    FROM read_csv('/data/sales_*.csv', header = true, union_by_name = true)
    WHERE order_date >= '{last_monday}' AND order_date < '{monday}'
    GROUP BY region ORDER BY revenue DESC
""").fetchdf()
print(region_stats.to_string(index=False))

conn.close()

-- 查看查询计划
EXPLAIN ANALYZE SELECT ...;

-- 检查是否使用了正确的列类型
-- 避免 VARCHAR 类型的数值列
SELECT typeof(column_name) FROM read_csv(...);

-- 确保使用了合适的文件格式
-- CSV < Parquet < DuckDB 内部表（最快）

-- 设置临时目录
SET temp_directory = '/disk/tmp';

-- 限制内存使用
SET memory_limit = '2GB';

-- 使用流式处理（不缓存中间结果）
SELECT * FROM read_csv('huge.csv') 
WHERE ...  -- 谓词下推，尽早过滤
;