263 lines
10 KiB
Python
263 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'] 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)
|