260 lines
10 KiB
Python
260 lines
10 KiB
Python
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']:
|
|
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()
|
|
|
|
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()
|
|
|
|
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)
|