lab_03/modules/aggrigation.py - master - nice_jam/SBT-database-cource

lab_03/modules/aggrigation.py
67 строк2 KB
Filippenko Pavel
change the structure of lab03, add replication set for mangodb service
26 мар 2025, 21:07
26 мар 2025, 21:078a42026
from bson.son import SON

# 1. Анализ продаж по категориям
def analyze_sales_by_category(db, ):
    orders_collection = db['orders']
    # products_collection = db['products']

    # Объединяем заказы с продуктами и подсчитываем общую сумму продаж для каждой категории
    pipeline = [
        {
            '$lookup': {
                'from': 'products',
                'localField': 'product_id',  
                'foreignField': '_id',
                'as': 'product_info'
            }
        },
        {
            '$unwind': '$product_info'
        },
        {
            '$group': {
                '_id': '$product_info.category',  # Группируем по категории
                'total_sales': {
                    '$sum': {'$multiply': ['$quantity', '$product_info.price']}  # Подсчитываем общую сумму продаж
                }
            }
        },
        {
            '$sort': SON([('total_sales', -1)])  # Сортируем по убыванию суммы продаж
        }
    ]

    results = orders_collection.aggregate(pipeline)
    return results

# 2. Выявление популярного товара
def find_most_popular_product(db):
    orders_collection = db['orders']

    # Подсчитываем количество заказов для каждого товара
    pipeline = [
        {
            '$group': {
                '_id': '$product_id', 
                'total_orders': {'$sum': '$quantity'}
            }
        },
        {
            '$sort': SON([('total_orders', -1)])  # Сортируем по убыванию количества заказов
        },
        {
            '$limit': 1  # Берем только самый популярный товар
        }
    ]

    product_rank = list(orders_collection.aggregate(pipeline))
    product_id = product_rank[0]['_id']
    total_orders = product_rank[0]['total_orders']

    # Получаем название продукта из коллекции products
    product_info = db['products'].find_one({'_id': product_id})

    if product_rank:
        return {'product_name': product_info['category'], 'total_orders': total_orders}
    else:
        return None