Lazada Data Scraping Price Data and Proxy Planning

mkdir -p lazada-pilot/examples
cd lazada-pilot
python -m venv .venv
. .venv/bin/activate
python -m pip install playwright
python -m playwright install chromium

New-Item -ItemType Directory -Force .\lazada-pilot\examples
Set-Location .\lazada-pilot
py -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install playwright
python -m playwright install chromium

"""Single-URL Playwright sample for an authorized public product page.

This sample intentionally stops on human-verification pages and does not solve
CAPTCHAs. It emits one JSON row per JSON-LD offer when possible. The adapter
keeps price meaning, extraction method, transport status, parsing status, and
field validation status separate. It is a design sample and has not been run
against a live Lazada page. It performs no automatic retries; a 429 Retry-After value is recorded for later scheduling. It requires a pre-approved URL list, rechecks the final URL after redirects, and writes no screenshots or raw JSON-LD by default.
"""

from __future__ import annotations

import argparse
import json
import os
import re
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from time import perf_counter
from urllib.parse import urlparse

from playwright.sync_api import Page, sync_playwright


MARKETPLACE_HOSTS = {
    "lazada.sg": ("lazada.sg",),
    "lazada.com.my": ("lazada.com.my",),
    "lazada.co.th": ("lazada.co.th",),
    "lazada.vn": ("lazada.vn",),
    "lazada.com.ph": ("lazada.com.ph",),
    "lazada.co.id": ("lazada.co.id",),
}
ALLOWED_HOST_SUFFIXES = tuple(
    suffix for hosts in MARKETPLACE_HOSTS.values() for suffix in hosts
)


class SampleFailure(RuntimeError):
    def __init__(self, error_type: str, message: str):
        super().__init__(message)
        self.error_type = error_type


def proxy_from_environment() -> dict[str, str] | None:
    server = os.getenv("ROLA_PROXY_SERVER")
    username = os.getenv("ROLA_PROXY_USERNAME")
    password = os.getenv("ROLA_PROXY_PASSWORD")
    supplied = [server, username, password]
    if not any(supplied):
        return None
    if not all(supplied):
        raise RuntimeError("Set all three ROLA_PROXY_* variables or none of them.")
    return {"server": server, "username": username, "password": password}


def normalize_url(url: str) -> str:
    parsed = urlparse(url)
    return parsed._replace(fragment="").geturl()


def host_matches_marketplace(host: str, marketplace: str) -> bool:
    allowed_hosts = MARKETPLACE_HOSTS.get(marketplace.lower())
    return bool(allowed_hosts) and any(
        host == suffix or host.endswith("." + suffix) for suffix in allowed_hosts
    )


def validate_approved_urls(approved_urls: list[str]) -> None:
    if not approved_urls:
        raise ValueError("Provide at least one pre-approved Lazada URL.")
    for approved_url in approved_urls:
        parsed = urlparse(approved_url)
        host = (parsed.hostname or "").lower()
        if parsed.scheme != "https" or not host or not any(
            host == suffix or host.endswith("." + suffix)
            for suffix in ALLOWED_HOST_SUFFIXES
        ):
            raise ValueError(
                "Every approved URL must be an HTTPS Lazada storefront URL."
            )


def validate_target(url: str, approved_urls: list[str]) -> None:
    parsed = urlparse(url)
    host = (parsed.hostname or "").lower()
    if parsed.scheme != "https" or not host or not any(
        host == suffix or host.endswith("." + suffix)
        for suffix in ALLOWED_HOST_SUFFIXES
    ):
        raise ValueError("Use an approved HTTPS Lazada storefront URL.")
    approved_set = {normalize_url(item) for item in approved_urls}
    if normalize_url(url) not in approved_set:
        raise ValueError("URL is outside the pre-approved URL set.")


def validate_marketplace(url: str, marketplace: str) -> None:
    parsed = urlparse(url)
    host = (parsed.hostname or "").lower()
    marketplace = marketplace.lower()
    if marketplace not in MARKETPLACE_HOSTS:
        raise ValueError("Use a marketplace with a verified Lazada host mapping.")
    if not host_matches_marketplace(host, marketplace):
        raise ValueError("Marketplace does not match the URL host.")

def classify_blocking_page(page: Page) -> None:
    """Best-effort check of the main document and accessible frames."""
    texts = []
    try:
        texts.append(page.locator("body").inner_text(timeout=10_000).lower())
        for frame in page.frames[1:]:
            try:
                texts.append(frame.locator("body").inner_text(timeout=2_000).lower())
            except Exception:
                continue
    except Exception as exc:
        raise SampleFailure(
            "page_classification_error",
            "Could not inspect the main document for verification or login state.",
        ) from exc

    combined = "\n".join(texts)
    verification_signals = (
        "verify you are human",
        "complete the verification",
        "human verification",
        "完成验证",
        "人机验证",
        "验证您是人类",
    )
    if any(signal in combined for signal in verification_signals):
        raise SampleFailure(
            "verification",
            "Human verification detected. Stop; do not automate CAPTCHA solving.",
        )

    login_signals = (
        "log in to continue",
        "login to continue",
        "sign in to continue",
        "please log in",
        "请登录",
        "登入",
    )
    page_url = getattr(page, "url", "").lower()
    login_url = any(token in page_url for token in ("/login", "/signin", "/sign-in"))
    if any(signal in combined for signal in login_signals) and (
        login_url or "password" in combined
    ):
        raise SampleFailure(
            "login_required",
            "A login page was detected. Stop; do not automate login without authorization.",
        )

def wait_for_product_json_ld(page: Page) -> None:
    try:
        page.locator('script[type="application/ld+json"]').first.wait_for(
            state="attached",
            timeout=15_000,
        )
    except Exception as exc:
        raise SampleFailure(
            "json_ld_timeout",
            "Product JSON-LD did not appear within 15 seconds.",
        ) from exc


def product_json_ld(page: Page) -> dict:
    """Return one Product node, or fail when the page is ambiguous."""

    def candidates(data: object):
        if isinstance(data, list):
            for item in data:
                yield from candidates(item)
        elif isinstance(data, dict):
            node_type = data.get("@type")
            if node_type == "Product" or (
                isinstance(node_type, list) and "Product" in node_type
            ):
                yield data
            if "@graph" in data:
                yield from candidates(data["@graph"])

    products = []
    for node in page.locator('script[type="application/ld+json"]').all_text_contents():
        try:
            data = json.loads(node)
        except json.JSONDecodeError:
            continue
        products.extend(candidates(data))

    if not products:
        raise SampleFailure(
            "json_ld_missing",
            "No Product JSON-LD found; inspect the current authorized page.",
        )
    if len(products) > 1:
        raise SampleFailure(
            "multiple_products",
            f"Found {len(products)} Product JSON-LD nodes; target selection is not implemented.",
        )
    return products[0]

def offer_list(product: dict) -> list[dict]:
    offers = product.get("offers") or []
    if isinstance(offers, dict):
        offers = [offers]
    return [offer for offer in offers if isinstance(offer, dict)]


def price_value(offer: dict) -> object | None:
    raw = offer.get("price")
    if raw is None:
        raw = offer.get("lowPrice")
    return raw


def numeric_price(raw_price: object) -> str | None:
    """Parse machine numbers and unambiguous currency display strings."""

    if raw_price is None:
        return None
    if isinstance(raw_price, (int, float, Decimal)):
        try:
            value = Decimal(str(raw_price))
            if not value.is_finite() or value <= 0:
                return None
            return str(value.quantize(Decimal("0.01")))
        except InvalidOperation:
            return None

    text = str(raw_price).strip()
    if not text or re.search(r"\d\s*[-–—]\s*\d|\bto\b", text, re.IGNORECASE):
        return None

    cleaned = re.sub(r"[^0-9.,]", "", text)
    if not cleaned:
        return None

    if "," in cleaned and "." in cleaned:
        decimal_separator = "," if cleaned.rfind(",") > cleaned.rfind(".") else "."
        grouping_separator = "." if decimal_separator == "," else ","
        integer_part, fractional_part = cleaned.rsplit(decimal_separator, 1)
        groups = integer_part.split(grouping_separator)
        if len(groups) > 1 and (
            not groups[0] or any(len(group) != 3 for group in groups[1:])
        ):
            return None
        if not fractional_part.isdigit() or len(fractional_part) not in (1, 2):
            return None
        cleaned = "".join(groups) + "." + fractional_part
    elif "," in cleaned or "." in cleaned:
        separator = "," if "," in cleaned else "."
        if cleaned.count(separator) > 1:
            groups = cleaned.split(separator)
            if not groups[0] or any(len(group) != 3 for group in groups[1:]):
                return None
            cleaned = "".join(groups)
        else:
            whole, fraction = cleaned.split(separator)
            if len(fraction) == 3:
                return None
            if not whole or not fraction or len(fraction) not in (1, 2):
                return None
            cleaned = whole + "." + fraction

    try:
        value = Decimal(cleaned)
        if not value.is_finite() or value <= 0:
            return None
        return str(value.quantize(Decimal("0.01")))
    except InvalidOperation:
        return None
def seller_name(offer: dict) -> str | None:
    seller = offer.get("seller")
    if isinstance(seller, dict):
        return seller.get("name")
    return seller if isinstance(seller, str) else None


def context_value(value: object) -> str | None:
    if value is None:
        return None
    if isinstance(value, str):
        return value
    return json.dumps(value, ensure_ascii=False, sort_keys=True)


def stock_status(offer: dict) -> str:
    availability = str(offer.get("availability") or "").lower()
    if "instock" in availability:
        return "in_stock"
    if "outofstock" in availability:
        return "out_of_stock"
    if "preorder" in availability:
        return "preorder"
    return "unknown"


def offer_type(offer: dict) -> str:
    value = offer.get("@type")
    if isinstance(value, list):
        return ",".join(str(item) for item in value)
    return str(value or "Offer")


def transport_status(http_status: int | None) -> str:
    if http_status is None:
        return "no_response"
    return "http_2xx" if 200 <= http_status < 300 else "http_error"


def retry_after_value(response) -> str | None:
    if response is None:
        return None
    return response.headers.get("retry-after")


def classify_response(response) -> None:
    if response is None:
        raise SampleFailure("no_response", "Navigation returned no HTTP response.")
    if not 200 <= response.status < 300:
        detail = f"Target returned HTTP {response.status}."
        retry_after = retry_after_value(response)
        if response.status == 429 and retry_after:
            detail += f" Retry-After: {retry_after}."
        raise SampleFailure("http_error", detail)


def error_type_for(error: Exception) -> str:
    if isinstance(error, SampleFailure):
        return error.error_type
    if isinstance(error, ValueError):
        return "invalid_target"
    error_name = type(error).__name__.lower()
    error_text = str(error).lower()
    if "timeout" in error_name or "timeout" in error_text:
        return "timeout"
    if any(
        token in error_text
        for token in (
            "dns",
            "name not resolved",
            "ssl",
            "tls",
            "certificate",
            "net::err_",
        )
    ):
        return "network_error"
    return "runtime_error"


def failure_row(args, response, error: Exception, started_at: float) -> dict:
    http_status = response.status if response is not None else None
    error_type = error_type_for(error)
    return {
        "marketplace": args.marketplace,
        "source_url": args.url,
        "product_id": None,
        "title": None,
        "seller_name": None,
        "variant_id": None,
        "variant_label": None,
        "raw_price": None,
        "numeric_price": None,
        "currency": None,
        "price_type": "unknown",
        "eligibility_context": None,
        "delivery_context": None,
        "price_source": "json_ld",
        "extraction_method": "json_ld",
        "original_price": None,
        "discount_text": None,
        "stock_status": "unknown",
        "captured_at_utc": datetime.now(timezone.utc).isoformat(),
        "route_label": args.route_label,
        "http_status": http_status,
        "retry_after": retry_after_value(response),
        "transport_status": transport_status(http_status),
        "status": error_type,
        "validation_status": "invalid",
        "validation_errors": [f"{error_type}: {str(error)}"],
        "error_type": error_type,
        "elapsed_ms": int((perf_counter() - started_at) * 1000),
    }


def validation_result(row: dict) -> tuple[str, list[str]]:
    """Separate transport/parse evidence from business-record acceptance."""

    errors: list[str] = []
    hard_failure = False

    if row.get("transport_status") != "http_2xx":
        errors.append("transport_status: only http_2xx can be a valid product record")
        hard_failure = True

    marketplace = str(row.get("marketplace") or "").lower()
    parsed_url = urlparse(str(row.get("source_url") or ""))
    host = (parsed_url.hostname or "").lower()
    if parsed_url.scheme != "https" or not host:
        errors.append("source_url: expected an HTTPS URL with a hostname")
        hard_failure = True
    elif marketplace not in MARKETPLACE_HOSTS or not host_matches_marketplace(
        host, marketplace
    ):
        errors.append("marketplace: unsupported or does not match source_url host")
        hard_failure = True

    required = (
        "marketplace",
        "product_id",
        "title",
        "seller_name",
        "raw_price",
        "numeric_price",
        "currency",
        "price_type",
        "delivery_context",
    )
    for field in required:
        if row.get(field) in (None, ""):
            errors.append(f"{field}: required")

    currency = row.get("currency")
    if not isinstance(currency, str) or not re.fullmatch(r"[A-Z]{3}", currency):
        errors.append("currency: expected a three-letter uppercase code")

    valid_price_types = {
        "listed",
        "promotional",
        "member",
        "voucher-adjusted",
        "quantity-tier",
        "shipping-inclusive",
        "checkout",
    }
    conditional_price_types = {"member", "voucher-adjusted"}
    price_type = row.get("price_type")
    if price_type == "unknown" or price_type not in valid_price_types:
        errors.append("price_type: insufficient or unsupported evidence")
    elif price_type in conditional_price_types and not str(
        row.get("eligibility_context") or ""
    ).strip():
        errors.append("eligibility_context: required for conditional price")

    if row.get("status") == "aggregate_offer":
        errors.append("status: AggregateOffer is not selected SKU evidence")
    if not row.get("variant_id"):
        errors.append("variant_id: selected SKU evidence is missing")
    if row.get("numeric_price") in (None, ""):
        errors.append("numeric_price: selected price is missing")

    if hard_failure:
        return "invalid", errors
    if errors:
        return "partial", errors
    return "valid", []


def rows_from_product(product: dict, args, http_status: int | None) -> list[dict]:
    offers = offer_list(product) or [{}]
    rows = []
    for offer in offers:
        raw_value = price_value(offer)
        raw_price = None if raw_value is None else str(raw_value)
        currency = offer.get("priceCurrency") or product.get("priceCurrency")
        kind = offer_type(offer)
        is_aggregate = "AggregateOffer" in kind
        status = "parsed_without_offer"
        if offer:
            if is_aggregate:
                status = "aggregate_offer"
            elif raw_price is None:
                status = "parsed_without_price"
            else:
                status = "live_parsed"
        row = {
            "marketplace": args.marketplace,
            "source_url": args.url,
            "product_id": product.get("sku") or product.get("productID"),
            "title": product.get("name"),
            "seller_name": seller_name(offer),
            "variant_id": offer.get("sku"),
            "variant_label": offer.get("name") if offer.get("name") != product.get("name") else None,
            "raw_price": raw_price,
            "numeric_price": numeric_price(raw_value),
            "currency": currency,
            "price_type": str(offer.get("priceType") or product.get("priceType") or "unknown"),
            "eligibility_context": context_value(offer.get("eligibleCustomerType")),
            "delivery_context": context_value(offer.get("areaServed")),
            "price_source": "json_ld",
            "extraction_method": "json_ld",
            "original_price": None,
            "discount_text": None,
            "stock_status": stock_status(offer),
            "captured_at_utc": datetime.now(timezone.utc).isoformat(),
            "route_label": args.route_label,
            "http_status": http_status,
            "transport_status": transport_status(http_status),
            "status": status,
            "validation_status": "not_run",
            "validation_errors": [],
            "retry_after": None,
            "error_type": None,
            "elapsed_ms": None,
        }
        if args.include_raw_json_ld:
            row["raw_product_json_ld"] = product
        row["validation_status"], row["validation_errors"] = validation_result(row)
        rows.append(row)
    return rows


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--url", required=True)
    parser.add_argument("--marketplace", required=True)
    parser.add_argument("--approved-url", dest="approved_urls", action="append", required=True)
    parser.add_argument("--include-raw-json-ld", action="store_true")
    parser.add_argument("--output", default="lazada_product.json")
    parser.add_argument("--route-label", default=None)
    args = parser.parse_args()
    args.route_label = args.route_label or (
        "rola" if os.getenv("ROLA_PROXY_SERVER") else "direct"
    )
    args.marketplace = args.marketplace.lower()
    started_at = perf_counter()
    response = None

    try:
        validate_approved_urls(args.approved_urls)
        validate_target(args.url, args.approved_urls)
        validate_marketplace(args.url, args.marketplace)
        with sync_playwright() as playwright:
            browser = playwright.chromium.launch(
                headless=True,
                proxy=proxy_from_environment(),
            )
            try:
                page = browser.new_page(locale="en-US")
                response = page.goto(
                    args.url,
                    wait_until="domcontentloaded",
                    timeout=30_000,
                )
                classify_response(response)
                final_url = page.url
                validate_target(final_url, args.approved_urls)
                validate_marketplace(final_url, args.marketplace)
                args.url = final_url
                classify_blocking_page(page)
                wait_for_product_json_ld(page)
                product = product_json_ld(page)
                rows = rows_from_product(
                    product,
                    args,
                    response.status if response else None,
                )
                elapsed_ms = int((perf_counter() - started_at) * 1000)
                for row in rows:
                    row["retry_after"] = retry_after_value(response)
                    row["elapsed_ms"] = elapsed_ms
                Path(args.output).write_text(
                    json.dumps(rows, indent=2),
                    encoding="utf-8",
                )
            finally:
                browser.close()
    except Exception as exc:
        failure = failure_row(args, response, exc, started_at)
        Path(args.output).write_text(
            json.dumps([failure], indent=2),
            encoding="utf-8",
        )
        raise
if __name__ == "__main__":
    main()

{
  "marketplace": "lazada.sg",
  "source_url": "https://www.lazada.sg/example-authorized-product",
  "product_id": "DEMO-1001",
  "title": "Demo Wireless Mouse",
  "seller_name": "Example Seller",
  "currency": "SGD",
  "delivery_context": "SG",
  "variants": [
    {
      "variant_id": "DEMO-1001-BLK",
      "variant_label": "Black",
      "raw_price": "S$29.90",
      "original_price": "S$39.90",
      "discount_text": "25% off",
      "price_type": "promotional",
      "eligibility_context": null,
      "delivery_context": "SG",
      "stock_status": "in_stock"
    },
    {
      "variant_id": "DEMO-1001-WHT",
      "variant_label": "White",
      "raw_price": "S$31.90",
      "original_price": null,
      "discount_text": null,
      "price_type": "listed",
      "eligibility_context": null,
      "delivery_context": "SG",
      "stock_status": "in_stock"
    }
  ]
}

"""Normalize and validate a fictional Lazada-shaped fixture without network traffic."""

from __future__ import annotations

import json
import re
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
from urllib.parse import urlparse


ROOT = Path(__file__).resolve().parent
SOURCE = ROOT / "lazada_fixture.json"
OUTPUT = ROOT / "normalized_output.json"

VALID_PRICE_TYPES = {
    "listed",
    "promotional",
    "member",
    "voucher-adjusted",
    "quantity-tier",
    "shipping-inclusive",
    "checkout",
    "unknown",
}
CONDITIONAL_PRICE_TYPES = {"member", "voucher-adjusted"}
VALID_STOCK_STATUSES = {"in_stock", "out_of_stock", "preorder", "unknown"}
REQUIRED_FIELDS = (
    "marketplace",
    "source_url",
    "product_id",
    "title",
    "seller_name",
    "variant_id",
    "variant_label",
    "raw_price",
    "numeric_price",
    "currency",
    "price_type",
    "delivery_context",
    "price_source",
    "extraction_method",
    "stock_status",
    "captured_at_utc",
    "route_label",
    "transport_status",
)


def numeric_price(raw_price: object) -> str | None:
    """Parse machine numbers and unambiguous currency display strings."""

    if raw_price is None:
        return None
    if isinstance(raw_price, (int, float, Decimal)):
        try:
            value = Decimal(str(raw_price))
            if not value.is_finite() or value <= 0:
                return None
            return str(value.quantize(Decimal("0.01")))
        except InvalidOperation:
            return None

    text = str(raw_price).strip()
    if not text or re.search(r"\d\s*[-–—]\s*\d|\bto\b", text, re.IGNORECASE):
        return None

    cleaned = re.sub(r"[^0-9.,]", "", text)
    if not cleaned:
        return None

    if "," in cleaned and "." in cleaned:
        decimal_separator = "," if cleaned.rfind(",") > cleaned.rfind(".") else "."
        grouping_separator = "." if decimal_separator == "," else ","
        integer_part, fractional_part = cleaned.rsplit(decimal_separator, 1)
        groups = integer_part.split(grouping_separator)
        if len(groups) > 1 and (
            not groups[0] or any(len(group) != 3 for group in groups[1:])
        ):
            return None
        if not fractional_part.isdigit() or len(fractional_part) not in (1, 2):
            return None
        cleaned = "".join(groups) + "." + fractional_part
    elif "," in cleaned or "." in cleaned:
        separator = "," if "," in cleaned else "."
        if cleaned.count(separator) > 1:
            groups = cleaned.split(separator)
            if not groups[0] or any(len(group) != 3 for group in groups[1:]):
                return None
            cleaned = "".join(groups)
        else:
            whole, fraction = cleaned.split(separator)
            if len(fraction) == 3:
                return None
            if not whole or not fraction or len(fraction) not in (1, 2):
                return None
            cleaned = whole + "." + fraction

    try:
        value = Decimal(cleaned)
        if not value.is_finite() or value <= 0:
            return None
        return str(value.quantize(Decimal("0.01")))
    except InvalidOperation:
        return None

def validate_record(row: dict[str, Any]) -> list[str]:
    """Return contract failures instead of assuming a normalized row is valid."""

    errors: list[str] = []

    for field in REQUIRED_FIELDS:
        value = row.get(field)
        if value is None or (isinstance(value, str) and not value.strip()):
            errors.append(f"{field}: required")

    marketplace = str(row.get("marketplace") or "").lower()
    parsed_url = urlparse(str(row.get("source_url") or ""))
    host = (parsed_url.hostname or "").lower()
    if parsed_url.scheme != "https":
        errors.append("source_url: must use https")
    if marketplace and host and not (host == marketplace or host.endswith(f".{marketplace}")):
        errors.append("source_url: hostname does not match marketplace")

    currency = row.get("currency")
    if not isinstance(currency, str) or not re.fullmatch(r"[A-Z]{3}", currency):
        errors.append("currency: expected a three-letter uppercase code")

    price_type = str(row.get("price_type") or "")
    if price_type not in VALID_PRICE_TYPES:
        errors.append("price_type: unsupported value")
    elif price_type == "unknown":
        errors.append("price_type: insufficient evidence")
    elif price_type in CONDITIONAL_PRICE_TYPES and not str(
        row.get("eligibility_context") or ""
    ).strip():
        errors.append("eligibility_context: required for conditional price")

    try:
        amount = Decimal(str(row.get("numeric_price")))
        if amount <= 0:
            errors.append("numeric_price: must be greater than zero")
    except Exception:
        errors.append("numeric_price: invalid decimal")

    if row.get("stock_status") not in VALID_STOCK_STATUSES:
        errors.append("stock_status: unsupported value")

    transport_status = str(row.get("transport_status") or "")
    http_status = row.get("http_status")
    if transport_status.startswith("http_") and not (
        isinstance(http_status, int) and 200 <= http_status < 300
    ):
        errors.append("transport_status: HTTP error cannot be valid")

    return errors


def main() -> None:
    source = json.loads(SOURCE.read_text(encoding="utf-8"))
    captured_at = datetime.now(timezone.utc).isoformat()
    rows = []

    for variant in source["variants"]:
        row: dict[str, Any] = {
            "marketplace": source["marketplace"],
            "source_url": source["source_url"],
            "product_id": source["product_id"],
            "title": source["title"],
            "seller_name": source["seller_name"],
            "variant_id": variant["variant_id"],
            "variant_label": variant["variant_label"],
            "raw_price": variant["raw_price"],
            "numeric_price": numeric_price(variant["raw_price"]),
            "currency": source["currency"],
            "price_type": variant["price_type"],
            "eligibility_context": variant["eligibility_context"],
            "delivery_context": variant["delivery_context"],
            "price_source": "fixture",
            "extraction_method": "fixture_normalizer",
            "original_price": variant["original_price"],
            "discount_text": variant["discount_text"],
            "stock_status": variant["stock_status"],
            "captured_at_utc": captured_at,
            "route_label": "fixture-no-network",
            "http_status": None,
            "transport_status": "offline_no_network",
            "status": "fixture_candidate",
            "validation_status": "pending",
            "validation_errors": [],
        }
        errors = validate_record(row)
        row["validation_errors"] = errors
        row["validation_status"] = "valid" if not errors else "invalid"
        row["status"] = "fixture_validated" if not errors else "fixture_rejected"
        rows.append(row)

    OUTPUT.write_text(json.dumps(rows, indent=2), encoding="utf-8")
    valid_count = sum(row["validation_status"] == "valid" for row in rows)
    print(f"Validated {valid_count} variant rows")
    print(f"Contract validation: {valid_count}/{len(rows)} passed")
    for row in rows:
        if row["validation_errors"]:
            print(
                f"Rejected {row['variant_id']}: "
                + "; ".join(row["validation_errors"])
            )
    print(f"Output: {OUTPUT}")


if __name__ == "__main__":
    main()

export ROLA_PROXY_SERVER="http://gateway.example:port"
export ROLA_PROXY_USERNAME="generated_username"
export ROLA_PROXY_PASSWORD="generated_password"
export TARGET_URL="https://www.lazada.sg/approved-product-url"
export LAZADA_MARKETPLACE="lazada.sg"
python examples/lazada_playwright_sample.py \
  --url "$TARGET_URL" \
  --approved-url "$TARGET_URL" \
  --marketplace "$LAZADA_MARKETPLACE" \
  --route-label "sg-residential-sticky-a" \
  --output lazada_product.json

$env:ROLA_PROXY_SERVER = "http://gateway.example:port"
$env:ROLA_PROXY_USERNAME = "generated_username"
$env:ROLA_PROXY_PASSWORD = "generated_password"
$env:TARGET_URL = "https://www.lazada.sg/approved-product-url"
$env:LAZADA_MARKETPLACE = "lazada.sg"
python .\examples\lazada_playwright_sample.py --url $env:TARGET_URL --approved-url $env:TARGET_URL --marketplace $env:LAZADA_MARKETPLACE --route-label "sg-residential-sticky-a" --output .\lazada_product.json

python examples/normalize_lazada_fixture.py