"""Server-rendered SEO layer.
The app is a client-rendered SPA, but every page's HTML leaves through FastAPI,
so the whole machine-readable layer -- title, description, canonical, social
tags and schema.org JSON-LD -- is stamped into the server-side before a
single line of JavaScript runs. Crawlers that never execute JS still see all of
it. Product pages also get their BODY rendered here (`product_body`): the same
H1, meta line, awards, verdict and price table the SPA draws, with the SPA's
own class names, so a client that runs no JavaScript reads the page and one
that does sees no change when React mounts over it. The rule is that the served
markup shows exactly what the SPA shows in its default state (no airports
chosen); anything else is cloaking. The detail object is also seeded into the
page as `window.__DFP_PRODUCT__`, so the SPA's first commit is the finished
page rather than its skeleton.
Everything injected here is built from untrusted strings (product names come
from retailer feeds), so escaping is not optional: HTML attributes are
entity-escaped, and JSON-LD forbids a raw "<" so a name containing
"" cannot break out of its tag.
"""
import html
import json
import re
from collections.abc import Collection
from datetime import UTC, datetime
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.config import settings
from app.models import Listing, Location, PriceObservation, Product
from app.models.editorial import ArticleOut
from app.models.hubs import AirportDetail, BrandDetail, DatasetFacts, AirportCategoryDetail
from app.models.schemas import AwardOut, PriceOut, ProductDetail, ProductSummary
from app.services import catalog_queries, editorial
# URL shapes live in urls.py (mirrored by lib/urls.ts); re-exported here because
# the routes and tests have always imported them from the SEO layer.
from app.services.urls import ( # noqa: F401
airport_path,
airport_query,
airport_tab,
brand_page_path,
brand_path,
category_path,
parse_airport_slug,
parse_product_slug,
product_path,
airport_category_path,
category_from_slug,
feature_flags,
savings_from_path,
slugify,
)
SITE_NAME = "Duty Free Professor"
FEED_TITLE = f"{SITE_NAME}: new in duty free"
DEFAULT_TITLE = "Duty Free Professor: compare airport duty-free prices"
DEFAULT_DESCRIPTION = (
"Compare duty-free prices across airport shops worldwide. "
"Set your trip and see where along your route it's cheapest to buy."
)
_HOST_RE = re.compile(r"^[a-z0-9.-]+(:\d+)?$")
#: The SPA's empty mount point in the built shell; product_body fills it.
ROOT_SLOT = ''
def public_base(host_header: str | None) -> str:
"""An origin derived from the Host header the proxy passed through."""
host = (host_header or "").lower()
if not _HOST_RE.match(host):
return ""
scheme = "http" if host.startswith(("127.", "172.", "192.168.", "localhost")) else "https"
return f"{scheme}://{host}"
def site_base(host_header: str | None, configured: str, production: bool) -> str:
"""The origin every absolute URL is built from.
Configured (`PUBLIC_BASE_URL`) wins. Without it, a dev or test process may
trust the Host header; a production one must not, and emits relative URLs
until the value is set (the startup log says so).
"""
if configured:
return configured.rstrip("/")
return "" if production else public_base(host_header)
# JSON-LD keys whose values are URLs on this site. Payloads are written with
# site-relative paths and made absolute at apply time, so the static heads can
# be module constants and every URL still carries the configured origin.
_URL_KEYS = {"url", "item", "@id", "logo", "image", "target"}
def absolutise(payload, base: str):
"""Prefix site-relative URL values in a JSON-LD payload with the origin."""
if isinstance(payload, dict):
return {
k: (base + v if k in _URL_KEYS and isinstance(v, str) and v.startswith("/") else absolutise(v, base))
for k, v in payload.items()
}
if isinstance(payload, list):
return [absolutise(v, base) for v in payload]
return payload
def _jsonld(payload: dict) -> str:
# < keeps any "" inside a value from terminating the tag.
body = json.dumps(payload, ensure_ascii=False).replace("<", "\\u003c")
return f''
class Head:
"""One page's injected head content."""
def __init__(
self,
title: str,
description: str,
canonical_path: str | None = None,
jsonld: list[dict] | None = None,
noindex: bool = False,
image: str | None = None,
preload_image: str | None = None,
body: str | None = None,
seed: dict | None = None,
last_modified: datetime | None = None,
seed_key: str = "__DFP_PRODUCT__",
og_type: str = "website",
) -> None:
self.title = title
#: Open Graph type: "website" for every page but an article.
self.og_type = og_type
self.description = description
self.preload_image = preload_image
self.canonical_path = canonical_path
self.jsonld = jsonld or []
self.noindex = noindex
self.image = image
#: Server-rendered markup for the SPA's root element (see product_body).
self.body = body
#: The page's data object, seeded for the SPA so it need not refetch it,
#: under `window.` (one key per page type; queries.ts reads them).
self.seed = seed
self.seed_key = seed_key
#: When the page's data last changed (the HTTP Last-Modified); None = process start.
self.last_modified = last_modified
def apply(self, template: str, base: str) -> str:
"""Stamp this head into the built index.html."""
page = re.sub(
r".*?", f"{html.escape(self.title)}", template, count=1
)
page = re.sub(
r'')
canonical = f"{base}{self.canonical_path}" if base and self.canonical_path else None
if canonical:
extra.append(f'')
extra.append(
f''
)
if base:
extra.append(
f''
)
extra.append(f'')
extra.append(f'')
extra.append(
f''
)
extra.append(f'')
if self.image:
extra.append(f'')
if self.preload_image:
# The LCP element is only discovered after the SPA fetches its
# data; preloading it from the head starts the download at parse
# time instead.
extra.append(
f''
)
for payload in self.jsonld:
extra.append(_jsonld(absolutise(payload, base)))
if self.seed is not None:
seed = json.dumps(self.seed, ensure_ascii=False).replace("<", "\\u003c")
extra.append(f"")
page = page.replace("", " " + "\n ".join(extra) + "\n ", 1)
if self.body:
page = page.replace(ROOT_SLOT, f'
{self.body}
', 1)
return page
def _truncate(text: str, limit: int = 155) -> str:
return text if len(text) <= limit else text[: limit - 1].rstrip() + "…"
def product_head(
db: Session, product_id: int, medal_assets: Collection[str] = ()
) -> Head | None:
"""Title, description, schema.org Product markup and the rendered body for one bottle.
Everything derives from the same `ProductDetail` the API hands the SPA, so
the head, the body and the page a user sees cannot disagree.
"""
detail = catalog_queries.get_product(db, product_id)
if detail is None:
return None
return head_for_detail(detail, medal_assets)
def gtin_property(gtin: str | None) -> str | None:
"""schema.org names the barcode property by its width; a 12-digit UPC
published as gtin13 fails validation, and the generic `gtin` is ignored by
the checkers that matter."""
if not gtin or not gtin.isdigit():
return None
return {8: "gtin8", 12: "gtin12", 13: "gtin13", 14: "gtin14"}.get(len(gtin))
def availability_of(flags: list[bool | None]) -> str | None:
"""schema.org availability from stock flags, or None when we do not know.
Only an explicit flag counts. Roughly a third of observations carry no
stock signal at all, and defaulting those to InStock would assert a fact
nobody observed; omitting the property is what the vocabulary is for.
"""
known = [f for f in flags if f is not None]
if not known:
return None
if any(known):
return "https://schema.org/InStock"
return "https://schema.org/OutOfStock"
def offer_jsonld(price: PriceOut) -> dict:
offer: dict = {
"@type": "Offer",
"price": f"{price.price:.2f}",
"priceCurrency": price.currency,
"seller": {"@type": "Organization", "name": price.retailer_name},
"availableAtOrFrom": {"@type": "Place", "name": _place(price)},
}
if (availability := availability_of([price.in_stock])) is not None:
offer["availability"] = availability
if price.url:
offer["url"] = price.url
return offer
def head_for_detail(detail: ProductDetail, medal_assets: Collection[str] = ()) -> Head:
shop_count = detail.location_count
low, high = detail.cheapest_usd, detail.dearest_usd
awards = [
f"{a.medal}, {a.competition} {a.year}" if a.medal else f"{a.competition} {a.year}"
for a in detail.awards
]
path = product_path(detail.id, detail.name)
title = (
f"{detail.name}: duty-free prices at {shop_count} shops | {SITE_NAME}"
if shop_count > 1
else f"{detail.name}: duty-free price | {SITE_NAME}"
)
bits = [f"Compare {detail.name} across airport duty-free shops."]
if low is not None and shop_count > 1:
bits.append(f"From ${low:,.2f} to ${high:,.2f} at {shop_count} shops,")
bits.append("every price dated.")
elif low is not None:
bits.append(f"Currently ${low:,.2f}, price dated.")
description = _truncate(" ".join(bits))
schema: dict = {
"@context": "https://schema.org",
"@type": "Product",
"@id": f"{path}#product",
"name": detail.name,
"url": path,
# Same factual sentence as the meta description: prices and coverage,
# never marketing copy (we store facts, not expression).
"description": description,
}
if detail.brand:
schema["brand"] = {"@type": "Brand", "name": detail.brand}
if gtin_key := gtin_property(detail.gtin):
schema[gtin_key] = detail.gtin
if detail.image_url:
schema["image"] = detail.image_url
if detail.category:
schema["category"] = detail.category
if awards:
schema["award"] = awards
if detail.prices:
# The price of record (structure review, 9 Sep): the shop's own price
# in its own currency is the price on the page and the only price in
# the markup. Each Offer carries the observation as quoted. The
# combined offer states a range only when every shop quoted one
# currency, from those quotes; with mixed currencies it carries none,
# because a USD range would be our conversion, which the page shows
# second and labelled. Before this the range was USD from our rates
# while the offers beneath it were in pounds and dirhams, which is the
# incongruence Mark warned tanks rankings.
priced = detail.prices
aggregate: dict = {
"@type": "AggregateOffer",
"offerCount": shop_count,
"url": path,
"offers": [offer_jsonld(p) for p in priced],
}
currencies = {p.currency for p in priced}
if len(currencies) == 1:
quotes = [p.price for p in priced]
aggregate["priceCurrency"] = priced[0].currency
aggregate["lowPrice"] = f"{min(quotes):.2f}"
aggregate["highPrice"] = f"{max(quotes):.2f}"
if (availability := availability_of([p.in_stock for p in priced])) is not None:
aggregate["availability"] = availability
schema["offers"] = aggregate
breadcrumbs = {
"@context": "https://schema.org",
"@type": "BreadcrumbList",
"itemListElement": [
{"@type": "ListItem", "position": 1, "name": "Products", "item": "/products"},
*([{"@type": "ListItem", "position": 2, "name": detail.category,
"item": category_path(detail.category)}] if detail.category else []),
{"@type": "ListItem",
"position": 3 if detail.category else 2,
"name": detail.name},
],
}
return Head(
title=title,
description=description,
canonical_path=path,
jsonld=[schema, breadcrumbs],
image=detail.image_url,
preload_image=detail.image_url,
body=product_body(detail, medal_assets),
seed=detail.model_dump(mode="json"),
last_modified=max((p.observed_at for p in detail.prices), default=None),
)
# --- the rendered body ------------------------------------------------------
#
# Every function below mirrors one piece of the React app. The pairings are
# named so a change on one side can be carried to the other:
# fmt_* <- web/src/lib/format.ts
# bottle_mark <- components/BottleMark.tsx
# medal_html <- components/Medal.tsx
# header_html <- components/SiteHeader.tsx + AnnouncementBar.tsx (default state)
# product_body <- pages/ProductPage.tsx + components/PriceTable.tsx (no airports chosen)
# The markup uses the SPA's class names so the stylesheet already loaded in the
# head styles it, and React's mount replaces it with identical geometry.
#: Mirrors Medal.tsx MEDAL_ART_VERSION. Bump both when medal artwork changes.
MEDAL_ART_VERSION = 2
# Intl.NumberFormat("en-US", {style: "currency"}) prefixes, for the currencies the
# collectors have seen or are about to. Anything else renders as "CODE 1,234.50",
# which is also what Intl does for a code it has no symbol for.
_CURRENCY_PREFIX = {
"USD": "$", "EUR": "€", "GBP": "£", "JPY": "¥", "KRW": "₩", "INR": "₹", "ILS": "₪",
"VND": "₫", "PHP": "₱", "CAD": "CA$", "AUD": "A$", "HKD": "HK$", "MXN": "MX$",
"NZD": "NZ$", "TWD": "NT$", "CNY": "CN¥", "BRL": "R$", "XAF": "FCFA",
}
# Currencies Intl formats with no minimum fraction digits.
_ZERO_DECIMAL = {"JPY", "KRW", "ISK", "VND", "CLP", "XAF", "XOF", "UGX", "PYG", "RWF"}
def fmt_usd(value: float | None) -> str:
"""format.ts formatUsd."""
return "-" if value is None else f"${value:,.2f}"
def fmt_local(value: float, currency: str) -> str:
"""format.ts formatLocal (en-US Intl currency output)."""
prefix = _CURRENCY_PREFIX.get(currency, f"{currency} ")
number = f"{value:,.2f}"
if currency in _ZERO_DECIMAL:
number = number.rstrip("0").rstrip(".")
return f"{prefix}{number}"
def fmt_size(size_ml: int | None) -> str | None:
"""format.ts formatSize."""
if not size_ml:
return None
if size_ml >= 1000:
litres = size_ml / 1000
return f"{litres:.0f}L" if size_ml % 1000 == 0 else f"{litres:.1f}L"
return f"{size_ml}ml"
def fmt_observed(seen: datetime, now: datetime) -> str:
"""format.ts formatObservedAt. Relative, like the SPA; the