"""Duty-free shops running on Shopify. Several airport operators run their storefronts on Shopify, which publishes a complete product feed at /products.json. One adapter therefore covers several airports at once, and adding another Shopify shop is a line of configuration rather than new code. """ import re from collections.abc import Iterator from dataclasses import dataclass from app.services.collectors.base import ( Collector, ShopSpec, RawListing, facts_only, gone, listing_ref, quantity_options, report_skip, ) from app.services.collectors.fetch import FetchError, SourceBlocked, fetch_json from app.services.collectors.robots import check_allowed from app.services.normalize import clean_gtin, gtin_from_sku, parse_abv, parse_quantity_ml from app.services.taxonomy import vertical_from_hints PARSER_VERSION = "shopify/2026-09-17" PAGE_SIZE = 250 MAX_PAGES = 12 FEED_PATH = "/products.json" # Which tree a product hangs in, judged on the shop's own shelf names (product_type # and tags), never the title. The shelves are English at one shop and Spanish at # three: "Licores" is the whole drinks shelf at the Attenza shops, and until it # was listed here only 181 of Panama's 395 drinks passed (issue, 2026-09-03). # Word-start matching, so "ron" (rum) cannot fire inside "electrónica". _SHELF_RULES: list[tuple[str, re.Pattern[str]]] = [ ("liquor", re.compile( r"\b(spirits?|liquor|liqueurs?|licor(?:es)?|whisk|wines?|vinos?|champa|cavas?|" r"prosecco|espumantes?|sparkling|vodkas?|gin\b|ginebras?|rum\b|rums\b|ron\b|rones\b|" r"tequilas?|mezcal|cognac|brandy|brandies|armagnac|calvados|pisco|grappa|beers?|" r"cervezas?|ciders?|sidra|aperitif|aperitivo|cordial(?:es)?|vermouth|vermut|amaro|" r"sake|soju|baijiu|porto?\b|sherry|jerez|icewine|eau de vie|pineau|anise|anis\b|" r"aguardiente|reposado|a[nñ]ejo|blanco tequila|bitters?)", re.I)), ("beauty", re.compile( r"\b(perfum|parfum|cologne|fragan|fragr|eau de|edp|edt|unisex|body mist|brumas?|" r"skin ?care|rostro|cuerpo|capilar|cabello|labios|ojos|cremas?|sueros?|serums?|" r"t[oó]nicos?|lociones|lotions?|exfoliantes?|mascarillas?|desmaquillantes?|" r"limpiador|cleansers?|sunscreen|protecci[oó]n|after sun|u[nñ]as|nails?|" r"maquillaje|make-?up|cosm[eé]tic|beauty|belleza|cuidado|shampoos?|conditioners?|" r"styling|hair|face|body|hand|mano\b|tratamientos?)", re.I)), ] # A shop's own "not on the shelf" markers: hidden by an automation, switched off. _NOT_OFFERED_TAG_RE = re.compile(r"^(auto-oculto|apagar|oculto|hidden)\b", re.I) def shelf_vertical(product: dict) -> str | None: """liquor | beauty from product_type and tags; None means out of our scope.""" tags = product.get("tags") tag_text = " ".join(tags) if isinstance(tags, list) else str(tags or "") haystack = f"{product.get('product_type') or ''} {tag_text}" for vertical, pattern in _SHELF_RULES: if pattern.search(haystack): return vertical return None def published_options(product: dict, variant: dict) -> list[tuple[str, str]]: """`(the shop's option name, this variant's value)` from `product.options[n].name` and the variant's `option1..3`, in the shop's order; the placeholder single option ("Title" / "Default Title") is no option. Used by the collector and by `backfill options` over stored fragments, so both read a fragment the same way.""" names = {} for index, option in enumerate(product.get("options") or [], start=1): if isinstance(option, dict): names[int(option.get("position") or index)] = str(option.get("name") or "").strip() values = {position: variant.get(f"option{position}") for position in (1, 2, 3)} if not any(values.values()): # A fragment that carries only the variant's title (the platform writes it as the option # values joined with " / "): the same fields, read from there. parts = [part.strip() for part in str(variant.get("title") or "").split(" / ")] values = {position: part for position, part in enumerate(parts, start=1) if part} out: list[tuple[str, str]] = [] for position in (1, 2, 3): value = values.get(position) if value is None or not str(value).strip() or str(value).strip().lower() == "default title": continue out.append((names.get(position) or f"option{position}", str(value).strip())) return out def not_offered(product: dict) -> bool: tags = product.get("tags") tags = tags if isinstance(tags, list) else [t.strip() for t in str(tags or "").split(",")] return any(_NOT_OFFERED_TAG_RE.match(t) for t in tags) @dataclass(frozen=True, slots=True) class ShopifyShop: code: str name: str base_url: str currency: str iata: str | None = None city: str | None = None country: str | None = None class ShopifyCollector(Collector): """One Shopify storefront. Instances are registered per shop.""" parser_version = PARSER_VERSION def __init__(self, shop: ShopifyShop, retailer_slug: str, retailer_name: str, operator: str | None = None) -> None: self.shop = shop self.slug = f"shopify-{shop.code.lower()}" self.retailer_slug = retailer_slug self.retailer_name = retailer_name self.operator = operator or retailer_name self.homepage = shop.base_url def shops(self) -> list[ShopSpec]: return [ ShopSpec( code=self.shop.code, iata=self.shop.iata, name=self.shop.name, city=self.shop.city, country=self.shop.country, currency=self.shop.currency, ) ] def listing(self, product: dict, variant: dict) -> RawListing | None: """One variant of one product as a listing; None when it has no price.""" title = (product.get("title") or "").strip() if not title: return None price = variant.get("price") if price in (None, "", "0.00"): return None try: price_value = float(price) except (TypeError, ValueError): return None if price_value <= 0: return None # The shop's options as the fields they are (identity rules v6): the name is the product # title alone. The variant title was glued on until 2026-09-17, and a name rule then had # to read the shade back off a " / " tail. options = published_options(product, variant) measured = " ".join([title, *quantity_options(options)]) was = variant.get("compare_at_price") handle = product.get("handle") return RawListing( source_sku=str(variant.get("id")), name=title, options=options, brand=(product.get("vendor") or None), gtin=clean_gtin(variant.get("barcode")) or gtin_from_sku(variant.get("sku")), price=price_value, currency=self.shop.currency, was_price=float(was) if was else None, shop_code=self.shop.code, quantity_ml=parse_quantity_ml(measured), abv=parse_abv(measured), feed_categories=[c for c in (product.get("product_type"),) if c], url=f"{self.shop.base_url}/products/{handle}" if handle else None, in_stock=bool(variant.get("available")), # The product minus its variants (each variant is its own record) # and minus body_html/images, which are expression, not facts. raw={ "product": facts_only({k: v for k, v in product.items() if k != "variants"}), "variant": facts_only(variant), }, # The shop's shelf decides the family; our category hints only # confirm it, because a Spanish shelf name says nothing to them. vertical=shelf_vertical(product) or vertical_from_hints( str(product.get("product_type") or "") ), ) def listing_from_product_json(self, listing, payload: object) -> RawListing | None: """The published variant as /products/.json describes it now, offline.""" ref = listing_ref(listing) product = payload.get("product") if isinstance(payload, dict) else None if not isinstance(product, dict): return None for variant in product.get("variants") or []: if str(variant.get("id")) == ref.source_sku: return self.listing(product, variant) return None def read_one(self, listing) -> RawListing | None: ref = listing_ref(listing) handle = ref.url.rstrip("/").rsplit("/", 1)[-1] if ref.url else "" if not handle: return None robots = check_allowed(self.shop.base_url, [FEED_PATH], fresh=False) url = f"{self.shop.base_url}/products/{handle}.json" if not robots.allows(url): raise SourceBlocked(f"{self.shop.code}: robots.txt disallows {url}") try: payload = fetch_json(url, delay=robots.delay_for(1.0)) except FetchError as exc: if gone(exc): return None raise return self.listing_from_product_json(ref, payload) def collect(self, *, limit: int | None = None, delay: float = 1.0) -> Iterator[RawListing]: robots = check_allowed(self.shop.base_url, [FEED_PATH]) delay = robots.delay_for(delay) produced = 0 for page in range(1, MAX_PAGES + 1): url = f"{self.shop.base_url}{FEED_PATH}?limit={PAGE_SIZE}&page={page}" if not robots.allows(url): return payload = fetch_json(url, delay=delay if page > 1 else 0.0) product_variants = payload.get("products", []) if isinstance(payload, dict) else [] if not product_variants: return for product in product_variants: if shelf_vertical(product) is None: continue if not_offered(product): report_skip( "not_offered", source_sku=str(product.get("id")), url=f"{self.shop.base_url}/products/{product.get('handle')}", ) continue for variant in product.get("variants") or []: listing = self.listing(product, variant) if listing is None: report_skip( "no_price", source_sku=str(variant.get("id")), url=f"{self.shop.base_url}/products/{product.get('handle')}", ) continue yield listing produced += 1 if limit and produced >= limit: return SHOPS: list[tuple[ShopifyShop, str, str, str]] = [ ( ShopifyShop("YUL", "Montreal-Trudeau", "https://www.montrealdutyfree.ca", "CAD", iata="YUL", city="Montreal", country="Canada"), "ari", "Aer Rianta International", "ARI", ), ( ShopifyShop("PTY", "Panama Tocumen", "https://pa.attenza.net", "USD", iata="PTY", city="Panama City", country="Panama"), "attenza", "Attenza Duty Free", "Motta Internacional", ), ( ShopifyShop("BOG", "Bogota El Dorado", "https://co.attenza.net", "USD", iata="BOG", city="Bogota", country="Colombia"), "attenza", "Attenza Duty Free", "Motta Internacional", ), ( ShopifyShop("SAL", "San Salvador", "https://es.attenza.net", "USD", iata="SAL", city="San Salvador", country="El Salvador"), "attenza", "Attenza Duty Free", "Motta Internacional", ), ] def shopify_collectors() -> list[ShopifyCollector]: return [ShopifyCollector(shop, slug, name, op) for shop, slug, name, op in SHOPS]