"""Extime, the Paris airports storefront.

Operated by Extime (the ADP and Lagardere joint venture) and covering both
Charles de Gaulle and Orly: the storefront is one "paris" tenant and publishes
no per-airport split, so what we collect is Paris, not CDG alone.

Two things about this source shape the adapter:

* **The sitemap is the only honest classifier.** Parking spaces and sightseeing
  tours live at exactly the same `/en/paris/product/<slug>-<id>` shape as a
  bottle of whisky, and their id ranges overlap, so nothing about a URL tells
  you what it points at. Only its sitemap shard does. We therefore walk
  `shopping.xml` and treat everything else on the site as out of scope.
* **Size lives only in the page's data payload.** "Single Malt Whisky Ten Years
  Old" carries no size in its name and none in its schema.org markup; the
  bottle is 100cl, and only the embedded payload says so. A collector that
  skipped that payload would file a litre bottle as sizeless and compare it
  against a 70cl elsewhere, which is the mistake that has already cost this
  project a day.
"""

import json
import logging
import re
from collections.abc import Iterator

from app.services.collectors.base import Collector, ShopSpec, RawListing, facts_only, gone, listing_ref, report_skip
from app.services.collectors.fetch import FetchError, PageGone, SourceBlocked, fetch
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 Quantity, parse_quantity
from app.services.taxonomy import vertical_from_hints

PARSER_VERSION = "extime/2026-09-05"

logger = logging.getLogger(__name__)

BASE = "https://www.extime.com"
SHOPPING_INDEX = "/en/paris/sitemap/shopping.xml"
PRODUCT_PATH = "/en/paris/product/"

# The trailing number is the identity. Ids run 8 to 9 digits today; the looser
# bound is deliberate, so a shorter one is collected rather than silently lost.
_PRODUCT_URL_RE = re.compile(
    r"^https://www\.extime\.com/en/paris/product/[a-z0-9-]+-(\d{6,})$"
)
_LOC_RE = re.compile(r"<loc>\s*([^<\s]+)\s*</loc>", re.I)
# A family's variants each get their own URL, and every one of them renders
# the whole family. The slug without its id identifies the family.
_SLUG_STEM_RE = re.compile(r"/([a-z0-9-]+?)-\d{6,}$")
# The payload arrives as many self.__next_f.push([1,"..."]) calls whose strings
# concatenate into one document, with objects split across the joins.
_PUSH_RE = re.compile(r"self\.__next_f\.push\((\[.*?\])\)</script>", re.S)
_JSONLD_RE = re.compile(
    r'<script type="application/ld\+json"[^>]*>(.*?)</script>', re.S
)

_UNIT_TO_ML = {"ml": 1.0, "cl": 10.0, "l": 1000.0, "lt": 1000.0, "litre": 1000.0}


def sitemap_locs(xml: str) -> list[str]:
    """Every <loc> in a sitemap or sitemap index. Kept parseable offline."""
    return [m.strip() for m in _LOC_RE.findall(xml)]


def product_urls(xml: str) -> list[str]:
    """Product URLs from one shopping shard, in file order, de-duplicated.

    Anything in a shopping shard that is not shaped like a product is dropped
    rather than guessed at, and the caller logs the discrepancy.
    """
    seen: set[str] = set()
    out: list[str] = []
    for loc in sitemap_locs(xml):
        if _PRODUCT_URL_RE.match(loc) and loc not in seen:
            seen.add(loc)
            out.append(loc)
    return out


def slug_stem(url: str) -> str | None:
    """The family a product URL belongs to, ignoring which variant it is."""
    match = _SLUG_STEM_RE.search(url)
    return match.group(1) if match else None


def parse_jsonld_product(page: str) -> dict | None:
    """The schema.org ProductVariant block, if the page carries one."""
    for raw in _JSONLD_RE.findall(page):
        try:
            data = json.loads(raw)
        except ValueError:
            continue
        for node in data if isinstance(data, list) else [data]:
            if isinstance(node, dict) and node.get("@type") == "Product":
                return node
    return None


def jsonld_offer(product: dict) -> dict:
    """The offer node. It is always a single-element LIST, never an object."""
    offers = product.get("offers")
    if isinstance(offers, list):
        return offers[0] if offers and isinstance(offers[0], dict) else {}
    return offers if isinstance(offers, dict) else {}


def rsc_payload(page: str) -> str:
    """Reassemble the streamed data payload.

    Each push carries a fragment as a JSON string. The fragments must be
    decoded BEFORE they are joined: concatenating the raw source and decoding
    afterwards splits escape sequences across the seam and yields nothing,
    silently.
    """
    chunks: list[str] = []
    for match in _PUSH_RE.finditer(page):
        try:
            arg = json.loads(match.group(1))
        except ValueError:
            continue
        if isinstance(arg, list) and len(arg) > 1 and arg[0] == 1 and isinstance(arg[1], str):
            chunks.append(arg[1])
    return "".join(chunks)


def rsc_page_object(page: str) -> dict | None:
    """The product record embedded in the payload: main_offer plus variations."""
    payload = rsc_payload(page)
    marker = payload.find('"main_offer"')
    if marker < 0:
        return None
    start = payload.rfind("{", 0, marker)
    if start < 0:
        return None
    try:
        obj, _ = json.JSONDecoder().raw_decode(payload[start:])
    except ValueError:
        return None
    return obj if isinstance(obj, dict) and "main_offer" in obj else None


def capacity_ml(value: object, unit: object) -> int | None:
    """A declared capacity in millilitres. 100cl is a litre, not a hundred."""
    factor = _UNIT_TO_ML.get(str(unit or "").strip().lower())
    if factor is None:
        return None
    try:
        millilitres = float(value) * factor
    except (TypeError, ValueError):
        return None
    return int(round(millilitres)) if 1 <= millilitres <= 20_000 else None


def size_label(quantity_ml: int) -> str:
    return f"{quantity_ml / 1000:g}L" if quantity_ml >= 1000 else f"{quantity_ml / 10:g}cl"


def _number(value: object) -> float | None:
    try:
        return float(str(value).replace(",", "."))
    except (TypeError, ValueError):
        return None


_PER_QUANTITY_RE = re.compile(r"^\s*(\d*(?:[.,]\d+)?)\s*(ml|cl|l|lt|litre)\s*$", re.I)


def implied_quantity_ml(variation: dict, price: float | None) -> int | None:
    """A second reading of the size, from fields that only agree if it is right.

    The live shape, pinned 2026-09-05 from a raw page (`tests/fixtures/
    extime_cheirosa_76.html`): each price tier carries `price_per_quantity`
    (a number) and `price_per_quantity_type` (the reference quantity, e.g.
    "100ml"), so tier price / rate * reference = the size. `net_weight` is a
    number with `net_weight_unit` beside it and counts only when that unit is
    a volume; grams say nothing about millilitres. The older dict-or-string
    readings stay for the shapes seen before the fixture existed.
    """
    for tier in (variation.get("duty_free"), variation.get("duty_paid")):
        if not isinstance(tier, dict):
            continue
        rate, tier_price = _number(tier.get("price_per_quantity")), _number(tier.get("price"))
        match = _PER_QUANTITY_RE.match(str(tier.get("price_per_quantity_type") or ""))
        if rate and tier_price and match:
            quantity = _number(match.group(1)) or 1.0
            millilitres = tier_price / rate * quantity * _UNIT_TO_ML[match.group(2).lower()]
            if 1 <= millilitres <= 20_000:
                return int(round(millilitres))
    for key in ("price_per_quantity", "price_per_unit"):
        per = variation.get(key)
        if isinstance(per, dict):
            rate = _number(per.get("price") or per.get("value") or per.get("amount"))
            unit = per.get("unit") or per.get("quantity_unit") or per.get("capacity_unit")
            quantity = _number(per.get("quantity") or per.get("capacity")) or 1.0
        elif isinstance(per, str):
            match = re.search(r"(\d+(?:[.,]\d+)?)\s*[^\d/]*/\s*(\d*(?:[.,]\d+)?)\s*(ml|cl|l|lt|litre)\b", per, re.I)
            if not match:
                continue
            rate, unit = _number(match.group(1)), match.group(3)
            quantity = _number(match.group(2)) or 1.0
        else:
            continue
        factor = _UNIT_TO_ML.get(str(unit or "").strip().lower())
        if rate and price and factor:
            millilitres = price / rate * quantity * factor
            if 1 <= millilitres <= 20_000:
                return int(round(millilitres))
    weight = variation.get("net_weight")
    if isinstance(weight, dict):
        return capacity_ml(weight.get("value") or weight.get("quantity"), weight.get("unit"))
    if isinstance(weight, (int, float)) and not isinstance(weight, bool):
        return capacity_ml(weight, variation.get("net_weight_unit"))
    if isinstance(weight, str):
        match = re.search(r"(\d+(?:[.,]\d+)?)\s*(ml|cl|l|lt|litre)\b", weight, re.I)
        if match:
            return capacity_ml(match.group(1).replace(",", "."), match.group(2))
    return None


def checked_capacity_ml(variation: dict, price: float | None) -> int | None:
    """The declared capacity, or None when a second reading contradicts it.

    A declared 100cl once became a litre bottle compared against 70cl rows
    elsewhere; where the page offers a second reading, disagreement beyond
    the 700-vs-750 sloppiness means we do not know the size, and an unknown
    size is honest where a guessed one is not.
    """
    declared = capacity_ml(variation.get("capacity"), variation.get("capacity_unit"))
    implied = implied_quantity_ml(variation, price)
    if declared is None or implied is None:
        return declared
    if abs(declared - implied) / max(declared, implied) > 0.12:
        return None
    return declared


def quantity_of_variation(variation: dict, product_name: str, price: float | None = None) -> tuple[Quantity | None, str | None, bool]:
    """The variation's own structured quantity: the declared capacity when the page states
    one and the price-per-unit reading does not contradict it (`checked_capacity_ml`, the
    same gate `size_ml` passes: the 7,624 ml mist fixture never becomes a quantity), or,
    only when nothing about the size is stated anywhere, capacity included, the net weight,
    read as a bare number beside `net_weight_unit` or as a dict of the two. The third value
    says whether the net weight was the source, for `quantity_source` on the raw record.
    """
    capacity = variation.get("capacity")
    capacity_unit = variation.get("capacity_unit")
    capacity_ml_value = checked_capacity_ml(variation, price)
    if capacity_ml_value is None and capacity_ml(capacity, capacity_unit) is not None:
        return None, None, False  # a contradicted capacity: the name decides
    if capacity_ml_value is not None:
        text = variation.get("name") or f"{capacity} {capacity_unit}"
        return Quantity(value=float(capacity_ml_value), unit="ml", form="single", state="stated"), text, False
    variation_states = parse_quantity(str(variation.get("name") or "")).state == "stated"
    product_states = parse_quantity(product_name or "").state == "stated"
    if variation_states or product_states:
        return None, None, False
    weight = variation.get("net_weight")
    if isinstance(weight, dict):
        raw_value = weight.get("value") if weight.get("value") is not None else weight.get("quantity")
        unit = weight.get("unit")
    else:
        raw_value = weight
        unit = variation.get("net_weight_unit")
    value = _number(raw_value)
    unit_l = str(unit or "").strip().lower()
    if value is None or unit_l not in ("g", "gr", "kg"):
        return None, None, False
    grams = value * 1000.0 if unit_l == "kg" else value
    return Quantity(value=grams, unit="g", form="single", state="stated"), f"{value:g} {unit}", True


def published_options(variation: dict) -> list[tuple[str, str]]:
    """The variation's own label as the shop's option field (`variation.name`: "50 ml" on a
    page of several sizes). A size label is the quantity and ingest reads it as one; anything
    else the shop puts there is an option of the variant. The capacity fields stay the typed
    quantity (`quantity_of_variation`)."""
    label = str(variation.get("name") or "").strip()
    return [("variation", label)] if label else []


def variation_price(variation: dict) -> tuple[float | None, float | None]:
    """The duty-free price, and the crossed-out one where a promotion runs.

    Duty-paid is deliberately ignored: this site compares what a traveller pays
    at the gate, and quoting the higher figure would flatter every other shop.
    """
    duty_free = variation.get("duty_free")
    if not isinstance(duty_free, dict):
        return None, None
    try:
        price = float(duty_free["price"])
    except (KeyError, TypeError, ValueError):
        return None, None
    was = duty_free.get("price_crossed") or variation.get("price_crossed")
    try:
        was_price = float(was) if was else None
    except (TypeError, ValueError):
        was_price = None
    return price, (was_price if was_price and was_price > price else None)


def in_stock_of(variation: dict) -> bool | None:
    stock = variation.get("stock")
    if isinstance(stock, bool) or not isinstance(stock, (int, float)):
        return None
    return stock > 0


class Extime(Collector):
    slug = "extime-paris"
    retailer_slug = "extime"
    retailer_name = "Extime"
    operator = "Extime (ADP and Lagardere)"
    homepage = BASE
    parser_version = PARSER_VERSION

    def shops(self) -> list[ShopSpec]:
        return [
            ShopSpec(
                code="CDG",
                iata="CDG",
                # Named for what it is: one storefront serving both Paris
                # airports, with no per-airport split published.
                name="Paris Charles de Gaulle and Orly",
                city="Paris",
                country="France",
                currency="EUR",
            )
        ]

    def _shard_urls(self, delay: float) -> list[str]:
        index = fetch(f"{BASE}{SHOPPING_INDEX}", accept="application/xml", delay=0.0).text
        shards = [u for u in sitemap_locs(index) if u.endswith(".xml")]
        if not shards:
            raise FetchError(f"{BASE}{SHOPPING_INDEX}: no shards listed")
        return shards

    def _listings(self, url: str, page: str) -> list[RawListing]:
        product = parse_jsonld_product(page)
        if product is None:
            return []
        offer = jsonld_offer(product)
        name = (product.get("name") or offer.get("name") or "").strip()
        if not name:
            return []
        brand = (product.get("brand") or {}).get("name") if isinstance(product.get("brand"), dict) else None

        record = rsc_page_object(page) or {}
        main = record.get("main_offer") if isinstance(record.get("main_offer"), dict) else {}
        variations = [v for v in (record.get("variations") or []) if isinstance(v, dict)]
        categories = [c for c in (main.get("categories_name") or []) if isinstance(c, str)]
        gtin = clean_gtin(main.get("gtin"))
        base_sku = str(main.get("sku") or offer.get("sku") or "").strip()
        if not base_sku:
            return []
        abv = parse_abv(main.get("product_line")) or parse_abv(name)

        priced = [(v, *variation_price(v)) for v in variations]
        priced = [(v, p, w) for v, p, w in priced if p is not None]

        # No readable price in the payload: fall back to the advertised
        # "starting at" figure, and only when there is a single variant for it
        # to belong to. A multi-variant page without prices is skipped rather
        # than pinned to its cheapest member.
        if not priced:
            if len(variations) > 1:
                report_skip("multi_variant_unpriced", source_sku=base_sku, url=url)
                return []
            try:
                fallback = float(offer["lowPrice"])
            except (KeyError, TypeError, ValueError):
                report_skip("no_price", source_sku=base_sku, url=url)
                return []
            priced = [(variations[0] if variations else {}, fallback, None)]

        out: list[RawListing] = []
        multi = len(priced) > 1
        # The first category name is the tree the product hangs in
        # ("Beverage", "Beauty"), which is what a vertical is.
        vertical = vertical_from_hints(categories[0]) if categories else None
        raw_common = {"jsonld": facts_only(product), "main_offer": facts_only(main)}
        for variation, price, was_price in priced:
            quantity_ml = checked_capacity_ml(variation, price)
            if quantity_ml is None and implied_quantity_ml(variation, price) is None:
                quantity_ml = parse_quantity_ml(name)
            quantity, quantity_text, from_net_weight = quantity_of_variation(variation, name, price)
            full_name = name
            sku = base_sku
            if multi and quantity_ml:
                # Each size is its own row, so one bottle can never be priced
                # from another's cheaper sibling.
                full_name = f"{name} {size_label(quantity_ml)}"
                sku = f"{base_sku}::{quantity_ml}"
            raw_payload = {**raw_common, "variation": facts_only(variation)}
            if from_net_weight:
                raw_payload["quantity_source"] = "extime.net_weight"
            out.append(
                RawListing(
                    source_sku=sku,
                    name=full_name,
                    brand=(brand or main.get("brand_name") or None),
                    # The barcode identifies the product, so it belongs to a
                    # single-variant listing only: shared across sizes it would
                    # merge them back into one.
                    gtin=None if multi else gtin,
                    price=price,
                    was_price=was_price,
                    currency="EUR",
                    shop_code="CDG",
                    in_stock=in_stock_of(variation),
                    quantity_ml=quantity_ml,
                    quantity=quantity,
                    quantity_text=quantity_text,
                    abv=abv,
                    feed_categories=categories,
                    country_of_origin=variation.get("product_country") or None,
                    url=url,
                    price_type="promo" if was_price else "list",
                    is_exclusive=looks_exclusive(f"{name} {' '.join(categories)}"),
                    raw=raw_payload,
                    vertical=vertical,
                    options=published_options(variation),
                )
            )
        return out

    def read_one(self, listing) -> RawListing | None:
        ref = listing_ref(listing)
        if not ref.url:
            return None
        robots = check_allowed(BASE, [SHOPPING_INDEX, PRODUCT_PATH], fresh=False)
        if not robots.allows(ref.url):
            raise SourceBlocked(f"{BASE}: robots.txt disallows {ref.url}")
        try:
            page = fetch(ref.url, accept="text/html", delay=robots.delay_for(1.5)).text
        except FetchError as exc:
            if gone(exc):
                return None
            raise
        # A family page renders every size; the published row is one of them.
        return next(
            (row for row in self._listings(ref.url, page) if row.source_sku == ref.source_sku),
            None,
        )

    def collect(self, *, limit: int | None = None, delay: float = 1.0) -> Iterator[RawListing]:
        # Their robots.txt names no crawl delay and no bot-specific group
        # today, but it is four lines long and could gain one tomorrow.
        robots = check_allowed(BASE, [SHOPPING_INDEX, PRODUCT_PATH])
        crawl_delay = robots.delay_for(max(delay, 1.5))

        urls: list[str] = []
        for shard in self._shard_urls(delay):
            try:
                xml = fetch(shard, accept="application/xml", delay=crawl_delay).text
            except PageGone as exc:
                # A sitemap shard the index still lists: skip it by name and read the rest.
                logger.warning("extime_page_gone source=%s detail=%s", self.slug, exc)
                continue
            except FetchError:
                continue
            urls.extend(product_urls(xml))
            if limit is not None and len(urls) >= limit:
                break

        produced = 0
        # One product family is reachable from several URLs, one per variant,
        # and each of those pages renders the entire family. Left alone that
        # fetches the same bottles repeatedly and records every price several
        # times over, so a family is read once and its rows emitted once.
        families: set[str] = set()
        emitted: set[str] = set()
        for url in urls:
            if limit is not None and produced >= limit:
                return
            family = slug_stem(url)
            if family and family in families:
                continue
            try:
                page = fetch(url, accept="text/html", delay=crawl_delay).text
            except SourceBlocked:
                raise
            except PageGone as exc:
                logger.warning("extime_page_gone source=%s detail=%s", self.slug, exc)
                continue
            except FetchError:
                continue
            listings = self._listings(url, page)
            if listings and family:
                families.add(family)
            for listing in listings:
                if listing.source_sku in emitted:
                    continue
                emitted.add(listing.source_sku)
                yield listing
                produced += 1
                if limit is not None and produced >= limit:
                    return