"""Read queries behind the catalog API. Routers stay thin; the shaping of a product plus its latest price at every shop lives here. Every query scopes its shops through `publishable(db)`: visible shops only, minus any shop whose source verify has blocked (Decision 10), so one rule decides what the public site shows. """ from collections import Counter from datetime import datetime from decimal import Decimal from sqlalchemy import Select, and_, case, distinct, func, or_, select, tuple_ from sqlalchemy.orm import Session, aliased from app.models import Award, Brand, Listing, ProductLine, Shop, PriceObservation, ProductVariant, Retailer, Source from app.models.hubs import ( AirportCategoryDetail, AirportDetail, AirportShop, AirportSummary, BrandAirport, BrandDetail, BrandSummary, CategoryPageLink, DatasetFacts, ) from app.services import airport_featured, airport_guides, award_picker, editorial, overrides, taxonomy, verify from app.services import places from app.models.places import ShopPlace from app.services.collectors.registry import COLLECTORS from app.services.featured import SavingRecord, pick_featured from app.services.urls import airport_category_path, airport_path, brand_page_path, category_slug, line_path from app.models.schemas import ( TopAward, AwardOut, CategoryCount, LineAirportOut, LineAttributeOut, LineOptionOut, LinePriceBlockOut, LineSelectorOut, LineShownOut, LineVariantOut, ProductLineDetail, PriceOut, ProductDetail, ProductSuggestion, ProductSummary, ShopPrice, ) def _latest_observation_subquery( variant_ids: list[int] | None = None, shop_ids: list[int] | None = None ): """The most recent observation per listing. `variant_ids` confines the ranking to those product variants' listings, `shop_ids` to those shops' listings (a route). The answer for them is the same; the cost is not: ranking every observation in the catalogue to price the eight featured cards took about 450 of the home page's 600 ms on staging (11 Sep), a two-airport savings page 0.7 to 1.2 s, and both grow with every collection.""" ranked = select( PriceObservation.id.label("obs_id"), PriceObservation.listing_id.label("listing_id"), func.row_number() .over( partition_by=PriceObservation.listing_id, # id breaks same-timestamp ties, else "latest" is nondeterministic. order_by=(PriceObservation.observed_at.desc(), PriceObservation.id.desc()), ) .label("rn"), ) if variant_ids is not None or shop_ids is not None: listings = select(Listing.id).where(live_listings()) if variant_ids is not None: listings = listings.where(Listing.variant_id.in_(variant_ids)) if shop_ids is not None: listings = listings.where(Listing.shop_id.in_(shop_ids)) ranked = ranked.where(PriceObservation.listing_id.in_(listings)) ranked = ranked.subquery() return select(ranked.c.obs_id).where(ranked.c.rn == 1).subquery() def _available(): return PriceObservation.in_stock.isnot(False) def live_listings(): """The listings the site reads: not ignored by a person (Stream L, LT3 e). An ignored listing keeps collecting prices and stays out of every price, brand and comparable read here and on the collectors page; the trip comparison, the home counts, IndexNow, coverage and the audit still query `Listing` directly (a filed `do` for that lane).""" return Listing.ignored_at.is_(None) def blocked_shop_ids(db: Session | None) -> list[int]: """The shops whose prices the site must not show as current (Decision 10). An uncleared correctness failure (MISMATCH_*, PARSE_FAIL: `verify.blocked_sources`) blocks publication of its source until a human clears the check. The block is a fact about that source's parser on that host, so it covers every shop the source writes to: the failing checks' own listings name the shops exactly, and the collector's declared shops cover a platform collector that serves several. Nothing else hides a row: a retailer's refusal (BLOCKED) and the freshness findings (PRICE_MOVED, GONE) keep their prices visible with their date. Empty when nothing is blocked, the common case, at the cost of one small query; memoised on the session so one request asks once and every query in it agrees. """ if db is None or not hasattr(db, "scalars"): return [] info = getattr(db, "info", None) cache = info if isinstance(info, dict) else None if cache is not None and "blocked_shop_ids" in cache: return cache["blocked_shop_ids"] blocked = verify.blocked_sources(db) ids: set[int] = set() if blocked: listing_ids = {check.listing_id for checks in blocked.values() for check in checks} ids.update(db.scalars(select(Listing.shop_id).where(Listing.id.in_(listing_ids)))) ids.update(_declared_shop_ids(db, list(blocked))) out = sorted(ids) if cache is not None: cache["blocked_shop_ids"] = out return out def _declared_shop_ids(db: Session, source_ids: list[int]) -> list[int]: """The shops these sources' collectors declare, by retailer slug and code (the pair `ingest.ensure_retailer_and_shops` creates them under).""" wanted: list[tuple[str, str]] = [] for slug in db.scalars(select(Source.slug).where(Source.id.in_(source_ids))): collector = COLLECTORS.get(slug) if collector is None: continue wanted.extend((collector.retailer_slug, spec.code) for spec in collector.shops()) if not wanted: return [] return list( db.scalars( select(Shop.id) .join(Retailer, Retailer.id == Shop.retailer_id) .where(tuple_(Retailer.slug, Shop.code).in_(wanted)) ) ) def publishable(db: Session | None): """The shops the site shows: visible, and not publication-blocked. Hidden shops exist nowhere on the site; their collectors still run. A shop blocked by verify's tripwire is hidden the same way, everywhere at once, until a human clears the check (`blocked_shop_ids`); the site never shows its prices as current in one place and not another. """ condition = Shop.visible.is_(True) blocked = blocked_shop_ids(db) if blocked: condition = and_(condition, Shop.id.not_in(blocked)) return condition def shown_category(): """The storefront's category rule (taxonomy.HIDDEN_CATEGORIES): a hidden category's product variants are listed, counted and mapped nowhere, like a hidden airport's prices.""" return or_(ProductVariant.category.is_(None), ProductVariant.category.not_in(taxonomy.HIDDEN_CATEGORIES)) def _like_pattern(query: str) -> str: """Escape LIKE wildcards so a shopper's % or _ is a character, not an operator.""" return "%" + query.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%" def _prefix_pattern(query: str) -> str: """`_like_pattern` anchored at the start: the words begin the text.""" return _like_pattern(query)[1:] #: Under this many characters the header search suggests nothing ("a" is everything). SUGGEST_MIN_CHARS = 2 def suggest_product_variants(db: Session, query: str, limit: int = 6) -> list[ProductSuggestion]: """The header search's product suggestions while a shopper types. The same match as `list_product_variants` (the words anywhere in the name or the brand, accents folded, over the visible listings), so "see all results" holds what was suggested (unless the shopper's airports narrow that page). Ranked for typing: names that begin with the words, then brands that do, then the most widely stocked, then the shortest name. No prices: `list_product_variants` spends about half a second per search on the price aggregation over the whole catalogue (measured on staging, 11 Sep), too slow per keystroke; this answers in about a tenth of that.""" q = query.strip() if len(q) < SUGGEST_MIN_CHARS: return [] name = func.unaccent(ProductVariant.name) brand = func.unaccent(func.coalesce(ProductVariant.brand, "")) anywhere = func.unaccent(_like_pattern(q)) starting = func.unaccent(_prefix_pattern(q)) shops = places.unit_count() rank = case((name.ilike(starting, escape="\\"), 0), (brand.ilike(starting, escape="\\"), 1), else_=2) rows = db.execute( select(ProductVariant.id, ProductVariant.name, ProductVariant.brand, ProductVariant.category, ProductVariant.thumb_url, shops) .join(Listing, Listing.variant_id == ProductVariant.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where( publishable(db), shown_category(), live_listings(), or_(name.ilike(anywhere, escape="\\"), brand.ilike(anywhere, escape="\\")), ) .group_by(ProductVariant.id) .order_by(rank, shops.desc(), func.length(ProductVariant.name), ProductVariant.id) .limit(limit * 3) ).all() # One suggestion per product line (Stream K5): the line's representative among the matches # (most places, then lowest id), at the rank of its best match; its path opens the line. slugs = line_slugs_of(db, [r[0] for r in rows]) by_line: dict[object, list] = {} order: list[object] = [] for row in rows: key = slugs.get(row[0]) or ("variant", row[0]) if key not in by_line: order.append(key) by_line.setdefault(key, []).append(row) out = [] for key in order[:limit]: pid, n, b, c, t, count = min(by_line[key], key=lambda r: (-(r[5] or 0), r[0])) out.append(ProductSuggestion(id=pid, name=n, brand=b, category=c, thumb_url=t, shop_count=count, path=line_path(slugs[pid], pid) if pid in slugs else None)) return out def _summary_base( db: Session, at_codes: list[str] | None = None, variant_ids: list[int] | None = None ) -> Select: latest = _latest_observation_subquery(variant_ids) stmt = ( select( ProductVariant.id, ProductVariant.name, ProductVariant.brand, ProductVariant.brand_id, ProductVariant.gtin, ProductVariant.category, ProductVariant.quantity_ml, ProductVariant.abv, ProductVariant.is_exclusive, ProductVariant.thumb_url, ProductVariant.image_source, places.unit_count().label("shop_count"), # Airport shops only: a product in one airport plus the online # catalogue cannot be compared airport-to-airport, and counting it # as comparable overstated that number by nearly a third. places.unit_count() .filter(Shop.is_catalogue_only.is_(False)) .label("airport_count"), # A sold-out shelf is not the cheapest place to buy: out-of-stock # observations only count when nothing in stock exists at all. func.coalesce( func.min(PriceObservation.price_usd).filter(_available()), func.min(PriceObservation.price_usd), ).label("cheapest_usd"), func.coalesce( func.max(PriceObservation.price_usd).filter(_available()), func.max(PriceObservation.price_usd), ).label("dearest_usd"), ) .join(Listing, Listing.variant_id == ProductVariant.id) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(PriceObservation.id.in_(select(latest.c.obs_id)), publishable(db), shown_category(), live_listings()) .group_by(ProductVariant.id) ) if at_codes: # Everything downstream -- price, saving, shop count -- is then relative # to where this shopper is actually travelling. stmt = stmt.where(Shop.code.in_(at_codes)) return stmt def _priced_shops( db: Session, variant_ids: list[int], at_codes: list[str] | None ) -> dict[int, list[tuple[str, str | None, float, bool | None]]]: """Where each product is sold with its price, cheapest first, one row per shop. A card that SHOWS the comparison answers more of the shopper's question than a bare price with a promise behind a click.""" if not variant_ids: return {} latest = _latest_observation_subquery(variant_ids) stmt = ( select( Listing.variant_id, Shop.name, Shop.iata, Shop.code, Shop.is_catalogue_only, PriceObservation.price_usd, PriceObservation.in_stock, ) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where( Listing.variant_id.in_(variant_ids), PriceObservation.id.in_(select(latest.c.obs_id)), PriceObservation.price_usd.isnot(None), publishable(db), live_listings(), ) .order_by( Listing.variant_id, case((PriceObservation.in_stock.is_(False), 1), else_=0), PriceObservation.price_usd.asc(), # Two shops at the same price keep one order, whatever plan the # database picks (a tie at BCN and MAD swapped with the query's shape). Shop.code, ) ) if at_codes: stmt = stmt.where(Shop.code.in_(at_codes)) out: dict[int, list[tuple[str, str | None, float, bool | None]]] = {} seen: set[tuple[int, str]] = set() for variant_id, name, iata, code, catalogue_only, price, in_stock in db.execute(stmt): # Bars have room for a short code, not a shop's full trading name. label = iata or ("Online" if catalogue_only else (code if len(code) <= 4 else name)) # Two listings can share a shop; the first (cheapest, in-stock-first) # row is that shop's price on the card. if (variant_id, label) in seen: continue seen.add((variant_id, label)) out.setdefault(variant_id, []).append((label, iata, float(price), in_stock)) return out def _top_award(award: Award) -> TopAward: return TopAward( competition_slug=award.competition_slug, competition=award.competition, medal=award.medal, year=award.year, ) def _top_awards( db: Session, variant_ids: list[int], at_codes: list[str] | None = None ) -> dict[int, TopAward]: """Each product's featured medal for its card corner, by the strategy in `award_picker` (the latest result per competition, then level, recency and the visitor's airports). The first version kept the best medal a bottle ever won, so a 2024 Double Gold outranked every later result for good.""" if not variant_ids: return {} rows = db.scalars(select(Award).where(Award.variant_id.in_(variant_ids))).all() held: dict[int, list[Award]] = {} for award in rows: held.setdefault(award.variant_id, []).append(award) out: dict[int, TopAward] = {} for pid, awards in held.items(): picked = award_picker.pick(awards, visitor_airports=at_codes or (), listing=True) if picked.featured is not None: out[pid] = _top_award(picked.featured) return out def _award_counts(db: Session, variant_ids: list[int]) -> dict[int, int]: if not variant_ids: return {} rows = db.execute( select(Award.variant_id, func.count(Award.id)) .where(Award.variant_id.in_(variant_ids)) .group_by(Award.variant_id) ).all() return dict(rows) def list_product_variants( db: Session, *, query: str | None = None, category: str | None = None, family: str | None = None, multi_only: bool = False, awarded_only: bool = False, exclusives_only: bool = False, with_image_only: bool = False, at_codes: list[str] | None = None, only_ids: list[int] | None = None, stocked_at: list[int] | None = None, brand_ids: list[int] | None = None, sort: str = "featured", limit: int = 24, offset: int = 0, ) -> tuple[int, list[ProductSummary]]: # A known handful of product variants (the featured cards, a picker's list) ranks only # their own observations; a browse ranks the catalogue's. stmt = _summary_base(db, at_codes, variant_ids=only_ids or None) if stocked_at: # An airport page: only product variants carried at these shops, but every # figure on the card still spans all our airports. Unlike `at_codes` # this narrows WHICH product variants, not where their prices come from, so a # bottle at Heathrow still shows that it is cheaper in Dubai. stmt = stmt.where( ProductVariant.id.in_(select(Listing.variant_id).where(Listing.shop_id.in_(stocked_at), live_listings())) ) if brand_ids: # A brand page: the fold row and its aliases (Decision 6), never a name match. stmt = stmt.where(ProductVariant.brand_id.in_(brand_ids)) if query: # unaccent so "patron" finds Patrón; escaped so % and _ are literals. pattern = func.unaccent(_like_pattern(query)) stmt = stmt.where( or_( func.unaccent(ProductVariant.name).ilike(pattern, escape="\\"), func.unaccent(func.coalesce(ProductVariant.brand, "")).ilike(pattern, escape="\\"), ) ) if category: stmt = stmt.where(ProductVariant.category == category) if family: # A department ("All drinks"): every shown category of the family. An # unknown family names no category and so matches nothing. stmt = stmt.where(ProductVariant.category.in_(taxonomy.categories_of(family))) if awarded_only: stmt = stmt.where(ProductVariant.id.in_(select(Award.variant_id))) if exclusives_only: stmt = stmt.where(ProductVariant.is_exclusive.is_(True)) if with_image_only: stmt = stmt.where(ProductVariant.thumb_url.isnot(None)) if only_ids: stmt = stmt.where(ProductVariant.id.in_(only_ids)) if multi_only: stmt = stmt.having( places.unit_count() .filter(Shop.is_catalogue_only.is_(False)) > 1 ) total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0 if sort == "price": stmt = stmt.order_by( func.min(PriceObservation.price_usd).asc().nulls_last(), ProductVariant.id ) elif sort == "name": stmt = stmt.order_by(ProductVariant.name.asc(), ProductVariant.id) elif sort == "compared": stmt = stmt.order_by( places.unit_count().desc(), func.min(PriceObservation.price_usd).asc().nulls_last(), ProductVariant.id, ) else: # "featured" (the default): lead with product variants that look and read as # complete -- a photo, a medal, and a real cross-shop comparison. A grid # of grey placeholders is a bad first impression and is also, genuinely, # a worse set of results for a shopper. awarded = select(Award.variant_id).distinct().subquery() stmt = stmt.order_by( case((ProductVariant.thumb_url.isnot(None), 0), else_=1), case((ProductVariant.id.in_(select(awarded.c.variant_id)), 0), else_=1), places.unit_count().desc(), func.min(PriceObservation.price_usd).asc().nulls_last(), ProductVariant.id, ) rows = db.execute(stmt.limit(limit).offset(offset)).all() awards = _award_counts(db, [r.id for r in rows]) top_awards = _top_awards(db, [r.id for r in rows], at_codes) priced_at = _priced_shops(db, [r.id for r in rows], at_codes) brand_slugs = brand_page_slugs(db, [r.brand_id for r in rows]) line_slugs = line_slugs_of(db, [r.id for r in rows]) items = [ ProductSummary( id=r.id, name=r.name, brand=r.brand, brand_slug=brand_slugs.get(r.brand_id), gtin=r.gtin, category=r.category, quantity_ml=r.quantity_ml, abv=float(r.abv) if r.abv is not None else None, is_exclusive=r.is_exclusive, thumb_url=r.thumb_url, image_source=r.image_source, shop_count=r.shop_count, cheapest_usd=float(r.cheapest_usd) if r.cheapest_usd is not None else None, dearest_usd=float(r.dearest_usd) if r.dearest_usd is not None else None, award_count=awards.get(r.id, 0), best_shop=(priced_at.get(r.id) or [(None, None, 0.0, None)])[0][0], best_shop_iata=(priced_at.get(r.id) or [(None, None, 0.0, None)])[0][1], shop_labels=[ label for label, _, _, _ in (priced_at.get(r.id) or [])[:4] ], top_prices=[ ShopPrice(label=label, usd=round(price, 2), in_stock=in_stock) for label, _, price, in_stock in (priced_at.get(r.id) or [])[:4] ], top_award=top_awards.get(r.id), path=card_path(line_slugs.get(r.id), r.id, priced_at.get(r.id) or []), ) for r in rows ] return total, items def line_slugs_of(db: Session, variant_ids: list[int]) -> dict[int, str]: """`{variant_id: the live product line's slug}`, aliases followed, hidden lines left out: where a card or a suggestion opens. One query for the lines, one per alias hop (rare).""" if not variant_ids or not hasattr(db, "execute"): return {} rows = db.execute( select(ProductVariant.id, ProductLine.id, ProductLine.slug, ProductLine.alias_of_id, ProductLine.hidden) .join(ProductLine, ProductLine.id == ProductVariant.product_line_id) .where(ProductVariant.id.in_(variant_ids)) ).all() out: dict[int, str] = {} for variant_id, _line_id, slug, alias_of_id, hidden in rows: hops = 0 while alias_of_id is not None and hops < MERGE_CHAIN_LIMIT: target = db.get(ProductLine, alias_of_id) if target is None: break slug, alias_of_id, hidden, hops = target.slug, target.alias_of_id, target.hidden, hops + 1 if not hidden and alias_of_id is None: out[variant_id] = slug return out def card_path(line_slug: str | None, variant_id: int, priced: list[tuple]) -> str | None: """The card deep link contract (plan W1): the line, this variant, and the airports the card's comparison shows (its first four priced rows, in the card's order; a catalogue has no code). The card list is already scoped to the shopper's airports when they chose some, so this is their set wherever it stocks the variant, else the card's own.""" if not line_slug: return None codes: list[str] = [] for _label, iata, _price, _in_stock in priced[:4]: if iata and iata not in codes: codes.append(iata) return line_path(line_slug, variant_id, codes) # How many forwards to follow before giving up: a merge chain is one or two # hops in practice, and a cycle (which nothing should write) must not hang a # request. MERGE_CHAIN_LIMIT = 8 def resolve_variant_id(db: Session, variant_id: int) -> int: """The id a request for `variant_id` should be answered under. Every merged product must 301 to its survivor (build plan §6, Decision 6): the page route and the product API both pass their id through here first. Follows `product_variants.merged_into_id` (migration #3) hop by hop, bounded, and returns the last id that forwards nowhere; an unknown id comes back as itself so the caller's 404 still fires. """ scalar = getattr(db, "scalar", None) if scalar is None: # No session behind the request (the route tests fake the dependency): # the identity, exactly as before migration #3. return variant_id current = variant_id for _ in range(MERGE_CHAIN_LIMIT): target = scalar(select(ProductVariant.merged_into_id).where(ProductVariant.id == current)) if not target or target == current: return current current = target return current def get_product( db: Session, variant_id: int, at_codes: list[str] | None = None ) -> ProductDetail | None: """The product page's object. `at_codes` (the visitor's airports) only order the medals; the server-rendered page passes none, so it is deterministic.""" product = db.get(ProductVariant, variant_id) if product is None: return None latest = _latest_observation_subquery() rows = db.execute( select(PriceObservation, Shop, Retailer, Listing) .join(Listing, PriceObservation.listing_id == Listing.id) .join(Shop, Listing.shop_id == Shop.id).outerjoin(ShopPlace, places.primary_on()) .join(Retailer, Shop.retailer_id == Retailer.id) .where( Listing.variant_id == variant_id, PriceObservation.id.in_(select(latest.c.obs_id)), publishable(db), live_listings(), ) .order_by( case((PriceObservation.in_stock.is_(False), 1), else_=0), PriceObservation.price_usd.asc().nulls_last(), ) ).all() # One row per shop: two source SKUs occasionally resolve to one product at # one shop, and showing the same airport twice reads as a data error. # Rows are ordered best-buyable-first, so the first per shop wins. seen_shops: set[int] = set() deduped = [] for row in rows: if row[1].id in seen_shops: continue seen_shops.add(row[1].id) deduped.append(row) rows = deduped prices = [ PriceOut( shop_code=shop.code, shop_name=shop.name, shop_iata=shop.iata, shop_city=shop.city, shop_country=shop.country, shop_path=( airport_path(shop.iata, shop.city, shop.name) if shop.iata and not shop.is_catalogue_only else None ), retailer_name=retailer.name, is_catalogue_only=shop.is_catalogue_only, currency=obs.currency, price=float(obs.price), price_usd=float(obs.price_usd) if obs.price_usd is not None else None, was_price=float(obs.was_price) if obs.was_price is not None else None, price_type=obs.price_type, in_stock=obs.in_stock, observed_at=obs.observed_at, url=listing.url, ) for obs, shop, retailer, listing in rows ] # The page shows the retained medals (the latest result per competition), # featured first; seo.py's `award` property reads the same list. The count # stays the number of medals held. A pin will arrive through # award_picker.pin_from_override once the overrides table exists (R's # migration #4); the read path only ever falls back, a GET never writes. held = db.scalars(select(Award).where(Award.variant_id == variant_id)).all() picked = award_picker.pick(held, visitor_airports=at_codes or ()) awards = [AwardOut.model_validate(a) for a in picked.retained] usd = [p.price_usd for p in prices if p.price_usd is not None] return ProductDetail( id=product.id, name=product.name, brand=product.brand, brand_slug=brand_page_slugs(db, [product.brand_id]).get(product.brand_id), gtin=product.gtin, category=product.category, quantity_ml=product.quantity_ml, abv=float(product.abv) if product.abv is not None else None, is_exclusive=product.is_exclusive, thumb_url=product.thumb_url, image_url=product.image_url, image_source=product.image_source, country_of_origin=product.country_of_origin, shop_count=len({p.shop_name for p in prices}), cheapest_usd=min(usd) if usd else None, dearest_usd=max(usd) if usd else None, award_count=len(held), top_award=_top_award(picked.featured) if picked.featured is not None else None, prices=prices, awards=awards, ) # --- the product line page (Stream K5; plan W1, W20) ---------------------------- def line_by_slug(db: Session, slug: str) -> tuple[ProductLine | None, str | None]: """`(line, None)` for a live line; `(None, )` for an alias line or a slug the redirect table retired (K6's `publish.redirect_for`), aliases followed; `(None, None)` when nothing answers. A hidden line answers nowhere.""" from app.services import publish line = db.scalar(select(ProductLine).where(ProductLine.slug == slug)) if line is None: forward = publish.redirect_for(slug, "product_line", db) return (None, forward) if forward and forward != slug else (None, None) hops = 0 target = line while target.alias_of_id is not None and hops < MERGE_CHAIN_LIMIT: target, hops = db.get(ProductLine, target.alias_of_id), hops + 1 if target is None or publish.is_hidden(target): return None, None if target.id != line.id: return None, target.slug return line, None def _attribute_text(attribute) -> str: value = attribute.value if attribute.kind == "quantity": if value.form == "set" and value.set_contents: return value.set_contents if value.value is None or value.unit is None: return "" unit = {"pcs": " pcs"}.get(value.unit, f" {value.unit}") if value.pack_count and value.pack_unit_value: return f"{value.pack_count} x {value.pack_unit_value:g}{unit}" if value.unit == "ml" and float(value.value) >= 1000: return f"{float(value.value) / 1000:g} L" # a litre bottle is sold as 1 L, never 1000 ml return f"{float(value.value):g}{unit}" if isinstance(value, bool): return attribute.label if isinstance(value, (int, float, Decimal)): return f"{float(value):g}{attribute.unit or ''}" if attribute.unit != "years" else f"{float(value):g} years" return str(value) def line_of_variant(db: Session, variant_id: int) -> tuple[int, str] | None: """Where an old variant address lands (plan W1): the survivor's id (a merged-away id followed) and the slug of the live line that holds it, aliases followed; None when it has no line to show (unknown id, no line, a hidden line).""" resolved = resolve_variant_id(db, variant_id) variant = db.get(ProductVariant, resolved) if variant is None or variant.product_line_id is None or variant.merged_into_id is not None: return None line = db.get(ProductLine, variant.product_line_id) if line is None: return None found, forward = line_by_slug(db, line.slug) slug = found.slug if found is not None else forward return (resolved, slug) if slug else None def representative_variant(variants: list[LineVariantOut]) -> int | None: """The variant a card stands for (plan W20): among the line's publishable variants, the most comparison units, then the lowest id. Never recency: a card must not change which bottle it shows because a collection ran.""" if not variants: return None return min(variants, key=lambda v: (-v.units, v.id)).id def price_block(variants: list[LineVariantOut], chosen: int | None, airports: list[str]) -> LinePriceBlockOut: """The W20 cases, in order. A price prints only when it is unambiguous: one variant's own prices, each with the date it was seen; never a family's cheapest as the family's price and never a range. A catalogue shop never prices a page; a variant at one place never overrides a choice. Rian, 17 Sep, on a line page with nothing chosen: *"it says 'choose your quantity to see its price' but there is no ability to choose it right there in that section... if you choose the quantity, now it shows you a list of airports you can select. But if you click it, that takes you to the airport page. What I'm looking for is a price at that airport displayed on that page."* So: * with no variant chosen the block offers EVERY variant to choose from (`variants_offered`), marking the ones stocked at the shopper's airports; it used to offer only those, which is an empty list when no airport is chosen -- the state he was looking at; * with a variant chosen and no airports the case is `every_place`, and it prints that variant's price at every shop that stocks it, on the page. It was `ask_airport`, which named the airports and linked away to them; * with airports chosen the comparison is unchanged and `elsewhere_prices` carries every other shop's price, so the page can show them behind a disclosure without the shopper giving up the airports they chose. """ by_id = {v.id: v for v in variants} variant = by_id.get(chosen) if chosen is not None else None if variant is None: if not variants: return LinePriceBlockOut(case="no_price", airports=airports) at = [v.id for v in variants if airports and set(v.airports) & set(airports)] return LinePriceBlockOut(case="ask_variant", airports=airports, variants_at_airports=at, variants_offered=[v.id for v in variants]) shelf = [p for p in variant.prices if not p.is_catalogue_only] places_ = [LineAirportOut(iata=code, path=next((p.shop_path for p in shelf if p.shop_iata == code), None)) for code in variant.airports] if not shelf: return LinePriceBlockOut(case="no_price", variant_id=variant.id, airports=airports) if airports: here = [p for p in shelf if p.shop_iata in airports] if here: missing = [code for code in airports if code not in variant.airports] return LinePriceBlockOut(case="chosen_airports", variant_id=variant.id, airports=airports, prices=here, missing_airports=missing, elsewhere_prices=[p for p in shelf if p.shop_iata not in airports]) return LinePriceBlockOut(case="none_at_chosen", variant_id=variant.id, airports=airports, missing_airports=list(airports), elsewhere=places_, elsewhere_prices=shelf) if variant.units <= 1: return LinePriceBlockOut(case="single_place", variant_id=variant.id, prices=shelf, elsewhere=places_) return LinePriceBlockOut(case="every_place", variant_id=variant.id, prices=shelf, elsewhere=places_) def get_product_line( db: Session, slug: str, variant_id: int | None = None, airports: list[str] | None = None ) -> ProductLineDetail | None: """The product line page's object (plan W1, W20): the line and its brand; every live variant with a publishable listing, its attributes through the one accessor, its newest price per shop (one row per shop, best buyable first) and its comparison units; the selectors from the picked attributes (quantity always), the shown attributes as facts about what shops carry; the chosen variant (`variant_id` when it belongs to the line, a merged-away id followed to its survivor, else the only variant); the price block; the representative for cards. One query for the prices; `airports` only chooses the block and never filters the variants.""" from app.services import attributes, publish from app.services.product_lines import PICKED, SHOWN line = db.scalar(select(ProductLine).where(ProductLine.slug == slug)) if line is None or line.alias_of_id is not None or publish.is_hidden(line): return None airports = [a.upper() for a in (airports or [])] members = list(db.scalars( select(ProductVariant).where(ProductVariant.product_line_id == line.id, ProductVariant.merged_into_id.is_(None)) .order_by(ProductVariant.id) )) member_ids = [v.id for v in members] rows = [] if member_ids: latest = _latest_observation_subquery(member_ids) rows = db.execute( select(PriceObservation, Shop, Retailer, Listing) .join(Listing, PriceObservation.listing_id == Listing.id) .join(Shop, Listing.shop_id == Shop.id) .join(Retailer, Shop.retailer_id == Retailer.id) .where( Listing.variant_id.in_(member_ids), PriceObservation.id.in_(select(latest.c.obs_id)), publishable(db), live_listings(), ) .order_by( Listing.variant_id, case((PriceObservation.in_stock.is_(False), 1), else_=0), PriceObservation.price_usd.asc().nulls_last(), Shop.code, ) ).all() primary = places.primary_of(db) prices_of: dict[int, list[PriceOut]] = {} shops_of: dict[int, set[int]] = {} iata_of: dict[int, str | None] = {} seen: set[tuple[int, int]] = set() newest = None for obs, shop, retailer, listing in rows: if (listing.variant_id, shop.id) in seen: continue seen.add((listing.variant_id, shop.id)) iata_of[shop.id] = shop.iata if not shop.is_catalogue_only: shops_of.setdefault(listing.variant_id, set()).add(shop.id) newest = obs.observed_at if newest is None or obs.observed_at > newest else newest prices_of.setdefault(listing.variant_id, []).append(PriceOut( shop_code=shop.code, shop_name=shop.name, shop_iata=shop.iata, shop_city=shop.city, shop_country=shop.country, shop_path=airport_path(shop.iata, shop.city, shop.name) if shop.iata and not shop.is_catalogue_only else None, retailer_name=retailer.name, is_catalogue_only=shop.is_catalogue_only, currency=obs.currency, price=float(obs.price), price_usd=float(obs.price_usd) if obs.price_usd is not None else None, was_price=float(obs.was_price) if obs.was_price is not None else None, price_type=obs.price_type, in_stock=obs.in_stock, observed_at=obs.observed_at, url=listing.url, )) held: dict[int, list[Award]] = {} for award in db.scalars(select(Award).where(Award.variant_id.in_(member_ids or [0]))): held.setdefault(award.variant_id, []).append(award) variants: list[LineVariantOut] = [] for v in members: if v.id not in prices_of: continue # a variant only at a hidden or blocked shop exists nowhere on the site shop_ids = shops_of.get(v.id, set()) picked = award_picker.pick(held.get(v.id, []), visitor_airports=()) # A country code like "WOR" or "AE" means nothing to a shopper: only real names are facts. attrs = [LineAttributeOut(kind=a.kind, label=a.label, text=_attribute_text(a), display=a.display) for a in attributes.of(v) if not (a.kind == "country_of_origin" and len(str(a.value)) <= 3)] units = places.count_units(shop_ids, primary, iata_of) variants.append(LineVariantOut( id=v.id, name=v.name, gtin=v.gtin, quantity_ml=v.quantity_ml, attributes=[a for a in attrs if a.text], thumb_url=v.thumb_url, image_url=v.image_url, image_source=v.image_source, is_exclusive=bool(v.is_exclusive), country_of_origin=v.country_of_origin, units=units, comparable=units >= 2, airports=sorted({iata_of[s] for s in shop_ids if iata_of.get(s)}), prices=prices_of[v.id], award_count=len(held.get(v.id, [])), top_award=_top_award(picked.featured) if picked.featured is not None else None, awards=[AwardOut.model_validate(a) for a in picked.retained], )) selectors: list[LineSelectorOut] = [] shown: list[LineShownOut] = [] kinds: dict[str, tuple[str, str]] = {} for variant in variants: for a in variant.attributes: kinds.setdefault(a.kind, (a.label, a.display)) for kind, (label, display) in kinds.items(): if display == PICKED or kind == "quantity": options: dict[str, list[int]] = {} for variant in variants: text = next((a.text for a in variant.attributes if a.kind == kind), "") if text: options.setdefault(text, []).append(variant.id) if len(options) > 1 or kind == "quantity": selectors.append(LineSelectorOut(kind=kind, label=label, options=[LineOptionOut(text=t, variant_ids=ids) for t, ids in options.items()])) elif display == SHOWN: values = sorted({a.text for variant in variants for a in variant.attributes if a.kind == kind}) shown.append(LineShownOut(kind=kind, label=label, values=values)) ids = {v.id for v in variants} chosen = None if variant_id is not None: resolved = resolve_variant_id(db, variant_id) chosen = resolved if resolved in ids else None if chosen is None and len(variants) == 1: chosen = variants[0].id categories = Counter(v.category for v in members if v.category) brand = db.get(Brand, line.brand_id) return ProductLineDetail( id=line.id, slug=line.slug, name=line.name, path=line_path(line.slug), brand=brand.name if brand is not None else None, brand_slug=brand_page_slugs(db, [line.brand_id]).get(line.brand_id), category=categories.most_common(1)[0][0] if categories else None, indexed=publish.is_indexed(line), variants=variants, selectors=selectors, shown=shown, chosen_variant_id=chosen, representative_variant_id=representative_variant(variants), airports=airports, price_block=price_block(variants, chosen, airports), last_observed_at=newest, ) def featured_savings( db: Session, at_codes: list[str] | None = None, total: int = 8 ) -> list[ProductSummary]: """The home page's lead cards, selected by the agreed rules (see featured.py).""" latest = _latest_observation_subquery() awarded = select(Award.variant_id).distinct().subquery() rows = db.execute( select( ProductVariant.id, func.coalesce( func.min(PriceObservation.price_usd).filter(_available()), func.min(PriceObservation.price_usd), ).label("cheapest"), func.max(PriceObservation.price_usd).label("dearest"), ProductVariant.id.in_(select(awarded.c.variant_id)).label("has_award"), ProductVariant.thumb_url.isnot(None).label("has_image"), places.unit_count().label("shop_count"), ) .join(Listing, Listing.variant_id == ProductVariant.id) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where( PriceObservation.id.in_(select(latest.c.obs_id)), PriceObservation.price_usd.isnot(None), publishable(db), shown_category(), live_listings(), *( [Shop.code.in_(at_codes)] if at_codes else [] ), ) .group_by(ProductVariant.id) .having( places.unit_count() .filter(Shop.is_catalogue_only.is_(False)) > 1 ) ).all() records = [ SavingRecord( variant_id=r.id, saving_usd=float(r.dearest) - float(r.cheapest), saving_pct=(float(r.dearest) - float(r.cheapest)) / float(r.dearest) if float(r.dearest) > 0 else 0.0, awarded=bool(r.has_award), has_image=bool(r.has_image), shop_count=int(r.shop_count), ) for r in rows if r.cheapest is not None and r.dearest is not None ] order = pick_featured(records, total=total) if not order: return [] _, items = list_product_variants(db, at_codes=at_codes, only_ids=order, limit=len(order)) by_id = {item.id: item for item in items} return [by_id[pid] for pid in order if pid in by_id] def similar_product_variants( db: Session, variant_id: int, limit: int = 8, at_codes: list[str] | None = None, ) -> list[ProductSummary]: """Other bottles a shopper looking at this one would plausibly consider: same category, comparable across shops, complete-looking first. Scoped to the shopper's airports when they have set any -- suggesting a bottle they cannot walk up to is worse than suggesting nothing. """ product = db.get(ProductVariant, variant_id) if product is None: return [] _, items = list_product_variants( db, category=product.category, at_codes=at_codes, sort="featured", limit=limit + 1, ) return [item for item in items if item.id != variant_id][:limit] # --- airports --------------------------------------------------------------- # # An airport page is keyed by IATA code and gathers every visible storefront # there (one today, but two operators at one airport is the normal case in # the world), so each query below takes the airport's shop ids. SORTS = ("featured", "compared", "price", "name") def airport_shops(db: Session, iata: str | None = None) -> list[Shop]: """Visible airport storefronts that hold stock, in IATA order; one airport's when `iata` is given. The same rule as the airport picker: a configured store we hold nothing for has no page.""" stmt = ( select(Shop) .where( Shop.is_catalogue_only.is_(False), Shop.iata.isnot(None), publishable(db), Shop.id.in_(select(Listing.shop_id).where(live_listings()).distinct()), ) .order_by(Shop.iata, Shop.id) ) if iata: stmt = stmt.where(Shop.iata == iata.upper()) return list(db.scalars(stmt)) def _airport_shop_views(db: Session, shops: list[Shop]) -> list[AirportShop]: ids = [loc.id for loc in shops] rows = db.execute( select( Listing.shop_id, func.count(distinct(Listing.variant_id)), func.max(PriceObservation.observed_at), ) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .where(Listing.shop_id.in_(ids), live_listings()) .group_by(Listing.shop_id) ).all() by_id = {shop_id: (product_variants, last) for shop_id, product_variants, last in rows} return [ AirportShop( code=loc.code, retailer_name=loc.retailer.name, product_variants=by_id.get(loc.id, (0, None))[0], last_collected_at=by_id.get(loc.id, (0, None))[1], ) for loc in shops ] def _airport_summary(db: Session, shops: list[Shop], shop_views: list[AirportShop]) -> AirportSummary: first = shops[0] city = next((loc.city for loc in shops if loc.city), None) country = next((loc.country for loc in shops if loc.country), None) # Several shops can carry the same bottle; the airport's count is of # distinct product variants, not of shelves. product_variants = db.scalar( select(func.count(distinct(Listing.variant_id))).where( Listing.shop_id.in_([loc.id for loc in shops]), live_listings() ) ) or 0 return AirportSummary( iata=first.iata, path=airport_path(first.iata, city, first.name), name=first.name, city=city, country=country, currency=first.currency, product_variants=product_variants, last_collected_at=max((s.last_collected_at for s in shop_views if s.last_collected_at), default=None), ) def list_airports(db: Session) -> list[AirportSummary]: """Every airport with a page, for the sitemap and the airports index.""" grouped: dict[str, list[Shop]] = {} for loc in airport_shops(db): grouped.setdefault(loc.iata, []).append(loc) return [ _airport_summary(db, shops, _airport_shop_views(db, shops)) for shops in grouped.values() ] def _airport_rows(db: Session, shop_ids: list[int]): """One row per product carried at the airport: the card figures across every airport plus `here_usd`, what it costs at this one.""" stocked = select(Listing.variant_id).where(Listing.shop_id.in_(shop_ids), live_listings()) here = Listing.shop_id.in_(shop_ids) stmt = ( _summary_base(db) .add_columns( func.coalesce( func.min(PriceObservation.price_usd).filter(here, _available()), func.min(PriceObservation.price_usd).filter(here), ).label("here_usd") ) .where(ProductVariant.id.in_(stocked)) ) return db.execute(stmt).all() def airport_detail( db: Session, iata: str, *, category: str | None = None, multi_only: bool = False, awarded_only: bool = False, exclusives_only: bool = False, sort: str = "featured", limit: int = 24, offset: int = 0, sample: int = 8, ) -> AirportDetail | None: """Everything the airport page shows, or None when we hold no visible shop at that code (a hidden airport is a 404, not a teaser). The three feature flags are the storefront's own chips, applied to the full list only.""" shops = airport_shops(db, iata) if not shops: return None ids = [loc.id for loc in shops] shop_views = _airport_shop_views(db, shops) summary = _airport_summary(db, shops, shop_views) sort = sort if sort in SORTS else "featured" rows = _airport_rows(db, ids) # "Cheapest here" is the airport's card: this shop's latest price is the # lowest we know for a bottle sold at two or more of our airports. The saving # is against the dearest of them, the same figure the product page shows. winners = [] for r in rows: if r.airport_count < 2 or r.here_usd is None or r.cheapest_usd is None: continue if float(r.here_usd) <= float(r.cheapest_usd) + 0.005: winners.append((float(r.dearest_usd or r.here_usd) - float(r.here_usd), r.id)) winners.sort(key=lambda w: (-w[0], w[1])) top_ids = [pid for saving, pid in winners if saving > 0.5][:sample] savings: list[ProductSummary] = [] if top_ids: _, found = list_product_variants(db, only_ids=top_ids, limit=len(top_ids)) by_id = {item.id: item for item in found} savings = [by_id[pid] for pid in top_ids if pid in by_id] _, exclusive_items = list_product_variants( db, stocked_at=ids, exclusives_only=True, sort="featured", limit=sample ) # Featured by family (services/airport_featured.py): the same rows, the home page's picker # per family, one summaries fetch for every card. featured = airport_featured.featured_at( db, rows, lambda wanted: list_product_variants(db, only_ids=wanted, limit=len(wanted))[1] ) total, items = list_product_variants( db, stocked_at=ids, category=category, multi_only=multi_only, awarded_only=awarded_only, exclusives_only=exclusives_only, sort=sort, limit=limit, offset=offset, ) counted = Counter(r.category for r in rows if r.category) writeup = editorial.airport_writeup(db, iata) category_pages = category_page_links(db, iata) return AirportDetail( **summary.model_dump(), shops=shop_views, comparable=sum(1 for r in rows if r.airport_count > 1), cheapest_here=len(winners), exclusives=sum(1 for r in rows if r.is_exclusive), categories=[CategoryCount(category=c, count=n) for c, n in counted.most_common()], savings=savings, exclusive_items=exclusive_items, featured=featured, category_pages=category_pages, category=category, multi_only=multi_only, awarded_only=awarded_only, exclusives_only=exclusives_only, sort=sort, total=total, limit=limit, offset=offset, items=items, writeup=editorial.article_out(writeup) if writeup else None, guide=airport_guides.guide_for(iata, db), ) # --- category at airport ---------------------------------------------------- # # The pairing page (`/airports//`) exists only where the pair holds at # least coverage.CATEGORY_AT_AIRPORT_MIN_PRODUCTS published product variants, read at request time: # one count, `category_pair_counts`, feeds the page's 404, the airport page's rail of links # and the sitemap, so they can never disagree. def category_pair_counts(db: Session, iata: str | None = None) -> list[tuple[str, str, int]]: """`(iata, category, distinct product variants)` over the shops the site shows and the categories it shows, one airport's when `iata` is given; biggest first within an airport. No session (a sitemap built from stubs in a test) means no pairs, as `blocked_shop_ids` answers.""" if db is None or not hasattr(db, "execute"): return [] stmt = ( select(Shop.iata, ProductVariant.category, func.count(distinct(ProductVariant.id))) .join(Listing, Listing.shop_id == Shop.id) .join(ProductVariant, ProductVariant.id == Listing.variant_id) .where( Shop.is_catalogue_only.is_(False), Shop.iata.isnot(None), ProductVariant.category.isnot(None), publishable(db), shown_category(), ) .group_by(Shop.iata, ProductVariant.category) .order_by(Shop.iata, func.count(distinct(ProductVariant.id)).desc(), ProductVariant.category) ) if iata: stmt = stmt.where(Shop.iata == iata.upper()) return [(code, category, int(n)) for code, category, n in db.execute(stmt).all()] def category_page_links(db: Session, iata: str, airport_path_: str | None = None) -> list[CategoryPageLink]: """The category pages one airport has, biggest first (the airport page's rail).""" from app.services import coverage # lazy: coverage imports this module if airport_path_ is None: shops = airport_shops(db, iata) if not shops: return [] airport_path_ = _airport_summary(db, shops, _airport_shop_views(db, shops)).path return [ CategoryPageLink(category=category, slug=category_slug(category), path=airport_category_path(airport_path_, category), count=n) for _, category, n in coverage.qualifying_pairs(db, iata) if category_slug(category) ] def category_page_rows(db: Session) -> list[tuple[str, datetime | None]]: """(path, lastmod) for every pairing page that exists, for the sitemap.""" from app.services import coverage # lazy: coverage imports this module airports = {a.iata: a for a in list_airports(db)} return [ (airport_category_path(airports[code].path, category), airports[code].last_collected_at) for code, category, _ in coverage.qualifying_pairs(db) if code in airports and category_slug(category) ] def airport_category_detail( db: Session, iata: str, category: str, *, multi_only: bool = False, awarded_only: bool = False, exclusives_only: bool = False, sort: str = "featured", limit: int = 24, offset: int = 0, sample: int = 4, ) -> AirportCategoryDetail | None: """Everything the pairing page shows, or None where the airport is hidden, the category has no address, or the pair is under the coverage bar (a 404, never a thin page).""" from app.services import coverage # lazy: coverage imports this module shops = airport_shops(db, iata) if not shops or not category_slug(category): return None counts = {c: n for _, c, n in category_pair_counts(db, iata)} if not coverage.category_at_airport_on() or counts.get(category, 0) < coverage.CATEGORY_AT_AIRPORT_MIN_PRODUCTS: return None ids = [loc.id for loc in shops] shop_views = _airport_shop_views(db, shops) summary = _airport_summary(db, shops, shop_views) sort = sort if sort in SORTS else "featured" rows = [r for r in _airport_rows(db, ids) if r.category == category] winners = 0 for r in rows: if r.airport_count < 2 or r.here_usd is None or r.cheapest_usd is None: continue if float(r.here_usd) <= float(r.cheapest_usd) + 0.005 and float(r.dearest_usd or r.here_usd) - float(r.here_usd) > 0.5: winners += 1 _, exclusive_items = list_product_variants( db, stocked_at=ids, category=category, exclusives_only=True, sort="featured", limit=sample ) total, items = list_product_variants( db, stocked_at=ids, category=category, multi_only=multi_only, awarded_only=awarded_only, exclusives_only=exclusives_only, sort=sort, limit=limit, offset=offset, ) siblings = [link for link in category_page_links(db, iata, summary.path) if link.category != category] return AirportCategoryDetail( # The airport's identity, then the pair's own address and the category's count: the # first rehearsal on a copy kept the airport's path here and the canonical pointed at # the airport page (tests/test_airport_category.py pins the path). **{**summary.model_dump(), "path": airport_category_path(summary.path, category), "product_variants": counts[category]}, shops=shop_views, category=category, category_slug=category_slug(category), airport_path=summary.path, comparable=sum(1 for r in rows if r.airport_count > 1), cheapest_here=winners, exclusives=sum(1 for r in rows if r.is_exclusive), exclusive_items=exclusive_items, siblings=siblings, multi_only=multi_only, awarded_only=awarded_only, exclusives_only=exclusives_only, sort=sort, total=total, limit=limit, offset=offset, items=items, ) # --- brands ----------------------------------------------------------------- # # A brand page is keyed by the brands table's slug (migration #3): the fold row # plus any alias rows pointing at it through alias_of_id. Every brand has a page from the # moment its row exists (plan W18); only a brand a person hid has none. The review's floor # (Adam, 10 Sep: "three or more where stocked at two or more airports") once decided # existence; it now decides only whether the page is SUGGESTED for indexing # (`publish.brand_candidates`): a thin page stays out of search by `noindex, follow`, not by # not existing. Read that way because the page is a comparison: three bottles at three # airports beat five at one. One definition of the floor, `_brand_counts`. BRAND_PAGE_MIN_PRODUCTS = 3 BRAND_PAGE_MIN_AIRPORTS = 2 def hidden_brand_ids(db) -> list[int]: """The brands a person HID from the site (the review gate, the catalogue decisions §2.8: `overrides` entity `brand`, field `review`, value `hidden`), read once per session like `blocked_shop_ids`. Mechanical eligibility stays the first gate; this is the second. A stub session without a bind (the pure-logic suites) hides none.""" if db is None or not hasattr(db, "get_bind"): return [] info = getattr(db, "info", None) cache = info if isinstance(info, dict) else None if cache is not None and "hidden_brand_ids" in cache: return cache["hidden_brand_ids"] out = sorted(overrides.hidden_ids(db, "brand")) if cache is not None: cache["hidden_brand_ids"] = out return out def brand_by_slug(db: Session, slug: str) -> Brand | None: """The canonical brand row for a slug: an alias resolves to its brand_of, followed to the end of the chain so an old address answers ONE 301 (plan W18: redirects are flat). On the staging copy an alias of an alias answered two hops, `/brands/glen-moray` to its first brand and only then to the second.""" brand = db.scalar(select(Brand).where(Brand.slug == slug)) hops = 0 while brand is not None and brand.alias_of_id and brand.alias_of_id != brand.id and hops < MERGE_CHAIN_LIMIT: target = db.get(Brand, brand.alias_of_id) if target is None: break brand, hops = target, hops + 1 return brand def _brand_ids(db: Session, brand: Brand) -> list[int]: aliases = db.scalars(select(Brand.id).where(Brand.alias_of_id == brand.id)).all() return [brand.id, *[a for a in aliases if a != brand.id]] def _brand_rows(db: Session, brand_ids: list[int]): """The brand's product variants at visible airports: each product's latest priced observation per airport shop, with the shop's identity.""" latest = _latest_observation_subquery() return db.execute( select( ProductVariant.id, ProductVariant.category, ProductVariant.is_exclusive, Shop.id, Shop.iata, Shop.city, Shop.name, func.coalesce( func.min(PriceObservation.price_usd).filter(_available()), func.min(PriceObservation.price_usd), ), func.max(PriceObservation.observed_at), ) .join(Listing, Listing.variant_id == ProductVariant.id) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where( ProductVariant.brand_id.in_(brand_ids), shown_category(), PriceObservation.id.in_(select(latest.c.obs_id)), Shop.is_catalogue_only.is_(False), Shop.iata.isnot(None), publishable(db), ) .group_by(ProductVariant.id, Shop.id) ).all() def _brand_summary(brand: Brand, rows) -> BrandSummary: return BrandSummary( slug=brand.slug, path=brand_page_path(brand.slug), name=brand.name, product_variants=len({r[0] for r in rows}), last_collected_at=max((r[8] for r in rows if r[8]), default=None), ) def _brand_counts( db: Session, *, min_product_variants: int = BRAND_PAGE_MIN_PRODUCTS, min_airports: int = BRAND_PAGE_MIN_AIRPORTS, brand_ids: list[int] | None = None, ): """The standard brands (canonical brand rows) that meet the floor for a page: (brand id, product variants, last observed). The one definition of "this brand has a page": `min_product_variants` published product variants priced, between them, at `min_airports` distinct airports (by IATA, so two shops at one airport are one airport), and not hidden by a person (`hidden_brand_ids`). `brand_ids` narrows it to the brands a list of cards names.""" brand_of = func.coalesce(Brand.alias_of_id, Brand.id) stmt = ( select(brand_of.label("brand_of"), func.count(distinct(ProductVariant.id)), func.max(PriceObservation.observed_at)) .join(ProductVariant, ProductVariant.brand_id == Brand.id) .join(Listing, Listing.variant_id == ProductVariant.id) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(publishable(db), shown_category(), Shop.is_catalogue_only.is_(False)) .group_by("brand_of") .having( func.count(distinct(ProductVariant.id)) >= min_product_variants, places.unit_count() >= min_airports, ) ) if brand_ids is not None: stmt = stmt.where(brand_of.in_(brand_ids)) hidden = hidden_brand_ids(db) if hidden: stmt = stmt.where(brand_of.not_in(hidden)) return db.execute(stmt).all() def brand_page_slugs(db: Session, brand_ids: list[int | None]) -> dict[int, str]: """For each product `brand_id` (a fold row or an alias), the slug of its brand_of's page, when the brand_of has one. A product page links its brand name to `/brands/` only through this. Every brand has a page from discovery (plan W18), so every brand gets its slug; only a brand a person hid gets none, and its name links to a catalogue search instead. Before W18 the floor decided existence, and a blind slug would have put a 404 link on about a quarter of the visible product pages: the reason links still come from here and nowhere else. """ ids = sorted({i for i in brand_ids if i}) if not ids: return {} brand_of = aliased(Brand) rows = db.execute( select(Brand.id, brand_of.id, brand_of.slug) .join(brand_of, brand_of.id == func.coalesce(Brand.alias_of_id, Brand.id)) .where(Brand.id.in_(ids)) ).all() brands = {brand_id: (brand_of_id, slug) for brand_id, brand_of_id, slug in rows} if not brands: return {} hidden = set(hidden_brand_ids(db)) return {brand_id: slug for brand_id, (brand_of_id, slug) in brands.items() if brand_of_id not in hidden} def list_brands( db: Session, min_product_variants: int = BRAND_PAGE_MIN_PRODUCTS, min_airports: int = BRAND_PAGE_MIN_AIRPORTS ) -> list[BrandSummary]: """The brands index: canonical rows over the floor, not hidden. The index is a curated hub, not the rule of existence (plan W18): a brand under the floor still has its page, linked from its product variants (`brand_page_slugs`); on the 17 Sep staging copy the floor lists 364 brands where every brand with a priced variant would list 1,498, most of them unreviewed spellings. The sitemap lists only the indexed ones (`seo.sitemap_entries`).""" counted = _brand_counts(db, min_product_variants=min_product_variants, min_airports=min_airports) if not counted: return [] brand_ids = [h for h, _, _ in counted] brands = {b.id: b for b in db.scalars(select(Brand).where(Brand.id.in_(brand_ids)))} families = _brand_families(db, brand_ids) out = [ BrandSummary(slug=brands[h].slug, path=brand_page_path(brands[h].slug), name=brands[h].name, product_variants=n, last_collected_at=last, family=families.get(h)) for h, n, last in counted if h in brands ] return sorted(out, key=lambda b: b.slug) def _brand_families(db: Session, brand_ids: list[int]) -> dict[int, str | None]: """Each brand_of's family, from where its visible product variants sit (the same listings `_brand_counts` counts), so a menu lists a whisky brand_of under Drinks and a fragrance brand_of under Beauty without a typed list of either.""" if not brand_ids: return {} brand_of = func.coalesce(Brand.alias_of_id, Brand.id) rows = db.execute( select(brand_of.label("brand_of"), ProductVariant.category, func.count(distinct(ProductVariant.id))) .join(ProductVariant, ProductVariant.brand_id == Brand.id) .join(Listing, Listing.variant_id == ProductVariant.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(publishable(db), shown_category(), Shop.is_catalogue_only.is_(False), brand_of.in_(brand_ids)) .group_by("brand_of", ProductVariant.category) ).all() per_brand: dict[int, dict[str | None, int]] = {} for h, category, n in rows: per_brand.setdefault(h, {})[category] = n return {h: taxonomy.majority_family(counts) for h, counts in per_brand.items()} def brand_detail( db: Session, slug: str, *, category: str | None = None, sort: str = "featured", limit: int = 24, offset: int = 0, min_product_variants: int = 0, min_airports: int = 0, ) -> BrandDetail | None: """Everything the brand page shows, or None when the slug is unknown or a person hid the brand (the page route answers 302 for that, `publish.hidden_forward`). A brand has a page from the moment its row exists (plan W18): the floor (`_brand_counts`) no longer decides existence, only whether the page is SUGGESTED for indexing (`publish.brand_candidates`). `min_product_variants` / `min_airports` stay for a caller that wants the old gate.""" from app.services import publish brand = brand_by_slug(db, slug) if brand is None or publish.is_hidden(brand): return None if (min_product_variants, min_airports) != (0, 0) and not _brand_counts( db, min_product_variants=min_product_variants, min_airports=min_airports, brand_ids=[brand.id]): return None ids = _brand_ids(db, brand) rows = _brand_rows(db, ids) summary = _brand_summary(brand, rows) sort = sort if sort in SORTS else "featured" # Per product: its price at each airport; per airport: how many of the # brand's bottles it stocks, and for how many comparable ones it is the # cheapest place we know. by_product: dict[int, dict[int, float]] = {} airports: dict[int, dict] = {} for pid, category_name, is_exclusive, loc_id, iata, city, name, usd, last in rows: entry = airports.setdefault(loc_id, {"iata": iata, "city": city, "name": name, "product_variants": set(), "cheapest_for": 0, "last": None}) entry["product_variants"].add(pid) if last and (entry["last"] is None or last > entry["last"]): entry["last"] = last if usd is not None: by_product.setdefault(pid, {})[loc_id] = float(usd) comparable = 0 for pid, prices in by_product.items(): if len(prices) < 2: continue comparable += 1 low = min(prices.values()) for loc_id, usd in prices.items(): if usd <= low + 0.005: airports[loc_id]["cheapest_for"] += 1 # One row per IATA even if two shops share an airport. per_iata: dict[str, BrandAirport] = {} for loc_id, a in sorted(airports.items()): row = per_iata.get(a["iata"]) if row is None: per_iata[a["iata"]] = BrandAirport( iata=a["iata"], path=airport_path(a["iata"], a["city"], a["name"]), name=a["name"], product_variants=len(a["product_variants"]), cheapest_for=a["cheapest_for"], last_collected_at=a["last"], ) else: row.product_variants = max(row.product_variants, len(a["product_variants"])) row.cheapest_for += a["cheapest_for"] airport_list = sorted(per_iata.values(), key=lambda a: (-a.product_variants, -a.cheapest_for, a.iata)) categories = Counter(category_name for pid, category_name, *_ in {(r[0], r[1]) for r in rows} if category_name) exclusives = len({r[0] for r in rows if r[2]}) total, items = list_product_variants( db, brand_ids=ids, category=category, sort=sort, limit=limit, offset=offset ) return BrandDetail( **summary.model_dump(), airports=airport_list, comparable=comparable, exclusives=exclusives, categories=[CategoryCount(category=c, count=n) for c, n in categories.most_common()], category=category, sort=sort, total=total, limit=limit, offset=offset, items=items, ) def newest_product_variants(db: Session, limit: int = 50) -> list[tuple[ProductSummary, object]]: """The product variants most recently added to the catalogue at a visible airport, newest first, each with the moment it was first recorded (the feed's items).""" rows = db.execute( select(ProductVariant.id, ProductVariant.created_at) .where( ProductVariant.id.in_( select(Listing.variant_id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(publishable(db)) ) ) .order_by(ProductVariant.created_at.desc(), ProductVariant.id.desc()) .limit(limit) ).all() if not rows: return [] _, items = list_product_variants(db, only_ids=[pid for pid, _ in rows], limit=len(rows)) by_id = {item.id: item for item in items} return [(by_id[pid], created) for pid, created in rows if pid in by_id] def dataset_facts(db: Session) -> DatasetFacts: """The public data page's figures: visible airports only, like /api/stats.""" visible_listing = ( select(Listing.id, Listing.variant_id, Listing.shop_id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(publishable(db)) .subquery() ) obs = select(PriceObservation).where( PriceObservation.listing_id.in_(select(visible_listing.c.id)) ).subquery() first, last, observations, currencies = db.execute( select( func.min(obs.c.observed_at), func.max(obs.c.observed_at), func.count(obs.c.id), func.count(distinct(obs.c.currency)), ) ).one() product_variants = select(func.count(distinct(visible_listing.c.variant_id))) return DatasetFacts( observations=observations or 0, product_variants=db.scalar(product_variants) or 0, airports=db.scalar( places.unit_join(select(places.unit_count()).select_from(Shop)).where( Shop.id.in_(select(visible_listing.c.shop_id)), Shop.is_catalogue_only.is_(False), ) ) or 0, retailers=db.scalar( select(func.count(distinct(Shop.retailer_id))).where( Shop.id.in_(select(visible_listing.c.shop_id)) ) ) or 0, currencies=currencies or 0, with_barcode=db.scalar( select(func.count(ProductVariant.id)).where( ProductVariant.id.in_(select(visible_listing.c.variant_id)), ProductVariant.gtin.isnot(None) ) ) or 0, awards=db.scalar( select(func.count(Award.id)).where( Award.variant_id.in_(select(visible_listing.c.variant_id)) ) ) or 0, first_observed_at=first, last_observed_at=last, airport_list=list_airports(db), categories=category_counts(db), ) def category_counts(db: Session) -> list[CategoryCount]: # Counted through visible listings so the numbers match what browsing that # category actually shows. Each count carries its family (taxonomy's vertical # and the shopper's word for it) so the mega menu groups shelves without a # copy of the taxonomy in the SPA; a category outside the vocabulary has none. rows = db.execute( select(ProductVariant.category, func.count(func.distinct(ProductVariant.id))) .join(Listing, Listing.variant_id == ProductVariant.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(ProductVariant.category.isnot(None), publishable(db), shown_category()) .group_by(ProductVariant.category) .order_by(func.count(func.distinct(ProductVariant.id)).desc()) ).all() return [ CategoryCount( category=c, count=n, family=taxonomy.vertical_of(c), family_label=taxonomy.family_label(c), ) for c, n in rows ]