selenium_scraper.py

#!/usr/bin/env python3
"""
transfermarkt-scraper — Selenium edition (secondary engine)
==============================================================

The same scrape as playwright_scraper.py, driven through Selenium. It must
agree with its twin on exit codes, run status, and whether a run crashes or
spends money — those decisions live in page_flow.py and
output_writer.finish_run(), so this file is browser plumbing and nothing
else. See playwright_scraper.py's docstring for what is different about
Transfermarkt and what was and was not measured about it.

Two limits of this engine, stated here rather than left to be discovered.
Neither is a bug in this code and neither can be fixed from here:

  * **Selenium cannot use an authenticated remote CDP endpoint.**
    chromedriver's `debuggerAddress` takes a bare `host:port` with nowhere
    to put a password. A credentialed --cdp-endpoint is refused with exit 2
    rather than connected to and silently failing.
  * **Selenium cannot authenticate a proxy at all.** `--proxy-server=`
    accepts no credentials. They are stripped and a warning says so.

There is also no --concurrency here: parallel page fetching (--mode
transfers only) lives in playwright_scraper.py.

Usage
-----
    python selenium_scraper.py --mode market-values --pages 3
    python selenium_scraper.py --mode club-squad --club-id 281

Requires: pip install -r requirements.txt -r requirements-selenium.txt
          Selenium 4 fetches a matching chromedriver itself; a local Chrome
          or Chromium must be installed.
"""

import argparse
import logging
import re
import sys
import time
from dataclasses import dataclass, field
from urllib.parse import urlparse
from typing import List, Optional
from urllib.parse import urlsplit

from selenium import webdriver
from selenium.common.exceptions import (TimeoutException, WebDriverException,
                                        JavascriptException)
from selenium.webdriver.chrome.options import Options

from captcha_solver import (detect_aws_waf, detect_recaptcha_v3,
                            detect_recaptcha_in_page, reconcile_detections,
                            solve_recaptcha, AWS_WAF_COOKIE, INJECT_TOKEN_JS)
from product_parser import (parse_market_values, parse_club_squad,
                            parse_transfers, parse_player_detail,
                            detect_bot_challenge, page_states_rows, page_url,
                            next_page_link,
                            site_host, is_supported_host, unsupported_reason,
                            HOSTS, player_url, club_squad_url,
                            MARKET_VALUES_URL, TRANSFERS_URL, transfers_url)
from output_writer import dedupe_by_key, finish_run, RemoteAPIError, EXIT_REMOTE_API_ERROR
import page_flow
from proxy_pool import (from_args as proxy_pool_from_args, mask, ROTATE_MODES,
                        ProxyError, split_credentials)
import env_config

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("selenium_scraper")

PAGE_LOAD_TIMEOUT = 60
SCRIPT_TIMEOUT = 30
PAGINATED_MODES = ("market-values", "transfers")

_PROXY_ERROR_MARKERS = (
    "ERR_PROXY_CONNECTION_FAILED", "ERR_TUNNEL_CONNECTION_FAILED",
    "ERR_PROXY_AUTH_UNSUPPORTED", "ERR_PROXY_AUTH_REQUESTED",
    "ERR_UNEXPECTED_PROXY_AUTH", "ERR_PROXY_CERTIFICATE_INVALID",
)

_CREDENTIALS_IN_URL_RE = re.compile(r"([a-z][a-z0-9+.\-]*://)[^\s/@]+:[^\s/@]+@",
                                    re.IGNORECASE)


def _mask_credentials(text: str) -> str:
    return _CREDENTIALS_IN_URL_RE.sub(r"\1***:***@", text or "")


def _chrome_ua(version: str) -> str:
    return (f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
            f"(KHTML, like Gecko) Chrome/{version} Safari/537.36")


def _cdp_host_port(endpoint: str) -> str:
    """`host:port` for chromedriver's debuggerAddress, or exit 2 with a reason."""
    parts = urlsplit(endpoint if "//" in endpoint else f"//{endpoint}")
    if parts.username or parts.password:
        logger.error(
            "This --cdp-endpoint carries credentials (%s), and Selenium "
            "cannot send them: chromedriver's debuggerAddress is a bare "
            "host:port. Use playwright_scraper.py or puppeteer_scraper.py "
            "for a credentialed endpoint such as the Scraping Browser API.",
            _mask_credentials(endpoint))
        sys.exit(2)
    host = parts.hostname or endpoint
    port = f":{parts.port}" if parts.port else ""
    return f"{host}{port}"


@dataclass
class PageOutcome:
    page_num: int
    url: str
    final_url: Optional[str] = None
    rows: List = field(default_factory=list)
    blocked_by: Optional[str] = None
    load_failed: bool = False
    state: Optional[str] = None
    # A served page that shows rows the parser could not read: a markup
    # change, never an empty listing (page_flow.STOP_PARSER_FOUND_NOTHING).
    parse_failed: bool = False
    # The served page's own next-page link (None when it has none):
    # decides whether a repeated page is the listing's end or a page
    # lost (page_flow.PageSequence).
    next_url: Optional[str] = None
    # The target answered 5xx on every try (page_flow.STOP_TARGET_HTTP_5XX).
    server_error: bool = False

    @property
    def ok(self) -> bool:
        # A "not found" answer is the run's answer on page 1, but on a later
        # page it is a page the run did not get.
        lost = self.state == "not_found" and self.page_num > 1
        return (not self.load_failed and self.blocked_by is None
                and not self.parse_failed and not self.server_error
                and not lost)


class _Session:
    """One Chrome driver, relaunchable onto a different exit."""

    def __init__(self, args, pool):
        self.args, self.pool = args, pool
        self.remote = bool(args.cdp_endpoint)
        self.driver = None

    def open(self):
        options = Options()
        # Selenium's equivalent of the `domcontentloaded` that
        # playwright_scraper.py and puppeteer_scraper.py both navigate with.
        # The default strategy is "normal", which blocks until the `load`
        # event, and on this site that event does not arrive: measured
        # 2026-09-16, www.transfermarkt.com timed out at 60s on every attempt
        # ("timeout: Timed out receiving message from renderer") while
        # https://example.com returned in 0.2s through the same driver, and
        # while both other engines fetched the same page in under two
        # seconds. A listing page's rows are in the first response here (see
        # page_flow.py), so waiting for the last tracker pixel buys nothing
        # and cost this engine every live run it ever attempted.
        options.page_load_strategy = "eager"
        if self.remote:
            options.debugger_address = _cdp_host_port(self.args.cdp_endpoint)
            logger.info("Attaching to an existing browser at %s.",
                        options.debugger_address)
            self.driver = webdriver.Chrome(options=options)
            self._apply_timeouts()
            return self

        if self.args.headless:
            options.add_argument("--headless=new")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        options.add_argument("--window-size=1600,1000")
        options.add_argument("--disable-blink-features=AutomationControlled")

        if self.pool:
            scrubbed, credentials = split_credentials(self.pool.current)
            options.add_argument(f"--proxy-server={scrubbed}")
            logger.info("Using proxy exit %s", mask(self.pool.current))
            if credentials:
                logger.warning(
                    "This proxy has credentials and SELENIUM CANNOT SEND "
                    "THEM. They have been stripped, so requests will go out "
                    "unauthenticated. Use playwright_scraper.py or "
                    "puppeteer_scraper.py for an authenticated proxy.")

        self.driver = webdriver.Chrome(options=options)
        self._apply_timeouts()

        version = self.driver.capabilities.get("browserVersion", "")
        if version:
            try:
                self.driver.execute_cdp_cmd(
                    "Network.setUserAgentOverride",
                    {"userAgent": _chrome_ua(version)})
            except WebDriverException as e:
                logger.debug("Could not override the user agent: %s", e)

        if self.args.fingerprint:
            self._apply_fingerprint()
        return self

    def _apply_timeouts(self):
        self.driver.set_page_load_timeout(PAGE_LOAD_TIMEOUT)
        self.driver.set_script_timeout(SCRIPT_TIMEOUT)

    def _apply_fingerprint(self):
        from fingerprint_client import get_fingerprint, playwright_init_script, fingerprint_user_agent
        fp = get_fingerprint(self.args.twocaptcha_key, tags=self.args.fp_tags,
                             country=self.args.fp_country)
        ua = fingerprint_user_agent(fp)
        script = playwright_init_script(fp)
        try:
            if ua:
                self.driver.execute_cdp_cmd("Network.setUserAgentOverride", {"userAgent": ua})
            self.driver.execute_cdp_cmd(
                "Page.addScriptToEvaluateOnNewDocument", {"source": script})
            logger.info("Using 2captcha fingerprint %s (%s)", fp.get("id"), fp.get("country"))
        except WebDriverException as e:
            logger.warning("Could not apply the fingerprint over CDP (%s).", e)

    def relaunch(self):
        if self.remote:
            return
        self.close()
        self.open()

    def close(self):
        try:
            if self.driver is not None:
                self.driver.quit()
        except Exception as e:  # noqa: BLE001
            logger.debug("Ignoring error during driver teardown: %s", e)


def _driver(session):
    driver = session.driver

    def count(selector):
        try:
            return len(driver.find_elements("css selector", selector))
        except WebDriverException as e:
            logger.debug("count(%s) failed: %s", selector, e)
            return 0

    def sleep(ms):
        time.sleep(ms / 1000.0)

    def content():
        try:
            return driver.page_source
        except WebDriverException as e:
            logger.debug("page_source unavailable (page navigating?): %s", e)
            return None

    def current_url():
        try:
            return driver.current_url
        except WebDriverException:
            return ""

    return {"count": count, "sleep": sleep, "content": content,
            "current_url": current_url}


def _parse_for_mode(html: str, url: str, args, page_num: int) -> List:
    if args.mode == "market-values":
        return parse_market_values(html, url, page_num=page_num)
    if args.mode == "club-squad":
        return parse_club_squad(html, url, club_id=args.club_id)
    if args.mode == "transfers":
        return parse_transfers(html, url, page_num=page_num)
    if args.mode == "player":
        row = parse_player_detail(html, url)
        return [row] if row is not None else []
    raise ValueError(f"unknown mode {args.mode!r}")


def _next_page_href(session) -> Optional[str]:
    try:
        return session.driver.execute_script(
            "const a = document.querySelector(\"link[rel='next'], a[rel='next']\");"
            "return a ? (a.href || a.getAttribute('href')) : null;")
    except WebDriverException:
        return None


def handle_captcha_if_present(session, args, ready_selector: str,
                              proxy=None, *, budget) -> bool:
    driver = session.driver
    d = _driver(session)
    html = d["content"]()
    if html is None:
        return False

    already_rendered = d["count"](ready_selector)
    when_blocked = getattr(args, "solve_captcha", "when-blocked") == "when-blocked"

    # AWS WAF first, exactly as playwright_scraper.py does it: it is the one
    # challenge this site has been measured serving, and `window.gokuProps`
    # appears on nothing else. This engine used to skip the check entirely,
    # so it could detect a WAF page through the generic markers and then have
    # nothing to do about it.
    challenge = detect_aws_waf(html, d["current_url"]())
    if challenge is None:
        html_challenge = detect_recaptcha_v3(html, d["current_url"]())
        runtime_challenge = detect_recaptcha_in_page(
            lambda js: driver.execute_script(f"return ({js})();"),
            page_url=d["current_url"]())
        challenge = reconcile_detections(html_challenge, runtime_challenge)
    if not challenge:
        return False
    if when_blocked and already_rendered > 0:
        logger.info("%s detected via %s, but content is already on the "
                    "page — not solving it.", challenge.kind, challenge.source)
        return False
    # "Unsolvable" is a property of a PAGE -- it means the page
    # carries no widget. An AWS WAF CHALLENGE-action page is exactly that:
    # challenge.js only, no puzzle rendered, nothing for a solver to work
    # on. Sending it anyway buys a token for a widget that was never there,
    # and `createTask` validates little enough to take the money. A browser
    # that runs the script passes this by itself, which is why the wait
    # above is still the right thing to do for it.
    if challenge.is_aws_waf and not challenge.has_captcha_widget:
        # No widget, so nothing to buy ("unsolvable" is a property of a
        # page). But this action is a script a real browser runs by itself,
        # so give it the time it measurably takes before judging the page.
        logger.info("AWS WAF %s action and no captcha widget on the page — "
                    "not sending this to the solver API. Waiting up to %.0fs "
                    "for the challenge script to clear it by itself.",
                    challenge.aws_waf_action,
                    page_flow.AWS_CHALLENGE_SETTLE_MS / 1000)
        d = _driver(session)
        if page_flow.wait_for_waf_challenge(
                d["content"], d["sleep"],
                lambda h: detect_aws_waf(h, d["current_url"]()) is not None):
            logger.info("The AWS WAF challenge cleared by itself; nothing was "
                        "paid.")
            return True
        logger.info("The AWS WAF challenge was still up after %.0fs.",
                    page_flow.AWS_CHALLENGE_SETTLE_MS / 1000)
        return False

    logger.warning("%s detected via %s (sitekey=%s) — attempting to solve.",
                   challenge.kind, challenge.source, challenge.sitekey)
    if not args.twocaptcha_key:
        logger.warning("No 2captcha API key set, so this challenge is not "
                       "being solved. 2captcha does solve this type — set "
                       "TWOCAPTCHA_KEY (or pass --twocaptcha-key) to use it.")
        return False
    # Checked before the purchase, against the PAGE's allowance: a retry from
    # a fresh browser or another exit does not earn a second paid solve.
    if not budget.take():
        return False
    try:
        token = solve_recaptcha(challenge, args.twocaptcha_key,
                                api_version=args.captcha_api,
                                min_score=args.min_score,
                               proxy=proxy)
    except Exception as e:  # noqa: BLE001
        logger.error("Solving the challenge failed (%s).", e)
        return False
    if challenge.is_aws_waf:
        # AWS WAF reads its answer back from a COOKIE, not a form field, so
        # there is no `g-recaptcha-response` textarea for INJECT_TOKEN_JS to
        # fill. add_cookie() applies to the driver's current domain, which is
        # this page's, and survives the refresh below.
        host = urlparse(d["current_url"]()).hostname or ""
        try:
            driver.add_cookie({"name": AWS_WAF_COOKIE, "value": token,
                               "domain": host, "path": "/"})
        except WebDriverException as e:
            logger.error("Could not set %s (%s).", AWS_WAF_COOKIE, e)
            return False
        logger.info("Set %s for %s — refreshing to let the WAF re-check.",
                    AWS_WAF_COOKIE, host)
    else:
        try:
            driver.execute_script(f"return ({INJECT_TOKEN_JS})(arguments[0]);", token)
        except WebDriverException as e:
            logger.error("Could not inject the token (%s).", e)
            return False
    time.sleep(1.5)
    driver.refresh()
    return True


def _fetch_one_page(session, args, pool, page_num: int, url: str) -> PageOutcome:
    outcome = PageOutcome(page_num=page_num, url=url)
    d = _driver(session)
    ready_selector = page_flow.ready_selector(args.mode)
    ready_count = page_flow.ready_count(args.mode)
    content_timeout = page_flow.content_timeout_ms(args.mode)
    html, state, load_failed = None, "ok", False

    block_retries = args.proxy_block_retries if (pool and len(pool) > 1) else 0
    # ONE allowance for this page, across every block retry and rotation.
    budget = page_flow.SolveBudget()

    # A 5xx is fetched again on the same exit (bounded); a block moves on
    # to the next exit. Both under the one budget above.
    tries = page_flow.PageAttempts(block_retries, args.retry_delay)
    for block_attempt in tries:
        logger.info("Fetching page %d: %s", page_num, url)
        load_failed, exit_failed = False, None
        for attempt in range(1, args.retries + 1):
            try:
                session.driver.get(url)
                load_failed = False
                break
            except (TimeoutException, WebDriverException) as e:
                text = str(e)
                reason = next((m for m in _PROXY_ERROR_MARKERS if m in text), "")
                load_failed = True
                if reason:
                    exit_failed = reason
                    break
                if attempt < args.retries:
                    pause = args.retry_delay * (2 ** (attempt - 1))
                    logger.warning("Failed to load %s (attempt %d/%d) — "
                                   "retrying in %.1fs.", url, attempt,
                                   args.retries, pause)
                    time.sleep(pause)

        if exit_failed and block_attempt < block_retries:
            logger.warning("Exit %s is unusable (%s) — rotating (%d/%d).",
                           mask(pool.current), exit_failed,
                           block_attempt + 1, block_retries)
            pool.advance(f"unusable exit: {exit_failed}")
            session.relaunch()
            d = _driver(session)
            continue
        if load_failed:
            break

        if handle_captcha_if_present(session, args, ready_selector,
                                    proxy=pool.current if pool else None,
                                    budget=budget):
            time.sleep(1)

        html = d["content"]() or ""
        state = page_flow.classify(html, url=d["current_url"]())

        if state == "server_error" and tries.again_after_server_error(
                page_num, None):
            continue
        if not page_flow.should_retry(state):
            break

        if block_attempt < block_retries:
            logger.warning("Page %d came back as %s from %s — retrying from "
                           "another exit (%d/%d).", page_num, state,
                           mask(pool.current), block_attempt + 1, block_retries)
            pool.advance(f"{state} on page {page_num}")
            session.relaunch()
            d = _driver(session)

    if load_failed:
        logger.error("Gave up loading %s after %d attempt(s).", url, args.retries)
        outcome.load_failed = True
        return outcome

    if state == "content" and page_flow.redirected_off_entity(
            args.mode, url, d["current_url"]()):
        state = "not_found"
    outcome.state = state

    if state == "not_found":
        # The site's own answer that the address names nothing: a 404 page,
        # or a profile/squad URL redirected to an unrelated page. Not a
        # block (no retry, no solve) and not an empty listing.
        logger.error("Page %d: the site says %s does not exist (now at %s). "
                     "Check --club-id / --player-id / --url.", page_num, url,
                     d["current_url"]())
        outcome.final_url = d["current_url"]()
        return outcome

    if state == "server_error":
        # The target's own 5xx, still there after the same-exit retries:
        # a page not served. Not a block (exit 3) and not an empty page.
        debug_html = f"{args.out}_page{page_num}_debug.html"
        with open(debug_html, "w", encoding="utf-8") as f:
            f.write(html or "")
        logger.error("Page %d: the site answered with a server error "
                     "after %d retr%s — saved to %s.", page_num,
                     tries.server_retries_used,
                     "y" if tries.server_retries_used == 1 else "ies",
                     debug_html)
        outcome.server_error = True
        outcome.final_url = d["current_url"]()
        return outcome

    if state == "blocked":
        debug_html = f"{args.out}_page{page_num}_debug.html"
        with open(debug_html, "w", encoding="utf-8") as f:
            f.write(html or "")
        logger.error("Refused or unreachable (state=blocked) — saved to %s.", debug_html)
        outcome.blocked_by = "blocked"
        outcome.final_url = d["current_url"]()
        return outcome

    if state == "content":
        seen = page_flow.wait_for_count(d["count"], d["sleep"], ready_selector,
                                        ready_count, content_timeout)
        if seen >= ready_count:
            time.sleep(0.3)
        else:
            logger.info("No rows appeared within %.0fs (%d/%d matched) — if "
                        "this is a genuinely empty page, that is the expected "
                        "answer.", content_timeout / 1000, seen, ready_count)
        html = d["content"]() or html

    if args.dump_html:
        dump_path = (args.dump_html if page_num == 1
                     else f"{args.dump_html}.page{page_num}")
        with open(dump_path, "w", encoding="utf-8") as f:
            f.write(html)

    vendor = (detect_bot_challenge(html, url=d["current_url"]())
              if state != "content" else None)
    if not vendor and page_flow.counts_as_blocked(state):
        # Classified as a challenge (e.g. by its response header) with no
        # vendor marker in the body: still not a page to parse.
        vendor = state
    if vendor:
        debug_html = f"{args.out}_page{page_num}_debug.html"
        with open(debug_html, "w", encoding="utf-8") as f:
            f.write(html)
        try:
            session.driver.save_screenshot(f"{args.out}_page{page_num}_debug.png")
        except WebDriverException as e:
            logger.warning("Could not capture screenshot: %s", e)
        logger.error("Blocked by %s before parsing (%d bytes) — saved to %s.",
                     vendor, len(html), debug_html)
        outcome.blocked_by = vendor
        return outcome

    final_url = d["current_url"]() or url
    rows = _parse_for_mode(html, final_url, args, page_num)
    logger.info("Parsed %d row(s) from page %d.", len(rows), page_num)

    if rows and args.mode in ("market-values", "club-squad", "transfers"):
        priced = sum(1 for r in rows if r.price is not None)
        coverage = priced / len(rows)
        logger.info("Price coverage on page %d: %d/%d (%.0f%%).", page_num,
                    priced, len(rows), 100.0 * coverage)
        # See page_flow.PRICE_COVERAGE_FLOOR for why transfers is excluded.
        if (args.mode in ("market-values", "club-squad")
                and coverage < page_flow.PRICE_COVERAGE_FLOOR):
            logger.warning("Price coverage on page %d (%.0f%%) is below the "
                           "%.0f%% floor for --mode %s — check for a "
                           "parsing regression rather than assuming these "
                           "players are all free agents.",
                           page_num, 100.0 * coverage,
                           100.0 * page_flow.PRICE_COVERAGE_FLOOR, args.mode)

    if not rows:
        debug_html = f"{args.out}_page{page_num}_debug.html"
        with open(debug_html, "w", encoding="utf-8") as f:
            f.write(html)
        try:
            session.driver.save_screenshot(f"{args.out}_page{page_num}_debug.png")
        except WebDriverException as e:
            logger.warning("Could not capture screenshot: %s", e)
        logger.warning("0 rows parsed — saved what the browser saw to %s.", debug_html)

    if not rows and state == "content" and page_states_rows(html, args.mode):
        # The page shows rows and the parser read none of them: a markup
        # change, not an empty listing. Reported as a parser failure so the
        # run cannot end "complete" with exit 4.
        logger.error("Page %d shows rows but the parser read none — a "
                     "parser failure, not an empty result.", page_num)
        outcome.parse_failed = True

    outcome.rows = rows
    outcome.final_url = final_url
    outcome.next_url = next_page_link(html, outcome.final_url)
    return outcome


def scrape(args) -> int:
    outcomes: List[PageOutcome] = []
    seen_keys = set()
    pages = page_flow.PageSequence()
    blocked = False
    stop_reason = "single_page_mode" if args.mode not in PAGINATED_MODES else "completed"

    pool = proxy_pool_from_args(args)
    if pool and args.cdp_endpoint:
        logger.warning("Ignoring --proxy/--proxy-file: with --cdp-endpoint "
                       "the remote browser has its own exit.")
        pool = None
    if args.concurrency > 1:
        logger.warning("--concurrency is ignored in this engine: parallel "
                       "page fetching (--mode transfers) is implemented in "
                       "playwright_scraper.py. Running sequentially.")

    session = _Session(args, pool).open()
    try:
        first = _fetch_one_page(session, args, pool, 1, args.url)
        outcomes.append(first)

        if not first.ok:
            stop_reason = page_flow.failure_reason(first)
            blocked = first.blocked_by is not None
        elif first.state == "not_found":
            stop_reason = page_flow.STOP_NOT_FOUND
        elif args.mode not in PAGINATED_MODES:
            pass
        else:
            seen_keys.update(r.sku for r in first.rows if r.sku is not None)
            pages.add(1, first.rows, first.next_url)
            site_next = _next_page_href(session)
            addressable = page_flow.pagination_is_addressable(
                args.mode, first.final_url, site_next)
            planned = ([page_url(first.final_url, n) for n in range(2, args.pages + 1)]
                      if addressable else None)

            for page_num in range(2, args.pages + 1):
                if planned:
                    url = planned[page_num - 2]
                else:
                    href = _next_page_href(session)
                    if not href:
                        logger.info("No further pagination link on page %d "
                                    "— stopping here.", page_num - 1)
                        stop_reason = "pagination_exhausted"
                        break
                    url = href

                outcome = _fetch_one_page(session, args, pool, page_num, url)
                outcomes.append(outcome)
                if not outcome.ok:
                    stop_reason = page_flow.failure_reason(outcome)
                    blocked = outcome.blocked_by is not None
                    break

                # A page that only repeats earlier rows is not a fresh page:
                # the listing's end, or a page the site did not serve.
                repeat = pages.add(page_num, outcome.rows, outcome.next_url)
                if repeat:
                    stop_reason = repeat["stop_reason"]
                    break
                fresh_count = sum(1 for r in outcome.rows
                                  if r.sku is None or r.sku not in seen_keys)
                seen_keys.update(r.sku for r in outcome.rows if r.sku is not None)
                if not fresh_count:
                    logger.info("Page %d added no rows not already seen — "
                                "treating that as the end.", page_num)
                    stop_reason = "no_new_products"
                    break
                if page_flow.is_thin_page(len(outcome.rows), len(first.rows)):
                    logger.warning(
                        "Page %d returned only %d row(s), well under page "
                        "1's %d — possibly a page beyond the listing's "
                        "real depth, or a markup regression. Continuing; "
                        "this is informational only and does not change "
                        "the exit code.", page_num, len(outcome.rows),
                        len(first.rows))
                if page_num < args.pages:
                    if pool and pool.rotates_per_page():
                        pool.advance(f"per-page rotation after page {page_num}")
                        session.relaunch()
                    time.sleep(args.delay)
    finally:
        session.close()

    all_rows = []
    merged_seen = set()
    counted = [o for o in outcomes if pages.counts(o.page_num)]
    for oc in sorted(outcomes, key=lambda o: o.page_num):
        if not pages.counts(oc.page_num):
            logger.info("Page %d: not counted — it repeated an earlier "
                        "page (%d row(s) dropped).", oc.page_num, len(oc.rows))
            continue
        fresh = dedupe_by_key(oc.rows, merged_seen, key="sku")
        if len(fresh) < len(oc.rows):
            logger.info("Page %d: dropped %d duplicate row(s).",
                        oc.page_num, len(oc.rows) - len(fresh))
        all_rows.extend(fresh)

    ok_pages = [o for o in counted if o.ok]
    failed_pages = [o.page_num for o in counted if not o.ok]
    final_url = (max(ok_pages, key=lambda o: o.page_num).final_url
                 if ok_pages else args.url)

    return finish_run(all_rows, args.out, args.format, args.allow_empty,
                      blocked=blocked, stop_reason=stop_reason,
                      pages_requested=args.pages, pages_completed=len(ok_pages),
                      pages_failed=failed_pages, mode=args.mode,
                      source=site_host(final_url) or "transfermarkt.com",
                      start_url=args.url, final_url=final_url,
                      extra={**pages.meta(),
                             **page_flow.listing_filter_meta(args.mode, args.url)})


def parse_args():
    p = argparse.ArgumentParser(description="Transfermarkt scraper (Selenium edition)")
    p.add_argument("--mode", choices=["market-values", "club-squad", "transfers", "player"],
                   default="market-values")
    p.add_argument("--url", default=None)
    p.add_argument("--club-id", default=None)
    p.add_argument("--season", default=None)
    p.add_argument("--min-value", type=page_flow.bounded(int, 0, "--min-value"),
                   default=None, metavar="EUR",
                   help="Only --mode transfers: list transfers of players valued at least "
                        "this many EUR (0 = no floor). Without it the site's own default "
                        "view is read, which the site filters to EUR 500,000+ "
                        "(measured 2026-10-15).")
    p.add_argument("--player-id", default=None)
    p.add_argument("--pages", type=page_flow.bounded(int, 1, "--pages"), default=1)
    p.add_argument("--delay", type=page_flow.bounded(float, 0, "--delay"), default=2.0)
    p.add_argument("--concurrency", type=page_flow.bounded(int, 1, "--concurrency"), default=1,
                   help="Ignored in this engine — see playwright_scraper.py.")
    p.add_argument("--retries", type=page_flow.attempts, default=3)
    p.add_argument("--retry-delay", type=page_flow.bounded(float, 0, "--retry-delay"), default=2.0)
    p.add_argument("--format", choices=["json", "csv", "both"], default="both")
    p.add_argument("--out", default="transfermarkt_products")
    p.add_argument("--proxy", default=None)
    p.add_argument("--proxy-file", default=None)
    p.add_argument("--proxy-rotate", choices=list(ROTATE_MODES), default="per-run")
    p.add_argument("--proxy-sessions", type=page_flow.bounded(int, 1, "--proxy-sessions"), default=None,
                   help="With --proxy pointing at a 2Captcha proxy gateway, mint this many session-pinned exits from that one credential instead of keeping a file of them. Each session is a different exit address (measured 2026-09-17: ten sessions, ten distinct addresses). Has no effect with --proxy-file, and is refused for a non-2Captcha host.")
    p.add_argument("--proxy-shuffle", action="store_true")
    p.add_argument("--proxy-block-retries", type=page_flow.bounded(int, 0, "--proxy-block-retries"), default=2)
    p.add_argument("--twocaptcha-key", default=None)
    p.add_argument("--allow-empty", action="store_true")
    p.add_argument("--fingerprint", action="store_true")
    p.add_argument("--fp-tags", default="Windows")
    p.add_argument("--fp-country", default=None)
    p.add_argument("--captcha-api", choices=["v2", "v1"], default="v2")
    p.add_argument("--solve-captcha", choices=["when-blocked", "always"], default="when-blocked")
    p.add_argument("--min-score", type=float, default=0.7)
    p.add_argument("--cdp-endpoint", default=None,
                   help="A CREDENTIAL-FREE debuggerAddress only — Selenium "
                        "cannot authenticate on the WebSocket upgrade the "
                        "way Playwright/pyppeteer do. See this file's "
                        "docstring.")
    p.add_argument("--dump-html", default=None, metavar="PATH")
    p.add_argument("--headless", action="store_true", default=True)
    p.add_argument("--headful", dest="headless", action="store_false")
    args = p.parse_args()
    # Captured BEFORE the environment is applied: afterwards a URL from
    # `.env` is indistinguishable from one the user typed.
    typed_url = args.url
    env_config.apply(args)

    if not typed_url and args.url:
        why = page_flow.env_url_conflict(args.url, args.mode, args.club_id,
                                         args.player_id, args.season)
        if why:
            logger.warning("Not using TRANSFERMARKT_URL from the environment: "
                           "%s. Building the --mode %s URL instead.", why,
                           args.mode)
            args.url = None

    if not args.url:
        if args.mode == "market-values":
            args.url = MARKET_VALUES_URL
        elif args.mode == "transfers":
            args.url = TRANSFERS_URL
        elif args.mode == "club-squad":
            if not args.club_id:
                p.error("--mode club-squad needs --club-id (or --url)")
            args.url = club_squad_url(args.club_id, season=args.season)
        elif args.mode == "player":
            if not args.player_id:
                p.error("--mode player needs --player-id (or --url)")
            args.url = player_url(args.player_id)

    if args.min_value is not None:
        if args.mode != "transfers":
            p.error("--min-value applies to --mode transfers only")
        args.url = transfers_url(args.min_value, base_url=args.url)

    if not is_supported_host(args.url):
        why = unsupported_reason(args.url)
        p.error(f"{site_host(args.url) or args.url!r} {why}. Supported: "
                f"{', '.join(sorted(HOSTS))}.")
    if args.mode not in PAGINATED_MODES and args.pages != 1:
        logger.warning("--pages %d is ignored in --mode %s.", args.pages, args.mode)
        args.pages = 1
    return args


if __name__ == "__main__":
    args = parse_args()
    if args.fingerprint and not args.twocaptcha_key:
        logger.error("--fingerprint needs --twocaptcha-key.")
        sys.exit(2)
    try:
        sys.exit(scrape(args))
    except ProxyError as e:
        logger.error("%s", e)
        sys.exit(2)
    except RemoteAPIError as e:
        # The Fingerprint API call failed on its own terms -- not
        # Transfermarkt blocking a page (exit 3) and not an unexpected bug
        # (exit 1, which still gets a real traceback). See
        # output_writer.RemoteAPIError. (Selenium has no credentialed
        # --cdp-endpoint path to fail this way -- see _cdp_host_port, which
        # exits 2 for that as a usage error before any connection attempt.)
        logger.error("%s", e)
        sys.exit(EXIT_REMOTE_API_ERROR)