from datetime import datetime, timedelta from typing import Any from sqlalchemy import func from sqlalchemy.orm import Session from app.models.category import Category from app.models.order import Order, OrderStatus from app.models.product import Product, ProductStatus from app.models.user import User, UserRole from app.schemas.admin import TimePeriod class AdminDashboardService: """ Service for admin dashboard analytics. """ @staticmethod def get_date_range(period: TimePeriod) -> tuple: """ Get date range for a given time period. Args: period: Time period to get range for Returns: Tuple of (start_date, end_date) """ now = datetime.now() end_date = now if period == TimePeriod.TODAY: start_date = datetime(now.year, now.month, now.day, 0, 0, 0) elif period == TimePeriod.YESTERDAY: yesterday = now - timedelta(days=1) start_date = datetime(yesterday.year, yesterday.month, yesterday.day, 0, 0, 0) end_date = datetime(yesterday.year, yesterday.month, yesterday.day, 23, 59, 59) elif period == TimePeriod.LAST_7_DAYS: start_date = now - timedelta(days=7) elif period == TimePeriod.LAST_30_DAYS: start_date = now - timedelta(days=30) elif period == TimePeriod.THIS_MONTH: start_date = datetime(now.year, now.month, 1) elif period == TimePeriod.LAST_MONTH: if now.month == 1: start_date = datetime(now.year - 1, 12, 1) end_date = datetime(now.year, now.month, 1) - timedelta(days=1) else: start_date = datetime(now.year, now.month - 1, 1) end_date = datetime(now.year, now.month, 1) - timedelta(days=1) elif period == TimePeriod.THIS_YEAR: start_date = datetime(now.year, 1, 1) else: # ALL_TIME start_date = datetime(1900, 1, 1) # A long time ago return start_date, end_date @staticmethod def get_sales_summary(db: Session, period: TimePeriod) -> dict[str, Any]: """ Get sales summary for a given time period. Args: db: Database session period: Time period to get summary for Returns: Dictionary with sales summary data """ start_date, end_date = AdminDashboardService.get_date_range(period) # Get completed orders in the date range completed_statuses = [OrderStatus.DELIVERED, OrderStatus.COMPLETED] orders = db.query(Order).filter( Order.created_at.between(start_date, end_date), Order.status.in_(completed_statuses) ).all() # Get refunded orders in the date range refunded_orders = db.query(Order).filter( Order.created_at.between(start_date, end_date), Order.status == OrderStatus.REFUNDED ).all() # Calculate totals total_sales = sum(order.total_amount for order in orders) total_orders = len(orders) average_order_value = total_sales / total_orders if total_orders > 0 else 0 refunded_amount = sum(order.total_amount for order in refunded_orders) return { "period": period, "total_sales": total_sales, "total_orders": total_orders, "average_order_value": average_order_value, "refunded_amount": refunded_amount } @staticmethod def get_sales_over_time(db: Session, period: TimePeriod) -> dict[str, Any]: """ Get sales data over time for the given period. Args: db: Database session period: Time period to get data for Returns: Dictionary with sales over time data """ start_date, end_date = AdminDashboardService.get_date_range(period) # Determine the date grouping and format based on the period date_format = "%Y-%m-%d" # Default daily format delta = timedelta(days=1) # Default daily increment if period in [TimePeriod.THIS_YEAR, TimePeriod.ALL_TIME]: date_format = "%Y-%m" # Monthly format # Generate all dates in the range date_range = [] current_date = start_date while current_date <= end_date: date_range.append(current_date.strftime(date_format)) current_date += delta # Get completed orders in the date range completed_statuses = [OrderStatus.DELIVERED, OrderStatus.COMPLETED] orders = db.query(Order).filter( Order.created_at.between(start_date, end_date), Order.status.in_(completed_statuses) ).all() # Group orders by date date_sales = {} for date_str in date_range: date_sales[date_str] = {"date": date_str, "total_sales": 0, "order_count": 0} for order in orders: date_str = order.created_at.strftime(date_format) if date_str in date_sales: date_sales[date_str]["total_sales"] += order.total_amount date_sales[date_str]["order_count"] += 1 # Convert to list and sort by date data = list(date_sales.values()) data.sort(key=lambda x: x["date"]) total_sales = sum(item["total_sales"] for item in data) total_orders = sum(item["order_count"] for item in data) return { "period": period, "data": data, "total_sales": total_sales, "total_orders": total_orders } @staticmethod def get_top_categories(db: Session, period: TimePeriod, limit: int = 5) -> dict[str, Any]: """ Get top selling categories for the given period. Args: db: Database session period: Time period to get data for limit: Number of categories to return Returns: Dictionary with top category sales data """ start_date, end_date = AdminDashboardService.get_date_range(period) # Get completed orders in the date range completed_statuses = [OrderStatus.DELIVERED, OrderStatus.COMPLETED] # This is a complex query that would involve joining multiple tables # For simplicity, we'll use a more direct approach # Get all order items from completed orders orders = db.query(Order).filter( Order.created_at.between(start_date, end_date), Order.status.in_(completed_statuses) ).all() # Collect category sales data category_sales = {} total_sales = 0 for order in orders: for item in order.items: product = db.query(Product).filter(Product.id == item.product_id).first() if product and product.category_id: category_id = product.category_id category = db.query(Category).filter(Category.id == category_id).first() if category: if category_id not in category_sales: category_sales[category_id] = { "category_id": category_id, "category_name": category.name, "total_sales": 0 } item_total = item.unit_price * item.quantity - item.discount category_sales[category_id]["total_sales"] += item_total total_sales += item_total # Convert to list and sort by total sales categories = list(category_sales.values()) categories.sort(key=lambda x: x["total_sales"], reverse=True) # Calculate percentages and limit results for category in categories: category["percentage"] = (category["total_sales"] / total_sales * 100) if total_sales > 0 else 0 categories = categories[:limit] return { "period": period, "categories": categories, "total_sales": total_sales } @staticmethod def get_top_products(db: Session, period: TimePeriod, limit: int = 5) -> dict[str, Any]: """ Get top selling products for the given period. Args: db: Database session period: Time period to get data for limit: Number of products to return Returns: Dictionary with top product sales data """ start_date, end_date = AdminDashboardService.get_date_range(period) # Get completed orders in the date range completed_statuses = [OrderStatus.DELIVERED, OrderStatus.COMPLETED] # Get all order items from completed orders orders = db.query(Order).filter( Order.created_at.between(start_date, end_date), Order.status.in_(completed_statuses) ).all() # Collect product sales data product_sales = {} total_sales = 0 for order in orders: for item in order.items: product_id = item.product_id if product_id not in product_sales: product = db.query(Product).filter(Product.id == product_id).first() product_name = item.product_name if item.product_name else (product.name if product else "Unknown Product") product_sales[product_id] = { "product_id": product_id, "product_name": product_name, "quantity_sold": 0, "total_sales": 0 } product_sales[product_id]["quantity_sold"] += item.quantity item_total = item.unit_price * item.quantity - item.discount product_sales[product_id]["total_sales"] += item_total total_sales += item_total # Convert to list and sort by total sales products = list(product_sales.values()) products.sort(key=lambda x: x["total_sales"], reverse=True) # Limit results products = products[:limit] return { "period": period, "products": products, "total_sales": total_sales } @staticmethod def get_top_customers(db: Session, period: TimePeriod, limit: int = 5) -> dict[str, Any]: """ Get top customers for the given period. Args: db: Database session period: Time period to get data for limit: Number of customers to return Returns: Dictionary with top customer data """ start_date, end_date = AdminDashboardService.get_date_range(period) # Get completed orders in the date range completed_statuses = [OrderStatus.DELIVERED, OrderStatus.COMPLETED] # Get all orders from the period orders = db.query(Order).filter( Order.created_at.between(start_date, end_date), Order.status.in_(completed_statuses) ).all() # Collect customer data customer_data = {} total_sales = 0 for order in orders: user_id = order.user_id if user_id not in customer_data: user = db.query(User).filter(User.id == user_id).first() user_name = f"{user.first_name} {user.last_name}" if user and user.first_name else f"User {user_id}" customer_data[user_id] = { "user_id": user_id, "user_name": user_name, "order_count": 0, "total_spent": 0 } customer_data[user_id]["order_count"] += 1 customer_data[user_id]["total_spent"] += order.total_amount total_sales += order.total_amount # Convert to list and sort by total spent customers = list(customer_data.values()) customers.sort(key=lambda x: x["total_spent"], reverse=True) # Limit results customers = customers[:limit] return { "period": period, "customers": customers, "total_sales": total_sales } @staticmethod def get_dashboard_summary(db: Session) -> dict[str, Any]: """ Get a summary of key metrics for the dashboard. Args: db: Database session Returns: Dictionary with dashboard summary data """ # Get sales summary for last 30 days sales_summary = AdminDashboardService.get_sales_summary(db, TimePeriod.LAST_30_DAYS) # Count pending orders pending_orders = db.query(Order).filter(Order.status == OrderStatus.PENDING).count() # Count low stock products (less than 5 items) low_stock_products = db.query(Product).filter( Product.stock_quantity <= 5, Product.status != ProductStatus.DISCONTINUED ).count() # Count new customers in the last 30 days last_30_days = datetime.now() - timedelta(days=30) new_customers = db.query(User).filter( User.created_at >= last_30_days, User.role == UserRole.CUSTOMER ).count() # Count total products and customers total_products = db.query(Product).count() total_customers = db.query(User).filter(User.role == UserRole.CUSTOMER).count() return { "sales_summary": sales_summary, "pending_orders": pending_orders, "low_stock_products": low_stock_products, "new_customers": new_customers, "total_products": total_products, "total_customers": total_customers } @staticmethod def get_orders_by_status(db: Session) -> list[dict[str, Any]]: """ Get order counts by status. Args: db: Database session Returns: List of dictionaries with order status counts """ # Count orders by status status_counts = db.query( Order.status, func.count(Order.id).label('count') ).group_by(Order.status).all() # Calculate total orders total_orders = sum(count for _, count in status_counts) # Format results result = [] for status, count in status_counts: percentage = (count / total_orders * 100) if total_orders > 0 else 0 result.append({ "status": status.value, "count": count, "percentage": percentage }) # Sort by count (descending) result.sort(key=lambda x: x["count"], reverse=True) return result