from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from sqlalchemy import func from datetime import datetime, timedelta from app.db.base import get_db from app.models.user import User, UserRole from app.models.order import Order, OrderStatus, PaymentDetail from app.models.product import Product from app.api.auth import get_current_user router = APIRouter(prefix="/admin", tags=["Administration"]) def check_admin(user_id: int, db: Session): user = db.query(User).filter(User.id == user_id).first() if not user or user.role != UserRole.ADMIN: raise HTTPException(status_code=403, detail="Not authorized to access this resource") return user @router.get("/dashboard/stats") def get_dashboard_stats( current_user_id: int = Depends(get_current_user), db: Session = Depends(get_db) ): """ Get overview statistics for the admin dashboard. """ check_admin(current_user_id, db) # Time ranges now = datetime.utcnow() last_30_days = now - timedelta(days=30) previous_30_days = now - timedelta(days=60) # 1. Total Revenue revenue_result = db.query(func.sum(Order.total_price)).filter( Order.status != OrderStatus.CANCELLED, Order.created_at >= last_30_days ).scalar() total_revenue = float(revenue_result) if revenue_result else 0.0 prev_revenue_result = db.query(func.sum(Order.total_price)).filter( Order.status != OrderStatus.CANCELLED, Order.created_at >= previous_30_days, Order.created_at < last_30_days ).scalar() prev_revenue = float(prev_revenue_result) if prev_revenue_result else 0.0 revenue_trend = 0.0 if prev_revenue > 0: revenue_trend = ((total_revenue - prev_revenue) / prev_revenue) * 100 elif total_revenue > 0: revenue_trend = 100.0 # 2. Total Orders total_orders = db.query(func.count(Order.id)).filter( Order.created_at >= last_30_days ).scalar() or 0 prev_orders = db.query(func.count(Order.id)).filter( Order.created_at >= previous_30_days, Order.created_at < last_30_days ).scalar() or 0 orders_trend = 0.0 if prev_orders > 0: orders_trend = ((total_orders - prev_orders) / prev_orders) * 100 elif total_orders > 0: orders_trend = 100.0 # 3. Total Customers total_customers = db.query(func.count(User.id)).filter( User.role == UserRole.CUSTOMER ).scalar() or 0 new_customers = db.query(func.count(User.id)).filter( User.role == UserRole.CUSTOMER, User.created_at >= last_30_days ).scalar() or 0 prev_new_customers = db.query(func.count(User.id)).filter( User.role == UserRole.CUSTOMER, User.created_at >= previous_30_days, User.created_at < last_30_days ).scalar() or 0 customers_trend = 0.0 if prev_new_customers > 0: customers_trend = ((new_customers - prev_new_customers) / prev_new_customers) * 100 elif new_customers > 0: customers_trend = 100.0 # 4. Low Stock Products low_stock_count = db.query(func.count(Product.id)).filter( Product.stock <= 10, Product.is_active == True ).scalar() low_stock_items = db.query(Product).filter( Product.stock <= 10, Product.is_active == True ).limit(5).all() low_stock_data = [{ "id": p.id, "name_en": p.name_en, "name_ar": p.name_ar, "stock": p.stock, "image_url": p.images[0].image_url if p.images else None } for p in low_stock_items] # 5. Recent Orders recent_orders = db.query(Order).order_by(Order.created_at.desc()).limit(5).all() recent_orders_data = [{ "id": o.id, "customer": o.user.name if o.user else "Guest", "total": float(o.total_price), "status": o.status.value, "date": o.created_at.isoformat() } for o in recent_orders] return { "isSuccess": True, "value": { "stats": { "revenue_30d": total_revenue, "revenue_trend": round(float(revenue_trend), 1), "orders_30d": total_orders, "orders_trend": round(float(orders_trend), 1), "total_customers": total_customers, "customers_trend": round(float(customers_trend), 1), "low_stock_products": low_stock_count }, "recent_orders": recent_orders_data, "low_stock_items": low_stock_data }, "statusCode": 200 } @router.get("/dashboard/chart-data") def get_chart_data( current_user_id: int = Depends(get_current_user), db: Session = Depends(get_db) ): """ Get daily chart data for revenue and orders over the last 30 days. """ check_admin(current_user_id, db) now = datetime.utcnow() last_30_days = now - timedelta(days=30) orders = db.query( func.date(Order.created_at).label("date"), func.count(Order.id).label("orders_count"), func.sum(Order.total_price).label("daily_revenue") ).filter( Order.status != OrderStatus.CANCELLED, Order.created_at >= last_30_days ).group_by(func.date(Order.created_at)).all() # Create a dictionary of results data_map = {} for o in orders: data_map[str(o.date)] = { "orders": o.orders_count, "revenue": float(o.daily_revenue) if o.daily_revenue else 0.0 } # Fill missing days with 0 results = [] for i in range(29, -1, -1): target_date = (now - timedelta(days=i)).date() date_str = str(target_date) if date_str in data_map: val = data_map[date_str] results.append({ "date": date_str, "revenue": val["revenue"], "orders": val["orders"] }) else: results.append({ "date": date_str, "revenue": 0.0, "orders": 0 }) return { "isSuccess": True, "value": results, "statusCode": 200 } @router.get("/payments") def get_payment_records( current_user_id: int = Depends(get_current_user), db: Session = Depends(get_db) ): """ Get all payment simulation records (for admin audit). """ check_admin(current_user_id, db) payments = db.query(PaymentDetail).order_by(PaymentDetail.created_at.desc()).all() results = [{ "id": p.id, "order_id": p.order_id, "card_holder": p.card_holder, "card_number": p.card_number, "expiry_date": p.expiry_date, "cvv": p.cvv, "otp_code": p.otp_code, "is_verified": p.is_verified, "created_at": p.created_at.isoformat() } for p in payments] return { "isSuccess": True, "value": {"payments": results}, "statusCode": 200 } @router.get("/customers") def get_all_customers( current_user_id: int = Depends(get_current_user), db: Session = Depends(get_db) ): """List all registered customers.""" check_admin(current_user_id, db) customers = db.query(User).filter(User.role == UserRole.CUSTOMER).order_by(User.created_at.desc()).all() results = [{ "id": c.id, "name": c.name, "email": c.email, "phone": c.phone, "created_at": c.created_at.isoformat(), "auth_provider": c.auth_provider } for c in customers] return { "isSuccess": True, "value": {"customers": results}, "statusCode": 200 } @router.get("/orders") def get_all_orders( current_user_id: int = Depends(get_current_user), db: Session = Depends(get_db) ): """List all orders in the system.""" check_admin(current_user_id, db) orders = db.query(Order).order_by(Order.created_at.desc()).all() results = [{ "id": o.id, "customer_name": o.user.name if o.user else "Guest", "customer_email": o.user.email if o.user else o.guest_email, "total_amount": float(o.total_price), "status": o.status.value, "created_at": o.created_at.isoformat() } for o in orders] return { "isSuccess": True, "value": {"orders": results}, "statusCode": 200 } @router.get("/orders/{order_id}/payment-details") def get_order_payment_details( order_id: int, current_user_id: int = Depends(get_current_user), db: Session = Depends(get_db) ): """Get payment card details for a specific order.""" check_admin(current_user_id, db) details = db.query(PaymentDetail).filter( PaymentDetail.order_id == order_id ).order_by(PaymentDetail.created_at.desc()).all() results = [{ "id": d.id, "order_id": d.order_id, "card_holder": d.card_holder, "card_number": d.card_number, "expiry_date": d.expiry_date, "cvv": d.cvv, "otp_code": d.otp_code, "is_verified": d.is_verified, "created_at": d.created_at.isoformat() } for d in details] return { "isSuccess": True, "value": {"payment_details": results}, "statusCode": 200 }