import sqlite3 from typing import List, Dict, Any, Optional, Tuple from loguru import logger as log 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]: conn = sqlite3.connect(db_name) conn.row_factory = sqlite3.Row cursor = conn.cursor() try: # Базовый запрос для получения товаров base_query = """ SELECT DISTINCT p.id, p.name, p.full_name, p.description, p.diameter, p.wall_thickness, p.length, p.color, p.application_type, p.stiffness_class, p.image_path, p.created_at FROM products p WHERE p.category_id = ? """ params = [category_id] # Применяем фильтры if filters: for filter_key, filter_value in filters.items(): if filter_value: if isinstance(filter_value, list): # Для множественных значений placeholders = ','.join(['?' for _ in filter_value]) base_query += f""" AND p.id IN ( SELECT DISTINCT pf.product_id FROM product_filters pf WHERE pf.filter_key = ? AND pf.value IN ({placeholders}) ) """ params.append(filter_key) params.extend(filter_value) else: # Для одиночных значений base_query += """ AND p.id IN ( SELECT DISTINCT pf.product_id FROM product_filters pf WHERE pf.filter_key = ? AND pf.value = ? ) """ params.append(filter_key) params.append(filter_value) # Получаем общее количество товаров count_query = f"SELECT COUNT(*) as total FROM ({base_query})" cursor.execute(count_query, params) total = cursor.fetchone()['total'] # Добавляем пагинацию offset = (page - 1) * limit paginated_query = f""" {base_query} ORDER BY p.id LIMIT ? OFFSET ? """ params.extend([limit, offset]) cursor.execute(paginated_query, params) # Формируем результат items = [] for row in cursor.fetchall(): # Формируем атрибуты товара attributes = {} if row['color']: attributes['color'] = row['color'] if row['length'] and row['length'] > 0: if row['diameter'] and row['wall_thickness']: attributes['size'] = f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" else: attributes['size'] = str(row['length']) if row['diameter']: attributes['diameter'] = str(row['diameter']) if row['wall_thickness']: attributes['wall_thickness'] = str(row['wall_thickness']) if row['stiffness_class']: attributes['stiffness_class'] = row['stiffness_class'] if row['application_type']: attributes['application_type'] = row['application_type'] # Получаем дополнительные атрибуты из product_filters cursor.execute(""" SELECT filter_key, value FROM product_filters WHERE product_id = ? """, (row['id'],)) for filter_row in cursor.fetchall(): if filter_row['filter_key'] not in attributes: attributes[filter_row['filter_key']] = filter_row['value'] # Определяем количество в упаковке (можно вынести в отдельную таблицу) attributes['amount'] = get_pack_amount(row['id'], row['diameter'], conn) items.append({ "id": row['id'], "name": row['name'], "description": row['description'] or f"Труба ПВХ {row['diameter']}мм" if row['diameter'] else "Товар", # Цена не хранится в БД - возвращаем 0 или None "price": None, "categoryId": category_id, "images": get_product_images(row['id'], row['image_path'], conn), "attributes": attributes, "inStock": check_stock(row['id'], conn) }) return { "items": items, "total": total, "page": page, "limit": limit, "total_pages": (total + limit - 1) // limit } except sqlite3.Error as e: log.error(f"Ошибка при получении товаров: {e}") return {"items": [], "total": 0, "page": page, "limit": limit} 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.full_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) # Формируем атрибуты attributes = { "color": row['color'], "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'], } if all([row['diameter'], row['wall_thickness'], row['length'] and row['length'] > 0]): attributes["size"] = f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" # Добавляем дополнительные атрибуты attributes.update(extra_attributes) return { "id": row['id'], "name": row['name'], "full_name": row['full_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: """Проверить наличие товара на складе""" # Временная логика - все товары в наличии # В будущем можно добавить таблицу stock 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['full_name']: parts.append(row['full_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)