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: # Рекурсивный CTE для получения всех подкатегорий 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) # Обработка фильтров if filters: column_map = { 'diameter': 'p.diameter', 'length': 'p.length', 'wall_thickness': 'p.wall_thickness', 'color': 'p.color', 'application_type': 'p.application_type', 'stiffness_class': 'p.stiffness_class' } def cast_value(col_name, val): if col_name in ('diameter', 'length'): try: return int(val) except: return val if col_name == 'wall_thickness': try: return float(val) except: return val return val for key, value in filters.items(): if not value: continue if key in column_map: col = column_map[key] if isinstance(value, list): casted = [cast_value(key, v) for v in value] v_ph = ','.join(['?' for _ in casted]) base_where += f" AND {col} IN ({v_ph})" params.extend(casted) else: casted = cast_value(key, value) base_where += f" AND {col} = ?" params.append(casted) else: # Фильтры из product_filters if isinstance(value, list): v_ph = ','.join(['?' for _ in value]) base_where += f""" AND p.id IN ( SELECT product_id FROM product_filters WHERE filter_key = ? AND value IN ({v_ph}) )""" 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)) # Подсчёт общего количества count_query = f"SELECT COUNT(*) as total FROM products p {base_where}" cursor.execute(count_query, params) total = cursor.fetchone()['total'] # Пагинация 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() items = [] for row in rows: items.append({ "id": row['id'], "name": row['name'], "full_name": row['full_name'], "description": row['description'], "categoryId": row['category_id'], "images": [row['image_path']] if row['image_path'] else ["/images/products/default.jpg"], "attributes": { "diameter": row['diameter'], "length": row['length'], "wall_thickness": row['wall_thickness'], "color": row['color'], "application_type": row['application_type'] } }) 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.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)