#!/usr/bin/env python3
"""Find diverse SKUs with populated-attribute gains; archive live product pages.

Python 3.10+. Database work is SELECT/SHOW plus session settings and short
read-only transactions. Run --inspect-schema before your first selection.
The PDFs document the LIVE website, not the unpublished conversion database.
"""

from __future__ import annotations

import argparse
import copy
import getpass
import hashlib
import html
import json
import logging
import math
import os
import re
import sys
import time
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote, unquote, urlsplit

LOG = logging.getLogger("heilind")
SCHEMA = {
    "product_table": "product_entity",
    "entity_id": "entity_id",
    "sku": "sku",
    "value_table": "product_entity_attribute_value",
    "value_entity_id": "entity_id",
    "value_attribute_id": "attribute_id",
    "value": "value",
    "attribute_table": "eav_attributes",
    "attribute_id": "attribute_id",
    "attribute_code": "attribute_code",
}
DEFAULTS = {
    "connection": {
        "host": "YOUR_DATABASE_HOST", "port": 3306, "user": "YOUR_READ_ONLY_USER",
        "password_env": "ECOM_DB_PASSWORD", "ssl_ca": None,
        "connect_timeout_seconds": 10, "query_timeout_seconds": 30,
    },
    "databases": {"prod": "ecom_prod", "conversion": "ecom_conversion"},
    "schema": SCHEMA,
    "schema_overrides": {},
    "manufacturer": {
        "source": "auto", "column": None, "attribute_code": None,
        "label_lookup": None,
        "missing_values": ["", "null", "none", "n/a", "unknown", "unbranded"],
    },
    "selection": {
        "target_skus": 25, "min_added_attributes": 5, "min_percent_increase": 50,
        "min_conversion_attributes": 8, "max_per_manufacturer": 3,
        "batch_size": 250, "scan_limit": 50000, "segments": 20,
        "stop_when_enough": True,
        "empty_values": ["", "null", "[]", "{}"],
        "exclude_attribute_codes": ["sku", "manufacturer", "manufacturer_code",
                                    "mfg", "mfg_code", "brand", "brand_code"],
    },
    "capture": {
        "url_template": "https://www.heilind.com/{sku}.html", "sku_case": "lower",
        "viewport_width": 1440, "viewport_height": 1000,
        "navigation_timeout_seconds": 45, "content_selector": "main",
        "content_timeout_seconds": 20, "settle_seconds": 2,
        "between_pages_seconds": 2, "max_scroll_steps": 150,
        "expand_selectors": [], "open_native_details": True,
        "pdf_layout": "continuous", "max_single_page_height_px": 18000,
        "save_html_snapshot": False, "browser_channel": None,
    },
}


class ConfigError(RuntimeError):
    pass


class CaptureBlocked(RuntimeError):
    """Access controls/rate limits stop the capture run, without retries."""


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def merge(base: dict, override: dict) -> dict:
    result = copy.deepcopy(base)
    for key, value in override.items():
        if isinstance(value, dict) and isinstance(result.get(key), dict):
            result[key] = merge(result[key], value)
        else:
            result[key] = copy.deepcopy(value)
    return result


def ident(value: str) -> str:
    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", value):
        raise ConfigError(f"Invalid SQL identifier: {value!r}")
    return f"`{value}`"


def table(db: str, name: str) -> str:
    return f"{ident(db)}.{ident(name)}"


def text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="strict")
    return str(value).strip()


def populated(value: Any, empty: set[str]) -> bool:
    # 0 and False are real values, not missing values.
    return value is not None and text(value).casefold() not in empty


def product_url(sku: str, capture: dict) -> str:
    slug = sku.strip()
    case = capture["sku_case"]
    if case == "lower":
        slug = slug.lower()
    elif case == "upper":
        slug = slug.upper()
    elif case != "preserve":
        raise ConfigError("capture.sku_case must be lower, upper, or preserve")
    return capture["url_template"].format(sku=quote(slug, safe=""))


def safe_filename(rank: int, sku: str) -> str:
    stem = re.sub(r"[^A-Za-z0-9._-]+", "_", sku).strip("._")[:90] or "sku"
    digest = hashlib.sha256(sku.encode("utf-8")).hexdigest()[:8]
    return f"{rank:02d}_{stem}_{digest}"


def attribute_sets(rows: list[dict], codes: dict, settings: dict) -> dict:
    empty = {text(v).casefold() for v in settings["empty_values"]} | {""}
    excluded = {text(v).casefold() for v in settings["exclude_attribute_codes"]}
    result = defaultdict(set)
    for row in rows:
        code = codes.get(row["attribute_id"])
        if code and code.casefold() not in excluded and populated(row["raw_value"], empty):
            # Count the attribute, not the number of option IDs or repeated rows.
            result[row["entity_id"]].add(code)
    return result


def qualifies(prod: int, conversion: int, settings: dict) -> bool:
    gain = conversion - prod
    return (
        gain >= settings["min_added_attributes"]
        and conversion >= settings["min_conversion_attributes"]
        and (prod == 0 or gain * 100 >= prod * settings["min_percent_increase"])
    )


def rank_key(row: dict) -> tuple:
    # Absolute gain first; conversion breadth second; stable SKU tie-break.
    return (-row["added_attributes"], -row["conversion_count"], row["sku"].casefold())


def choose_diverse(candidates: list[dict], target: int, cap: int) -> list[dict]:
    groups = defaultdict(list)
    seen = set()
    for row in sorted(candidates, key=rank_key):
        sku_key = row["sku"].casefold()
        if sku_key in seen:
            continue
        seen.add(sku_key)
        groups[row["manufacturer_key"]].append(row)
    groups = {key: sorted(rows, key=rank_key)[:cap] for key, rows in groups.items()}
    result = []
    # One per manufacturer before a second from any manufacturer, etc.
    for depth in range(cap):
        layer = [rows[depth] for rows in groups.values() if len(rows) > depth]
        for row in sorted(layer, key=rank_key):
            result.append(copy.deepcopy(row))
            if len(result) == target:
                return result
    return result


def ranges(low: int, high: int, segments: int) -> list[list[int]]:
    if high < low:
        return []
    size = max(1, math.ceil((high - low + 1) / segments))
    return [[start, min(start + size - 1, high)] for start in range(low, high + 1, size)]


def query(conn, sql: str, params=()) -> list[dict]:
    with conn.cursor() as cursor:
        cursor.execute(sql, params)
        return cursor.fetchall()


def load_config(path: Path) -> dict:
    with path.open(encoding="utf-8-sig") as stream:
        config = merge(DEFAULTS, json.load(stream))
    s = config["selection"]
    for key in ("target_skus", "min_added_attributes", "max_per_manufacturer",
                "batch_size", "scan_limit", "segments"):
        if not isinstance(s[key], int) or isinstance(s[key], bool) or s[key] < 1:
            raise ConfigError(f"selection.{key} must be a positive integer")
    if s["min_percent_increase"] < 0 or s["min_conversion_attributes"] < 1:
        raise ConfigError("Invalid selection thresholds")
    if s["batch_size"] > 2000 or s["segments"] > 200:
        raise ConfigError("Use batch_size <= 2000 and segments <= 200")
    if "{sku}" not in config["capture"]["url_template"]:
        raise ConfigError("capture.url_template must contain {sku}")
    if config["capture"]["pdf_layout"] not in ("continuous", "paged"):
        raise ConfigError("capture.pdf_layout must be continuous or paged")
    for key in ("viewport_width", "viewport_height", "navigation_timeout_seconds",
                "content_timeout_seconds", "max_scroll_steps", "max_single_page_height_px"):
        if not isinstance(config["capture"][key], (int, float)) or config["capture"][key] <= 0:
            raise ConfigError(f"capture.{key} must be positive")
    for key in ("settle_seconds", "between_pages_seconds"):
        if not isinstance(config["capture"][key], (int, float)) or config["capture"][key] < 0:
            raise ConfigError(f"capture.{key} must not be negative")
    # Validate URL formatting early; values are escaped as a path component.
    product_url("COC120X10149X", config["capture"])
    return config


def connect(config: dict):
    try:
        import pymysql
    except ImportError as exc:
        raise ConfigError("Install dependencies: py -m pip install -r requirements.txt") from exc
    c = config["connection"]
    if c["host"].startswith("YOUR_") or c["user"].startswith("YOUR_"):
        raise ConfigError("Set connection.host and connection.user in your config first")
    password = os.environ.get(c["password_env"])
    if password is None:
        password = getpass.getpass(f"Database password for {c['user']}: ")
    kwargs = {
        "host": c["host"], "port": int(c["port"]), "user": c["user"],
        "password": password, "charset": "utf8mb4", "autocommit": True,
        "cursorclass": pymysql.cursors.DictCursor,
        "connect_timeout": int(c["connect_timeout_seconds"]),
        "read_timeout": int(c["query_timeout_seconds"]) + 15,
        "write_timeout": 15,
    }
    if c.get("ssl_ca"):
        kwargs.update(ssl_ca=c["ssl_ca"], ssl_verify_cert=True, ssl_verify_identity=True)
    conn = pymysql.connect(**kwargs)
    try:
        version = str(query(conn, "SELECT VERSION() AS version")[0]["version"])
        timeout = int(c["query_timeout_seconds"])
        if timeout < 1:
            raise ConfigError("query_timeout_seconds must be positive")
        if "mariadb" in version.lower():
            query(conn, "SET SESSION max_statement_time = %s", (timeout,))
        else:
            query(conn, "SET SESSION max_execution_time = %s", (timeout * 1000,))
        query(conn, "SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")
        LOG.info("Connected. Server-side SELECT timeout: %s seconds", timeout)
    except Exception:
        conn.close()
        raise
    return conn


class Database:
    def __init__(self, conn, name: str, schema: dict):
        self.conn, self.name, self.s = conn, name, schema
        for val in schema.values():
            ident(val)
        ident(name)
        self.codes = {}

    def t(self, key: str) -> str:
        return table(self.name, self.s[key])

    def columns(self, name: str) -> list[dict]:
        return query(self.conn, f"SHOW COLUMNS FROM {table(self.name, name)}")

    def indexes(self, name: str) -> list[dict]:
        return query(self.conn, f"SHOW INDEX FROM {table(self.name, name)}")

    def inspect(self) -> dict:
        names = [self.s[k] for k in ("product_table", "value_table", "attribute_table")]
        result = {}
        for name in names:
            result[name] = {"columns": self.columns(name), "indexes": self.indexes(name)}
        # This table is relevant to optional manufacturer-ID labels, if it exists.
        found = query(self.conn,
            "SELECT TABLE_NAME FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s",
            (self.name, "eav_attribute_option_value"))
        if found:
            result["eav_attribute_option_value"] = {
                "columns": self.columns("eav_attribute_option_value"),
                "indexes": self.indexes("eav_attribute_option_value")}
        return result

    def preflight(self):
        groups = {
            "product_table": ["entity_id", "sku"],
            "value_table": ["value_entity_id", "value_attribute_id", "value"],
            "attribute_table": ["attribute_id", "attribute_code"],
        }
        metadata = {}
        for key, fields in groups.items():
            cols = {r["Field"]: r for r in self.columns(self.s[key])}
            metadata[key] = cols
            missing = [self.s[f] for f in fields if self.s[f] not in cols]
            if missing:
                raise ConfigError(f"{self.name}.{self.s[key]} missing {missing}; run --inspect-schema")
        dtype = metadata["product_table"][self.s["entity_id"]]["Type"].lower()
        if not re.match(r"(tinyint|smallint|mediumint|int|bigint)\b", dtype):
            raise ConfigError("Keyset sampling requires an integer product entity_id")
        for key, col in (("product_table", "entity_id"), ("product_table", "sku"),
                         ("value_table", "value_entity_id")):
            indexes = self.indexes(self.s[key])
            leading = [r for r in indexes if r["Column_name"] == self.s[col]
                       and r["Seq_in_index"] == 1 and not r.get("Sub_part")
                       and str(r.get("Visible", "YES")).upper() != "NO"
                       and str(r.get("Ignored", "NO")).upper() != "YES"]
            if not leading:
                raise ConfigError(f"Missing usable leading index on {self.name}.{self.s[key]}"
                                  f"({self.s[col]}). Stop and review indexes with the DBA.")
            if col == "entity_id":
                unique = [r for r in leading if r["Non_unique"] == 0 and
                          sum(x["Key_name"] == r["Key_name"] for x in indexes) == 1]
                if not unique:
                    raise ConfigError(f"{self.name}: entity_id needs a single-column UNIQUE index")
        s = self.s
        rows = query(self.conn, f"SELECT {ident(s['attribute_id'])} AS attribute_id, "
                     f"{ident(s['attribute_code'])} AS code FROM {self.t('attribute_table')}")
        for row in rows:
            code = text(row["code"])
            if row["attribute_id"] in self.codes:
                raise ConfigError(f"Duplicate attribute ID in {self.name}")
            self.codes[row["attribute_id"]] = code

    def bounds(self):
        col = ident(self.s["entity_id"])
        row = query(self.conn, f"SELECT MIN({col}) AS lo, MAX({col}) AS hi "
                    f"FROM {self.t('product_table')}")[0]
        return row["lo"], row["hi"]

    def product_batch(self, start: int, end: int, limit: int, mfg_column=None):
        s = self.s
        extra = f", {ident(mfg_column)} AS mfg_raw" if mfg_column else ""
        return query(self.conn, f"SELECT {ident(s['entity_id'])} AS entity_id, "
                     f"{ident(s['sku'])} AS sku{extra} FROM {self.t('product_table')} "
                     f"WHERE {ident(s['entity_id'])} >= %s AND {ident(s['entity_id'])} <= %s "
                     f"ORDER BY {ident(s['entity_id'])} LIMIT %s", (start, end, limit))

    def by_skus(self, skus: list[str]) -> dict:
        if not skus:
            return {}
        s = self.s
        marks = ",".join(["%s"] * len(skus))
        rows = query(self.conn, f"SELECT {ident(s['entity_id'])} AS entity_id, "
                     f"{ident(s['sku'])} AS sku FROM {self.t('product_table')} "
                     f"WHERE {ident(s['sku'])} IN ({marks})", skus)
        result = {}
        for row in rows:
            key = text(row["sku"]).casefold()
            if key in result:
                raise ConfigError(f"Duplicate SKU in {self.name}: {row['sku']!r}")
            result[key] = row
        return result

    def values(self, ids: list) -> list[dict]:
        if not ids:
            return []
        s = self.s
        marks = ",".join(["%s"] * len(ids))
        return query(self.conn, f"SELECT {ident(s['value_entity_id'])} AS entity_id, "
                     f"{ident(s['value_attribute_id'])} AS attribute_id, "
                     f"{ident(s['value'])} AS raw_value FROM {self.t('value_table')} "
                     f"WHERE {ident(s['value_entity_id'])} IN ({marks}) "
                     f"AND {ident(s['value'])} IS NOT NULL", ids)


class Manufacturer:
    """Use actual conversion manufacturer data; never infer a SKU prefix."""
    COLUMNS = ("manufacturer_code", "mfg_code", "brand_code", "manufacturer", "mfg", "brand")
    ATTRIBUTES = ("manufacturer", "manufacturer_code", "mfg", "mfg_code", "brand", "brand_code")

    def __init__(self, db: Database, settings: dict):
        self.db, self.settings = db, copy.deepcopy(settings)
        self.source, self.column, self.attribute_id = settings["source"], None, None
        cols = {r["Field"] for r in db.columns(db.s["product_table"])}
        attr_matches = [(aid, code) for aid, code in db.codes.items()
                        if code.casefold() in self.ATTRIBUTES]
        if self.source == "auto":
            options = [("column", col) for col in self.COLUMNS if col in cols]
            if not options:
                options = [("attribute", code) for _, code in attr_matches]
            if len(options) != 1:
                raise ConfigError("Manufacturer source is missing or ambiguous. Set "
                                  "manufacturer.source to column or attribute and set column "
                                  "or attribute_code. Run --inspect-schema for column names. "
                                  f"Candidates: {options}")
            self.source, selected = options[0]
            self.settings["column" if self.source == "column" else "attribute_code"] = selected
        if self.source == "column":
            self.column = self.settings["column"]
            if self.column not in cols:
                raise ConfigError(f"Manufacturer column not found: {self.column!r}")
            self.description = f"{db.name}.{db.s['product_table']}.{self.column}"
        elif self.source == "attribute":
            code = self.settings["attribute_code"]
            matches = [aid for aid, value in db.codes.items() if value == code]
            if len(matches) != 1:
                raise ConfigError(f"Manufacturer attribute_code must identify one attribute: {code!r}")
            self.attribute_id = matches[0]
            self.description = f"{db.name} EAV attribute {code} (ID {self.attribute_id})"
        else:
            raise ConfigError("manufacturer.source must be auto, column, or attribute")
        self.missing = {text(v).casefold() for v in settings["missing_values"]} | {""}
        self.labels = {}
        if self.settings.get("label_lookup"):
            self.load_labels()
        LOG.info("Manufacturer grouping: %s", self.description)

    def load_labels(self):
        cfg = self.settings["label_lookup"]
        # Mapping is explicit: numeric product values are never guessed to be option IDs.
        source = table(self.db.name, cfg["table"])
        id_col, label_col = ident(cfg["id_column"]), ident(cfg["label_column"])
        filters, params = [], []
        if self.source == "attribute":
            if not cfg.get("attribute_id_column"):
                raise ConfigError("EAV label_lookup needs attribute_id_column for ownership. "
                                  "Leave label_lookup null to group by the stored manufacturer key.")
            filters.append(f"{ident(cfg['attribute_id_column'])} = %s")
            params.append(self.attribute_id)
        for col, value in cfg.get("filters", {}).items():
            filters.append(f"{ident(col)} = %s")
            params.append(value)
        where = " WHERE " + " AND ".join(filters) if filters else ""
        rows = query(self.db.conn, f"SELECT {id_col} AS mfg_id, {label_col} AS label "
                     f"FROM {source}{where} LIMIT 10001", params)
        if len(rows) > 10000:
            raise ConfigError("Manufacturer label lookup exceeds 10,000 rows; narrow its filters")
        for row in rows:
            key, value = text(row["mfg_id"]), text(row["label"])
            if key in self.labels and self.labels[key] != value:
                raise ConfigError("Ambiguous manufacturer labels; set label_lookup.filters for locale/store")
            self.labels[key] = value

    def for_products(self, products: list[dict], rows: list[dict]) -> dict:
        raw = defaultdict(set)
        if self.source == "column":
            for p in products:
                raw[p["entity_id"]].add(text(p.get("mfg_raw")))
        else:
            for row in rows:
                if row["attribute_id"] == self.attribute_id:
                    raw[row["entity_id"]].add(text(row["raw_value"]))
        result = {}
        for entity, values in raw.items():
            values = {v for v in values if v.casefold() not in self.missing}
            if len(values) != 1:
                continue  # Missing/ambiguous manufacturer is excluded, not a fake group.
            value = next(iter(values))
            if re.fullmatch(r"\s*\d+\s*,\s*\d+(?:\s*,\s*\d+)*\s*", value):
                continue  # Multiple stored option IDs are not one manufacturer.
            label = self.labels.get(value)
            if self.settings.get("label_lookup") and not label:
                continue  # Do not claim a resolved manufacturer for an orphaned ID.
            if label is None:
                label = f"Stored manufacturer key {value}" if value.isdigit() else value
            result[entity] = (value.casefold(), label)
        return result


def scan(conn, config: dict) -> dict:
    settings = config["selection"]
    sides = {}
    for side in ("prod", "conversion"):
        schema = merge(config["schema"], config["schema_overrides"].get(side, {}))
        sides[side] = Database(conn, config["databases"][side], schema)
        sides[side].preflight()
    prod, conv = sides["prod"], sides["conversion"]
    manufacturer = Manufacturer(conv, config["manufacturer"])
    low, high = conv.bounds()
    work = ranges(int(low), int(high), settings["segments"]) if low is not None else []
    stats = {"scanned_conversion_products": 0, "shared_products": 0,
             "qualifying_products": 0, "qualifying_without_manufacturer": 0,
             "unmapped_attribute_rows": 0, "started_at": now()}
    pool, seen = {}, set()
    stop_reason = "conversion ID ranges exhausted"
    while work and stats["scanned_conversion_products"] < settings["scan_limit"]:
        remaining_ranges = []
        for start, end in work:
            remaining = settings["scan_limit"] - stats["scanned_conversion_products"]
            if remaining <= 0:
                break
            size = min(settings["batch_size"], remaining)
            query(conn, "START TRANSACTION READ ONLY")
            try:
                products = conv.product_batch(start, end, size, manufacturer.column)
                if not products:
                    continue
                next_id = int(products[-1]["entity_id"]) + 1
                if len(products) == size and next_id <= end:
                    remaining_ranges.append([next_id, end])
                for p in products:
                    p["sku"] = text(p["sku"])
                    key = p["sku"].casefold()
                    if key in seen:
                        raise ConfigError(f"Duplicate conversion SKU encountered: {p['sku']!r}")
                    seen.add(key)
                stats["scanned_conversion_products"] += len(products)
                matched = prod.by_skus([p["sku"] for p in products if p["sku"]])
                shared = [p for p in products if p["sku"].casefold() in matched]
                stats["shared_products"] += len(shared)
                c_rows = conv.values([p["entity_id"] for p in shared])
                p_rows = prod.values([matched[p["sku"].casefold()]["entity_id"] for p in shared])
                c_sets = attribute_sets(c_rows, conv.codes, settings)
                p_sets = attribute_sets(p_rows, prod.codes, settings)
                mfgs = manufacturer.for_products(shared, c_rows)
                stats["unmapped_attribute_rows"] += sum(r["attribute_id"] not in conv.codes for r in c_rows)
                stats["unmapped_attribute_rows"] += sum(r["attribute_id"] not in prod.codes for r in p_rows)
                measured_at = now()
                for p in shared:
                    pid = matched[p["sku"].casefold()]["entity_id"]
                    p_codes, c_codes = p_sets.get(pid, set()), c_sets.get(p["entity_id"], set())
                    pn, cn = len(p_codes), len(c_codes)
                    if not qualifies(pn, cn, settings):
                        continue
                    if p["entity_id"] not in mfgs:
                        stats["qualifying_without_manufacturer"] += 1
                        continue
                    stats["qualifying_products"] += 1
                    mfg_key, label = mfgs[p["entity_id"]]
                    row = {"sku": p["sku"], "manufacturer": label, "manufacturer_key": mfg_key,
                           "prod_entity_id": pid, "conversion_entity_id": p["entity_id"],
                           "prod_count": pn, "conversion_count": cn, "added_attributes": cn - pn,
                           "percent_increase": round((cn - pn) / pn * 100, 2) if pn else None,
                           "prod_populated_attribute_codes": sorted(p_codes),
                           "conversion_populated_attribute_codes": sorted(c_codes),
                           "measured_at": measured_at,
                           "url": product_url(p["sku"], config["capture"]),
                           "capture_status": "pending"}
                    pool.setdefault(mfg_key, []).append(row)
                    pool[mfg_key] = sorted(pool[mfg_key], key=rank_key)[:settings["max_per_manufacturer"]]
            finally:
                conn.rollback()  # End short snapshot; never leave a long read transaction open.
            LOG.info("Scanned %s | shared %s | qualifying %s | manufacturers %s",
                     stats["scanned_conversion_products"], stats["shared_products"],
                     stats["qualifying_products"], len(pool))
        work = remaining_ranges
        available = sum(len(v) for v in pool.values())
        # Wait until one pass across all configured ID ranges before stopping.
        if available >= settings["target_skus"] and settings["stop_when_enough"]:
            stop_reason = "enough qualifying, manufacturer-diverse candidates after an ID-range pass"
            break
    else:
        if stats["scanned_conversion_products"] >= settings["scan_limit"]:
            stop_reason = "configured scan limit reached"
    candidates = [row for group in pool.values() for row in group]
    selected = choose_diverse(candidates, settings["target_skus"], settings["max_per_manufacturer"])
    warnings = []
    if len(selected) < settings["target_skus"]:
        warnings.append(f"Only {len(selected)} of {settings['target_skus']} requested SKUs qualified "
                        "within the scan and manufacturer limits. Thresholds were not relaxed.")
    if stats["qualifying_without_manufacturer"]:
        warnings.append(f"{stats['qualifying_without_manufacturer']} qualifying products lacked one "
                        "usable manufacturer value and were excluded.")
    if stats["unmapped_attribute_rows"]:
        warnings.append(f"{stats['unmapped_attribute_rows']} value rows had unmapped attribute IDs "
                        "and were excluded from counts.")
    stats.update(finished_at=now(), stop_reason=stop_reason)
    return {"format_version": 1, "databases": config["databases"],
            "manufacturer_source": manufacturer.description, "selection_settings": settings,
            "capture_settings": config["capture"], "scan": stats, "warnings": warnings,
            "selected": selected}


def write_json(path: Path, payload: dict):
    temporary = path.with_suffix(path.suffix + ".tmp")
    with temporary.open("w", encoding="utf-8") as stream:
        json.dump(payload, stream, indent=2, ensure_ascii=False, default=str)
        stream.write("\n")
    temporary.replace(path)


def summary_html(manifest: dict) -> str:
    esc = lambda value: html.escape(str(value), quote=True)
    rows = manifest["selected"]
    s = manifest["selection_settings"]
    makers = len({r["manufacturer_key"] for r in rows})
    body = []
    for i, r in enumerate(rows, 1):
        pct = "From zero" if r["percent_increase"] is None else f"+{r['percent_increase']:g}%"
        pdf = r.get("pdf_file")
        link = f'<a href="{esc(pdf)}">PDF</a>' if pdf else "-"
        body.append(f"<tr><td>{i}</td><td><a href=\"{esc(r['url'])}\">{esc(r['sku'])}</a></td>"
                    f"<td>{esc(r['manufacturer'])}</td><td>{r['prod_count']}</td>"
                    f"<td>{r['conversion_count']}</td><td class='gain'>+{r['added_attributes']}</td>"
                    f"<td>{pct}</td><td>{esc(r.get('capture_status', 'pending'))}<br>{link}</td></tr>")
    notices = "".join(f"<li>{esc(w)}</li>" for w in manifest.get("warnings", []))
    errors = "".join(f"<li><b>{esc(r['sku'])}</b>: {esc(r.get('capture_error') or r.get('capture_note'))}</li>"
                     for r in rows if r.get("capture_error") or r.get("capture_note"))
    return f'''<!doctype html>
<html lang="en"><head><meta charset="utf-8"><title>Heilind SKU enrichment selection</title>
<style>
@page {{ size: A4 landscape; margin: 15mm; }}
* {{ box-sizing: border-box; }}
body {{ color:#172e44; background:#fff; font:14px/1.5 Arial,sans-serif; margin:36px auto; max-width:1250px; padding:0 22px; }}
h1 {{ font-size:30px; line-height:1.15; margin:8px 0 12px; }}
.eyebrow {{ text-transform:uppercase; letter-spacing:2px; color:#44718e; font-size:11px; }}
.muted {{ color:#526579; }} .notice {{ background:#edf5fa; padding:14px 18px; border-left:4px solid #347ca2; }}
.metrics {{ display:flex; gap:36px; margin:24px 0; }} .metrics strong {{ font-size:27px; display:block; }}
table {{ border-collapse:collapse; width:100%; table-layout:fixed; font-size:12px; }}
th {{ color:#fff; background:#173d59; text-align:left; padding:10px 8px; }}
td {{ border-bottom:1px solid #dce4ea; padding:9px 8px; vertical-align:top; overflow-wrap:anywhere; }}
tr:nth-child(even) {{ background:#f5f8fa; }} tr {{ break-inside:avoid; }} thead {{ display:table-header-group; }}
a {{ color:#1c628c; text-decoration:none; }} .gain {{ font-weight:bold; color:#147154; }}
.notes {{ margin-top:22px; font-size:12px; }} h2 {{ font-size:18px; break-after:avoid; }}
@media print {{ body {{ margin:0; max-width:none; padding:0; }} .metrics {{ margin:14px 0; }} th,td {{ padding:7px; }} }}
</style></head><body>
<div class="eyebrow">Product data / selected examples</div><h1>Populated-attribute gains</h1>
<p class="muted">{esc(manifest['databases']['conversion'])} compared with {esc(manifest['databases']['prod'])}</p>
<div class="notice">The counts come from the databases. Product PDFs capture the current live website,
not a preview of conversion data. More populated fields do not by themselves prove accuracy or completeness.</div>
<div class="metrics"><div><strong>{len(rows)}</strong>selected SKUs</div><div><strong>{makers}</strong>manufacturer groups</div>
<div><strong>+{s['min_added_attributes']}</strong>minimum attribute gain</div><div><strong>{s['min_percent_increase']}%</strong>minimum increase*</div></div>
<table><colgroup><col style="width:4%"><col style="width:22%"><col style="width:22%"><col style="width:8%"><col style="width:10%"><col style="width:8%"><col style="width:10%"><col style="width:16%"></colgroup>
<thead><tr><th>#</th><th>SKU / live page</th><th>Manufacturer / key</th><th>Prod</th><th>Conversion</th><th>Gain</th><th>Increase</th><th>Capture</th></tr></thead>
<tbody>{''.join(body)}</tbody></table>
<div class="notes"><h2>Selection notes</h2><p>Each distinct populated attribute code is counted once per SKU within its own database.
Multiple selected options and duplicate value rows do not increase the count. Zero is a value.
Configured empty tokens and excluded attribute codes are not counted. Stored option IDs count as populated;
their labels and business correctness are not validated by this selection.</p>
<p>*For a zero production count, the percentage is undefined; the absolute-gain and conversion-minimum tests still apply.
Minimum conversion count: {s['min_conversion_attributes']}. Maximum per manufacturer: {s['max_per_manufacturer']}.
Manufacturer source: {esc(manifest['manufacturer_source'])}.</p>
<p>Excluded codes: {esc(', '.join(s['exclude_attribute_codes']) or '(none)')}. Empty tokens: {esc(repr(s['empty_values']))}.</p>
<p>This is a deterministic, ID-range-based search for examples, not a random sample or a global top-25 ranking.
SKU overlap uses case-insensitive matching in Python after database lookup; attribute codes remain exact.
Separate batches use short read-only snapshots, not one catalog-wide snapshot.</p>
<p>Scanned {manifest['scan']['scanned_conversion_products']:,} conversion products; {manifest['scan']['shared_products']:,} shared.
Stop reason: {esc(manifest['scan']['stop_reason'])}. Measurement window: {esc(manifest['scan']['started_at'])} to {esc(manifest['scan']['finished_at'])}.</p>
<p>Changed or split attribute definitions can affect the gain. Inspect attribute-code lists in selection.json before treating these examples as like-for-like enrichment.</p>
{'<h2>Warnings</h2><ul>'+notices+'</ul>' if notices else ''}
{'<h2>Capture notes</h2><ul>'+errors+'</ul>' if errors else ''}</div></body></html>'''


def checkpoint(out: Path, manifest: dict):
    write_json(out / "selection.json", manifest)
    (out / "selection_summary.html").write_text(summary_html(manifest), encoding="utf-8")


def check_page(page, status: int | None):
    if status in (401, 403, 429):
        raise CaptureBlocked(f"HTTP {status}; capture stopped. Resolve access/rate limits with the site owner.")
    if status is None or status >= 400:
        raise RuntimeError(f"Product page returned HTTP {status}")
    title = page.title()
    h1 = page.locator("h1").first.inner_text(timeout=3000) if page.locator("h1").count() else ""
    heading = f"{title}\n{h1}".casefold()
    if re.search(r"access denied|verify (?:you are|you're) human|just a moment|captcha|security check", heading):
        raise CaptureBlocked("Access/challenge page detected; capture stopped without attempting a bypass.")
    if re.search(r"page (?:was )?not found|404 not found|requested page.*not found|^404\b", heading):
        raise RuntimeError("A not-found page was returned instead of a product page")
    return title


def prepare_page(page, capture: dict):
    page.emulate_media(media="screen")
    selector = capture.get("content_selector")
    if selector:
        page.locator(selector).first.wait_for(state="visible", timeout=capture["content_timeout_seconds"] * 1000)
    for selector in capture.get("expand_selectors", []):
        # Only explicit, user-configured read-only product-content controls.
        page.locator(selector).first.click(timeout=10000)
    if capture.get("open_native_details"):
        page.locator("details").evaluate_all("nodes => nodes.forEach(node => node.open = true)")
    page.wait_for_timeout(capture["settle_seconds"] * 1000)
    bottom_stable = 0
    for _ in range(capture["max_scroll_steps"]):
        pos = page.evaluate("""() => ({y:window.scrollY, viewport:window.innerHeight,
            height:Math.max(document.body.scrollHeight,document.documentElement.scrollHeight)})""")
        at_bottom = pos["y"] + pos["viewport"] >= pos["height"] - 3
        bottom_stable = bottom_stable + 1 if at_bottom else 0
        if bottom_stable >= 3:
            break
        page.evaluate("window.scrollBy(0, Math.max(400, window.innerHeight * 0.8))")
        page.wait_for_timeout(200)
    else:
        raise RuntimeError("Scroll limit reached; complete lazy-loaded page could not be confirmed")
    readiness = page.evaluate("""async () => {
        document.querySelectorAll('img[loading="lazy"]').forEach(img => img.loading = 'eager');
        const ready = () => Array.from(document.images).every(img => img.complete);
        let timedOut = false;
        await Promise.race([
          Promise.all([document.fonts.ready, new Promise(resolve => {
            if (ready()) return resolve();
            const timer = setInterval(() => {if(ready()){clearInterval(timer);resolve();}},100);
            setTimeout(() => {clearInterval(timer);resolve();},9900);
          })]),
          new Promise(resolve => setTimeout(() => {timedOut=true;resolve();},10000))
        ]);
        return {pending:Array.from(document.images).filter(img=>!img.complete).length,
          broken:Array.from(document.images).filter(img=>img.complete && img.naturalWidth===0 && !!img.currentSrc).length,
          timedOut};
    }""")
    page.evaluate("window.scrollTo(0, 0)")
    page.wait_for_timeout(300)
    return readiness


def print_full_page(page, path: Path, capture: dict) -> str:
    page.emulate_media(media="screen")
    page.add_style_tag(content="html,body{overflow:visible!important;} *{-webkit-print-color-adjust:exact!important;print-color-adjust:exact!important;}")
    dims = page.evaluate("""() => ({width:Math.max(document.documentElement.scrollWidth,document.body.scrollWidth,window.innerWidth),
        height:Math.max(document.documentElement.scrollHeight,document.body.scrollHeight)})""")
    width, height = int(math.ceil(dims["width"])), int(math.ceil(dims["height"])) + 4
    layout = capture["pdf_layout"]
    if height > capture["max_single_page_height_px"]:
        layout = "paged"
    # A continuous PDF uses the document height, not the viewport height.
    # Very tall pages are paginated without limiting the page range.
    paper_height = height if layout == "continuous" else math.ceil(width * 11 / 8.5)
    page.pdf(path=str(path), width=f"{width}px", height=f"{paper_height}px",
             print_background=True, prefer_css_page_size=False, page_ranges="",
             margin={"top": "0", "right": "0", "bottom": "0", "left": "0"})
    return layout


def validate_pdf(path: Path) -> int:
    from pypdf import PdfReader
    reader = PdfReader(str(path))
    if not reader.pages:
        raise RuntimeError("PDF has no pages")
    return len(reader.pages)


def capture_one(context, row: dict, rank: int, out: Path, capture: dict):
    page = context.new_page()
    status = None
    try:
        page.set_default_navigation_timeout(capture["navigation_timeout_seconds"] * 1000)
        page.set_default_timeout(capture["content_timeout_seconds"] * 1000)
        response = page.goto(row["url"], wait_until="domcontentloaded")
        status = response.status if response else None
        check_page(page, status)
        requested, landed = urlsplit(row["url"]), urlsplit(page.url)
        if requested.hostname != landed.hostname or unquote(requested.path).casefold() != unquote(landed.path).casefold():
            raise RuntimeError(f"Unexpected redirect to {page.url}; verify this SKU's URL manually")
        readiness = prepare_page(page, capture)
        title = check_page(page, status)
        body = page.locator("body").inner_text(timeout=10000)
        if len(body.strip()) < 100:
            raise RuntimeError("Page is nearly empty; no product PDF saved")
        notes = []
        if row["sku"].casefold() not in body.casefold():
            notes.append("Full SKU was not found in visible page text; verify product identity in the PDF.")
        if readiness["pending"] or readiness["broken"] or readiness["timedOut"]:
            notes.append(f"Image/font readiness needs review: {readiness}")
        name = safe_filename(rank, row["sku"])
        relative = f"product_pdfs/{name}.pdf"
        path = out / relative
        partial = path.with_suffix(".partial.pdf")
        layout = print_full_page(page, partial, capture)
        page_count = validate_pdf(partial)
        partial.replace(path)
        if capture.get("save_html_snapshot"):
            (out / "html_snapshots").mkdir(exist_ok=True)
            (out / "html_snapshots" / f"{name}.html").write_text(page.content(), encoding="utf-8")
        row.update(capture_status="saved_needs_review" if notes else "saved",
                   capture_note=" ".join(notes), captured_at=now(), pdf_file=relative,
                   pdf_pages=page_count, pdf_layout=layout, page_title=title,
                   final_url=page.url, http_status=status)
        row.pop("capture_error", None)
    except CaptureBlocked:
        raise
    except Exception:
        # A client-rendered challenge may appear after navigation but before
        # the product-content selector becomes visible. Treat it as a stop,
        # not an ordinary selector timeout followed by more site requests.
        try:
            check_page(page, status)
        except CaptureBlocked:
            raise
        except Exception:
            pass
        raise
    finally:
        page.close()


def export_summary(context, out: Path):
    page = context.new_page()
    try:
        # The summary is local and self-contained; do not fetch linked live pages.
        page.goto((out / "selection_summary.html").resolve().as_uri())
        page.emulate_media(media="print")
        page.pdf(path=str(out / "selection_summary.pdf"), format="A4", landscape=True,
                 print_background=True, prefer_css_page_size=True, display_header_footer=True,
                 header_template="<span></span>",
                 footer_template='<div style="font:9px Arial;text-align:right;width:100%;padding-right:40px;color:#526579">'
                                 'Selection summary · <span class="pageNumber"></span> / <span class="totalPages"></span></div>')
        validate_pdf(out / "selection_summary.pdf")
    finally:
        page.close()


def combine_pdfs(out: Path, manifest: dict):
    from pypdf import PdfWriter
    writer = PdfWriter()
    try:
        summary = out / "selection_summary.pdf"
        if summary.exists():
            writer.append(str(summary), outline_item="Selection summary")
        for row in manifest["selected"]:
            if row.get("pdf_file") and row.get("capture_status") in ("saved", "saved_needs_review"):
                writer.append(str(out / row["pdf_file"]), outline_item=f"{row['sku']} - {row['manufacturer']}")
        if writer.pages:
            writer.add_metadata({"/Title": "Heilind selected SKU live-page captures"})
            with (out / "all_selected_product_pages.pdf").open("wb") as stream:
                writer.write(stream)
    finally:
        writer.close()


def capture_all(out: Path, manifest: dict, config: dict):
    try:
        from playwright.sync_api import sync_playwright
        import pypdf  # noqa: F401
    except ImportError as exc:
        raise ConfigError("Install dependencies: py -m pip install -r requirements.txt") from exc
    capture = config["capture"]
    manifest["capture_settings"] = copy.deepcopy(capture)
    (out / "product_pdfs").mkdir(exist_ok=True)
    with sync_playwright() as playwright:
        launch = {"headless": True}
        if capture.get("browser_channel"):
            launch["channel"] = capture["browser_channel"]
        browser = playwright.chromium.launch(**launch)
        context = browser.new_context(viewport={"width": capture["viewport_width"],
                                                "height": capture["viewport_height"]})
        try:
            for index, row in enumerate(manifest["selected"], 1):
                previous = row.get("pdf_file")
                if previous and row.get("capture_status") in ("saved", "saved_needs_review") and (out / previous).is_file():
                    try:
                        validate_pdf(out / previous)
                        LOG.info("Already captured: %s", row["sku"])
                        continue
                    except Exception:
                        LOG.warning("Existing PDF is invalid; recapturing %s", row["sku"])
                LOG.info("Capturing %s/%s: %s", index, len(manifest["selected"]), row["url"])
                try:
                    capture_one(context, row, index, out, capture)
                except CaptureBlocked as exc:
                    row.update(capture_status="blocked", capture_error=str(exc))
                    row.pop("pdf_file", None)
                    manifest["warnings"].append(str(exc))
                    LOG.error("%s", exc)
                    checkpoint(out, manifest)
                    break
                except Exception as exc:
                    row.update(capture_status="failed", capture_error=str(exc))
                    row.pop("pdf_file", None)
                    LOG.error("Capture failed for %s: %s", row["sku"], exc)
                checkpoint(out, manifest)
                time.sleep(capture["between_pages_seconds"])
            manifest["capture_finished_at"] = now()
            checkpoint(out, manifest)
            export_summary(context, out)
        finally:
            context.close()
            browser.close()
    combine_pdfs(out, manifest)


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=Path("config.example.json"))
    parser.add_argument("--inspect-schema", action="store_true", help="Show columns/indexes only; no selection or website requests")
    parser.add_argument("--select-only", action="store_true", help="Select SKUs and write JSON/HTML; do not open the website")
    parser.add_argument("--capture-only", type=Path, metavar="SELECTION_JSON", help="Capture/resume a saved selection without querying the DB")
    parser.add_argument("--output", type=Path, help="New output directory (must not already exist)")
    args = parser.parse_args(argv)
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S")
    config = load_config(args.config)
    if args.capture_only:
        if args.select_only or args.inspect_schema or args.output:
            raise ConfigError("--capture-only cannot be combined with --select-only, --inspect-schema, or --output")
        path = args.capture_only.resolve()
        with path.open(encoding="utf-8") as stream:
            manifest = json.load(stream)
        if manifest.get("format_version") != 1:
            raise ConfigError("Unrecognized selection.json format")
        out = path.parent
        # Retain selected URLs; current config can adjust capture timing/layout only.
        capture_all(out, manifest, config)
    else:
        conn = connect(config)
        try:
            if args.inspect_schema:
                result = {}
                for side, name in config["databases"].items():
                    schema = merge(config["schema"], config["schema_overrides"].get(side, {}))
                    result[side] = Database(conn, name, schema).inspect()
                print(json.dumps(result, indent=2, default=str))
                return 0
            stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
            out = (args.output or Path(f"heilind_sku_capture_{stamp}")).resolve()
            out.mkdir(parents=True, exist_ok=False)
            manifest = scan(conn, config)
        finally:
            conn.close()  # No DB connection remains open during browser capture.
        checkpoint(out, manifest)
        for warning in manifest["warnings"]:
            LOG.warning("%s", warning)
        if not manifest["selected"]:
            LOG.warning("No qualifying SKUs. Review %s", out / "selection_summary.html")
            return 2
        if not args.select_only:
            try:
                capture_all(out, manifest, config)
            except Exception:
                LOG.error("Selection preserved at %s. Use --capture-only to resume.", out / "selection.json")
                raise
    rows = manifest["selected"]
    saved = sum(r.get("capture_status") in ("saved", "saved_needs_review") for r in rows)
    LOG.info("Selected %s SKUs across %s manufacturer groups. PDFs saved: %s. Output: %s",
             len(rows), len({r["manufacturer_key"] for r in rows}), saved, out)
    partial_selection = len(rows) < manifest["selection_settings"]["target_skus"]
    needs_review = any(r.get("capture_status") == "saved_needs_review" for r in rows)
    if partial_selection or (not args.select_only and (saved < len(rows) or needs_review)):
        LOG.warning("Partial result or manual review required; see selection_summary.html")
        return 2
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        LOG.error("Interrupted. If selection.json exists, resume with --capture-only.")
        sys.exit(130)
    except Exception as error:
        LOG.error("%s", error)
        sys.exit(1)