From b9c1a1c01859c1356c7caeaaed3842ed171b5930 Mon Sep 17 00:00:00 2001 From: Kirill Date: Sun, 22 Mar 2026 16:53:24 +0300 Subject: [PATCH] feat(handlers): Get data from db Methods Added all methods from fronted documentation whitch needs --- db/handlers/catigory_handler.py | 124 ++++++++++++--- db/handlers/product_handler.py | 258 +++++++++++++++++++++++++++++++- 2 files changed, 360 insertions(+), 22 deletions(-) diff --git a/db/handlers/catigory_handler.py b/db/handlers/catigory_handler.py index 1adae80..28c81f5 100644 --- a/db/handlers/catigory_handler.py +++ b/db/handlers/catigory_handler.py @@ -1,23 +1,113 @@ import sqlite3 -from module.category import Category +from typing import List, Dict, Any, Optional +from loguru import logger as log -def GetAllCategories(db_name: str = 'pipe_store.db', parent_id: int = 1) -> list[Category]: +def get_all_categories_with_attributes(db_name: str = 'pipe_store.db') -> List[Dict[str, Any]]: conn = sqlite3.connect(db_name) conn.row_factory = sqlite3.Row cursor = conn.cursor() + + try: + # Получаем все категории (только основные, без parent_id) + cursor.execute(""" + SELECT id, name, image_path + FROM categories + WHERE parent_id IS NULL OR parent_id = 1 + ORDER BY id + """) + + categories = [] + + for cat_row in cursor.fetchall(): + category_id = cat_row['id'] + + # Получаем все фильтры для товаров в этой категории + cursor.execute(""" + SELECT DISTINCT f.id, f.name, f.filter_key + FROM filters f + WHERE f.filter_key IN ( + SELECT DISTINCT pf.filter_key + FROM product_filters pf + JOIN products p ON pf.product_id = p.id + WHERE p.category_id = ? + ) + ORDER BY f.id + """, (category_id,)) + + attributes = [] + + for filter_row in cursor.fetchall(): + filter_id = filter_row['id'] + filter_key = filter_row['filter_key'] + filter_name = filter_row['name'] + + # Получаем уникальные значения для этого фильтра + cursor.execute(""" + SELECT DISTINCT pf.value, COUNT(*) as count + FROM product_filters pf + JOIN products p ON pf.product_id = p.id + WHERE p.category_id = ? AND pf.filter_key = ? + GROUP BY pf.value + ORDER BY pf.value + """, (category_id, filter_key)) + + vals = [] + for val_row in cursor.fetchall(): + # Генерируем уникальный ID для значения + val_id = hash(f"{category_id}_{filter_key}_{val_row['value']}") % 1000000 + vals.append({ + "id": val_id, + "val": val_row['value'] + }) + + if vals: # Добавляем только если есть значения + attributes.append({ + "id": filter_id, + "name": filter_name, + "vals": vals + }) + + categories.append({ + "id": category_id, + "name": cat_row['name'], + "image": cat_row['image_path'] or f"/images/categories/{cat_row['name'].lower().replace(' ', '_')}.jpg", + "atributes": attributes + }) + + return categories + + except sqlite3.Error as e: + log.error(f"Ошибка при получении категорий: {e}") + return [] + finally: + conn.close() - categories = [] - - cursor.execute("SELECT id, name, description, image_path FROM categories WHERE parent_id = ?", [parent_id]) - - for row in cursor.fetchall(): - category = Category( - id = row["id"], - name = row["name"], - description= row["description"], - image_path= row["image_path"], - ) - categories.append(category.model_dump()) - - conn.close() - return categories +def get_category_by_id(category_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 id, name, description, image_path, parent_id + FROM categories + WHERE id = ? + """, (category_id,)) + + row = cursor.fetchone() + if row: + return { + "id": row['id'], + "name": row['name'], + "description": row['description'], + "image": row['image_path'], + "parent_id": row['parent_id'] + } + return None + + except sqlite3.Error as e: + log.error(f"Ошибка при получении категории {category_id}: {e}") + return None + finally: + conn.close() diff --git a/db/handlers/product_handler.py b/db/handlers/product_handler.py index 5209127..8c24bfe 100644 --- a/db/handlers/product_handler.py +++ b/db/handlers/product_handler.py @@ -1,11 +1,259 @@ import sqlite3 -from module.product import Product +from typing import List, Dict, Any, Optional, Tuple +from loguru import logger as log -def get_products(db_name: str = 'pipe_store.db', parent_id: int = 1) -> List[Product]: +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 - transaction = conn.cursor() + 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']: + attributes['size'] = f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" if row['diameter'] else 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'] + + # Определяем количество в упаковке (можно вынести в отдельную таблицу) + attributes['amount'] = get_pack_amount(row['id'], conn) + + items.append({ + "id": row['id'], + "name": row['name'], + "description": row['description'] or f"Труба ПВХ {row['diameter']}мм", + "price": calculate_price(row), # Функция для расчета цены + "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() - products = [] +def get_product_by_id(product_id: int, db_name: str = 'pipe_store.db') -> Optional[Dict[str, Any]]: + """ + Получить товар по ID с полной информацией + + Args: + product_id: ID товара + db_name: Имя базы данных + + Returns: + Dict с полной информацией о товаре или None + """ + 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) + + # Формируем полное описание + full_description = generate_full_description(row) + + return { + "id": row['id'], + "name": row['name'], + "full_name": row['full_name'], + "description": full_description, + "price": calculate_price(row), + "categoryId": row['category_id'], + "images": images, + "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'] else None, + "stiffness_class": row['stiffness_class'], + "application_type": row['application_type'], + "size": f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" if all([row['diameter'], row['wall_thickness'], row['length']]) else None + }, + "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() - transaction.execute("SELECT ... FROM products") +def get_product_images(product_id: int, main_image: Optional[str], conn: sqlite3.Connection) -> List[str]: + """ + Получить список изображений товара + (В текущей структуре только одно изображение, но можно расширить) + """ + images = [] + if main_image: + images.append(main_image) + + # Здесь можно добавить логику для получения дополнительных изображений + # Например, из отдельной таблицы product_images + + # Если нет изображения, возвращаем заглушку + if not images: + images.append("/images/products/default.jpg") + + return images + +def calculate_price(row: sqlite3.Row) -> float: + """ + Рассчитать цену товара на основе параметров + (В реальном проекте цена должна быть в базе данных) + """ + # Временная логика расчета цены + base_price = 500 + + if row['diameter']: + base_price += (row['diameter'] - 50) * 5 + + if row['length']: + base_price += row['length'] * 0.1 + + if row['stiffness_class'] == 'SN4': + base_price *= 1.2 + + return round(base_price, 2) + +def check_stock(product_id: int, conn: sqlite3.Connection) -> bool: + """ + Проверить наличие товара на складе + (В реальном проекте должна быть таблица stock) + """ + # Временная логика - все товары в наличии + return True + +def get_pack_amount(product_id: int, conn: sqlite3.Connection) -> str: + """ + Получить количество в упаковке + (В реальном проекте должно быть в базе данных) + """ + # Временная логика + return "4" + +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']: + parts.append(f"Размеры: {row['diameter']}x{row['wall_thickness']}x{row['length']} мм") + + if row['stiffness_class']: + parts.append(f"Класс жесткости: {row['stiffness_class']}") + + if row['color']: + parts.append(f"Цвет: {row['color']}") + + return " ".join(parts)