271 lines
10 KiB
Python
271 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:
|
|
# Рекурсивный 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)
|