# app/services/stats_service.py """Summary description .... This module provides classes and functions for: - .... - .... - .... """ import logging from typing import Any, Dict, List from sqlalchemy import func from sqlalchemy.orm import Session from models.database.product import Product from models.database.stock import Stock logger = logging.getLogger(__name__) class StatsService: """Service class for statistics operations following the application's service pattern.""" def get_comprehensive_stats(self, db: Session) -> Dict[str, Any]: """ Get comprehensive statistics with marketplace data. Args: db: Database session Returns: Dictionary containing all statistics data """ # Use more efficient queries with proper indexes total_products = db.query(Product).count() unique_brands = ( db.query(Product.brand) .filter(Product.brand.isnot(None), Product.brand != "") .distinct() .count() ) unique_categories = ( db.query(Product.google_product_category) .filter( Product.google_product_category.isnot(None), Product.google_product_category != "", ) .distinct() .count() ) # New marketplace statistics unique_marketplaces = ( db.query(Product.marketplace) .filter(Product.marketplace.isnot(None), Product.marketplace != "") .distinct() .count() ) unique_shops = ( db.query(Product.shop_name) .filter(Product.shop_name.isnot(None), Product.shop_name != "") .distinct() .count() ) # Stock statistics total_stock_entries = db.query(Stock).count() total_inventory = db.query(func.sum(Stock.quantity)).scalar() or 0 stats_data = { "total_products": total_products, "unique_brands": unique_brands, "unique_categories": unique_categories, "unique_marketplaces": unique_marketplaces, "unique_shops": unique_shops, "total_stock_entries": total_stock_entries, "total_inventory_quantity": total_inventory, } logger.info( f"Generated comprehensive stats: {total_products} products, {unique_marketplaces} marketplaces" ) return stats_data def get_marketplace_breakdown_stats(self, db: Session) -> List[Dict[str, Any]]: """ Get statistics broken down by marketplace. Args: db: Database session Returns: List of dictionaries containing marketplace statistics """ # Query to get stats per marketplace marketplace_stats = ( db.query( Product.marketplace, func.count(Product.id).label("total_products"), func.count(func.distinct(Product.shop_name)).label("unique_shops"), func.count(func.distinct(Product.brand)).label("unique_brands"), ) .filter(Product.marketplace.isnot(None)) .group_by(Product.marketplace) .all() ) stats_list = [ { "marketplace": stat.marketplace, "total_products": stat.total_products, "unique_shops": stat.unique_shops, "unique_brands": stat.unique_brands, } for stat in marketplace_stats ] logger.info( f"Generated marketplace breakdown stats for {len(stats_list)} marketplaces" ) return stats_list def get_product_count(self, db: Session) -> int: """Get total product count.""" return db.query(Product).count() def get_unique_brands_count(self, db: Session) -> int: """Get count of unique brands.""" return ( db.query(Product.brand) .filter(Product.brand.isnot(None), Product.brand != "") .distinct() .count() ) def get_unique_categories_count(self, db: Session) -> int: """Get count of unique categories.""" return ( db.query(Product.google_product_category) .filter( Product.google_product_category.isnot(None), Product.google_product_category != "", ) .distinct() .count() ) def get_unique_marketplaces_count(self, db: Session) -> int: """Get count of unique marketplaces.""" return ( db.query(Product.marketplace) .filter(Product.marketplace.isnot(None), Product.marketplace != "") .distinct() .count() ) def get_unique_shops_count(self, db: Session) -> int: """Get count of unique shops.""" return ( db.query(Product.shop_name) .filter(Product.shop_name.isnot(None), Product.shop_name != "") .distinct() .count() ) def get_stock_statistics(self, db: Session) -> Dict[str, int]: """ Get stock-related statistics. Args: db: Database session Returns: Dictionary containing stock statistics """ total_stock_entries = db.query(Stock).count() total_inventory = db.query(func.sum(Stock.quantity)).scalar() or 0 return { "total_stock_entries": total_stock_entries, "total_inventory_quantity": total_inventory, } def get_brands_by_marketplace(self, db: Session, marketplace: str) -> List[str]: """Get unique brands for a specific marketplace.""" brands = ( db.query(Product.brand) .filter( Product.marketplace == marketplace, Product.brand.isnot(None), Product.brand != "", ) .distinct() .all() ) return [brand[0] for brand in brands] def get_shops_by_marketplace(self, db: Session, marketplace: str) -> List[str]: """Get unique shops for a specific marketplace.""" shops = ( db.query(Product.shop_name) .filter( Product.marketplace == marketplace, Product.shop_name.isnot(None), Product.shop_name != "", ) .distinct() .all() ) return [shop[0] for shop in shops] def get_products_by_marketplace(self, db: Session, marketplace: str) -> int: """Get product count for a specific marketplace.""" return db.query(Product).filter(Product.marketplace == marketplace).count() # Create service instance following the same pattern as other services stats_service = StatsService()