#!/usr/bin/env python3
"""Download public code, models, and discussions for competitions in competitions_clean.csv.

Kaggle CLI cannot accept competition rules. This script does not join competitions
and does not download official competition data bundles. Public notebooks, the
competition Models tab, and public discussion threads are readable without joining.

Layout (slug is unique; titles are not)::

    competitions/<slug>/meta.json
    competitions/<slug>/code/<owner>__<kernel>/
    competitions/<slug>/models/<owner>__<model>__<framework>__<instance>/
    competitions/<slug>/discussion/<topic_id>.json
    competitions/<slug>/VERIFY.json

A 429, a 5xx, or an empty/unparseable body is retried with exponential backoff.
It is never treated as an empty page. A page is complete only after HTTP 200
and valid JSON.

The run stops before the next competition or the next new file when the
filesystem that holds competitions/ reaches 90% used. Already written files
stay in place. The next --all continues from VERIFY.json and per-file .ok
markers. A disk stop exits 0 and is not counted as a failed competition.
"""

from __future__ import annotations

import argparse
import json
import os
import random
import shutil
import sys
import time
import traceback
from pathlib import Path
from typing import Any, Callable

import requests

ROOT = Path(__file__).resolve().parent
CSV_PATH = ROOT / "competitions_clean.csv"
OUT_ROOT = ROOT / "competitions"
META_DIR = ROOT / "meta"
STATE_DIR = ROOT / "download_state"

KERNEL_LIST_CAP = 1100
MAX_RETRIES = 8
BACKOFF_START = 2.0
BACKOFF_CAP = 120.0
# Stop before the next download once this filesystem is this full.
DISK_STOP_RATIO = 0.90
DISK_STOP_MARKER = STATE_DIR / "stopped_disk.txt"

TOPICS_URL = "https://api.kaggle.com/v1/competitions.CompetitionApiService/ListCompetitionTopics"
MESSAGES_URL = "https://api.kaggle.com/v1/competitions.CompetitionApiService/ListTopicMessages"
MODELS_URL = "https://www.kaggle.com/api/i/models.ModelService/ListCompetitionModelInstances"


class Incomplete(RuntimeError):
    """A paged collection ended before the server said it was complete."""


class DiskLimit(SystemExit):
    """Filesystem is at the stop threshold. Exit 0 so the run can be resumed."""

    def __init__(self, message: str) -> None:
        self.message = message
        super().__init__(0)


def df_used_percent(used: int, free: int) -> int:
    """Match coreutils df: used space as a percent of used+available, rounded up."""
    denom = used + free
    if denom <= 0:
        return 100
    return (used * 100 + denom - 1) // denom


def disk_is_full(used: int, free: int) -> bool:
    return df_used_percent(used, free) >= int(DISK_STOP_RATIO * 100)


def disk_snapshot(path: Path | None = None) -> tuple[int, int, int, int]:
    usage = shutil.disk_usage(path or OUT_ROOT)
    return df_used_percent(usage.used, usage.free), usage.used, usage.free, usage.total


def ensure_disk_room(where: str) -> None:
    """Raise DiskLimit before writing more data once df would show the stop percent."""
    pct, used, free, total = disk_snapshot(OUT_ROOT)
    if not disk_is_full(used, free):
        return
    message = (
        f"disk {pct}% used ({used} used, {free} free, {total} total) at {where}; "
        f"stop threshold is {DISK_STOP_RATIO:.0%}"
    )
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    DISK_STOP_MARKER.write_text(message + "\n")
    print(f"STOP {message}", flush=True)
    raise DiskLimit(message)


def interpret_http(status: int, text: str | None) -> tuple[str, Any]:
    """Classify one HTTP response.

    Returns ('retry', None), ('ok', parsed_json), or ('fatal', message).
    429 and empty bodies are retries even when the body looks like an empty page.
    """
    body = text if text is not None else ""
    if status == 429 or status in (500, 502, 503, 504):
        return "retry", None
    if status != 200:
        snippet = body.strip().replace("\n", " ")[:300]
        return "fatal", f"HTTP {status}: {snippet}"
    if not body.strip():
        return "retry", None
    try:
        data = json.loads(body)
    except json.JSONDecodeError:
        return "retry", None
    if isinstance(data, dict):
        err = data.get("error")
        if isinstance(err, dict):
            code = err.get("code")
            status_name = str(err.get("status") or "")
            if code in (429, 500, 502, 503, 504) or "RESOURCE_EXHAUSTED" in status_name:
                return "retry", None
            return "fatal", json.dumps(err)[:300]
        if data.get("code") in (429, 500, 502, 503, 504):
            return "retry", None
    return "ok", data


def backoff_seconds(attempt: int) -> float:
    delay = min(BACKOFF_START * (2 ** (attempt - 1)), BACKOFF_CAP)
    return delay + random.uniform(0, delay * 0.25)


def request_json(
    transport: Callable[[], tuple[int, str | None]],
    *,
    what: str,
    sleep: Callable[[float], None] = time.sleep,
    max_retries: int = MAX_RETRIES,
) -> Any:
    """Call transport until a real HTTP 200 JSON body arrives.

    transport returns (status, text). A 429 with an empty body stays on this
    same call; the caller must not advance the page.
    """
    last = ""
    for attempt in range(1, max_retries + 1):
        try:
            status, text = transport()
        except requests.RequestException as exc:
            last = f"network {exc}"
            delay = backoff_seconds(attempt)
            print(f"RETRY {what} attempt {attempt}/{max_retries} {last} sleep {delay:.1f}s", flush=True)
            sleep(delay)
            continue
        kind, data = interpret_http(status, text)
        if kind == "ok":
            return data
        if kind == "fatal":
            raise RuntimeError(f"{what} failed: {data}")
        last = f"status {status} body_len {len(text or '')}"
        delay = backoff_seconds(attempt)
        print(f"RETRY {what} attempt {attempt}/{max_retries} {last} sleep {delay:.1f}s", flush=True)
        sleep(delay)
    raise Incomplete(f"{what} still rate-limited after {max_retries} tries ({last}); not treating it as an empty page")


def collect_topics(fetch_page: Callable[[int], tuple[int, str | None]], **kwargs: Any) -> list[dict]:
    """Page competition topics until totalCount is reached.

    fetch_page(page) -> (status, text). An empty 429 does not advance the page
    and does not finish the list.
    """
    page = 1
    seen: dict[int, dict] = {}
    total: int | None = None
    empty_tries = 0
    sleep = kwargs.get("sleep", time.sleep)
    while True:
        data = request_json(lambda p=page: fetch_page(p), what=f"topics page {page}", **kwargs)
        topics = data.get("topics") or []
        if "totalCount" in data:
            total = int(data["totalCount"])
        elif "total_count" in data:
            total = int(data["total_count"])
        if not topics:
            if total == 0:
                return []
            if total is not None and len(seen) >= total:
                return list(seen.values())
            # A 200 with no rows before totalCount is the same failure mode as a
            # 429 that came back without a body: do not advance and do not finish.
            empty_tries += 1
            if empty_tries > 3:
                raise Incomplete(
                    f"topics page {page} returned no rows, have {len(seen)} of {total}"
                )
            delay = backoff_seconds(empty_tries)
            print(
                f"RETRY topics page {page} empty before total {total}, have {len(seen)} sleep {delay:.1f}s",
                flush=True,
            )
            sleep(delay)
            continue
        empty_tries = 0
        added = 0
        for topic in topics:
            tid = int(topic["id"])
            if tid not in seen:
                seen[tid] = topic
                added += 1
        if total is not None and len(seen) >= total:
            return list(seen.values())
        if added == 0:
            raise Incomplete(f"topics page {page} added nothing, have {len(seen)} of {total}")
        page += 1
        if page > 100000:
            raise Incomplete("topics pagination did not end")


def collect_models(fetch_token: Callable[[str], tuple[int, str | None]], **kwargs: Any) -> list[dict]:
    """Page competition model instances by nextPageToken.

    The same token is retried on 429. An empty 429 is not the last page.
    """
    token = ""
    seen_tokens: set[str] = set()
    items: list[dict] = []
    seen_ids: set[int] = set()
    while True:
        data = request_json(lambda t=token: fetch_token(t), what=f"models token {token[:12] or 'first'}", **kwargs)
        batch = list(data.get("pinnedModelInstances") or []) + list(data.get("unpinnedModelInstances") or [])
        next_token = data.get("nextPageToken") or data.get("next_page_token") or ""
        for item in batch:
            iid = int(item["modelInstanceId"])
            if iid not in seen_ids:
                seen_ids.add(iid)
                items.append(item)
        if not next_token:
            return items
        if next_token in seen_tokens:
            raise Incomplete("model page token repeated")
        seen_tokens.add(next_token)
        token = next_token


def session_from_kaggle_json() -> requests.Session:
    path = Path.home() / ".kaggle" / "kaggle.json"
    cred = json.loads(path.read_text())
    sess = requests.Session()
    sess.auth = (cred["username"], cred["key"])
    sess.headers.update({"User-Agent": "kaggle-competition-downloader/1", "Content-Type": "application/json"})
    return sess


def post_text(sess: requests.Session, url: str, payload: dict, timeout: int = 90) -> tuple[int, str]:
    response = sess.post(url, json=payload, timeout=timeout)
    return response.status_code, response.text


def load_competitions(csv_path: Path) -> list[dict]:
    import csv

    with csv_path.open(newline="") as handle:
        return list(csv.DictReader(handle))


def load_views(topic_ids: set[int]) -> dict[int, int]:
    import csv

    path = META_DIR / "ForumTopics.csv"
    if not path.exists() or not topic_ids:
        return {}
    wanted = {str(i) for i in topic_ids}
    views: dict[int, int] = {}
    with path.open(newline="") as handle:
        for row in csv.DictReader(handle):
            if row["Id"] in wanted:
                raw = (row.get("TotalViews") or "").strip()
                views[int(row["Id"])] = int(raw) if raw else 0
                if len(views) == len(wanted):
                    break
    return views


def comment_to_dict(comment: Any) -> dict:
    post_date = getattr(comment, "post_date", None)
    return {
        "id": comment.id,
        "author_name": comment.author_name,
        "author_url": comment.author_url,
        "post_date": post_date.isoformat() if post_date else None,
        "content": comment.content,
        "votes": comment.votes,
        "replies": [comment_to_dict(reply) for reply in (comment.replies or [])],
    }


def count_comments(comments: list[dict]) -> int:
    total = 0
    for comment in comments:
        total += 1 + count_comments(comment.get("replies") or [])
    return total


def call_sdk(fn: Callable[[], Any], what: str) -> Any:
    """Retry a Kaggle SDK call. 429 and empty-body JSON errors stay on the same call."""

    def transport() -> tuple[int, str]:
        try:
            value = fn()
        except requests.HTTPError as exc:
            response = getattr(exc, "response", None)
            status = response.status_code if response is not None else 429
            text = response.text if response is not None else ""
            return status, text
        except requests.RequestException as exc:
            raise
        except json.JSONDecodeError:
            return 429, ""
        except ValueError as exc:
            if "Expecting value" in str(exc) or "JSON" in str(exc):
                return 429, ""
            raise
        return 200, json.dumps({"_sdk": True, "value_id": id(value)})

    # The transport above cannot pass the SDK object through JSON. Call directly.
    last = ""
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            return fn()
        except requests.HTTPError as exc:
            response = getattr(exc, "response", None)
            status = response.status_code if response is not None else 0
            text = response.text if response is not None else ""
            kind, _ = interpret_http(status or 429, text)
            if kind != "retry":
                raise
            last = f"HTTP {status} body_len {len(text)}"
        except requests.RequestException as exc:
            last = str(exc)
        except json.JSONDecodeError:
            last = "empty json body"
        except ValueError as exc:
            if "Expecting value" not in str(exc):
                raise
            last = "empty json body"
        delay = backoff_seconds(attempt)
        print(f"RETRY {what} attempt {attempt}/{MAX_RETRIES} {last} sleep {delay:.1f}s", flush=True)
        time.sleep(delay)
    raise Incomplete(f"{what} still failing after {MAX_RETRIES} tries ({last}); not treating it as empty")


def build_code_index(comp_ids: set[str], dest: Path) -> dict[str, list[dict]]:
    """Public kernels for each competition.

    A kernel counts once, not once per version. It is included when any of its
    versions sources the competition and MadePublicDate is set. The downloaded
    file is the current public notebook.
    """
    import csv

    versions_path = META_DIR / "KernelVersions.csv"
    if not versions_path.exists():
        raise FileNotFoundError(versions_path)

    version_comps: dict[str, set[str]] = {}
    with (META_DIR / "KernelVersionCompetitionSources.csv").open(newline="") as handle:
        for row in csv.DictReader(handle):
            cid = row["SourceCompetitionId"]
            if cid in comp_ids:
                version_comps.setdefault(row["KernelVersionId"], set()).add(cid)
    print(f"index versions {len(version_comps)}", flush=True)

    kernel_comps: dict[str, set[str]] = {}
    with versions_path.open(newline="") as handle:
        reader = csv.DictReader(handle)
        fields = set(reader.fieldnames or [])
        kernel_col = "ScriptId" if "ScriptId" in fields else "KernelId"
        if kernel_col not in fields or "Id" not in fields:
            raise RuntimeError(f"unexpected KernelVersions columns: {reader.fieldnames}")
        for row in reader:
            comps = version_comps.get(row["Id"])
            if comps:
                kernel_comps.setdefault(row[kernel_col], set()).update(comps)
    print(f"index kernels {len(kernel_comps)}", flush=True)

    pending: dict[str, dict] = {}
    with (META_DIR / "Kernels.csv").open(newline="") as handle:
        for row in csv.DictReader(handle):
            if not (row.get("MadePublicDate") or "").strip():
                continue
            comps = kernel_comps.get(row["Id"])
            if not comps:
                continue
            pending[row["Id"]] = {
                "author": row["AuthorUserId"],
                "slug": row["CurrentUrlSlug"],
                "comps": comps,
            }
    author_ids = {item["author"] for item in pending.values()}
    names: dict[str, str] = {}
    with (META_DIR / "Users.csv").open(newline="") as handle:
        for row in csv.DictReader(handle):
            if row["Id"] in author_ids:
                names[row["Id"]] = row["UserName"]
                if len(names) == len(author_ids):
                    break

    by_comp: dict[str, list[dict]] = {cid: [] for cid in comp_ids}
    missing_authors = 0
    for kernel_id, item in pending.items():
        username = names.get(item["author"])
        if not username or not item["slug"]:
            missing_authors += 1
            continue
        record = {"ref": f"{username}/{item['slug']}", "kernel_id": kernel_id}
        for cid in item["comps"]:
            by_comp[cid].append(record)
    merged = read_code_index(dest) or {}
    for cid, records in by_comp.items():
        merged[str(cid)] = records
    write_code_index(merged, dest)
    set_code_index_cache(merged)
    print(f"index wrote {dest} competitions {len(merged)} missing_authors {missing_authors}", flush=True)
    return merged


def code_index_covers(have: set[str], need: set[str]) -> bool:
    return bool(need) and need <= have


def read_code_index(path: Path | None = None) -> dict[str, list[dict]] | None:
    """Load the derived kernel cache. None when the file is not there."""
    path = path or (STATE_DIR / "code_index.jsonl")
    if not path.exists():
        return None
    found: dict[str, list[dict]] = {}
    with path.open() as handle:
        for line in handle:
            line = line.strip()
            if not line:
                continue
            row = json.loads(line)
            found[str(row["competition_id"])] = list(row.get("kernels") or [])
    return found


def write_code_index(by_comp: dict[str, list[dict]], dest: Path) -> None:
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(dest.name + ".tmp")
    with tmp.open("w") as handle:
        for cid in sorted(by_comp, key=lambda item: (0, int(item)) if str(item).isdigit() else (1, str(item))):
            handle.write(
                json.dumps({"competition_id": str(cid), "kernels": by_comp[cid]}, ensure_ascii=False) + "\n"
            )
    tmp.replace(dest)


_CODE_INDEX: dict[str, list[dict]] | None = None
_CODE_INDEX_LOADED = False


def set_code_index_cache(index: dict[str, list[dict]] | None) -> None:
    global _CODE_INDEX, _CODE_INDEX_LOADED
    _CODE_INDEX = index
    _CODE_INDEX_LOADED = index is not None


def cached_code_index() -> dict[str, list[dict]] | None:
    """Read code_index.jsonl at most once per process."""
    global _CODE_INDEX, _CODE_INDEX_LOADED
    if _CODE_INDEX_LOADED:
        return _CODE_INDEX
    _CODE_INDEX = read_code_index()
    _CODE_INDEX_LOADED = True
    return _CODE_INDEX


def ensure_code_index(comp_ids: set[str], dest: Path) -> dict[str, list[dict]]:
    """Use the local cache when it already lists every competition.

    The meta CSVs are the source. This cache is the join of those CSVs, and a
    partial file (the two debug competitions) does not count. A full file is
    never rebuilt on --all.
    """
    existing = read_code_index(dest)
    have = set(existing) if existing is not None else set()
    if code_index_covers(have, comp_ids):
        assert existing is not None
        set_code_index_cache(existing)
        print(f"code index ready: {len(comp_ids)} competitions, using local cache", flush=True)
        return existing
    print(
        f"local code index has {len(have & comp_ids)} of {len(comp_ids)} competitions; "
        "scanning local meta CSVs once",
        flush=True,
    )
    return build_code_index(comp_ids, dest)


def load_code_index(competition_id: str) -> list[dict] | None:
    index = cached_code_index()
    if index is None:
        return None
    return index.get(str(competition_id), [])


def list_kernels(api: Any, slug: str) -> tuple[list[dict], bool]:
    """Return public kernels from the code page, and whether the list hit the API cap."""
    found: list[dict] = []
    seen: set[str] = set()
    page = 1
    while True:
        def once(page: int = page) -> Any:
            return api.kernels_list_with_response(
                competition=slug, page=page, page_size=100, sort_by="dateCreated"
            )

        response = call_sdk(once, f"{slug} kernels page {page}")
        kernels = response.kernels or []
        if not kernels:
            break
        for kernel in kernels:
            if kernel.ref and kernel.ref not in seen:
                seen.add(kernel.ref)
                found.append({"ref": kernel.ref, "title": kernel.title or ""})
        if len(kernels) < 100:
            break
        page += 1
        if len(found) >= KERNEL_LIST_CAP:
            return found, True
    return found, False


def pull_kernel(api: Any, ref: str, dest: Path) -> None:
    marker = dest / ".ok"
    if marker.exists() and any(p.is_file() and p.stat().st_size > 0 and p.name != ".ok" for p in dest.iterdir()):
        return
    ensure_disk_room(f"kernel {ref}")
    if dest.exists():
        shutil.rmtree(dest)
    dest.mkdir(parents=True, exist_ok=True)

    def once() -> None:
        api.kernels_pull(ref, str(dest), metadata=True, quiet=True)

    call_sdk(once, f"pull {ref}")
    files = [p for p in dest.iterdir() if p.is_file() and p.name != ".ok" and p.stat().st_size > 0]
    if not files:
        raise Incomplete(f"{ref} pull returned no source file")
    marker.write_text(ref + "\n")


def list_model_instances(sess: requests.Session, competition_id: str) -> list[dict]:
    def fetch(token: str) -> tuple[int, str]:
        payload: dict[str, Any] = {"competitionId": str(competition_id), "pageSize": 100}
        if token:
            payload["pageToken"] = token
        return post_text(sess, MODELS_URL, payload)

    return collect_models(fetch)


def model_ref(item: dict) -> str | None:
    url = item.get("downloadUrl") or ""
    # /models/owner/model/Framework/instance/version/download
    parts = [part for part in url.split("/") if part]
    if len(parts) >= 6 and parts[0] == "models" and parts[-1] == "download":
        return "/".join(parts[1:-1])
    return None


def download_model(api: Any, ref: str, dest: Path) -> Path:
    dest.mkdir(parents=True, exist_ok=True)
    existing = list(dest.glob("*.tar.gz"))
    if existing and existing[0].stat().st_size > 0 and (dest / ".ok").exists():
        return existing[0]
    ensure_disk_room(f"model {ref}")

    def once() -> str:
        return api.model_instance_version_download(ref, path=str(dest), force=False, quiet=True, untar=False)

    path = Path(call_sdk(once, f"model {ref}"))
    if not path.exists() or path.stat().st_size <= 0:
        raise Incomplete(f"model {ref} download is empty")
    (dest / ".ok").write_text(ref + "\n")
    return path


def list_topic_stubs(sess: requests.Session, slug: str) -> list[dict]:
    def fetch(page: int) -> tuple[int, str]:
        return post_text(sess, TOPICS_URL, {"competitionName": slug, "page": page})

    return collect_topics(fetch)


def raw_messages(sess: requests.Session, slug: str, topic_id: int) -> list[dict]:
    data = request_json(
        lambda: post_text(
            sess,
            MESSAGES_URL,
            {"competitionName": slug, "topicId": topic_id, "pageSize": -1},
        ),
        what=f"{slug} topic {topic_id} messages",
    )
    return list(data.get("messages") or [])


def fetch_comment_tree(api: Any, topic_id: int) -> tuple[Any, list[Any]]:
    collected: list[Any] = []
    seen: set[int] = set()
    token: str | None = None
    topic = None
    while True:
        def once(token: str | None = token) -> Any:
            return api.forums_topic_show(topic_id, page_size=100, page_token=token)

        topic, comments, next_token = call_sdk(once, f"topic {topic_id} comments")
        comments = comments or []
        if not comments and topic is not None and topic.comment_count > len(seen) and not next_token:
            raise Incomplete(
                f"topic {topic_id} expected {topic.comment_count} comments, got {len(seen)}, and the page was empty"
            )
        for comment in comments:
            if comment.id not in seen:
                seen.add(comment.id)
                collected.append(comment)
        if not next_token:
            break
        token = next_token
    if topic is not None and count_comments([comment_to_dict(c) for c in collected]) < topic.comment_count:
        raise Incomplete(
            f"topic {topic_id} saved {len(seen)} comments, server commentCount is {topic.comment_count}"
        )
    return topic, collected


def safe_name(text: str) -> str:
    cleaned = "".join(ch if ch.isalnum() or ch in "-_." else "_" for ch in text)
    return cleaned.strip("._") or "item"


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2))
    tmp.replace(path)


def download_competition(row: dict, api: Any, sess: requests.Session, force: bool = False) -> dict:
    slug = row["slug"]
    dest = OUT_ROOT / slug
    dest.mkdir(parents=True, exist_ok=True)
    write_json(
        dest / "meta.json",
        {
            "title": row["title"],
            "slug": slug,
            "url": row["url"],
            "id": int(row["id"]),
            "title_imputed": row.get("title_imputed") == "1",
        },
    )
    report: dict[str, Any] = {"slug": slug, "title": row["title"], "id": int(row["id"])}

    pct, _used, _free, _total = disk_snapshot()
    print(f"CODE {slug} disk {pct}%", flush=True)
    live_kernels, capped = list_kernels(api, slug)
    meta_kernels = load_code_index(row["id"])
    kernels_by_ref: dict[str, dict] = {}
    for kernel in live_kernels:
        kernels_by_ref[kernel["ref"]] = {"ref": kernel["ref"], "title": kernel.get("title") or "", "source": "live"}
    if meta_kernels:
        for kernel in meta_kernels:
            kernels_by_ref.setdefault(kernel["ref"], {"ref": kernel["ref"], "title": "", "source": "meta"})
    kernels = list(kernels_by_ref.values())
    code_dir = dest / "code"
    code_dir.mkdir(exist_ok=True)
    code_errors = []
    saved = 0
    for kernel in kernels:
        ref = kernel["ref"]
        owner, kernel_slug = ref.split("/", 1)
        folder = code_dir / f"{safe_name(owner)}__{safe_name(kernel_slug)}"
        try:
            if force and folder.exists():
                shutil.rmtree(folder)
            pull_kernel(api, ref, folder)
            saved += 1
        except DiskLimit:
            raise
        except Exception as exc:
            code_errors.append({"ref": ref, "error": str(exc)})
            print(f"CODE FAIL {ref}: {exc}", flush=True)
    index_missing = meta_kernels is None
    code_ok = saved == len(kernels) and not code_errors and not (capped and index_missing)
    write_json(
        code_dir / "manifest.json",
        {
            "capped": capped,
            "live": len(live_kernels),
            "meta": None if meta_kernels is None else len(meta_kernels),
            "kernels": kernels,
            "errors": code_errors,
        },
    )
    report["code"] = {
        "expected": len(kernels),
        "live": len(live_kernels),
        "meta": None if meta_kernels is None else len(meta_kernels),
        "downloaded": saved,
        "capped": capped,
        "ok": code_ok,
        "errors": code_errors,
    }
    if capped and index_missing:
        report["code"]["error"] = (
            f"kernel list hit the {KERNEL_LIST_CAP} cap and the Meta Kaggle code index is not built"
        )

    print(f"MODELS {slug}", flush=True)
    models = list_model_instances(sess, row["id"])
    model_dir = dest / "models"
    model_dir.mkdir(exist_ok=True)
    model_errors = []
    downloaded_models = 0
    expected_files = 0
    for item in models:
        ref = model_ref(item)
        record = {
            "model_instance_id": item.get("modelInstanceId"),
            "title": item.get("modelTitle"),
            "url": item.get("modelInstanceUrl"),
            "download_url": item.get("downloadUrl"),
            "ref": ref,
            "private": bool(item.get("isPrivate")),
        }
        if item.get("isPrivate") or not ref:
            model_errors.append({**record, "error": "no public download"})
            continue
        expected_files += 1
        owner, model, framework, instance, version = ref.split("/")
        folder = model_dir / f"{safe_name(owner)}__{safe_name(model)}__{safe_name(framework)}__{safe_name(instance)}"
        try:
            path = download_model(api, ref, folder)
            record["file"] = str(path.relative_to(dest))
            record["bytes"] = path.stat().st_size
            downloaded_models += 1
        except DiskLimit:
            raise
        except Exception as exc:
            model_errors.append({**record, "error": str(exc)})
            print(f"MODEL FAIL {ref}: {exc}", flush=True)
        item["_record"] = record
    write_json(
        model_dir / "manifest.json",
        {"instances": [item.get("_record") or {"model_instance_id": item.get("modelInstanceId")} for item in models], "errors": model_errors},
    )
    report["models"] = {
        "expected": len(models),
        "downloadable": expected_files,
        "downloaded": downloaded_models,
        "ok": downloaded_models == expected_files and not model_errors,
        "errors": model_errors,
    }

    print(f"DISCUSSION {slug}", flush=True)
    stubs = list_topic_stubs(sess, slug)
    views = load_views({int(topic["id"]) for topic in stubs})
    disc_dir = dest / "discussion"
    if force and disc_dir.exists():
        shutil.rmtree(disc_dir)
    disc_dir.mkdir(exist_ok=True)
    disc_errors = []
    saved_topics = 0
    expected_comments = 0
    saved_comments = 0
    index = []
    for stub in stubs:
        topic_id = int(stub["id"])
        try:
            topic, comments = fetch_comment_tree(api, topic_id)
            messages = raw_messages(sess, slug, topic_id)
            comment_dicts = [comment_to_dict(comment) for comment in comments]
            expected_comments += int(topic.comment_count or 0)
            saved_comments += count_comments(comment_dicts)
            raw_by_id = {int(message["id"]): message.get("rawMarkdown") or "" for message in messages if "id" in message}
            post_date = topic.post_date.isoformat() if topic.post_date else stub.get("postDate")
            payload = {
                "id": topic.id,
                "title": topic.title,
                "url": topic.url or stub.get("topicUrl"),
                "author_name": topic.author_name,
                "post_date": post_date,
                "votes": topic.votes,
                "comment_count": topic.comment_count,
                "views": views.get(topic_id),
                "content": topic.content,
                "messages": messages,
                "comments": comment_dicts,
                "raw_markdown_by_id": {str(k): v for k, v in raw_by_id.items()},
            }
            write_json(disc_dir / f"{topic_id}.json", payload)
            index.append(
                {
                    "id": topic.id,
                    "title": topic.title,
                    "comment_count": topic.comment_count,
                    "views": views.get(topic_id),
                    "votes": topic.votes,
                }
            )
            saved_topics += 1
        except DiskLimit:
            raise
        except Exception as exc:
            disc_errors.append({"id": topic_id, "error": str(exc)})
            print(f"DISCUSSION FAIL {topic_id}: {exc}", flush=True)
    write_json(disc_dir / "index.json", {"total": len(stubs), "topics": index, "errors": disc_errors})
    report["discussion"] = {
        "expected_topics": len(stubs),
        "saved_topics": saved_topics,
        "expected_comments": expected_comments,
        "saved_comments": saved_comments,
        "ok": saved_topics == len(stubs) and saved_comments == expected_comments and not disc_errors,
        "errors": disc_errors,
    }
    report["ok"] = report["code"]["ok"] and report["models"]["ok"] and report["discussion"]["ok"]
    write_json(dest / "VERIFY.json", report)
    print(json.dumps({"slug": slug, "ok": report["ok"], "code": report["code"], "models": report["models"], "discussion": report["discussion"]}, ensure_ascii=False), flush=True)
    return report


def check_model_paging(sess: requests.Session) -> None:
    """Two pages of UBC-OCEAN must be different instances. A 429 must not end the list."""
    first = request_json(
        lambda: post_text(sess, MODELS_URL, {"competitionId": "45867", "pageSize": 1}),
        what="UBC-OCEAN models page 1",
    )
    batch = list(first.get("pinnedModelInstances") or []) + list(first.get("unpinnedModelInstances") or [])
    token = first.get("nextPageToken") or ""
    if len(batch) != 1 or not token:
        raise SystemExit(f"paging check failed on first page: n={len(batch)} token={bool(token)}")
    second = request_json(
        lambda: post_text(sess, MODELS_URL, {"competitionId": "45867", "pageSize": 1, "pageToken": token}),
        what="UBC-OCEAN models page 2",
    )
    batch2 = list(second.get("pinnedModelInstances") or []) + list(second.get("unpinnedModelInstances") or [])
    if len(batch2) != 1 or batch2[0]["modelInstanceId"] == batch[0]["modelInstanceId"]:
        raise SystemExit(f"paging check failed on second page: {batch2}")
    print(f"paging ok {batch[0]['modelInstanceId']} -> {batch2[0]['modelInstanceId']}", flush=True)


def self_test() -> None:
    assert interpret_http(429, "")[0] == "retry"
    assert interpret_http(429, "{}")[0] == "retry"
    assert interpret_http(429, '{"topics":[],"totalCount":0}')[0] == "retry"
    assert interpret_http(200, "")[0] == "retry"
    assert interpret_http(200, "not-json")[0] == "retry"
    assert interpret_http(503, "")[0] == "retry"
    kind, data = interpret_http(200, '{"topics":[],"totalCount":0}')
    assert kind == "ok" and data["totalCount"] == 0
    kind, data = interpret_http(404, '{"error":{"code":404,"message":"missing"}}')
    assert kind == "fatal"
    assert backoff_seconds(1) >= BACKOFF_START

    calls: list[int] = []
    script = {
        1: [(429, ""), (429, "{}"), (200, json.dumps({"topics": [{"id": 1}], "totalCount": 2}))],
        2: [(200, ""), (200, json.dumps({"topics": [{"id": 2}], "totalCount": 2}))],
    }

    def fetch_page(page: int) -> tuple[int, str]:
        calls.append(page)
        return script[page].pop(0)

    topics = collect_topics(fetch_page, sleep=lambda _s: None, max_retries=6)
    assert [topic["id"] for topic in topics] == [1, 2], topics
    assert calls == [1, 1, 1, 2, 2], calls

    def empty_first(_page: int) -> tuple[int, str]:
        return 200, '{"topics":[],"totalCount":0}'

    assert collect_topics(empty_first, sleep=lambda _s: None) == []

    def short_page(page: int) -> tuple[int, str]:
        if page == 1:
            return 200, json.dumps({"topics": [{"id": 1}], "totalCount": 3})
        return 200, json.dumps({"topics": [], "totalCount": 3})

    try:
        collect_topics(short_page, sleep=lambda _s: None, max_retries=2)
        raise AssertionError("short page was accepted")
    except Incomplete:
        pass

    model_calls: list[str] = []
    model_script = [
        (429, ""),
        (200, json.dumps({"unpinnedModelInstances": [{"modelInstanceId": 1}], "nextPageToken": "t2"})),
        (429, "{}"),
        (200, json.dumps({"unpinnedModelInstances": [{"modelInstanceId": 2}]})),
    ]

    def fetch_token(token: str) -> tuple[int, str]:
        model_calls.append(token)
        return model_script.pop(0)

    models = collect_models(fetch_token, sleep=lambda _s: None, max_retries=4)
    assert [item["modelInstanceId"] for item in models] == [1, 2]
    assert model_script == []
    assert df_used_percent(90, 10) == 90
    assert df_used_percent(174, 826) == 18
    assert disk_is_full(90, 10)
    assert disk_is_full(891, 109)
    assert not disk_is_full(89, 11)
    assert not disk_is_full(890, 110)
    assert disk_is_full(1, 0)
    assert code_index_covers({"1", "2"}, {"1", "2"})
    assert not code_index_covers({"6538", "26183"}, {str(i) for i in range(3)})
    import tempfile

    index_dir = Path(tempfile.mkdtemp())
    index_path = index_dir / "code_index.jsonl"
    write_code_index({"10": [{"ref": "a/b", "kernel_id": "9"}], "2": []}, index_path)
    loaded = read_code_index(index_path)
    assert loaded == {"2": [], "10": [{"ref": "a/b", "kernel_id": "9"}]}
    assert code_index_covers(set(loaded or {}), {"2", "10"})
    assert not code_index_covers(set(loaded or {}), {"2", "10", "11"})
    print("self-test ok", flush=True)


def main() -> None:
    parser = argparse.ArgumentParser(description="Download public competition code, models, and discussions.")
    parser.add_argument("--slug", action="append", help="Competition slug to download. Repeatable.")
    parser.add_argument("--all", action="store_true", help="Download every row in the CSV. Resumes when VERIFY.json is ok.")
    parser.add_argument("--csv", type=Path, default=CSV_PATH)
    parser.add_argument("--force", action="store_true", help="Replace files already downloaded for the selected slug.")
    parser.add_argument("--self-test", action="store_true", help="Run pager tests, including 429 empty bodies.")
    parser.add_argument("--check-paging", action="store_true", help="Check live model pagination on UBC-OCEAN.")
    parser.add_argument("--build-code-index", action="store_true", help="Build the public-kernel index for --slug, or for every row.")
    args = parser.parse_args()

    if args.self_test:
        self_test()
        if not args.slug and not args.check_paging and not args.all:
            return

    rows = load_competitions(args.csv)
    by_slug = {row["slug"]: row for row in rows}
    if args.build_code_index:
        if args.slug:
            ids = {by_slug[slug]["id"] for slug in args.slug if slug in by_slug}
            missing = [slug for slug in args.slug if slug not in by_slug]
            if missing:
                raise SystemExit(f"not in csv: {', '.join(missing)}")
        else:
            ids = {row["id"] for row in rows}
        build_code_index(ids, STATE_DIR / "code_index.jsonl")
        if not args.slug and not args.check_paging and not args.all:
            return
    if args.all and not args.build_code_index:
        pct, used, free, total = disk_snapshot()
        print(
            f"disk {pct}% used ({used} used, {free} free, {total} total); stop at {DISK_STOP_RATIO:.0%}",
            flush=True,
        )
        if disk_is_full(used, free):
            ensure_disk_room("before code index")
        ensure_code_index({row["id"] for row in rows}, STATE_DIR / "code_index.jsonl")
    sess = session_from_kaggle_json()
    if args.check_paging:
        check_model_paging(sess)
        if not args.slug and not args.all:
            return
    if args.all:
        selected = [row["slug"] for row in rows]
    else:
        selected = args.slug or []
    if not selected:
        parser.error("Pass --slug or --all")

    from kaggle.api.kaggle_api_extended import KaggleApi

    api = KaggleApi()
    api.authenticate()
    failed = []
    summary_path = STATE_DIR / "summary.jsonl"
    summary_path.parent.mkdir(parents=True, exist_ok=True)
    pct, used, free, total = disk_snapshot()
    print(
        f"disk {pct}% used ({used} used, {free} free, {total} total); stop at {DISK_STOP_RATIO:.0%}",
        flush=True,
    )
    if disk_is_full(used, free):
        ensure_disk_room("before downloads")
    elif DISK_STOP_MARKER.exists():
        DISK_STOP_MARKER.unlink()
    for slug in selected:
        if slug not in by_slug:
            raise SystemExit(f"{slug} is not in {args.csv}")
        verify_path = OUT_ROOT / slug / "VERIFY.json"
        if args.all and not args.force and verify_path.exists():
            previous = json.loads(verify_path.read_text())
            if previous.get("ok"):
                continue
        try:
            ensure_disk_room(f"competition {slug}")
            report = download_competition(by_slug[slug], api, sess, force=args.force)
        except DiskLimit:
            return
        except Exception:
            traceback.print_exc()
            failed.append(slug)
            with summary_path.open("a") as handle:
                handle.write(json.dumps({"slug": slug, "ok": False, "error": "exception"}) + "\n")
            continue
        with summary_path.open("a") as handle:
            handle.write(json.dumps({"slug": slug, "ok": report["ok"], "code": report["code"]["downloaded"], "models": report["models"]["downloaded"], "topics": report["discussion"]["saved_topics"]}) + "\n")
        if not report["ok"]:
            failed.append(slug)
    if failed:
        raise SystemExit(f"incomplete: {len(failed)} competitions, first: {', '.join(failed[:10])}")


if __name__ == "__main__":
    main()