import duckdb
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta

# 连接到 DuckDB 数据库（持久化存储）
db_path = Path("ecommerce.duckdb")
con = duckdb.connect(str(db_path))

# 原地读取 CSV（无需先加载到内存）
con.execute("""
    CREATE TABLE IF NOT EXISTS orders AS 
    SELECT * FROM read_csv_auto('orders.csv')
""")
con.execute("""
    CREATE TABLE IF NOT EXISTS products AS 
    SELECT * FROM read_csv_auto('products.csv')
""")
con.execute("""
    CREATE TABLE IF NOT EXISTS users AS 
    SELECT * FROM read_csv_auto('users.csv')
""")

# 查看表结构
print(con.execute("DESCRIBE orders").fetchdf())
print(con.execute("SELECT COUNT(*) FROM orders").fetchone())

# 创建周度销售视图（按周自动切片）
con.execute("""
CREATE VIEW IF NOT EXISTS v_weekly_sales AS
SELECT 
    DATE_TRUNC('week', order_date) AS week,
    COUNT(*) AS total_orders,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value,
    COUNT(DISTINCT user_id) AS active_users,
    COUNT(DISTINCT product_id) AS unique_products
FROM orders
WHERE order_date >= CURRENT_DATE - INTERVAL '12 weeks'
GROUP BY DATE_TRUNC('week', order_date)
ORDER BY week DESC
""")

# 验证视图
print(con.execute("SELECT * FROM v_weekly_sales LIMIT 5").fetchdf())

def generate_weekly_report(db_path: str) -> dict:
    """生成周度经营报告"""
    con = duckdb.connect(db_path)
    
    # 获取本周和上周数据
    current_week = con.execute("""
        SELECT * FROM v_weekly_sales 
        WHERE week = DATE_TRUNC('week', CURRENT_DATE)
    """).fetchone()
    
    prev_week = con.execute("""
        SELECT * FROM v_weekly_sales 
        WHERE week = (
            SELECT MAX(week) FROM v_weekly_sales 
            WHERE week < DATE_TRUNC('week', CURRENT_DATE)
        )
    """).fetchone()
    
    # 环比计算函数
    def pct_change(curr, prev):
        if prev is None or prev == 0: 
            return None
        return round((curr - prev) / prev * 100, 2)
    
    report = {
        "week": current_week[0].strftime('%Y-%m-%d') if current_week else None,
        "total_orders": current_week[1] if current_week else 0,
        "total_revenue": round(current_week[2], 2) if current_week else 0,
        "revenue_growth": pct_change(current_week[2], prev_week[2]) if current_week and prev_week else None,
        "avg_order_value": round(current_week[3], 2) if current_week else 0,
        "avg_order_growth": pct_change(current_week[3], prev_week[3]) if current_week and prev_week else None,
        "active_users": current_week[4] if current_week else 0,
        "user_growth": pct_change(current_week[4], prev_week[4]) if current_week and prev_week else None,
    }
    
    # 热销品类 TOP5
    report["top_categories"] = con.execute("""
        SELECT p.category, SUM(o.amount) as revenue
        FROM orders o 
        JOIN products p ON o.product_id = p.id
        WHERE DATE_TRUNC('week', o.order_date) = DATE_TRUNC('week', CURRENT_DATE)
        GROUP BY p.category
        ORDER BY revenue DESC
        LIMIT 5
    """).fetchall()
    
    # 用户留存分析（窗口函数）
    report["retention"] = con.execute("""
        SELECT 
            first_week,
            COUNT(DISTINCT user_id) as new_users,
            SUM(CASE WHEN weeks_active >= 2 THEN 1 ELSE 0 END) as retained_2w,
            SUM(CASE WHEN weeks_active >= 3 THEN 1 ELSE 0 END) as retained_3w
        FROM (
            SELECT 
                user_id,
                DATE_TRUNC('week', MIN(order_date)) as first_week,
                COUNT(DISTINCT DATE_TRUNC('week', order_date)) as weeks_active
            FROM orders
            WHERE order_date >= CURRENT_DATE - INTERVAL '12 weeks'
            GROUP BY user_id
        )
        GROUP BY first_week
        ORDER BY first_week DESC
        LIMIT 4
    """).fetchall()
    
    con.close()
    return report

import json
from pathlib import Path

def export_report(report: dict, output_dir: str):
    """导出为多种格式"""
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    
    # JSON 格式（供 API 调用）
    (out / "report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    
    # Markdown 格式（可直接发微信/飞书）
    md = f"""# 周度经营报告 {report['week']}

## 核心指标
- **订单数**: {report['total_orders']:,}（{'↑' if report['revenue_growth'] and report['revenue_growth'] > 0 else '↓'} {abs(report['revenue_growth']):.1f}%）
- **营收**: ¥{report['total_revenue']:,.2f}（{'↑' if report['revenue_growth'] and report['revenue_growth'] > 0 else '↓'} {abs(report['revenue_growth']):.1f}%）
- **客单价**: ¥{report['avg_order_value']:.2f}（{'↑' if report['avg_order_growth'] and report['avg_order_growth'] > 0 else '↓'} {abs(report['avg_order_growth']):.1f}%）
- **活跃用户**: {report['active_users']:,}

## 热销品类 TOP5
"""
    for i, (cat, rev) in enumerate(report['top_categories'], 1):
        md += f"{i}. {cat}: ¥{rev:,.2f}\n"
    
    (out / "report.md").write_text(md)
    print(f"✅ 报告已生成: {out}")

# 使用示例
if __name__ == "__main__":
    report = generate_weekly_report("ecommerce.duckdb")
    export_report(report, "reports/2026-W31")

# main.py
from pathlib import Path
import duckdb
from datetime import datetime

def main():
    # 1. 加载数据（幂等）
    con = duckdb.connect("ecommerce.duckdb")
    
    # 自动增量更新
    csv_files = list(Path("data").glob("orders_*.csv"))
    for f in sorted(csv_files)[-1:]:  # 只加载最新文件
        con.execute(f"INSERT INTO orders SELECT * FROM read_csv_auto('{f}')")
    
    # 2. 生成报告
    report = generate_weekly_report("ecommerce.duckdb")
    
    # 3. 导出
    week_str = report['week'].replace('-', '')
    export_report(report, f"reports/{week_str}")
    
    con.close()
    print("🎉 周报生成完成")

if __name__ == "__main__":
    main()