# src/cache/download_cache.py import locale import sys from sqlalchemy import select, func from sqlalchemy.orm import selectinload from tqdm import tqdm from src.cache.cache import PRODUCTS_CACHE, CATEGORIES_CACHE, PRODUCTS_SEO_CACHE, CATEGORIES_SEO_CACHE from src.category.models import Category from src.database import async_session from src.product.models import Product, ProductCategory, ProductStatus from src.store.models import ProductStore, Store from src.review.models import ProductReview, StoreReview, ReviewStatusEnum locale.setlocale(locale.LC_ALL, "uk_UA.UTF-8") async def load_categories_cache(db): CATEGORIES_CACHE.clear() CATEGORIES_SEO_CACHE.clear() # Очищуємо SEO-кеш # Завантажуємо всі категорії (включно з неактивними для перевірки ієрархії) result = await db.execute(select(Category)) all_categories = result.scalars().all() # Створюємо словник для швидкого пошуку батьківських категорій categories_dict = {c.category_id: c for c in all_categories} def has_enabled_parent_chain(category): """Перевіряє чи всі батьківські категорії активні""" current = category while current.parent_category_id: parent = categories_dict.get(current.parent_category_id) if not parent or parent.status != 1: return False current = parent return True # Фільтруємо тільки активні категорії з активними батьками categories = [c for c in all_categories if c.status == 1 and has_enabled_parent_chain(c)] # Рахуємо товари по категоріях з PRODUCTS_CACHE (тільки товари з валідними цінами) product_counts = {} for product_data in PRODUCTS_CACHE.values(): for cat in product_data.get("categories", []): cat_id = cat["id"] product_counts[cat_id] = product_counts.get(cat_id, 0) + 1 # Сортуємо категорії за назвою з урахуванням локалі categories.sort(key=lambda c: locale.strxfrm(c.name or "")) # Формуємо словник категорій for c in categories: product_count = product_counts.get(c.category_id) category_data = { "id_category": c.category_id, "name": c.name, "description": c.description, "seo_keyword": c.seo_keyword, "meta_title": c.meta_title, "meta_description": c.meta_description, "image": c.image, "status": c.status, "sort_order": c.sort_order, "icon": c.icon, "viewed": c.viewed, "date_added": c.date_added.isoformat() if c.date_added else None, "date_modify": c.date_modify.isoformat() if c.date_modify else None, "product_count": product_count, "total_product_count": product_count if product_count else 0, # буде обчислено пізніше "parent_id": c.parent_category_id if c.parent_category_id not in (0,) else None, "children": [], } # Зберігаємо в основний кеш за ID CATEGORIES_CACHE[c.category_id] = category_data # Зберігаємо в SEO-кеш за seo_keyword (якщо він є) if c.seo_keyword: CATEGORIES_SEO_CACHE[c.seo_keyword] = category_data # Будуємо дерево for c in list(CATEGORIES_CACHE.values()): parent_id = c["parent_id"] if parent_id is not None and parent_id in CATEGORIES_CACHE: CATEGORIES_CACHE[parent_id]["children"].append(c) # Сортуємо дітей у кожного батька for c in CATEGORIES_CACHE.values(): c["children"].sort(key=lambda x: locale.strxfrm(x["name"] or "")) # # Рекурсивно рахуємо total_product_count # def calculate_total_product_count(category): # total = category["product_count"] # for child in category["children"]: # total += calculate_total_product_count(child) # category["total_product_count"] = total # return total # # for c in CATEGORIES_CACHE.values(): # if c["parent_id"] is None: # тільки для кореневих # calculate_total_product_count(c) print(f"✅ Завантажено {len(CATEGORIES_CACHE)} категорій з метриками товарів в оперативку") print(f"✅ Створено SEO-кеш для {len(CATEGORIES_SEO_CACHE)} категорій") async def load_products_cache(batch_size=50_000): PRODUCTS_CACHE.clear() PRODUCTS_SEO_CACHE.clear() # Очищуємо SEO-кеш async with async_session() as session: total_result = await session.execute( select(Product.product_id).where(Product.status == ProductStatus.ON) ) total = len(total_result.scalars().all()) # Отримуємо кількість відгуків для кожного товару product_reviews_query = await session.execute( select( ProductReview.product_id, func.count(ProductReview.product_review_id).label('reviews_count') ) .where(ProductReview.status == ReviewStatusEnum.approved) .group_by(ProductReview.product_id) ) product_reviews_counts = {row.product_id: row.reviews_count for row in product_reviews_query} # Отримуємо кількість відгуків для кожного магазину store_reviews_query = await session.execute( select( StoreReview.store_id, func.count(StoreReview.store_review_id).label('reviews_count') ) .where(StoreReview.status == ReviewStatusEnum.approved) .group_by(StoreReview.store_id) ) store_reviews_counts = {row.store_id: row.reviews_count for row in store_reviews_query} # Отримуємо всі фільтри для всіх товарів одним запросом from src.product.models import ProductFilter, Filter, FilterGroup all_filters_query = await session.execute( select( ProductFilter.product_id, ProductFilter.product_filter_id, ProductFilter.filter_id, Filter.name.label('filter_name'), Filter.sort_order.label('filter_sort_order'), FilterGroup.filter_group_id, FilterGroup.name.label('group_name'), FilterGroup.sort_order.label('group_sort_order') ) .join(Filter, ProductFilter.filter_id == Filter.filter_id) .join(FilterGroup, Filter.filter_group_id == FilterGroup.filter_group_id) .order_by(ProductFilter.product_id, FilterGroup.sort_order, Filter.sort_order) ) # Групуємо фільтри по product_id products_filters = {} for row in all_filters_query: if row.product_id not in products_filters: products_filters[row.product_id] = [] filter_item = { "product_filter_id": row.product_filter_id, "filter_id": row.filter_id, "filter_data": { "filter_id": row.filter_id, "filter_group_id": row.filter_group_id, "name": row.filter_name, "sort_order": row.filter_sort_order, "filter_group_data": { "filter_group_id": row.filter_group_id, "name": row.group_name, "sort_order": row.group_sort_order } } } products_filters[row.product_id].append(filter_item) offset = 0 pbar = tqdm(total=total, desc="Кешуємо продукти") while offset < total: result = await session.execute( select(Product) .where(Product.status == ProductStatus.ON) .options( selectinload(Product.categories).selectinload(ProductCategory.category), selectinload(Product.images), selectinload(Product.attributes), selectinload(Product.manufacturer), selectinload(Product.product_links).selectinload(ProductStore.store).selectinload(Store.photos), ) .offset(offset) .limit(batch_size) ) products = result.scalars().unique().all() skipped_count = 0 for p in products: reviews_count = product_reviews_counts.get(p.product_id, 0) # Створюємо список магазинів stores_list = [ { "id": link.store.store_id, "name": link.store.name, "description": link.store.description, "phone": link.store.phone, "email": link.store.email, "status": link.store.status, "date_added": link.store.date_added.isoformat() if link.store.date_added else None, "date_modify": link.store.date_modify.isoformat() if link.store.date_modify else None, "rating": link.store.rating, "seo_keyword": link.store.seo_keyword, "meta_title": link.store.meta_title, "meta_description": link.store.meta_description, "meta_keyword": link.store.meta_keyword, "logo": link.store.logo, "viewed": link.store.viewed, "reviews_count": store_reviews_counts.get(link.store.store_id, 0), # Додано: кількість відгуків магазину "photos": [photo.image for photo in link.store.photos], "price": link.price, "price_old": link.price_old, } for link in p.product_links if link.store is not None ] # Сортуємо магазини від меншої ціни до більшої stores_list.sort(key=lambda store: store["price"] if store["price"] is not None else float('inf')) # Фільтруємо магазини: залишаємо тільки з ціною > 0 stores_list = [store for store in stores_list if store["price"] is not None and store["price"] > 0] # Якщо немає жодного магазину з валідною ціною — пропускаємо товар if not stores_list: skipped_count += 1 continue # Знаходимо мінімальну ціну серед магазинів min_price = min(store["price"] for store in stores_list) product_data = { "product_id": p.product_id, "name": p.name, "description": p.description, "seo_keyword": p.seo_keyword, "meta_title": p.meta_title, "meta_description": p.meta_description, "meta_keyword": p.meta_keyword, "image": p.image, "status": p.status.value if p.status else None, "date_added": p.date_added.isoformat() if p.date_added else None, "date_modify": p.date_modify.isoformat() if p.date_modify else None, "price": min_price, # Встановлюємо мінімальну ціну з магазинів "price_old": p.price_old, "special": p.price_old is not None, "model": p.model, "rating": p.rating, "manufacturer": { "id": p.manufacturer.manufacturer_id if p.manufacturer else None, "name": p.manufacturer.name if p.manufacturer else None, }, "viewed": p.viewed, "reviews_count": reviews_count, # Додано: кількість відгуків про товар "categories": [ {"id": pc.category.category_id, "name": pc.category.name} for pc in p.categories ], "images": [img.image for img in p.images], "attributes": [ { "group": attr.group_name, "name": attr.name, "value": attr.value, "sort_order": attr.sort_order, } for attr in p.attributes ], "stores": stores_list, # Використовуємо відсортований список "filters": products_filters.get(p.product_id, []) # Додаємо фільтри товару } # Зберігаємо в основний кеш за ID PRODUCTS_CACHE[p.product_id] = product_data # Зберігаємо в SEO-кеш за seo_keyword (якщо він є) if p.seo_keyword: PRODUCTS_SEO_CACHE[p.seo_keyword] = product_data offset += batch_size pbar.update(len(products)) pbar.close() print(f"✅ Завантажено {len(PRODUCTS_CACHE)} товарів з метриками відгуків та фільтрами в оперативку (пропущено {total - len(PRODUCTS_CACHE)} без ціни)") print(f"✅ Створено SEO-кеш для {len(PRODUCTS_SEO_CACHE)} товарів") # print(f"✅ Загружено фільтри для {len(products_filters)} товарів") async def update_product_cache(product_id): """Оновлює один конкретний товар в кеші з БД""" async with async_session() as session: # Спочатку перевіряємо, чи існує товар product_exists = await session.execute( select(Product.product_id).where(Product.product_id == product_id) ) if not product_exists.scalar_one_or_none(): # Товар не існує - видаляємо з кешу, якщо він там є if product_id in PRODUCTS_CACHE: old_seo = PRODUCTS_CACHE[product_id].get("seo_keyword") del PRODUCTS_CACHE[product_id] if old_seo and old_seo in PRODUCTS_SEO_CACHE: del PRODUCTS_SEO_CACHE[old_seo] print(f"🗑️ Товар {product_id} видалено з кешу") return # Отримуємо кількість відгуків для цього товару reviews_count_result = await session.execute( select(func.count(ProductReview.product_review_id)) .where( ProductReview.product_id == product_id, ProductReview.status == ReviewStatusEnum.approved ) ) reviews_count = reviews_count_result.scalar() or 0 # Отримуємо кількість відгуків для магазинів цього товару store_reviews_query = await session.execute( select( StoreReview.store_id, func.count(StoreReview.store_review_id).label('reviews_count') ) .join(ProductStore, StoreReview.store_id == ProductStore.store_id) .where( ProductStore.product_id == product_id, StoreReview.status == ReviewStatusEnum.approved ) .group_by(StoreReview.store_id) ) store_reviews_counts = {row.store_id: row.reviews_count for row in store_reviews_query} # Завантажуємо товар з усіма зв'язками result = await session.execute( select(Product) .options( selectinload(Product.categories).selectinload(ProductCategory.category), selectinload(Product.images), selectinload(Product.attributes), selectinload(Product.manufacturer), selectinload(Product.product_links).selectinload(ProductStore.store).selectinload(Store.photos), ) .where(Product.product_id == product_id) ) product = result.scalar_one_or_none() if not product: return # Якщо товар не має статус ON, видаляємо його з кешу if product.status != ProductStatus.ON: if product_id in PRODUCTS_CACHE: old_seo = PRODUCTS_CACHE[product_id].get("seo_keyword") del PRODUCTS_CACHE[product_id] if old_seo and old_seo in PRODUCTS_SEO_CACHE: del PRODUCTS_SEO_CACHE[old_seo] print(f"🗑️ Товар {product_id} ({product.name}) видалено з кешу (статус не ON)") return # Створюємо список магазинів stores_list = [ { "id": link.store.store_id, "name": link.store.name, "description": link.store.description, "phone": link.store.phone, "email": link.store.email, "status": link.store.status, "date_added": link.store.date_added.isoformat() if link.store.date_added else None, "date_modify": link.store.date_modify.isoformat() if link.store.date_modify else None, "rating": link.store.rating, "seo_keyword": link.store.seo_keyword, "meta_title": link.store.meta_title, "meta_description": link.store.meta_description, "meta_keyword": link.store.meta_keyword, "logo": link.store.logo, "viewed": link.store.viewed, "reviews_count": store_reviews_counts.get(link.store.store_id, 0), "photos": [photo.image for photo in link.store.photos], "price": link.price, "price_old": link.price_old, } for link in product.product_links if link.store is not None ] # Сортуємо магазини від меншої ціни до більшої stores_list.sort(key=lambda store: store["price"] if store["price"] is not None else float('inf')) # Фільтруємо магазини: залишаємо тільки з ціною > 0 stores_list = [store for store in stores_list if store["price"] is not None and store["price"] > 0] # Якщо немає жодного магазину з валідною ціною — видаляємо з кешу if not stores_list: if product_id in PRODUCTS_CACHE: old_seo = PRODUCTS_CACHE[product_id].get("seo_keyword") del PRODUCTS_CACHE[product_id] if old_seo and old_seo in PRODUCTS_SEO_CACHE: del PRODUCTS_SEO_CACHE[old_seo] print(f"🗑️ Товар {product_id} ({product.name}) видалено з кешу (немає магазинів з ціною)") return # Знаходимо мінімальну ціну серед магазинів min_price = min(store["price"] for store in stores_list) # Формуємо дані товару product_data = { "product_id": product.product_id, "name": product.name, "description": product.description, "seo_keyword": product.seo_keyword, "meta_title": product.meta_title, "meta_description": product.meta_description, "meta_keyword": product.meta_keyword, "image": product.image, "status": product.status.value if product.status else None, "date_added": product.date_added.isoformat() if product.date_added else None, "date_modify": product.date_modify.isoformat() if product.date_modify else None, "price": min_price, # Встановлюємо мінімальну ціну з магазинів "price_old": product.price_old, "special": product.price_old is not None, "model": product.model, "rating": product.rating, "manufacturer": { "id": product.manufacturer.manufacturer_id if product.manufacturer else None, "name": product.manufacturer.name if product.manufacturer else None, }, "viewed": product.viewed, "reviews_count": reviews_count, "categories": [ {"id": pc.category.category_id, "name": pc.category.name} for pc in product.categories ], "images": [img.image for img in product.images], "attributes": [ { "group": attr.group_name, "name": attr.name, "value": attr.value, "sort_order": attr.sort_order, } for attr in product.attributes ], "stores": stores_list, # Використовуємо відсортований список } # Додаємо дані про фільтри товару from src.product.models import ProductFilter, Filter, FilterGroup filters_result = await session.execute( select( ProductFilter.product_filter_id, ProductFilter.filter_id, Filter.name.label('filter_name'), Filter.sort_order.label('filter_sort_order'), FilterGroup.filter_group_id, FilterGroup.name.label('group_name'), FilterGroup.sort_order.label('group_sort_order') ) .join(Filter, ProductFilter.filter_id == Filter.filter_id) .join(FilterGroup, Filter.filter_group_id == FilterGroup.filter_group_id) .where(ProductFilter.product_id == product_id) .order_by(FilterGroup.sort_order, Filter.sort_order) ) filters_data = [] for row in filters_result: filter_item = { "product_filter_id": row.product_filter_id, "filter_id": row.filter_id, "filter_data": { "filter_id": row.filter_id, "filter_group_id": row.filter_group_id, "name": row.filter_name, "sort_order": row.filter_sort_order, "filter_group_data": { "filter_group_id": row.filter_group_id, "name": row.group_name, "sort_order": row.group_sort_order } } } filters_data.append(filter_item) # Додаємо фільтри до product_data product_data["filters"] = filters_data # Видаляємо старий SEO-запис, якщо він змінився old_product_data = PRODUCTS_CACHE.get(product_id) if old_product_data: old_seo = old_product_data.get("seo_keyword") if old_seo and old_seo != product.seo_keyword and old_seo in PRODUCTS_SEO_CACHE: del PRODUCTS_SEO_CACHE[old_seo] # Зберігаємо в основний кеш за ID PRODUCTS_CACHE[product_id] = product_data # Зберігаємо в SEO-кеш за seo_keyword (якщо він є) if product.seo_keyword: PRODUCTS_SEO_CACHE[product.seo_keyword] = product_data print(f"✅ Оновлено товар {product_id} ({product.name}) в кеші") async def update_category_cache(category_id): """Оновлює одну конкретну категорію в кеші з БД""" async with async_session() as session: # Спочатку перевіряємо, чи існує категорія category_exists = await session.execute( select(Category.category_id).where(Category.category_id == category_id) ) if not category_exists.scalar_one_or_none(): # Категорія не існує - видаляємо з кешу, якщо вона там є if category_id in CATEGORIES_CACHE: old_seo = CATEGORIES_CACHE[category_id].get("seo_keyword") # Видаляємо з батьківського children, якщо є parent_id = CATEGORIES_CACHE[category_id].get("parent_id") if parent_id and parent_id in CATEGORIES_CACHE: CATEGORIES_CACHE[parent_id]["children"] = [ child for child in CATEGORIES_CACHE[parent_id]["children"] if child["id_category"] != category_id ] del CATEGORIES_CACHE[category_id] if old_seo and old_seo in CATEGORIES_SEO_CACHE: del CATEGORIES_SEO_CACHE[old_seo] print(f"🗑️ Категорію {category_id} видалено з кешу") return # Завантажуємо всі категорії для перевірки ланцюжка батьків all_categories_result = await session.execute(select(Category)) all_categories = all_categories_result.scalars().all() categories_dict = {c.category_id: c for c in all_categories} def has_enabled_parent_chain(category): """Перевіряє чи всі батьківські категорії активні""" current = category while current.parent_category_id: parent = categories_dict.get(current.parent_category_id) if not parent or parent.status != 1: return False current = parent return True # Завантажуємо категорію result = await session.execute( select(Category).where(Category.category_id == category_id) ) category = result.scalar_one_or_none() if not category: return # Якщо категорія неактивна або її батьківська категорія неактивна, видаляємо з кешу if category.status != 1 or not has_enabled_parent_chain(category): if category_id in CATEGORIES_CACHE: old_seo = CATEGORIES_CACHE[category_id].get("seo_keyword") # Видаляємо з батьківського children, якщо є parent_id = CATEGORIES_CACHE[category_id].get("parent_id") if parent_id and parent_id in CATEGORIES_CACHE: CATEGORIES_CACHE[parent_id]["children"] = [ child for child in CATEGORIES_CACHE[parent_id]["children"] if child["id_category"] != category_id ] del CATEGORIES_CACHE[category_id] if old_seo and old_seo in CATEGORIES_SEO_CACHE: del CATEGORIES_SEO_CACHE[old_seo] print(f"🗑️ Категорію {category_id} видалено з кешу (неактивна або неактивний батько)") return # Рахуємо товари в категорії з PRODUCTS_CACHE (тільки товари з валідними цінами) product_count = sum( 1 for product_data in PRODUCTS_CACHE.values() if any(cat["id"] == category_id for cat in product_data.get("categories", [])) ) # Формуємо дані категорії category_data = {"id_category": category.category_id, "name": category.name, "description": category.description, "seo_keyword": category.seo_keyword, "meta_title": category.meta_title, "meta_description": category.meta_description, "image": category.image, "status": category.status, "sort_order": category.sort_order, "icon": category.icon, "date_added": category.date_added.isoformat() if category.date_added else None, "date_modify": category.date_modify.isoformat() if category.date_modify else None, "product_count": product_count, "total_product_count": 0, "parent_id": category.parent_category_id if category.parent_category_id not in (0,) else None, "children": [ child_data for child_data in CATEGORIES_CACHE.values() if child_data.get("parent_id") == category_id ]} # Отримуємо дітей цієї категорії з кешу # Сортуємо дітей category_data["children"].sort(key=lambda x: locale.strxfrm(x["name"] or "")) # Видаляємо старий SEO-запис, якщо він змінився old_category_data = CATEGORIES_CACHE.get(category_id) if old_category_data: old_seo = old_category_data.get("seo_keyword") if old_seo and old_seo != category.seo_keyword and old_seo in CATEGORIES_SEO_CACHE: del CATEGORIES_SEO_CACHE[old_seo] # Видаляємо зі старого батьківського children old_parent_id = old_category_data.get("parent_id") if old_parent_id and old_parent_id in CATEGORIES_CACHE: CATEGORIES_CACHE[old_parent_id]["children"] = [ child for child in CATEGORIES_CACHE[old_parent_id]["children"] if child["id_category"] != category_id ] # Зберігаємо в основний кеш за ID CATEGORIES_CACHE[category_id] = category_data # Зберігаємо в SEO-кеш за seo_keyword (якщо він є) if category.seo_keyword: CATEGORIES_SEO_CACHE[category.seo_keyword] = category_data # Додаємо до нового батьківського children parent_id = category_data.get("parent_id") if parent_id and parent_id in CATEGORIES_CACHE: # Перевіряємо, чи не додана вже ця категорія до батьківських дітей if not any(child["id_category"] == category_id for child in CATEGORIES_CACHE[parent_id]["children"]): CATEGORIES_CACHE[parent_id]["children"].append(category_data) # Пересортуємо дітей у батька CATEGORIES_CACHE[parent_id]["children"].sort(key=lambda x: locale.strxfrm(x["name"] or "")) # Рекурсивно рахуємо total_product_count для цієї категорії та її батьків def calculate_total_product_count(category_data): total = category_data["product_count"] for child in category_data["children"]: total += calculate_total_product_count(child) category_data["total_product_count"] = total return total calculate_total_product_count(category_data) # Оновлюємо total_product_count у батьківських категоріях current_parent_id = parent_id while current_parent_id and current_parent_id in CATEGORIES_CACHE: calculate_total_product_count(CATEGORIES_CACHE[current_parent_id]) current_parent_id = CATEGORIES_CACHE[current_parent_id].get("parent_id") print(f"✅ Оновлено категорію {category_id} ({category.name}) в кеші")