114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
import sqlite3
|
|
from typing import List, Dict, Any, Optional
|
|
from loguru import logger as log
|
|
|
|
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()
|
|
|
|
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()
|