6271a715a1
Some checks failed
Build and Deploy / build-and-deploy (push) Has been cancelled
Split from coop monolith. Includes: - Market/browse/product blueprints - Product sync API - Suma scraping pipeline - Templates for market, browse, and product views - Dockerfile and CI workflow for independent deployment
658 lines
23 KiB
Python
658 lines
23 KiB
Python
from __future__ import annotations
|
|
from typing import Dict, List, Optional
|
|
|
|
from sqlalchemy import select, and_
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from config import config # if unused elsewhere, you can remove this import
|
|
|
|
# ORM models
|
|
from models.market import (
|
|
Product, ProductImage, ProductSection,
|
|
Listing, ListingItem,
|
|
NavTop, NavSub,
|
|
ProductSticker, ProductLabel,
|
|
ProductAttribute, ProductNutrition, ProductAllergen, ProductLike
|
|
|
|
)
|
|
from sqlalchemy import func, case
|
|
|
|
|
|
# ---------- helpers ----------
|
|
def _regular_price_of(p: Product) -> Optional[float]:
|
|
try:
|
|
return (
|
|
float(p.regular_price)
|
|
if p.regular_price is not None
|
|
else (
|
|
float(p.special_price)
|
|
if p.special_price is not None
|
|
else None
|
|
)
|
|
)
|
|
except Exception:
|
|
return None
|
|
|
|
# ---------- NAV ----------
|
|
async def db_nav(session) -> Dict:
|
|
tops = (await session.execute(select(NavTop))).scalars().all()
|
|
subs = (await session.execute(select(NavSub))).scalars().all()
|
|
|
|
subs_by_top: Dict[int, List[Dict]] = {}
|
|
for s in subs:
|
|
sub_name = (s.label or s.slug or "").strip()
|
|
subs_by_top.setdefault(s.top_id, []).append({
|
|
"label": s.label,
|
|
"name": sub_name, # back-compat for callers expecting "name"
|
|
"slug": s.slug,
|
|
"href": s.href,
|
|
})
|
|
|
|
cats: Dict[str, Dict] = {}
|
|
for t in tops:
|
|
top_label = (t.label or t.slug or "").strip()
|
|
cats[top_label] = {
|
|
"label": t.label,
|
|
"name": top_label, # back-compat
|
|
"slug": t.slug,
|
|
"subs": sorted(subs_by_top.get(t.id, []), key=lambda x: (x["name"] or "").lower()),
|
|
}
|
|
return {"cats": cats}
|
|
|
|
|
|
async def db_product_full(session, slug: str, user_id=0) -> Optional[dict]:
|
|
|
|
liked_product_ids_subq = (
|
|
select(ProductLike.product_slug)
|
|
.where(
|
|
and_(
|
|
ProductLike.user_id == user_id,
|
|
ProductLike.deleted_at.is_(None)
|
|
)
|
|
)
|
|
)
|
|
|
|
is_liked_case = case(
|
|
(and_(
|
|
(Product.slug.in_(liked_product_ids_subq)),
|
|
Product.deleted_at.is_(None)
|
|
), True),
|
|
else_=False
|
|
).label("is_liked")
|
|
|
|
q = (
|
|
select(Product, is_liked_case)
|
|
.where(Product.slug == slug, Product.deleted_at.is_(None))
|
|
.options(
|
|
selectinload(Product.images.and_(ProductImage.deleted_at.is_(None))),
|
|
selectinload(Product.sections.and_(ProductSection.deleted_at.is_(None))),
|
|
selectinload(Product.labels.and_(ProductLabel.deleted_at.is_(None))),
|
|
selectinload(Product.stickers.and_(ProductSticker.deleted_at.is_(None))),
|
|
selectinload(Product.attributes.and_(ProductAttribute.deleted_at.is_(None))),
|
|
selectinload(Product.nutrition.and_(ProductNutrition.deleted_at.is_(None))),
|
|
selectinload(Product.allergens.and_(ProductAllergen.deleted_at.is_(None))),
|
|
)
|
|
)
|
|
result = await session.execute(q)
|
|
|
|
row = result.first() if result is not None else None
|
|
p, is_liked = row if row else (None, None)
|
|
if not p:
|
|
return None
|
|
|
|
gallery = [
|
|
img.url
|
|
for img in sorted(p.images, key=lambda i: (i.kind or "gallery", i.position or 0))
|
|
if (img.kind or "gallery") == "gallery"
|
|
]
|
|
embedded = [
|
|
img.url
|
|
for img in sorted(p.images, key=lambda i: i.position or 0)
|
|
if (img.kind or "") == "embedded"
|
|
]
|
|
all_imgs = [
|
|
img.url
|
|
for img in sorted(p.images, key=lambda i: i.position or 0)
|
|
if (img.kind or "") == "all"
|
|
]
|
|
return {
|
|
"id": p.id,
|
|
"slug": p.slug,
|
|
"title": p.title,
|
|
"brand": p.brand,
|
|
"image": p.image,
|
|
"description_short": p.description_short,
|
|
"description_html": p.description_html,
|
|
"suma_href": p.suma_href,
|
|
"rrp": float(p.rrp) if p.rrp is not None else None,
|
|
"special_price": float(p.special_price) if p.special_price is not None else None,
|
|
"special_price_raw": p.special_price_raw,
|
|
"special_price_currency": p.special_price_currency,
|
|
"regular_price": _regular_price_of(p),
|
|
"regular_price_raw": p.regular_price_raw,
|
|
"regular_price_currency": p.regular_price_currency,
|
|
"rrp_raw": p.rrp_raw,
|
|
"rrp_currency": p.rrp_currency,
|
|
"price_per_unit_raw": p.price_per_unit_raw,
|
|
"price_per_unit": p.price_per_unit,
|
|
"price_per_unit_currency": p.price_per_unit_currency,
|
|
"oe_list_price": p.oe_list_price,
|
|
"images": gallery,
|
|
"embedded_image_urls": embedded,
|
|
"all_image_urls": all_imgs,
|
|
"sections": [{"title": s.title, "html": s.html} for s in p.sections],
|
|
"stickers": [v.name.strip().lower() for v in p.stickers if v.name],
|
|
"labels": [v.name for v in p.labels if v.name],
|
|
"ean": p.ean,
|
|
"sku": p.sku,
|
|
"unit_size": p.unit_size,
|
|
"pack_size": p.pack_size,
|
|
"case_size_raw": p.case_size_raw,
|
|
"case_size_count": p.case_size_count,
|
|
"case_size_item_qty": p.case_size_item_qty,
|
|
"case_size_item_unit": p.case_size_item_unit,
|
|
"info_table": {a.key: a.value for a in p.attributes if a.key},
|
|
"nutrition": [{"key": n.key, "value": n.value, "unit": n.unit} for n in p.nutrition if n.key],
|
|
"allergens": [{"name": a.name, "contains": a.contains} for a in p.allergens if a.name],
|
|
"is_liked": is_liked,
|
|
"deleted_at": p.deleted_at
|
|
}
|
|
|
|
|
|
async def db_product_full_id(session, id:int, user_id=0) -> Optional[dict]:
|
|
liked_product_ids_subq = (
|
|
select(ProductLike.product_slug)
|
|
.where(
|
|
and_(
|
|
ProductLike.user_id == user_id,
|
|
ProductLike.deleted_at.is_(None)
|
|
)
|
|
)
|
|
)
|
|
|
|
is_liked_case = case(
|
|
(
|
|
(Product.slug.in_(liked_product_ids_subq)),
|
|
True
|
|
),
|
|
else_=False
|
|
).label("is_liked")
|
|
|
|
q = (
|
|
select(Product, is_liked_case)
|
|
.where(Product.id == id)
|
|
.options(
|
|
selectinload(Product.images.and_(ProductImage.deleted_at.is_(None))),
|
|
selectinload(Product.sections.and_(ProductSection.deleted_at.is_(None))),
|
|
selectinload(Product.labels.and_(ProductLabel.deleted_at.is_(None))),
|
|
selectinload(Product.stickers.and_(ProductSticker.deleted_at.is_(None))),
|
|
selectinload(Product.attributes.and_(ProductAttribute.deleted_at.is_(None))),
|
|
selectinload(Product.nutrition.and_(ProductNutrition.deleted_at.is_(None))),
|
|
selectinload(Product.allergens.and_(ProductAllergen.deleted_at.is_(None))),
|
|
)
|
|
)
|
|
result = await session.execute(q)
|
|
|
|
row = result.first() if result is not None else None
|
|
p, is_liked = row if row else (None, None)
|
|
if not p:
|
|
return None
|
|
|
|
gallery = [
|
|
img.url
|
|
for img in sorted(p.images, key=lambda i: (i.kind or "gallery", i.position or 0))
|
|
if (img.kind or "gallery") == "gallery"
|
|
]
|
|
embedded = [
|
|
img.url
|
|
for img in sorted(p.images, key=lambda i: i.position or 0)
|
|
if (img.kind or "") == "embedded"
|
|
]
|
|
all_imgs = [
|
|
img.url
|
|
for img in sorted(p.images, key=lambda i: i.position or 0)
|
|
if (img.kind or "") == "all"
|
|
]
|
|
return {
|
|
"id": p.id,
|
|
"slug": p.slug,
|
|
"title": p.title,
|
|
"brand": p.brand,
|
|
"image": p.image,
|
|
"description_short": p.description_short,
|
|
"description_html": p.description_html,
|
|
"suma_href": p.suma_href,
|
|
"rrp": float(p.rrp) if p.rrp is not None else None,
|
|
"special_price": float(p.special_price) if p.special_price is not None else None,
|
|
"special_price_raw": p.special_price_raw,
|
|
"special_price_currency": p.special_price_currency,
|
|
"regular_price": _regular_price_of(p),
|
|
"regular_price_raw": p.regular_price_raw,
|
|
"regular_price_currency": p.regular_price_currency,
|
|
"rrp_raw": p.rrp_raw,
|
|
"rrp_currency": p.rrp_currency,
|
|
"price_per_unit_raw": p.price_per_unit_raw,
|
|
"price_per_unit": p.price_per_unit,
|
|
"price_per_unit_currency": p.price_per_unit_currency,
|
|
"oe_list_price": p.oe_list_price,
|
|
"images": gallery,
|
|
"embedded_image_urls": embedded,
|
|
"all_image_urls": all_imgs,
|
|
"sections": [{"title": s.title, "html": s.html} for s in p.sections],
|
|
"stickers": [v.name.strip().lower() for v in p.stickers if v.name],
|
|
"labels": [v.name for v in p.labels if v.name],
|
|
"ean": p.ean,
|
|
"sku": p.sku,
|
|
"unit_size": p.unit_size,
|
|
"pack_size": p.pack_size,
|
|
"case_size_raw": p.case_size_raw,
|
|
"case_size_count": p.case_size_count,
|
|
"case_size_item_qty": p.case_size_item_qty,
|
|
"case_size_item_unit": p.case_size_item_unit,
|
|
"info_table": {a.key: a.value for a in p.attributes if a.key},
|
|
"nutrition": [{"key": n.key, "value": n.value, "unit": n.unit} for n in p.nutrition if n.key],
|
|
"allergens": [{"name": a.name, "contains": a.contains} for a in p.allergens if a.name],
|
|
"is_liked": is_liked,
|
|
"deleted_at": p.deleted_at
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
# ---------- PRODUCTS LISTING ----------
|
|
|
|
async def db_products_nocounts(
|
|
session,
|
|
top_slug: str | None,
|
|
sub_slug: str | None,
|
|
selected_brands: Optional[List[str]] = None,
|
|
selected_stickers: Optional[List[str]] = None,
|
|
selected_labels: Optional[List[str]] = None,
|
|
page: int = 1,
|
|
search: Optional[str] = None,
|
|
sort: Optional[str] = None,
|
|
page_size: int = 20,
|
|
liked: bool = None,
|
|
user_id: int=0
|
|
) -> Dict:
|
|
BLOCKED_SLUGS = set((config().get("blacklist", {}).get("product", []) or []))
|
|
base_conditions = []
|
|
if BLOCKED_SLUGS:
|
|
base_conditions.append(
|
|
~Product.slug.in_(BLOCKED_SLUGS),
|
|
)
|
|
|
|
if top_slug:
|
|
|
|
q_list = (
|
|
select(Listing.id)
|
|
.join(NavTop, Listing.top)
|
|
.outerjoin(NavSub, Listing.sub)
|
|
.where(
|
|
Listing.deleted_at.is_(None),
|
|
NavTop.deleted_at.is_(None),
|
|
NavTop.slug == top_slug,
|
|
NavSub.deleted_at.is_(None),
|
|
NavSub.slug == sub_slug if sub_slug else Listing.sub_id.is_(None),
|
|
)
|
|
)
|
|
|
|
listing_id = (await session.execute(q_list)).scalars().first()
|
|
if not listing_id:
|
|
return {"total_pages": 1, "items": []}
|
|
|
|
base_conditions.append(Product.slug.in_(
|
|
select(ListingItem.slug).where(ListingItem.listing_id == listing_id, ListingItem.deleted_at.is_(None))
|
|
))
|
|
|
|
base_ids_subq = select(Product.id).where(*base_conditions, Product.deleted_at.is_(None))
|
|
base_ids = (await session.execute(base_ids_subq)).scalars().all()
|
|
|
|
if not base_ids:
|
|
return {"total_pages": 1, "items": []}
|
|
sel_brands = [(b or "").strip().lower() for b in (selected_brands or []) if (b or "").strip()]
|
|
sel_stickers = [(s or "").strip().lower() for s in (selected_stickers or []) if (s or "").strip()]
|
|
sel_labels = [(l or "").strip().lower() for l in (selected_labels or []) if (l or "").strip()]
|
|
search_q = (search or "").strip().lower()
|
|
|
|
filter_conditions = []
|
|
if sel_brands:
|
|
filter_conditions.append(func.lower(Product.brand).in_(sel_brands))
|
|
for sticker_name in sel_stickers:
|
|
filter_conditions.append(
|
|
Product.stickers.any(
|
|
and_(
|
|
func.lower(ProductSticker.name) == sticker_name,
|
|
ProductSticker.deleted_at.is_(None)
|
|
)
|
|
)
|
|
)
|
|
for label_name in sel_labels:
|
|
filter_conditions.append(
|
|
Product.labels.any(
|
|
and_(
|
|
func.lower(ProductLabel.name) == label_name,
|
|
ProductLabel.deleted_at.is_(None),
|
|
)
|
|
)
|
|
)
|
|
if search_q:
|
|
filter_conditions.append(func.lower(Product.description_short).contains(search_q))
|
|
if liked:
|
|
liked_subq = liked_subq = (
|
|
select(ProductLike.product_slug)
|
|
.where(
|
|
and_(
|
|
ProductLike.user_id == user_id,
|
|
ProductLike.deleted_at.is_(None)
|
|
)
|
|
)
|
|
.subquery()
|
|
)
|
|
filter_conditions.append(Product.slug.in_(liked_subq))
|
|
|
|
filtered_count_query = select(func.count(Product.id)).where(Product.id.in_(base_ids), *filter_conditions)
|
|
total_filtered = (await session.execute(filtered_count_query)).scalars().one()
|
|
total_pages = max(1, (total_filtered + page_size - 1) // page_size)
|
|
page = max(1, page)
|
|
|
|
|
|
liked_product_slugs_subq = (
|
|
select(ProductLike.product_slug)
|
|
.where(
|
|
and_(
|
|
ProductLike.user_id == user_id,
|
|
ProductLike.deleted_at.is_(None)
|
|
)
|
|
)
|
|
)
|
|
is_liked_case = case(
|
|
(Product.slug.in_(liked_product_slugs_subq), True),
|
|
else_=False
|
|
).label("is_liked")
|
|
|
|
q_filtered = select(Product, is_liked_case).where(Product.id.in_(base_ids), *filter_conditions).options(
|
|
selectinload(Product.images),
|
|
selectinload(Product.sections),
|
|
selectinload(Product.labels),
|
|
selectinload(Product.stickers),
|
|
selectinload(Product.attributes),
|
|
selectinload(Product.nutrition),
|
|
selectinload(Product.allergens),
|
|
)
|
|
|
|
sort_mode = (sort or "az").strip().lower()
|
|
if sort_mode == "az":
|
|
q_filtered = q_filtered.order_by(func.lower(Product.title), Product.slug)
|
|
elif sort_mode == "za":
|
|
q_filtered = q_filtered.order_by(func.lower(Product.title).desc(), Product.slug.desc())
|
|
elif sort_mode in ("price-asc", "price_asc", "price-low", "price-low-high", "low-high", "lo-hi"):
|
|
q_filtered = q_filtered.order_by(
|
|
case((Product.regular_price.is_(None), 1), else_=0),
|
|
Product.regular_price.asc(),
|
|
func.lower(Product.title),
|
|
Product.slug
|
|
)
|
|
elif sort_mode in ("price-desc", "price_desc", "price-high", "price-high-low", "high-low", "hi-lo"):
|
|
q_filtered = q_filtered.order_by(
|
|
case((Product.regular_price.is_(None), 1), else_=0),
|
|
Product.regular_price.desc(),
|
|
func.lower(Product.title),
|
|
Product.slug
|
|
)
|
|
else:
|
|
q_filtered = q_filtered.order_by(func.lower(Product.title), Product.slug)
|
|
|
|
offset_val = (page - 1) * page_size
|
|
q_filtered = q_filtered.offset(offset_val).limit(page_size)
|
|
products_page = (await session.execute(q_filtered)).all()
|
|
|
|
items: List[Dict] = []
|
|
for p, is_liked in products_page:
|
|
gallery_imgs = sorted((img for img in p.images), key=lambda i: (i.kind or "gallery", i.position or 0))
|
|
gallery = [img.url for img in gallery_imgs if (img.kind or "gallery") == "gallery"]
|
|
embedded = [img.url for img in sorted(p.images, key=lambda i: i.position or 0) if (img.kind or "") == "embedded"]
|
|
all_imgs = [img.url for img in sorted(p.images, key=lambda i: i.position or 0) if (img.kind or "") == "all"]
|
|
|
|
items.append({
|
|
"slug": p.slug,
|
|
"title": p.title,
|
|
"brand": p.brand,
|
|
"description_short": p.description_short,
|
|
"description_html": p.description_html,
|
|
"image": p.image,
|
|
"rrp": float(p.rrp) if p.rrp is not None else None,
|
|
"special_price": float(p.special_price) if p.special_price is not None else None,
|
|
"special_price_raw": p.special_price_raw,
|
|
"special_price_currency": p.special_price_currency,
|
|
"regular_price": _regular_price_of(p),
|
|
"regular_price_raw": p.regular_price_raw,
|
|
"regular_price_currency": p.regular_price_currency,
|
|
"rrp_raw": p.rrp_raw,
|
|
"rrp_currency": p.rrp_currency,
|
|
"price_per_unit_raw": p.price_per_unit_raw,
|
|
"price_per_unit": p.price_per_unit,
|
|
"price_per_unit_currency": p.price_per_unit_currency,
|
|
"images": gallery,
|
|
"embedded_image_urls": embedded,
|
|
"all_image_urls": all_imgs,
|
|
"sections": [{"title": s.title, "html": s.html} for s in p.sections],
|
|
"labels": [l.name for l in p.labels if l.name],
|
|
"stickers": [s.name.strip().lower() for s in p.stickers if s.name],
|
|
"info_table": {a.key: a.value for a in p.attributes if a.key},
|
|
"nutrition": [{"key": n.key, "value": n.value, "unit": n.unit} for n in p.nutrition if n.key],
|
|
"allergens": [{"name": a.name, "contains": a.contains} for a in p.allergens if a.name],
|
|
"ean": p.ean,
|
|
"sku": p.sku,
|
|
"unit_size": p.unit_size,
|
|
"pack_size": p.pack_size,
|
|
"is_liked": is_liked,
|
|
})
|
|
|
|
return {
|
|
"total_pages": total_pages,
|
|
"items": items,
|
|
}
|
|
|
|
|
|
async def db_products_counts(
|
|
session,
|
|
top_slug: str | None,
|
|
sub_slug: str | None,
|
|
search: Optional[str] = None,
|
|
user_id: int=0
|
|
) -> Dict:
|
|
BLOCKED_SLUGS = set((config().get("blacklist", {}).get("product", []) or []))
|
|
base_conditions = []
|
|
|
|
if top_slug:
|
|
q_list = select(Listing.id).where(
|
|
Listing.deleted_at.is_(None),
|
|
Listing.top.has(slug=top_slug),
|
|
Listing.sub.has(slug=sub_slug) if sub_slug else Listing.sub_id.is_(None),
|
|
)
|
|
listing_id = (await session.execute(q_list)).scalars().first()
|
|
if not listing_id:
|
|
return {
|
|
"brands": [],
|
|
"stickers": [],
|
|
"labels": [],
|
|
"liked_count": 0,
|
|
"search_count": 0,
|
|
}
|
|
|
|
listing_slug_subquery = select(ListingItem.slug).where(ListingItem.listing_id == listing_id, ListingItem.deleted_at.is_(None))
|
|
|
|
if BLOCKED_SLUGS:
|
|
base_conditions.append(
|
|
and_(
|
|
Product.slug.in_(listing_slug_subquery),
|
|
~Product.slug.in_(BLOCKED_SLUGS),
|
|
)
|
|
)
|
|
else:
|
|
base_conditions.append(Product.slug.in_(listing_slug_subquery))
|
|
else:
|
|
if BLOCKED_SLUGS:
|
|
base_conditions.append(~Product.slug.in_(BLOCKED_SLUGS))
|
|
base_ids = (await session.execute(select(Product.id).where(*base_conditions, Product.deleted_at.is_(None)))).scalars().all()
|
|
if base_ids:
|
|
base_products_slugs = (await session.execute(
|
|
select(Product.slug).where(Product.id.in_(base_ids), Product.deleted_at.is_(None))
|
|
)).scalars().all()
|
|
if not base_products_slugs:
|
|
return {
|
|
"brands": [],
|
|
"stickers": [],
|
|
"labels": [],
|
|
"liked_count": 0,
|
|
"search_count": 0,
|
|
}
|
|
base_ids = (await session.execute(
|
|
select(Product.id).where(Product.slug.in_(base_products_slugs), Product.deleted_at.is_(None))
|
|
)).scalars().all()
|
|
else:
|
|
return {
|
|
"brands": [],
|
|
"stickers": [],
|
|
"labels": [],
|
|
"liked_count": 0,
|
|
"search_count": 0,
|
|
}
|
|
|
|
brands_list: List[Dict] = []
|
|
stickers_list: List[Dict] = []
|
|
labels_list: List[Dict] = []
|
|
liked_count = 0
|
|
search_count = 0
|
|
liked_product_slugs_subq = (
|
|
select(ProductLike.product_slug)
|
|
.where(ProductLike.user_id == user_id, ProductLike.deleted_at.is_(None))
|
|
)
|
|
liked_count = await session.scalar(
|
|
select(func.count(Product.id))
|
|
.where(
|
|
Product.id.in_(base_ids),
|
|
Product.slug.in_(liked_product_slugs_subq),
|
|
Product.deleted_at.is_(None)
|
|
)
|
|
)
|
|
|
|
liked_count = (await session.execute(
|
|
select(func.count())
|
|
.select_from(ProductLike)
|
|
.where(
|
|
ProductLike.user_id == user_id,
|
|
ProductLike.product_slug.in_(
|
|
select(Product.slug).where(Product.id.in_(base_ids))
|
|
),
|
|
ProductLike.deleted_at.is_(None)
|
|
)
|
|
)).scalar_one() if user_id else 0
|
|
|
|
# Brand counts
|
|
brand_count_rows = await session.execute(
|
|
select(Product.brand, func.count(Product.id))
|
|
.where(Product.id.in_(base_ids),
|
|
Product.brand.is_not(None),
|
|
func.trim(Product.brand) != "",
|
|
Product.deleted_at.is_(None)
|
|
)
|
|
.group_by(Product.brand)
|
|
)
|
|
for brand_name, count in brand_count_rows:
|
|
brands_list.append({"name": brand_name, "count": count})
|
|
brands_list.sort(key=lambda x: (-x["count"], x["name"].lower()))
|
|
|
|
# Sticker counts
|
|
sticker_count_rows = await session.execute(
|
|
select(ProductSticker.name, func.count(ProductSticker.product_id))
|
|
.where(
|
|
ProductSticker.product_id.in_(base_ids),
|
|
ProductSticker.deleted_at.is_(None)
|
|
)
|
|
.group_by(ProductSticker.name)
|
|
)
|
|
for sticker_name, count in sticker_count_rows:
|
|
if sticker_name:
|
|
stickers_list.append({"name": sticker_name.strip().lower(), "count": count})
|
|
stickers_list.sort(key=lambda x: (-x["count"], x["name"]))
|
|
|
|
# Label counts
|
|
label_count_rows = await session.execute(
|
|
select(ProductLabel.name, func.count(ProductLabel.product_id))
|
|
.where(
|
|
ProductLabel.product_id.in_(base_ids),
|
|
ProductLabel.deleted_at.is_(None)
|
|
)
|
|
.group_by(ProductLabel.name)
|
|
)
|
|
for label_name, count in label_count_rows:
|
|
if label_name:
|
|
labels_list.append({"name": label_name, "count": count})
|
|
labels_list.sort(key=lambda x: (-x["count"], x["name"]))
|
|
|
|
|
|
# Search count
|
|
search_q = (search or "").strip().lower()
|
|
if search_q:
|
|
search_count = (await session.execute(
|
|
select(func.count(Product.id))
|
|
.where(
|
|
Product.id.in_(base_ids),
|
|
func.lower(Product.description_short).contains(search_q),
|
|
Product.deleted_at.is_(None)
|
|
)
|
|
)).scalars().one()
|
|
else:
|
|
search_count = len(base_ids)
|
|
|
|
return {
|
|
"brands": brands_list,
|
|
"stickers": stickers_list,
|
|
"labels": labels_list,
|
|
"liked_count": liked_count,
|
|
"search_count": search_count,
|
|
}
|
|
|
|
async def db_products(
|
|
session,
|
|
top_slug: str | None,
|
|
sub_slug: str | None,
|
|
selected_brands: Optional[List[str]] = None,
|
|
selected_stickers: Optional[List[str]] = None,
|
|
selected_labels: Optional[List[str]] = None,
|
|
page: int = 1,
|
|
search: Optional[str] = None,
|
|
sort: Optional[str] = None,
|
|
page_size: int = 20,
|
|
liked: bool = None,
|
|
user_id: int=0
|
|
) -> Dict:
|
|
return {
|
|
**(await db_products_nocounts(
|
|
session,
|
|
top_slug=top_slug,
|
|
sub_slug=sub_slug,
|
|
selected_brands=selected_brands,
|
|
selected_stickers=selected_stickers,
|
|
selected_labels=selected_labels,
|
|
page=page,
|
|
search=search,
|
|
sort=sort,
|
|
page_size=page_size,
|
|
liked=liked,
|
|
user_id=user_id
|
|
)),
|
|
**(await db_products_counts(
|
|
session,
|
|
top_slug=top_slug,
|
|
sub_slug=sub_slug,
|
|
search=search,
|
|
user_id=user_id
|
|
)),
|
|
}
|
|
|
|
|