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

# Connect to DuckDB database (persistent storage)
db_path = Path("ecommerce.duckdb")
con = duckdb.connect(str(db_path))

# Read CSV in-place (no need to load into memory first)
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')
""")

# Check table structure
print(con.execute("DESCRIBE orders").fetchdf())
print(con.execute("SELECT COUNT(*) FROM orders").fetchone())

# Create weekly sales view (auto-slices by week)
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
""")

# Verify view
print(con.execute("SELECT * FROM v_weekly_sales LIMIT 5").fetchdf())

def generate_weekly_report(db_path: str) -> dict:
    """Generate weekly business report"""
    con = duckdb.connect(db_path)
    
    # Get current week and previous week data
    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()
    
    # Week-over-week calculation function
    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,
    }
    
    # Top 5 categories by revenue
    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()
    
    # User retention analysis (window functions)
    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):
    """Export in multiple formats"""
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    
    # JSON format (for API calls)
    (out / "report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2)
    )
    
    # Markdown format (can be sent directly via WeChat/Feishu)
    md = f"""# Weekly Report {report['week']}

## Core Metrics
- **Orders**: {report['total_orders']:,} ({'↑' if report['revenue_growth'] and report['revenue_growth'] > 0 else '↓'} {abs(report['revenue_growth']):.1f}%)
- **Revenue**: ${report['total_revenue']:,.2f} ({'↑' if report['revenue_growth'] and report['revenue_growth'] > 0 else '↓'} {abs(report['revenue_growth']):.1f}%)
- **Avg Order Value**: ${report['avg_order_value']:.2f} ({'↑' if report['avg_order_growth'] and report['avg_order_growth'] > 0 else '↓'} {abs(report['avg_order_growth']):.1f}%)
- **Active Users**: {report['active_users']:,}

## Top 5 Categories
"""
    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"✅ Report generated: {out}")

# Usage example
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. Load data (idempotent)
    con = duckdb.connect("ecommerce.duckdb")
    
    # Auto incremental update
    csv_files = list(Path("data").glob("orders_*.csv"))
    for f in sorted(csv_files)[-1:]:  # Only load the latest file
        con.execute(f"INSERT INTO orders SELECT * FROM read_csv_auto('{f}')")
    
    # 2. Generate report
    report = generate_weekly_report("ecommerce.duckdb")
    
    # 3. Export
    week_str = report['week'].replace('-', '')
    export_report(report, f"reports/{week_str}")
    
    con.close()
    print("🎉 Weekly report generated")

if __name__ == "__main__":
    main()