Spaces:
Sleeping
Sleeping
| 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 | |
| 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 | |
| } | |
| 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 | |
| } | |
| 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 | |
| } | |
| 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 | |
| } | |
| 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 | |
| } | |
| 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 | |
| } | |