Statistically the best crosshair (2026)

"""Pipeline: validate decoder -> enumerate -> extract -> decode -> dedupe -> write.

Run with:  python -m scraper
Resumable: all HTTP responses are cached in ./cache/, so re-runs only hit the
network for offsets not yet fetched.
"""

import logging

from . import decode, output
from .discover import enumerate_players
from .extract import extract_code
from .http_cache import CachedFetcher

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("scraper")


def main():
    decode.run_validation()  # fail fast if csxhair drifts

    fetcher = CachedFetcher()
    records = enumerate_players(fetcher)
    log.info("enumerated %d raw records", len(records))

    rows, skipped, seen = [], [], set()
    for rec in records:
        steamid = rec.get("steam_id")
        name = rec.get("nickname")
        if not steamid:
            skipped.append({"steamid64": None, "name": name, "reason": "missing steam_id"})
            continue
        if steamid in seen:  # dedupe by steamid64, keep first (rank order)
            continue
        seen.add(steamid)

        code = extract_code(rec)
        if not code:
            skipped.append({"steamid64": steamid, "name": name, "reason": "no share code found"})
            continue
        cvars = decode.decode_code(code)
        if cvars is None:
            skipped.append({"steamid64": steamid, "name": name,
                            "reason": f"code failed to decode: {code}"})
            continue
        rows.append({"name": name, "steamid64": steamid, "share_code": code, **cvars})

    for s in skipped:
        log.warning("skipped %s (%s): %s", s["name"], s["steamid64"], s["reason"])

    output.write_dataset(rows)
    output.write_manifest(len(records), rows, skipped, fetcher)
    log.info("wrote %s, %s, %s", *(str(p) for p in
             (output.config.CSV_PATH, output.config.JSON_PATH, output.config.MANIFEST_PATH)))
    output.print_summary(rows)


if __name__ == "__main__":
    main()

"""Central config for the procrosshairs.com scraper. Tweak everything here."""

from pathlib import Path

BASE_URL = "https://procrosshairs.com"

# Honest browser-style UA with a contact, per politeness requirements.
USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0 Safari/537.36 "
    "(crosshair-dataset-scraper; contact: baasil@elide.dev)"
)

# Politeness: minimum seconds between *network* requests (cache hits are free).
RATE_LIMIT_SECONDS = 1.0
REQUEST_TIMEOUT = 30

# Enumeration: the API pages 100 records per offset; stop on an empty page,
# but never walk past MAX_OFFSET as a runaway guard.
PAGE_SIZE = 100
MAX_OFFSET = 20_000

CACHE_DIR = Path("./cache")
DATA_DIR = Path("./data")
CSV_PATH = DATA_DIR / "crosshairs.csv"
JSON_PATH = DATA_DIR / "crosshairs.json"
MANIFEST_PATH = DATA_DIR / "manifest.json"

"""Disk-cached, rate-limited HTTP fetcher.

Every response body is written to CACHE_DIR keyed by sha256(url), alongside a
.meta.json with the original URL and status. Re-runs serve from disk and never
touch the network, which makes the whole pipeline resumable: kill it anywhere,
restart, and it picks up where the cache ends.
"""

import hashlib
import json
import logging
import time
import urllib.robotparser
from urllib.parse import urlparse

import requests

from . import config

log = logging.getLogger(__name__)


class FetchError(Exception):
    pass


class CachedFetcher:
    def __init__(self):
        self._session = requests.Session()
        self._session.headers["User-Agent"] = config.USER_AGENT
        self._last_request = 0.0
        self._robots = None
        config.CACHE_DIR.mkdir(parents=True, exist_ok=True)
        self.network_hits = 0
        self.cache_hits = 0

    # ---- robots.txt ------------------------------------------------------
    def _robots_allowed(self, url: str) -> bool:
        if self._robots is None:
            rp = urllib.robotparser.RobotFileParser()
            robots_url = f"{config.BASE_URL}/robots.txt"
            try:
                body = self.get(robots_url, check_robots=False)
                rp.parse(body.decode("utf-8", "replace").splitlines())
            except FetchError:
                rp.parse([])  # unreachable robots.txt -> allow
            self._robots = rp
        return self._robots.can_fetch(config.USER_AGENT, url)

    # ---- cache -----------------------------------------------------------
    @staticmethod
    def _cache_paths(url: str):
        key = hashlib.sha256(url.encode()).hexdigest()
        return config.CACHE_DIR / f"{key}.body", config.CACHE_DIR / f"{key}.meta.json"

    # ---- fetch -----------------------------------------------------------
    def get(self, url: str, check_robots: bool = True) -> bytes:
        """Return response body, from cache if present, else network (rate-limited)."""
        body_path, meta_path = self._cache_paths(url)
        if body_path.exists() and meta_path.exists():
            meta = json.loads(meta_path.read_text())
            if meta.get("status") == 200:
                self.cache_hits += 1
                return body_path.read_bytes()
            raise FetchError(f"cached non-200 ({meta.get('status')}) for {url}")

        if check_robots and urlparse(url).netloc == urlparse(config.BASE_URL).netloc:
            if not self._robots_allowed(url):
                raise FetchError(f"robots.txt disallows {url}")

        # <=1 request/sec, single-threaded by design (concurrency cap = 1).
        wait = config.RATE_LIMIT_SECONDS - (time.monotonic() - self._last_request)
        if wait > 0:
            time.sleep(wait)
        self._last_request = time.monotonic()

        log.info("GET %s", url)
        resp = self._session.get(url, timeout=config.REQUEST_TIMEOUT)
        self.network_hits += 1

        # Cache every response, including failures, so re-runs don't re-hit
        # known-bad URLs. Delete the cache entry to force a retry.
        body_path.write_bytes(resp.content)
        meta_path.write_text(json.dumps({"url": url, "status": resp.status_code}))

        if resp.status_code != 200:
            raise FetchError(f"HTTP {resp.status_code} for {url}")
        return resp.content

"""Player enumeration — THE FRAGILE PART. Read this before touching anything.

The site claims 1000+ players but the homepage HTML only embeds the top 100.
Enumeration options were investigated in this order (2026-06-05):

  1. /sitemap.xml          -> 404 ("Cannot GET /sitemap.xml"). Dead end.
     /robots.txt           -> only Cloudflare "content signal" comments; no
                              Sitemap: directive, no User-agent/Disallow rules.
  2. JSON API              -> WINNER. The homepage is an Astro site whose React
                              island (EliteCrosshairs) hydrates a TanStack Query
                              cache. Its bundle (/_astro/index.*.js) calls:
                                  GET /api/v1.2/crosshairs/{offset}
                              The path segment is an OFFSET, not a page number
                              (offset=1 starts at the 2nd-ranked player). Each
                              call returns {"data": [...]} with up to 100 records
                              containing steam_id, nickname, crosshair_code, and
                              crosshairImg (a CloudFront JWT URL). An offset past
                              the end returns {"data": []}.
  3. Pagination on listing -> none exists; the homepage has no page links.
  4. Homepage seed links   -> kept as FALLBACK below in case the API moves.

Strategy: walk offsets 0, 100, 200, ... until an empty page (or MAX_OFFSET).
Pages can overlap if rankings shift mid-run; the caller dedupes by steam_id.
If the API breaks entirely, fall back to scraping /player/ links + crosshair
alt text from the homepage HTML.
"""

from __future__ import annotations

import json
import logging
import re

from . import config
from .http_cache import CachedFetcher, FetchError

log = logging.getLogger(__name__)

# /player/{steamid64}/{name} links in served HTML (fallback path only).
PLAYER_LINK_RE = re.compile(r'href="/player/(\d{17})/([^"]+)"')
# alt="{name} Crosshair: CSGO-....." on listing images (fallback path only).
ALT_CODE_RE = re.compile(r'alt="([^"]*?)\s*Crosshair:\s*(CSGO(?:-[A-Za-z0-9]{5}){5})"')


def enumerate_players(fetcher: CachedFetcher) -> list[dict]:
    """Return raw player records ({steam_id, nickname, crosshair_code, crosshairImg, ...})."""
    try:
        return _enumerate_via_api(fetcher)
    except FetchError as exc:
        log.warning("API enumeration failed (%s); falling back to homepage HTML", exc)
        return _enumerate_via_homepage(fetcher)


def _enumerate_via_api(fetcher: CachedFetcher) -> list[dict]:
    records = []
    for offset in range(0, config.MAX_OFFSET, config.PAGE_SIZE):
        body = fetcher.get(f"{config.BASE_URL}/api/v1.2/crosshairs/{offset}")
        page = json.loads(body).get("data", [])
        log.info("offset %d -> %d records", offset, len(page))
        if not page:
            break  # walked off the end of the dataset
        records.extend(page)
        if len(page) < config.PAGE_SIZE:
            break  # short page == last page; saves one empty-page request
    else:
        log.warning("hit MAX_OFFSET=%d without an empty page; dataset may be truncated",
                    config.MAX_OFFSET)
    return records


def _enumerate_via_homepage(fetcher: CachedFetcher) -> list[dict]:
    """Worst-case seed set: ~100 players embedded in the homepage HTML."""
    html = fetcher.get(f"{config.BASE_URL}/").decode("utf-8", "replace")
    codes_by_name = {name.strip(): code for name, code in ALT_CODE_RE.findall(html)}
    records = []
    for steam_id, name in PLAYER_LINK_RE.findall(html):
        records.append({
            "steam_id": steam_id,
            "nickname": name,
            "crosshair_code": codes_by_name.get(name),
            "crosshairImg": None,
        })
    log.info("homepage fallback -> %d records", len(records))
    return records

"""Extract the CSGO share code from a raw player record.

Priority order:
  1. The API's `crosshair_code` field (same string the alt text carries).
  2. The CloudFront `crosshairImg` URL — its path is a JWT whose payload is
     base64url JSON: {"crosshairCode": "CSGO-...", "width": 128, "height": 128}.
Anything extracted is regex-validated here and *decode*-validated downstream.
"""

from __future__ import annotations

import base64
import json
import re
from urllib.parse import urlparse

CODE_RE = re.compile(r"CSGO(?:-[A-Za-z0-9]{5}){5}")


def extract_code(record: dict) -> str | None:
    # 1. Direct field (mirrors the image alt text).
    for candidate in (record.get("crosshair_code") or "",):
        m = CODE_RE.search(candidate)
        if m:
            return m.group(0)
    # 2. JWT payload inside the CloudFront image URL.
    img = record.get("crosshairImg")
    if img:
        code = _code_from_jwt_url(img)
        if code:
            return code
    return None


def _code_from_jwt_url(url: str) -> str | None:
    path = urlparse(url).path.lstrip("/")
    parts = path.split(".")
    if len(parts) != 3:
        return None
    payload = parts[1]
    try:
        raw = base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))
        code = json.loads(raw).get("crosshairCode", "")
    except (ValueError, json.JSONDecodeError):
        return None
    m = CODE_RE.search(code)
    return m.group(0) if m else None

"""Decode share codes into typed cvar columns via csxhair.

Rows are built from Crosshair's typed attributes (never by parsing the
cs2_commands strings). Booleans stay Python bools; CSV serialization lowercases
them to true/false.
"""

from __future__ import annotations

import logging

from csxhair import Crosshair

log = logging.getLogger(__name__)

# Output column order for CSV/JSON. share_code is the source of truth.
COLUMNS = [
    "name", "steamid64", "share_code",
    "gap", "size", "thickness", "outline_thickness", "alpha", "color", "style",
    "dot", "drawoutline", "usealpha", "t_style", "recoil", "gap_useweaponvalue",
    "fixed_gap", "splitdist", "inner_split_alpha", "outer_split_alpha",
    "split_size_ratio", "color_r", "color_g", "color_b",
]


def decode_code(code: str) -> dict | None:
    """Return the cvar columns for a share code, or None if it fails to decode."""
    try:
        c = Crosshair.decode(code)
    except Exception as exc:  # csxhair raises on bad checksums/chars
        log.debug("decode failed for %s: %s", code, exc)
        return None
    return {
        "gap": c.gap,
        "size": c.size,
        "thickness": c.thickness,
        "outline_thickness": c.outline_thickness,
        "alpha": c.alpha,
        "color": c.color,
        "style": c.style,
        "dot": c.dot,
        "drawoutline": c.draw_outline,
        "usealpha": c.use_alpha,
        "t_style": c.t,
        "recoil": c.recoil,
        "gap_useweaponvalue": c.gap_use_weapon_value,
        "fixed_gap": c.fixed_gap,
        "splitdist": c.dynamic_splitdist,
        "inner_split_alpha": c.dynamic_splitalpha_innermod,
        "outer_split_alpha": c.dynamic_splitalpha_outermod,
        "split_size_ratio": c.dynamic_maxdist_split_ratio,
        "color_r": c.red,
        "color_g": c.green,
        "color_b": c.blue,
    }


def run_validation():
    """Known-good decodes; abort the run if the decoder ever drifts."""
    donk = decode_code("CSGO-O36iX-6VGeQ-pnJLm-UWkab-osuWK")
    assert donk is not None, "donk code failed to decode"
    assert (donk["gap"], donk["color"], donk["style"]) == (-4.5, 1, 4), donk
    assert (donk["size"], donk["thickness"]) == (1.0, 1.0), donk
    assert (donk["color_r"], donk["color_g"], donk["color_b"]) == (0, 255, 0), donk
    assert (donk["alpha"], donk["usealpha"]) == (200, False), donk

    zywoo = decode_code("CSGO-FNOLG-fQcPX-V8P7K-VqtAf-ZbJaA")
    assert zywoo is not None, "ZywOo code failed to decode"
    assert (zywoo["gap"], zywoo["color"], zywoo["style"]) == (-3.0, 5, 4), zywoo
    assert (zywoo["size"], zywoo["thickness"]) == (1.5, 0.0), zywoo
    assert (zywoo["color_r"], zywoo["color_g"], zywoo["color_b"]) == (255, 255, 255), zywoo
    assert (zywoo["alpha"], zywoo["usealpha"]) == (255, True), zywoo
    log.info("validation passed (donk, ZywOo)")

"""Dataset + manifest writers and end-of-run summary stats."""

import csv
import json
import statistics
from collections import Counter
from datetime import datetime, timezone

from . import config
from .decode import COLUMNS


def _csv_cell(value):
    if isinstance(value, bool):
        return "true" if value else "false"  # booleans as true/false, not 0/1
    return value


def write_dataset(rows: list[dict]):
    config.DATA_DIR.mkdir(parents=True, exist_ok=True)
    with open(config.CSV_PATH, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=COLUMNS)
        writer.writeheader()
        for row in rows:
            writer.writerow({k: _csv_cell(row[k]) for k in COLUMNS})
    with open(config.JSON_PATH, "w") as f:
        json.dump(rows, f, indent=2)


def write_manifest(found: int, rows: list[dict], skipped: list[dict], fetcher):
    manifest = {
        "run_at": datetime.now(timezone.utc).isoformat(),
        "players_found": found,
        "players_decoded": len(rows),
        "players_skipped": len(skipped),
        "skipped": skipped,  # [{steamid64, name, reason}]
        "http": {"network_requests": fetcher.network_hits, "cache_hits": fetcher.cache_hits},
    }
    with open(config.MANIFEST_PATH, "w") as f:
        json.dump(manifest, f, indent=2)


def print_summary(rows: list[dict]):
    n = len(rows)
    print(f"\n=== summary ===\nrows: {n}")
    for field in ("style", "color"):
        counts = Counter(r[field] for r in rows)
        dist = ", ".join(f"{k}: {100 * v / n:.1f}%" for k, v in sorted(counts.items()))
        print(f"{field} distribution: {dist}")
    print(f"median gap:  {statistics.median(r['gap'] for r in rows)}")
    print(f"median size: {statistics.median(r['size'] for r in rows)}")