import duckdb
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
import os

app = FastAPI(title="Lightweight Analytics API")
DB_DIR = "./customer_dbs"

def get_conn(customer_id: str):
 """Each customer gets their own DuckDB database file"""
 os.makedirs(DB_DIR, exist_ok=True)
 db_path = f"{DB_DIR}/{customer_id}.db"
 conn = duckdb.connect(db_path)
 conn.execute("INSTALL 'httpfs'; LOAD 'httpfs';")
 conn.execute("INSTALL 'json'; LOAD 'json';")
 conn.execute("INSTALL 'spatial'; LOAD 'spatial';")
 return conn

@app.post("/api/{customer_id}/upload")
async def upload_data(customer_id: str, table_name: str, file: UploadFile = File(...)):
 """
 Customer uploads CSV/Parquet file, table auto-created.
 DuckDB infers column names and types automatically.
 """
 conn = get_conn(customer_id)
 temp_path = f"/tmp/{customer_id}_{file.filename}"

 with open(temp_path, "wb") as f:
 content = await file.read()
 f.write(content)

 try:
 conn.execute(f"""
 CREATE OR REPLACE TABLE {table_name} AS
 SELECT * FROM read_csv_auto('{temp_path}', header=true, all_varchar=false)
 """)
 row_count = conn.execute(f"SELECT count(*) FROM {table_name}").fetchone()[0]
 os.remove(temp_path)
 return {"status": "ok", "table": table_name, "rows": row_count}
 except Exception as e:
 os.remove(temp_path)
 raise HTTPException(status_code=400, detail=str(e))

# SQL Template Asset Library — each query is a competitive advantage
SQL_TEMPLATES = {
 "daily_sales": """
 SELECT
 strftime(order_date, '%Y-%m-%d') as date,
 sum(amount) as total_sales,
 count(distinct customer_id) as unique_customers,
 sum(amount) / nullif(count(*), 0) as avg_order_value
 FROM orders
 WHERE order_date >= current_date - interval '30 days'
 GROUP BY strftime(order_date, '%Y-%m-%d')
 ORDER BY date DESC
 """,
 "top_products": """
 SELECT
 product_name,
 sum(quantity) as units_sold,
 sum(amount) as revenue,
 count(distinct customer_id) as buyers
 FROM orders
 WHERE order_date >= current_date - interval '7 days'
 GROUP BY product_name
 ORDER BY revenue DESC
 LIMIT 10
 """,
 "customer_segments": """
 WITH customer_value AS (
 SELECT
 customer_id,
 sum(amount) as lifetime_value,
 count(*) as order_count,
 datediff('day', max(order_date), current_date) as days_since_last
 FROM orders
 GROUP BY customer_id
 )
 SELECT
 CASE
 WHEN lifetime_value > 10000 THEN 'High Value'
 WHEN lifetime_value > 3000 THEN 'Mid Value'
 ELSE 'Low Value'
 END as segment,
 count(*) as customer_count,
 avg(lifetime_value) as avg_lifetime_value,
 avg(days_since_last) as avg_recency
 FROM customer_value
 GROUP BY segment
 ORDER BY avg_lifetime_value DESC
 """,
 "inventory_warning": """
 SELECT
 product_name,
 stock_quantity,
 reorder_point,
 CASE WHEN stock_quantity <= reorder_point THEN '⚠️ Reorder Needed' ELSE 'OK' END as status
 FROM inventory
 WHERE stock_quantity <= reorder_point
 ORDER BY stock_quantity ASC
 """,
 "weekly_trend": """
 SELECT
 date_trunc('week', order_date) as week,
 sum(amount) as weekly_sales,
 count(*) as order_count,
 sum(amount) / nullif(count(*), 0) as avg_order_value
 FROM orders
 WHERE order_date >= current_date - interval '12 weeks'
 GROUP BY week
 ORDER BY week DESC
 """
}

class QueryRequest(BaseModel):
 query_name: str
 params: dict = {}

@app.post("/api/{customer_id}/query")
async def run_query(customer_id: str, req: QueryRequest):
 """Execute a predefined query and return JSON results"""
 if req.query_name not in SQL_TEMPLATES:
 raise HTTPException(status_code=404, detail="Query template not found")

 conn = get_conn(customer_id)
 sql = SQL_TEMPLATES[req.query_name]

 try:
 result = conn.execute(sql).fetchdf()
 return {
 "query": req.query_name,
 "columns": list(result.columns),
 "data": result.values.tolist(),
 "rows": len(result)
 }
 except Exception as e:
 raise HTTPException(status_code=400, detail=str(e))

import httpx
import json

# cron task: scan all customers at 09:00 daily, generate and push reports
async def generate_daily_reports():
 customers = ["Customer_A", "Customer_B", "Customer_C"]

 for cid in customers:
 conn = get_conn(cid)

 # Check if customer has data
 has_data = conn.execute(
 "SELECT count(*) FROM information_schema.tables WHERE table_name='orders'"
 ).fetchone()[0]
 if not has_data:
 continue

 # Single DuckDB query for multi-dimensional KPI aggregation
 report = conn.execute("""
 SELECT
 strftime(current_date, '%Y-%m-%d') as report_date,
 (SELECT sum(amount) FROM orders WHERE order_date = current_date) as today_sales,
 (SELECT sum(amount) FROM orders WHERE order_date = current_date - interval '1 day') as yesterday_sales,
 round(
 (SELECT sum(amount) FROM orders WHERE order_date = current_date)
 / nullif((SELECT sum(amount) FROM orders WHERE order_date = current_date - interval '1 day'), 0) * 100 - 100
 , 1) as mom_change,
 (SELECT count(*) FROM orders WHERE order_date = current_date) as today_orders,
 (SELECT count(DISTINCT customer_id) FROM orders WHERE order_date = current_date) as today_customers
 """).fetchdf()

 # Format push notification
 message = f"""
📊 Daily Report — {cid}
📅 Date: {report['report_date'][0]}
💰 Today's Sales: ¥{report['today_sales'][0]:,.2f}
📉 MoM Change: {report['mom_change'][0]}%
🛒 Orders: {report['today_orders'][0]}
👤 Customers: {report['today_customers'][0]}
 """

 # Push via Telegram Bot
 async with httpx.AsyncClient() as client:
 await client.post(
 f"https://api.telegram.org/bot{TOKEN}/sendMessage",
 data={"chat_id": CHAT_ID, "text": message}
 )

# Cloud server (recommended: 2C4G, Ubuntu 22.04)
apt update && apt install -y python3 python3-pip
pip install duckdb fastapi uvicorn httpx pydantic

# Configure systemd for auto-restart
cat > /etc/systemd/system/analytics-api.service << 'EOF'
[Unit]
Description=Lightweight Analytics API
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/analytics-api
ExecStart=/usr/bin/python3 -m uvicorn main:app --host 0.0.0.0 --port 8000
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

systemctl enable analytics-api
systemctl start analytics-api

# Daily report at 9 AM
crontab -e
# Add:
0 9 * * * cd /home/ubuntu/analytics-api && python3 -c "import asyncio; from main import generate_daily_reports; asyncio.run(generate_daily_reports())"

# Weekly report every Monday 9:30 AM
30 9 * * 1 cd /home/ubuntu/analytics-api && python3 generate_weekly.py

# In production, add authentication and rate limiting
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()
CUSTOMER_TOKENS = {
 "Customer_A": "token_a_xxxx",
 "Customer_B": "token_b_xxxx",
}

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
 token = credentials.credentials
 for cid, t in CUSTOMER_TOKENS.items():
 if t == token:
 return cid
 raise HTTPException(status_code=403, detail="Invalid token")

@app.get("/api/my/report")
async def get_report(customer_id: str = Depends(verify_token)):
 # Authenticated customer
 ...