"""The Shilla Duty Free, Seoul Incheon's online store, read through the browser sidecar.
What the probe established (2026-09-05, `.logs/runs/shilla-probe-2026-09-05.log`):
* **A text client is challenged; a browser is not.** Every product page answered
HTTP 403 with a Cloudflare challenge to the plain fetch, and HTTP 200 with the
full document to the sidecar under the same declared identity. Cloudflare
was judging the client, not the name. The browser is therefore the honest
client for this host, and the page is read as any browser would receive it.
* **The product page is server-rendered.** Name, brand, three USD price tiers,
a UPC in "REF.NO", size, stock and the category tree are all in the HTML
before any script runs. Nothing on the page is clicked.
* **Its scripts and imagery are robots-disallowed** (`/estore/_ui/`, `/medias/`,
`/files/` for `*`), so the sidecar refuses every request for them and the
document is read at `domcontentloaded`. That also means the category grids,
which those scripts draw, are not readable by a compliant bot, and their
loader is not ours to read either. Discovery is what the server renders:
the home page's product links and each product page's related-product
rail, walked breadth-first under the run's render budget.
* **Three price tiers.** "Price" (list), "Discount price" and "Price for online
member". Decision (`.logs/decisions-for-rian.md` #12, assumed): the discount
price is what a walk-up traveller pays and is published as `price_type`
"discount", with the list price as the crossed-out figure; the member tier
is recorded in raw and never published.
* **Stock is a visibility, not a text.** "Temporarily Sold Out" is in every
page's markup, hidden with `display: none` when the product is in stock.
Shop `ICN`, currency USD as the page displays it (`prdPriceDollar`).
"""
import html as html_lib
import logging
import re
from collections import deque
from collections.abc import Iterator
from app.services.collectors.base import (
Collector,
ShopSpec,
RawListing,
gone,
listing_ref,
report_skip,
)
from app.services.collectors.fetch import FetchError, SourceBlocked, fetch, render, render_budget
from app.services.collectors.robots import check_allowed
from app.services.normalize import clean_gtin, looks_exclusive, parse_abv, parse_quantity_ml
from app.services.quantity import parse_quantity
from app.services.taxonomy import vertical_from_hints
logger = logging.getLogger(__name__)
PARSER_VERSION = "shilla/2026-09-05"
BASE = "https://www.shilladfs.com"
HOME = "/estore/kr/en/?uiel=Desktop"
PRODUCT_PATH = "/estore/kr/en/p/"
PRICE_TYPE = "discount"
# The top-level categories the walk follows rails from; a page outside them is
# still recorded if it was reached, but its rail is not followed.
WALK_CATEGORIES = ("liquor", "skin care", "makeup", "fragrance/body/hair")
# Renders per run. At ten seconds a page (their 5 s, our floor) this is about
# an hour; the walk stops here whatever it has left in its queue.
RENDER_CAP = 200
_PRODUCT_URL_RE = re.compile(r"^https://www\.shilladfs\.com/estore/kr/en/p/(\d+)$")
_PRODUCT_LINK_RE = re.compile(r"/estore/kr/en/p/(\d+)")
_HIDDEN_INPUT_RE = re.compile(r'\s*(.*?)(.*?)\s*', re.S)
_REF_RE = re.compile(r'class="number pd-no">\s*REF\.NO\s*([0-9A-Za-z-]*)\s*/\s*SKU\.NO\s*([0-9A-Za-z-]*)')
_LIST_PRICE_RE = re.compile(r'
', re.S)
_PRODUCT_PAGE_MARKER = "pageType-ProductPage"
def product_urls_in(html: str) -> list[str]:
"""Every product link on a server-rendered page, in order, de-duplicated."""
seen: set[str] = set()
out: list[str] = []
for code in _PRODUCT_LINK_RE.findall(html):
if code not in seen:
seen.add(code)
out.append(f"{BASE}{PRODUCT_PATH}{code}")
return out
def product_code(url: str) -> str | None:
match = _PRODUCT_URL_RE.match(url.split("?", 1)[0])
return match.group(1) if match else None
def hidden_inputs(html: str) -> dict[str, str]:
"""The page's own hidden fields: product code, category tree, dollar price."""
return {k: html_lib.unescape(v) for k, v in _HIDDEN_INPUT_RE.findall(html)}
def _money(text: str | None) -> float | None:
if not text:
return None
try:
return float(text.replace(",", ""))
except ValueError:
return None
def price_tiers(html: str) -> dict[str, float | None]:
"""List, discount and online-member prices as the page states them, in USD."""
return {
"list": _money(m.group(1)) if (m := _LIST_PRICE_RE.search(html)) else None,
"discount": _money(m.group(1)) if (m := _DISCOUNT_RE.search(html)) else None,
"member": _money(m.group(1)) if (m := _MEMBER_RE.search(html)) else None,
}
def in_stock_of(html: str) -> bool | None:
"""Sold out only when the sold-out block is actually shown."""
match = _SOLD_OUT_RE.search(html)
if match is None:
return None
return "display: none" in match.group(1) or "display:none" in match.group(1)
def spec_rows(html: str) -> dict[str, str]:
return {
html_lib.unescape(k).strip(): html_lib.unescape(v).strip()
for k, v in _SPEC_ROW_RE.findall(html)
}
def listing_from_page(html: str, url: str) -> RawListing | None:
"""One RawListing from a server-rendered product page, or None if it is not one."""
if _PRODUCT_PAGE_MARKER not in html[:4000] and _PRODUCT_PAGE_MARKER not in html:
return None
fields = hidden_inputs(html)
code = fields.get("productCode") or product_code(url) or ""
title = _TITLE_RE.search(html)
if not code or title is None:
return None
brand = html_lib.unescape(re.sub(r"<[^>]+>", "", title.group(1))).strip() or None
name = html_lib.unescape(re.sub(r"<[^>]+>", "", title.group(2))).strip()
if not name:
return None
tiers = price_tiers(html)
# The hidden field and the visible discount tier are the same figure; the
# field names the currency (Dollar) where the visible one only shows "$".
price = _money(fields.get("prdPriceDollar")) or tiers["discount"]
if price is None:
report_skip("no_price", source_sku=code, url=url)
return None
was_price = tiers["list"] if tiers["list"] and tiers["list"] > price else None
ref = _REF_RE.search(html)
gtin = clean_gtin(ref.group(1)) if ref else None
sku_no = ref.group(2) if ref else None
categories = [
fields[k] for k in ("1depthCategory", "2depthCategory", "3depthCategory") if fields.get(k)
]
specs = spec_rows(html)
volume = specs.get("Weight,Volume") or specs.get("Volume") or ""
quantity_ml = parse_quantity_ml(volume) or parse_quantity_ml(name)
# The spec row states its own quantity ("10G", "100ML"); only a figure it actually
# reads becomes the structured hint, never an unparsed or silent read.
quantity_text = volume or None
q = parse_quantity(volume) if volume else None
quantity = q if q is not None and q.state == "stated" else None
full_name = f"{brand} {name}".strip() if brand and not name.lower().startswith(brand.lower()) else name
return RawListing(
source_sku=code,
name=full_name,
brand=brand,
gtin=gtin,
price=price,
was_price=was_price,
currency="USD",
shop_code="ICN",
in_stock=in_stock_of(html),
quantity_ml=quantity_ml,
quantity=quantity,
quantity_text=quantity_text,
abv=parse_abv(specs.get("Alcohol content") or specs.get("Alcohol") or name),
feed_categories=categories,
url=url,
price_type=PRICE_TYPE,
is_exclusive=looks_exclusive(name),
raw={
"product_code": code,
"sku_no": sku_no,
"ref_no": ref.group(1) if ref else None,
"brand": brand,
"name": name,
"categories": categories,
"tiers": tiers,
"hidden": {k: v for k, v in fields.items() if k not in ("netFunnelDomain", "netFunnelPort")},
"specs": {k: v for k, v in specs.items() if k in ("Weight,Volume", "Volume", "Alcohol content", "Country of origin", "Manufacturer, Distributor")},
},
vertical=vertical_from_hints(*categories) if categories else None,
)
def follows_rail(listing: RawListing) -> bool:
"""Whether a page's related-product rail is worth walking."""
top = (listing.feed_categories[0] if listing.feed_categories else "").lower()
return top in WALK_CATEGORIES
class Shilla(Collector):
slug = "shilla-icn"
retailer_slug = "shilla"
retailer_name = "The Shilla Duty Free"
operator = "Hotel Shilla"
homepage = BASE
parser_version = PARSER_VERSION
# The source's own pace between renders (fetch.render_wait); None keeps the
# host's Crawl-delay 5 and the run's delay as they are. The first run (5 Sep,
# 20 renders) drew no refusal.
render_floor_seconds: float | None = None
def shops(self) -> list[ShopSpec]:
return [
ShopSpec(
code="ICN",
iata="ICN",
name="Seoul Incheon (Shilla online store)",
city="Seoul",
country="South Korea",
currency="USD",
)
]
def _render_page(self, url: str, robots, delay: float):
return render(
url,
delay=delay,
floor=self.render_floor_seconds,
robots=robots,
wait_until="domcontentloaded",
settle_ms=500,
)
def collect(self, *, limit: int | None = None, delay: float = 10.0) -> Iterator[RawListing]:
robots = check_allowed(BASE, [HOME, PRODUCT_PATH])
# The home page serves to a text client; it is the one page here that does.
home = fetch(f"{BASE}{HOME}", accept="text/html", delay=0).text
queue: deque[str] = deque(product_urls_in(home))
seen: set[str] = set(queue)
logger.info("shilla_seed home_products=%d", len(queue))
produced = 0
cap = RENDER_CAP if limit is None else min(RENDER_CAP, limit)
with render_budget(cap, label=self.slug) as budget:
while queue and budget.count < budget.cap:
url = queue.popleft()
try:
page = self._render_page(url, robots, delay)
except SourceBlocked:
raise
except FetchError as exc:
logger.warning("shilla_render_failed url=%s error=%s", url, exc)
continue
listing = listing_from_page(page.html, url)
if listing is None:
report_skip("not_a_product_page", url=url, text_length=page.text_length)
continue
if follows_rail(listing):
for related in product_urls_in(page.html):
if related not in seen:
seen.add(related)
queue.append(related)
yield listing
produced += 1
if limit is not None and produced >= limit:
return
if queue:
logger.info("shilla_walk_stopped budget=%d queued=%d", budget.cap, len(queue))
def read_one(self, listing) -> RawListing | None:
ref = listing_ref(listing)
url = ref.url or f"{BASE}{PRODUCT_PATH}{ref.source_sku}"
robots = check_allowed(BASE, [PRODUCT_PATH], fresh=False)
try:
page = self._render_page(url, robots, 10.0)
except FetchError as exc:
if gone(exc):
return None
raise
row = listing_from_page(page.html, url)
return row if row and row.source_sku == ref.source_sku else None