"""Shops running Gebr. Heinemann's storefront platform. Several duty-free operations run on the same commerce platform with an identical catalogue API — Heinemann's own global catalogue, Keflavik, Sydney, and the Danish border shop among them. Rather than a near-duplicate module per shop, this is one collector parameterised by site, so adding another is a few lines of configuration. Each shop is a separate business on its own domain with its own robots.txt, and each was checked before being added here. Heinemann's own per-airport views on heinemann-shop.com are deliberately NOT included: that site asks crawlers not to take per-airport prices, so we take only its global catalogue and would rather ask than work around it. """ from collections.abc import Iterator from dataclasses import dataclass, field from urllib.parse import urlencode from app.services.collectors.base import Collector, LocationSpec, RawListing, facts_only, report_skip from app.services.collectors.fetch import SourceBlocked, fetch_json from app.services.collectors.robots import check_allowed from app.services.normalize import clean_gtin, parse_abv, parse_size_ml SEARCH_PATH = "/en/global/search/results" PARSER_VERSION = "heinemann-platform/2026-09-04" _VOLUME_UNITS = {"LTR": 1000.0, "L": 1000.0, "MLT": 1.0, "ML": 1.0, "CLT": 10.0} # The feed mixes real categories with packaging words; packaging is dropped # rather than shown to a shopper as a category. _PACKAGING_WORDS = { "bottle", "gift box", "gift set", "gift pack", "tin", "tube", "bag in box", "carton", "case", "can", "miniature", "multipack", "twinpack", } @dataclass(frozen=True, slots=True) class HeinemannSite: """One shop on the platform.""" slug: str base_url: str retailer_slug: str retailer_name: str location: LocationSpec # Alcohol category ids. Each site prefixes its own, e.g. auscat_/scacat_. category_ids: tuple[str, ...] = field(default=()) price_type: str = "list" def _category(item: dict) -> str | None: raw = (item.get("typeOfGoods") or "").strip() if not raw or raw.lower() in _PACKAGING_WORDS: return None return raw def _size_from_content_unit(item: dict) -> int | None: content = item.get("contentUnit") or {} unit = ((content.get("unit") or {}).get("code") or "").upper() quantity = content.get("quantity") factor = _VOLUME_UNITS.get(unit) if factor and isinstance(quantity, int | float) and quantity > 0: millilitres = round(float(quantity) * factor) if 10 <= millilitres <= 20000: return millilitres return None class HeinemannPlatformCollector(Collector): parser_version = PARSER_VERSION def __init__(self, site: HeinemannSite) -> None: self.site = site self.slug = site.slug self.retailer_slug = site.retailer_slug self.retailer_name = site.retailer_name self.operator = "Gebr. Heinemann" self.homepage = site.base_url def locations(self) -> list[LocationSpec]: return [self.site.location] def _page(self, category_id: str, page: int, delay: float) -> tuple[list[dict], int]: query = urlencode({"q": f":relevance:allCategories:{category_id}", "page": page}) payload = fetch_json(f"{self.site.base_url}{SEARCH_PATH}?{query}", delay=delay) if not isinstance(payload, dict): return [], 0 pagination = payload.get("pagination") or {} return payload.get("results") or [], int(pagination.get("numberOfPages") or 0) def read_one(self, listing) -> RawListing | None: """No per-listing read: every shop on this platform disallows the search endpoint we would need, and we do not read product pages.""" raise SourceBlocked(f"{self.slug}: robots.txt disallows {SEARCH_PATH}; no read path") def collect(self, *, limit: int | None = None, delay: float = 1.0) -> Iterator[RawListing]: # Each shop is its own domain with its own robots.txt, and permission # can be withdrawn between runs -- the shared global catalogue's was. delay = check_allowed(self.site.base_url, [SEARCH_PATH]).delay_for(delay) produced = 0 # Querying a parent category also returns its children's products, so # results are de-duplicated on the product code. seen: set[str] = set() first = True for category_id in self.site.category_ids: page, total_pages = 0, 1 while page < total_pages: results, total_pages = self._page(category_id, page, 0.0 if first else delay) first = False if not results: break for item in results: code = item.get("code") if not code or code in seen: continue seen.add(str(code)) price_block = item.get("price") or {} price = price_block.get("value") name = (item.get("name") or "").strip() if price is None or not name or float(price) <= 0: report_skip("no_price", source_sku=str(code)) continue strikethrough = (item.get("strikethroughPrice") or {}).get("value") stock_code = ((item.get("stock") or {}).get("stockLevelStatus") or {}).get( "code" ) url = item.get("url") yield RawListing( source_sku=str(code), name=name, brand=item.get("brand") or item.get("manufacturerName") or None, gtin=clean_gtin(item.get("gtin") or item.get("ean")), price=float(price), currency=price_block.get("currencyIso") or self.site.location.currency, was_price=float(strikethrough) if strikethrough else None, location_code=self.site.location.code, size_ml=_size_from_content_unit(item) or parse_size_ml(name), abv=( float(item["alcoholByVolume"]) if item.get("alcoholByVolume") else parse_abv(name) ), feed_categories=[c for c in (_category(item),) if c], is_exclusive=bool(item.get("travelRetailExclusive")), url=f"{self.site.base_url}{url}" if url else None, in_stock=(stock_code == "inStock") if stock_code else None, price_type=self.site.price_type, raw=facts_only(item), vertical="liquor", ) produced += 1 if limit and produced >= limit: return page += 1 SITES: tuple[HeinemannSite, ...] = ( HeinemannSite( slug="iceland-duty-free", base_url="https://www.islanddutyfree.is", retailer_slug="heinemann-iceland", retailer_name="Iceland Duty Free", location=LocationSpec( code="KEF", iata="KEF", name="Keflavik International", city="Reykjavik", country="Iceland", currency="ISK", ), # No single "all alcohol" node here -- the drinks node also holds soft # drinks -- so the four alcohol subtrees are queried directly. category_ids=("kefcat_5150", "kefcat_5010", "kefcat_1100", "kefcat_5500"), ), HeinemannSite( slug="sydney-duty-free", base_url="https://www.heinemanndutyfree.com.au", retailer_slug="heinemann-sydney", retailer_name="Heinemann Duty Free Sydney", location=LocationSpec( code="SYD", iata="SYD", name="Sydney Kingsford Smith", city="Sydney", country="Australia", currency="AUD", ), category_ids=("auscat_5000",), ), HeinemannSite( slug="bordershop-scandinavia", base_url="https://www.bordershop.com", retailer_slug="bordershop", retailer_name="BorderShop Puttgarden", location=LocationSpec( code="SCA", iata=None, name="BorderShop (Denmark/Germany)", city="Puttgarden", country="Germany", currency="DKK", ), # Spirits, wine, beer. category_ids=("scacat_5300", "scacat_5200", "scacat_5100"), ), ) def heinemann_platform_collectors() -> list[HeinemannPlatformCollector]: return [HeinemannPlatformCollector(site) for site in SITES]