"""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 from app.services.quantity import Quantity 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} # Heinemann's contentUnit.unit.code, by canonical unit. Only "L" is attested in the # fixtures; the rest come from the platform's own code list (no fragment holds them yet, # so they are pinned by a table-driven test rather than a captured page). _CONTENT_UNIT_CODES: dict[str, tuple[str, float]] = { "MLT": ("ml", 1.0), "ML": ("ml", 1.0), "CLT": ("ml", 10.0), "LTR": ("ml", 1000.0), "L": ("ml", 1000.0), "GRM": ("g", 1.0), "G": ("g", 1.0), "KGM": ("g", 1000.0), "KG": ("g", 1000.0), "PCE": ("pcs", 1.0), "PC": ("pcs", 1.0), "ST": ("pcs", 1.0), "EA": ("pcs", 1.0), } def _quantity_from_content_unit(item: dict) -> tuple[Quantity | None, str | None]: """The shop's own contentUnit as a Quantity, and its size wording verbatim. A code outside the known table is still reported (state "unparsed") rather than dropped, so a new code shows up as a fact to add, not silence.""" content = item.get("contentUnit") or {} code = ((content.get("unit") or {}).get("code") or "").upper() quantity = content.get("quantity") if not code or not isinstance(quantity, int | float) or quantity <= 0: return None, None text = f"{quantity:g} {code}" spec = _CONTENT_UNIT_CODES.get(code) if spec is None: return Quantity(state="unparsed", detail=f"contentUnit:{code}"), text unit, factor = spec value = float(int(quantity)) if unit == "pcs" else float(quantity) * factor return Quantity(value=value, unit=unit, form="single", state="stated"), text # 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") quantity, quantity_text = _quantity_from_content_unit(item) 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), quantity=quantity, quantity_text=quantity_text, 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]