DuckDB + Python + Cron：用 50 行代码打造自动化收入报表系统

# daily_income_report.py
import duckdb
import pandas as pd
from pathlib import Path
from datetime import datetime
import os

# ============================================================
# 步骤 1：批量读取所有 CSV 文件（使用 DuckDB 原生支持）
# ============================================================
data_dir = Path("data")
os.makedirs(data_dir, exist_ok=True)

# DuckDB 的 globbing 功能可以直接匹配多个文件，比 Pandas 快 10 倍以上
con = duckdb.connect()
result = con.execute("""
    SELECT 
        '*.csv' AS source_file,
        order_id,
        product_name,
        amount,
        region,
        cast(sale_date as date) as sale_date
    FROM '*.csv'
    WHERE amount > 0
    ORDER BY sale_date, region
""").df()

print(f"✅ 已读取 {len(result)} 条销售记录，来自 {result['source_file'].nunique()} 个文件")

# ============================================================
# 步骤 2：多维聚合分析（DuckDB SQL 的强大之处）
# ============================================================
# 将 DataFrame 注册为虚拟表，用 SQL 做复杂聚合
con.from_dataframe(result).execute("""
    CREATE OR REPLACE VIEW daily_summary AS
    SELECT 
        DATE(sale_date) AS sale_day,
        region,
        product_name,
        COUNT(DISTINCT order_id) AS order_count,
        SUM(amount) AS total_amount,
        AVG(amount) AS avg_order_value,
        COUNT(DISTINCT order_id) * AVG(amount) AS estimated_gmv
    FROM result
    GROUP BY sale_day, region, product_name
    ORDER BY sale_day, total_amount DESC
""")

# ============================================================
# 步骤 3：生成 Markdown 格式的日报报告
# ============================================================
today = datetime.now().strftime('%Y-%m-%d')
report_date = datetime.now().strftime('%Y%m%d')

os.makedirs("reports", exist_ok=True)

report = f"""# 📊 电商日报 - {today}

## 📈 今日概览

今日共处理 {len(result)} 笔交易，总销售额：{result['amount'].sum():,.2f} 元，涉及区域：{result['region'].nunique()} 个，产品线：{result['product_name'].nunique()} 款。

---

## 🌍 各区域销售额 TOP 10

{con.execute("""
    SELECT region, SUM(amount) AS total_amount
    FROM daily_summary
    GROUP BY region
    ORDER BY total_amount DESC
    LIMIT 10
""").read_csv()}

## 🏷️ 按品类分析

{con.execute("""
    SELECT product_name, 
           SUM(amount) AS total_amount,
           COUNT(DISTINCT order_id) AS order_count,
           AVG(amount) AS avg_price,
           COUNT(*) AS transactions
    FROM daily_summary
    GROUP BY product_name
    ORDER BY total_amount DESC
""").read_csv()}

## 💡 今日洞察

### 🔥 TOP 3 销售区域：
{con.execute("""
    SELECT region, SUM(amount) AS total_amount,
           ROUND(SUM(amount)/SUM(total_amount)*100,1)||'%' as share
    FROM daily_summary
    GROUP BY region
    ORDER BY total_amount DESC
    LIMIT 3
""").read_csv()}

### ⭐ 单客价值最高的产品：
{con.execute("""
    SELECT product_name, 
           ROUND(AVG(amount),2) as avg_order_value,
           COUNT(DISTINCT order_id) AS unique_orders
    FROM daily_summary
    GROUP BY product_name
    ORDER BY avg_order_value DESC
    LIMIT 5
""").read_csv()}

---

## 📅 每日趋势（最近 7 天）

{con.execute("""
    SELECT sale_day, 
           SUM(amount) as daily_revenue,
           COUNT(DISTINCT order_id) as daily_orders
    FROM daily_summary
    WHERE sale_day >= DATEADD('day', -7, current_date)
    GROUP BY sale_day
    ORDER BY sale_day
""").read_csv()}
"""

Path(f"reports/income_report_{report_date}.md").write_text(report, encoding='utf-8')
print(f"📄 报告已生成: reports/income_report_{report_date}.md")

# ============================================================
# 可选步骤 4：保存原始数据为 Parquet（供下次快速读取）
# ============================================================
con.execute("""
    COPY (SELECT * FROM result) TO '/data/parquet/raw_data.parquet' (FORMAT PARQUET)
""")
print("💾 原始数据已保存为 Parquet 格式，下次可直接读取加速")

crontab -e

0 8 * * * /usr/bin/python3 /path/to/daily_income_report.py >> /var/log/daily_income.log 2>&1

import sys
import smtplib

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(f"❌ 报告生成失败: {e}")
        # 发送告警邮件的逻辑...
        sys.exit(1)

-- 直接查数据库
SELECT * FROM postgres_connect('host=localhost dbname=mydb user=pass')

-- 或者读 S3 上的 Parquet
SELECT FROM 's3://bucket/data/*.parquet'

from fastapi import FastAPI
app = FastAPI()

@app.get("/sales/{region}")
def get_sales(region: str):
    return duckdb.query(f"SELECT * FROM daily_summary WHERE region='{region}'").df()