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