import sqlite3 from typing import List, Dict, Any, Optional, Tuple from loguru import logger as log from db.DBDataUtils import get_products_by_page, get_total_products_pages def get_products_with_filters( category_id: int, page: int = 1, limit: int = 9, filters: Dict[str, Any] = None, db_name: str = 'pipe_store.db' ) -> Dict[str, Any]: """ Получение товаров с фильтрацией и пагинацией. Все фильтры (включая диаметр, длину, цвет и т.д.) ищутся в таблице product_filters. """ print(f"Received filters: {filters}") conn = sqlite3.connect(db_name) conn.row_factory = sqlite3.Row cursor = conn.cursor() try: # 1. Получаем все ID категорий (включая подкатегории) hierarchy_query = """ WITH RECURSIVE category_hierarchy AS ( SELECT id FROM categories WHERE id = ? UNION ALL SELECT c.id FROM categories c JOIN category_hierarchy ch ON c.parent_id = ch.id ) SELECT id FROM category_hierarchy """ cursor.execute(hierarchy_query, (category_id,)) category_ids = [row['id'] for row in cursor.fetchall()] if not category_ids: return {"items": [], "total": 0, "page": page, "limit": limit, "total_pages": 0} placeholders_cats = ','.join(['?' for _ in category_ids]) base_where = f"WHERE p.category_id IN ({placeholders_cats})" params = list(category_ids) # 2. Обработка всех фильтров через product_filters if filters: for key, value in filters.items(): if not value: continue # Подзапрос для проверки наличия фильтра у товара if isinstance(value, list): # Множественный фильтр (например, несколько диаметров) placeholders = ','.join(['?' for _ in value]) base_where += f""" AND p.id IN ( SELECT product_id FROM product_filters WHERE filter_key = ? AND value IN ({placeholders}) ) """ params.append(key) params.extend([str(v) for v in value]) else: # Одиночный фильтр base_where += f""" AND p.id IN ( SELECT product_id FROM product_filters WHERE filter_key = ? AND value = ? ) """ params.append(key) params.append(str(value)) # 3. Подсчёт общего количества count_query = f"SELECT COUNT(*) as total FROM products p {base_where}" cursor.execute(count_query, params) print("SQL QUERY:", count_query) print("PARAMS:", params) total = cursor.fetchone()['total'] # 4. Пагинация offset = (page - 1) * limit query = f""" SELECT p.* FROM products p {base_where} ORDER BY p.id ASC LIMIT ? OFFSET ? """ cursor.execute(query, params + [limit, offset]) rows = cursor.fetchall() # 5. Формируем ответ items = [] for row in rows: # Вычисляем size size = None if row['diameter'] and row['wall_thickness'] and row['length'] and row['length'] > 0: size = f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" # Количество в упаковке amount = get_pack_amount(row['id'], row['diameter']) items.append({ "id": row['id'], "name": row['name'], "description": row['description'], "categoryId": row['category_id'], "images": [row['image_path']] if row['image_path'] else ["/images/products/default.jpg"], "attributes": { "color": row['color'], "amount": amount, "size": size, "diameter": str(row['diameter']) if row['diameter'] else None, "wall_thickness": str(row['wall_thickness']) if row['wall_thickness'] else None, "length": str(row['length']) if row['length'] and row['length'] > 0 else None, "stiffness_class": row['stiffness_class'], "application_type": row['application_type'] }, "inStock": True }) return { "items": items, "total": total, "page": page, "limit": limit, "total_pages": (total + limit - 1) // limit if total > 0 else 0 } except Exception as e: log.error(f"SQL Error in get_products_with_filters: {e}") return {"items": [], "total": 0, "page": page, "limit": limit, "total_pages": 0} finally: conn.close() def get_product_by_id(product_id: int, db_name: str = 'pipe_store.db') -> Optional[Dict[str, Any]]: """ Получить товар по ID с полной информацией """ conn = sqlite3.connect(db_name) conn.row_factory = sqlite3.Row cursor = conn.cursor() try: cursor.execute(""" SELECT p.id, p.name, p.description, p.diameter, p.wall_thickness, p.length, p.color, p.application_type, p.stiffness_class, p.image_path, p.created_at, p.category_id FROM products p WHERE p.id = ? """, (product_id,)) row = cursor.fetchone() if not row: return None images = get_product_images(product_id, row['image_path'], conn) # Получаем дополнительные фильтры cursor.execute(""" SELECT filter_key, value FROM product_filters WHERE product_id = ? """, (product_id,)) extra_attributes = {} for filter_row in cursor.fetchall(): extra_attributes[filter_row['filter_key']] = filter_row['value'] full_description = generate_full_description(row) # Вычисляем size и amount size = None if row['diameter'] and row['wall_thickness'] and row['length'] and row['length'] > 0: size = f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" amount = get_pack_amount(product_id, row['diameter']) # Формируем атрибуты attributes = { "color": row['color'], "amount": amount, "size": size, "diameter": str(row['diameter']) if row['diameter'] else None, "wall_thickness": str(row['wall_thickness']) if row['wall_thickness'] else None, "length": str(row['length']) if row['length'] and row['length'] > 0 else None, "stiffness_class": row['stiffness_class'], "application_type": row['application_type'], } # Добавляем дополнительные атрибуты attributes.update(extra_attributes) return { "id": row['id'], "name": row['name'], "description": full_description, "price": None, "categoryId": row['category_id'], "images": images, "attributes": attributes, "inStock": check_stock(product_id, conn), "created_at": row['created_at'] } except sqlite3.Error as e: log.error(f"Ошибка при получении товара {product_id}: {e}") return None finally: conn.close() def get_product_images(product_id: int, main_image: Optional[str], conn: sqlite3.Connection) -> List[str]: """Получить список изображений товара""" images = [] if main_image: images.append(main_image) if not images: images.append("/images/products/default.jpg") return images def check_stock(product_id: int, conn: sqlite3.Connection) -> bool: """Проверить наличие товара на складе""" # Временная логика - все товары в наличии return True def get_pack_amount(product_id: int, diameter: int = None, conn: sqlite3.Connection = None) -> str: """ Получить количество в упаковке """ if diameter: if diameter == 110: return "6-10" elif diameter == 160: return "2" return "1" def generate_full_description(row: sqlite3.Row) -> str: """Сгенерировать полное описание товара""" parts = [] if row['name']: parts.append(row['name']) if row['description']: parts.append(row['description']) if row['application_type']: parts.append(f"Тип применения: {row['application_type']} канализация") if row['diameter'] and row['wall_thickness'] and row['length'] and row['length'] > 0: parts.append(f"Размеры: {row['diameter']}x{row['wall_thickness']}x{row['length']} мм") elif row['diameter']: parts.append(f"Диаметр: {row['diameter']} мм") if row['stiffness_class']: parts.append(f"Класс жесткости: {row['stiffness_class']}") if row['color']: parts.append(f"Цвет: {row['color']}") return " ".join(parts) def get_product_list(page=1, limit=100, category_id=None): """ Получить список товаров с пагинацией (без фильтров). Возвращает список словарей с полями id, name, description. """ conn = None try: import sqlite3 conn = sqlite3.connect('pipe_store.db') conn.row_factory = sqlite3.Row rows = get_products_by_page(conn, page, limit, category_id) products = [] for row in rows: products.append({ 'id': row['id'], 'name': row['name'], 'description': row['description'] }) return products finally: if conn: conn.close()