#!/usr/bin/env python3
"""
bunny-export.py - list a Bunny Stream library and write the CSV that StreamVault's
"Bring videos over" tool reads.

What it does
  1. Reads every video (and the collections they sit in) from one Bunny Stream library.
  2. Works out, for each finished video, the link StreamVault's servers can download:
     the original file when Bunny hands it out (it does that only while Early-Play is switched on in
     the library's Encoding settings), otherwise Bunny's MP4 copy (240p up to 1080p, never more).
  3. Checks each link the way our servers will fetch it - no referrer, no token - and keeps the ones
     that answer. (Skip this with --no-check.)
  4. Writes bunny-export.csv. Drop that file onto Dashboard > Bring videos over.

What it never does
  - It never changes anything in your Bunny library. It only reads.
  - It never writes or prints your API key, and it sends the key only to video.bunnycdn.com.
  - It never contacts StreamVault. The CSV stays on your computer until you upload it.

Needs Python 3.8 or newer and nothing else (standard library only).

Three things from your Bunny dashboard (the library's API page, under Access Credentials):
  --library   the library ID (a number)
  --cdn       the library's CDN hostname, like vz-1a2b3c4d-5e6.b-cdn.net (or your own hostname for it)
  the API key of that library: the script asks for it with the typing hidden, or reads BUNNY_KEY.
  Use the key of this Stream library, not the account-wide key. The library's "Read-only API Key",
  listed under the full one on the same page, is enough: this script only reads.

Example
  python3 bunny-export.py --library 12345 --cdn vz-1a2b3c4d-5e6.b-cdn.net

Columns written: Video Title, Download URL, Size, Folder Name, Subtitle Download URLs, Description,
Bunny Video ID. Collections become Folder Name (they become projects in StreamVault). Captions you
already have at Bunny are listed in Subtitle Download URLs and come across as subtitle tracks. Line
breaks inside titles and descriptions become spaces.

Exit code: 0 when a CSV with at least one video was written, 1 for errors, 2 for a usage mistake,
3 when nothing could be exported.
Run `python3 bunny-export.py --selftest` to test this script against a built-in fake Bunny. It needs no
account and no network.
"""

from __future__ import annotations

import argparse
import csv
import getpass
import json
import os
import re
import socket
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor

__version__ = "1.0"
USER_AGENT = "streamvault-bunny-export/" + __version__

# Production values. Only --selftest changes them, and only inside its own process: there is
# deliberately no flag or environment variable that points your API key anywhere else.
API_BASE = "https://video.bunnycdn.com"
CDN_SCHEME = "https://"
_SELFTEST = False
_SELFTEST_HOST = ""

PAGE_SIZE = 100
MAX_PAGES = 5000          # 500,000 videos. A guard against a server that never stops paging.
# Bunny's status codes: 4 Finished, 7 JitSegmenting, 8 JitPlaylistsCreated (just-in-time encoding libraries).
# Everything else (0-3 still uploading or encoding, 5 Error, 6 UploadFailed) is left out and listed.
READY_STATUSES = (4, 7, 8)
MAX_MP4_HEIGHT = 1080     # Bunny's MP4 fallback copies stop at 1080p
# The MP4 sizes Bunny makes. 480p is one of them: a real library on 8 Nov 2026 listed "480p,720p,240p,360p"
# (in no particular order) and answered play_480p.mp4 with a 200. 1080p is made only when the source is that big.
MP4_HEIGHTS = (1080, 720, 480, 360, 240)
CHECK_WORKERS = 8
CHECK_TIMEOUT = 20
RETRY_DELAYS = (1, 2, 4, 8)
SLEEP = time.sleep

COLUMNS = [
    "Video Title", "Download URL", "Size", "Folder Name",
    "Subtitle Download URLs", "Description", "Bunny Video ID",
]
SKIPPED_COLUMNS = ["Bunny Video ID", "Video Title", "Reason"]

# The same shape StreamVault's importer reads: "[en] https://...". Anything else is left out.
_LANG_RE = re.compile(r"^[A-Za-z]{2,3}(?:-[A-Za-z]{2,4})?$")
_GUID_RE = re.compile(r"^[0-9A-Za-z-]{8,64}$")
_HOST_RE = re.compile(r"^(?=.{4,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$")


class ExportError(Exception):
    """Something the person running the script can act on. The message is shown as-is."""

    def __init__(self, message, auth=False):
        super().__init__(message)
        self.auth = auth


def say(message=""):
    try:
        print(message)
    except UnicodeEncodeError:
        print(message.encode("ascii", "replace").decode("ascii"))


# --------------------------------------------------------------------------------------
# Input
# --------------------------------------------------------------------------------------

def read_key():
    key = os.environ.get("BUNNY_KEY", "")
    if not key.strip():
        if not sys.stdin or not sys.stdin.isatty():
            raise ExportError(
                "No API key. Set the BUNNY_KEY environment variable, or run this in a terminal "
                "and it will ask for the key."
            )
        key = getpass.getpass("Bunny Stream API key for this library (typing is hidden): ")
    key = key.strip()
    if not key or len(key) > 200 or not re.fullmatch(r"[\x21-\x7e]+", key):
        raise ExportError(
            "That does not look like an API key. It should be one line of letters, digits and "
            "dashes, with no spaces."
        )
    return key


def normalise_host(raw):
    s = (raw or "").strip()
    s = re.sub(r"^[A-Za-z][A-Za-z0-9+.-]*://", "", s)
    s = s.split("/", 1)[0].split("?", 1)[0].strip().lower().rstrip(".")
    if _SELFTEST and s == _SELFTEST_HOST:
        return s
    if s in ("video.bunnycdn.com", "api.bunny.net"):
        raise ExportError(
            "%s is Bunny's API address. You need the library's CDN hostname instead, "
            "which looks like vz-1a2b3c4d-5e6.b-cdn.net." % s
        )
    if not _HOST_RE.match(s):
        raise ExportError(
            "'%s' is not a hostname. Use the library's CDN hostname, like vz-1a2b3c4d-5e6.b-cdn.net "
            "(or your own hostname for it)." % (raw or "").strip()
        )
    return s


# --------------------------------------------------------------------------------------
# Bunny's API
# --------------------------------------------------------------------------------------

def describe_network_error(exc):
    reason = getattr(exc, "reason", exc)
    if isinstance(reason, ssl.SSLCertVerificationError) or "CERTIFICATE_VERIFY_FAILED" in str(reason):
        return (
            "Python could not verify the security certificate. On a Mac with Python from python.org, "
            "run 'Install Certificates.command' (in /Applications/Python 3.x) and try again"
        )
    if isinstance(reason, (socket.timeout, TimeoutError)):
        return "the connection timed out"
    if isinstance(reason, socket.gaierror):
        return "the address could not be found - are you online?"
    return str(reason) or exc.__class__.__name__


def _retry_after(err):
    value = err.headers.get("Retry-After") if getattr(err, "headers", None) else None
    try:
        return max(0, min(int(value), 30)) if value is not None else None
    except ValueError:
        return None


def api_get(path, params, key, timeout=30):
    if not API_BASE.startswith("https://") and not _SELFTEST:
        raise ExportError("Refusing to send the API key over an unencrypted connection.")
    url = API_BASE + path + ("?" + urllib.parse.urlencode(params) if params else "")
    request = urllib.request.Request(
        url, headers={"AccessKey": key, "Accept": "application/json", "User-Agent": USER_AGENT}
    )
    tries = len(RETRY_DELAYS) + 1
    for attempt in range(tries):
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return json.loads(response.read().decode("utf-8"))
        except urllib.error.HTTPError as err:
            if err.code == 401:
                raise ExportError(
                    "Bunny did not accept the API key (HTTP 401). Use the API key of this Stream library, "
                    "from the library's API page, not the account-wide key. Check the library ID too.",
                    auth=True,
                )
            if err.code == 404:
                raise ExportError(
                    "Bunny has no such library for this key (HTTP 404). Check the library ID.", auth=True
                )
            if err.code in (429, 500, 502, 503, 504) and attempt < tries - 1:
                wait = _retry_after(err)
                SLEEP(RETRY_DELAYS[attempt] if wait is None else wait)
                continue
            raise ExportError("Bunny answered HTTP %d for %s." % (err.code, path))
        except (urllib.error.URLError, OSError) as err:
            message = describe_network_error(err)
            if "security certificate" in message or attempt >= tries - 1:
                raise ExportError("Could not reach Bunny: %s." % message)
            SLEEP(RETRY_DELAYS[attempt])
        except (ValueError, UnicodeDecodeError):
            raise ExportError("Bunny sent a reply this script could not read (not JSON).")
    raise ExportError("Could not reach Bunny.")


def list_items(path, key):
    """Every item of a paged Bunny list. Tolerant of the reply shape and of a server that ignores paging."""
    items, seen, page = [], set(), 1
    while page <= MAX_PAGES:
        data = api_get(path, {"page": page, "itemsPerPage": PAGE_SIZE}, key)
        batch = data.get("items") if isinstance(data, dict) else data
        if not isinstance(batch, list):
            raise ExportError("Bunny's reply for %s was not a list of videos." % path)
        fresh = []
        for entry in batch:
            if not isinstance(entry, dict):
                continue
            ident = entry.get("guid")
            if ident in seen:
                continue
            seen.add(ident)
            fresh.append(entry)
        items.extend(fresh)
        total = data.get("totalItems") if isinstance(data, dict) else None
        if not batch or not fresh:
            break
        if isinstance(total, int) and len(items) >= total:
            break
        if total is None and len(batch) < PAGE_SIZE:
            break
        page += 1
    return items


def fetch_collections(library, key):
    try:
        items = list_items("/library/%s/collections" % library, key)
    except ExportError as err:
        if err.auth:
            raise
        say("Warning: could not read the library's collections (%s). Every video will go in at the top level." % err)
        return {}
    names = {}
    for c in items:
        if isinstance(c.get("guid"), str) and isinstance(c.get("name"), str):
            names[c["guid"]] = c["name"]
    return names


# --------------------------------------------------------------------------------------
# Choosing a link
# --------------------------------------------------------------------------------------

def _height(token):
    m = re.match(r"^\s*(\d{3,4})p?\s*$", str(token))
    return int(m.group(1)) if m else 0


def best_mp4_height(video):
    if not video.get("hasMP4Fallback"):
        return 0
    raw = video.get("availableResolutions") or ""
    parts = raw if isinstance(raw, list) else str(raw).split(",")
    available = {h for h in (_height(p) for p in parts) if h}
    for height in MP4_HEIGHTS:
        if height in available:
            return height
    return 0


def candidates(video, cdn, prefer):
    """Download links to try, best first, as (url, kind)."""
    base = "%s%s/%s" % (CDN_SCHEME, cdn, video["guid"])
    found = {}
    # hasOriginal False is Bunny saying there is none. Missing means "not reported": try it.
    if video.get("hasOriginal") is not False:
        found["original"] = (base + "/original", "original")
    height = best_mp4_height(video)
    if height:
        found["mp4"] = ("%s/play_%dp.mp4" % (base, height), "mp4 %dp" % height)
    order = ("mp4", "original") if prefer == "mp4" else ("original", "mp4")
    return [found[k] for k in order if k in found]


def clean_text(value):
    return re.sub(r"\s+", " ", str(value if value is not None else "")).strip()


def safe_cell(text):
    """A leading space keeps Excel from running a title that starts with = + - or @ as a formula.
    StreamVault trims every cell, so the imported title is unchanged."""
    return " " + text if text[:1] in ("=", "+", "-", "@") else text


def captions_cell(video, cdn):
    tracks, seen = [], set()
    for track in video.get("captions") or []:
        lang = track.get("srclang") if isinstance(track, dict) else None
        if not isinstance(lang, str) or not _LANG_RE.match(lang) or lang.lower() in seen:
            continue
        seen.add(lang.lower())
        tracks.append("[%s] %s%s/%s/captions/%s.vtt" % (lang.lower(), CDN_SCHEME, cdn, video["guid"], lang))
    return " ".join(tracks)


# --------------------------------------------------------------------------------------
# Checking a link the way StreamVault's servers will fetch it
# --------------------------------------------------------------------------------------

def _size_from(headers, status):
    rng = headers.get("Content-Range")
    if rng:
        m = re.search(r"/(\d+)\s*$", rng)
        if m:
            return int(m.group(1))
    if status == 200:
        try:
            n = int(headers.get("Content-Length", ""))
            return n if n > 0 else None
        except ValueError:
            return None
    return None


def _fetch_status(url, method, headers):
    """(status or None, size or None, note) for one request. Tries twice on a network error or a 429/5xx."""
    for attempt in (0, 1):
        request = urllib.request.Request(url, method=method, headers=headers)
        try:
            with urllib.request.urlopen(request, timeout=CHECK_TIMEOUT) as response:
                if "text/html" in response.headers.get("Content-Type", "").lower():
                    return response.status, None, "webpage"
                return response.status, _size_from(response.headers, response.status), ""
        except urllib.error.HTTPError as err:
            if err.code in (429, 500, 502, 503, 504) and attempt == 0:
                SLEEP(1)
                continue
            return err.code, None, ""
        except (urllib.error.URLError, OSError) as err:
            if attempt == 0:
                SLEEP(1)
                continue
            return None, None, describe_network_error(err)
    return None, None, "no answer"


def probe(url):
    """(HTTP status or None, size in bytes or None, note). No referrer, no token, like our servers."""
    status, size, note = _fetch_status(url, "HEAD", {"User-Agent": USER_AGENT})
    if status in (405, 501):
        # This server refuses HEAD: ask for the first byte instead and read the size from Content-Range.
        status, size, note = _fetch_status(url, "GET", {"User-Agent": USER_AGENT, "Range": "bytes=0-0"})
    return status, size, note


class Problem:
    """Why a video is not in the CSV. `code` groups them in the summary."""

    def __init__(self, code, text):
        self.code, self.text = code, text


def problem_from_probe(kind, status, note):
    if note == "webpage":
        return Problem("webpage", "%s: Bunny answered with a web page, not a video" % kind)
    if status is None:
        return Problem("network", "%s: could not connect (%s)" % (kind, note))
    if status == 403:
        return Problem("forbidden", "%s: Bunny refused the download (HTTP 403)" % kind)
    if status == 404:
        return Problem("missing", "%s: not found at Bunny (HTTP 404)" % kind)
    return Problem("http", "%s: HTTP %d" % (kind, status))


_PRIORITY = ("forbidden", "webpage", "network", "http", "missing")


def resolve_one(video, cands, check):
    """-> (video, url, kind, size, Problem or None, original_refused).

    original_refused is True when Bunny turned the original file away with a 403 and the MP4 copy
    answered instead: the one fallback a person may want to know about (see summarise)."""
    if not cands:
        return video, None, None, None, Problem("nofile", "no original file and no MP4 copy at Bunny"), False
    if not check:
        return video, cands[0][0], cands[0][1], None, None, False
    problems = []
    refused = False
    for url, kind in cands:
        status, size, note = probe(url)
        if status is not None and 200 <= status < 300 and not note:
            return video, url, kind, size, None, refused
        problems.append(problem_from_probe(kind, status, note))
        if kind == "original" and status == 403:
            refused = True
    worst = min(problems, key=lambda p: _PRIORITY.index(p.code) if p.code in _PRIORITY else 99)
    return video, None, None, None, Problem(worst.code, "; ".join(p.text for p in problems)), False


# --------------------------------------------------------------------------------------
# The export
# --------------------------------------------------------------------------------------

def skipped_path(out):
    root, ext = os.path.splitext(out)
    return root + "-skipped" + (ext or ".csv")


def write_csv(path, header, rows):
    # utf-8-sig so Excel shows accented titles correctly; QUOTE_ALL because StreamVault's reader also
    # splits on semicolons and tabs, and a quoted cell is the only way a title keeps them.
    with open(path, "w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.writer(handle, quoting=csv.QUOTE_ALL, lineterminator="\r\n")
        writer.writerow(header)
        writer.writerows(rows)


def run_export(library, cdn, key, out, prefer="original", check=True, limit=None):
    say("Reading library %s from Bunny..." % library)
    videos = list_items("/library/%s/videos" % library, key)
    if limit:
        videos = videos[:limit]
    collections = fetch_collections(library, key)
    say("Found %d video%s." % (len(videos), "" if len(videos) == 1 else "s"))

    work, skipped = [], []
    for video in videos:
        guid = video.get("guid")
        title = clean_text(video.get("title")) or (guid if isinstance(guid, str) else "")
        if not isinstance(guid, str) or not _GUID_RE.match(guid):
            skipped.append(("", title, Problem("unfinished", "Bunny sent no usable video ID")))
            continue
        if video.get("status") not in READY_STATUSES:
            skipped.append((guid, title, Problem("unfinished", "not finished encoding at Bunny (status %s)" % video.get("status"))))
            continue
        work.append((video, candidates(video, cdn, prefer)))

    if check and work:
        say("Checking %d link%s the way StreamVault's servers will fetch them (no referrer, no token)..." % (len(work), "" if len(work) == 1 else "s"))
    results = []
    with ThreadPoolExecutor(max_workers=CHECK_WORKERS if check else 1) as pool:
        for i, res in enumerate(pool.map(lambda w: resolve_one(w[0], w[1], check), work), 1):
            results.append(res)
            if check and i % 100 == 0 and i < len(work):
                say("  checked %d of %d" % (i, len(work)))

    rows, kinds, fell_back = [], {}, 0
    for video, url, kind, size, problem, original_refused in results:
        guid = video["guid"]
        title = clean_text(video.get("title")) or guid
        if problem is not None:
            skipped.append((guid, title, problem))
            continue
        folder = clean_text(collections.get(video.get("collectionId") or "", ""))
        description = video.get("description")
        rows.append([
            safe_cell(title), url, str(size) if size else "", safe_cell(folder),
            captions_cell(video, cdn), safe_cell(clean_text(description) if isinstance(description, str) else ""), guid,
        ])
        base_kind = "original" if kind == "original" else "mp4"
        kinds[base_kind] = kinds.get(base_kind, 0) + 1
        if original_refused and base_kind == "mp4":
            fell_back += 1

    write_csv(out, COLUMNS, rows)
    skipped_file = skipped_path(out)
    write_csv(skipped_file, SKIPPED_COLUMNS, [[g, safe_cell(t), safe_cell(p.text)] for g, t, p in skipped])
    summarise(out, skipped_file, len(videos), rows, kinds, skipped, check, fell_back)
    return rows, skipped


def summarise(out, skipped_file, total, rows, kinds, skipped, check, fell_back=0):
    say()
    say("Exported %d of %d video%s to %s" % (len(rows), total, "" if total == 1 else "s", out))
    if rows:
        parts = []
        if kinds.get("original"):
            parts.append("%d original file%s" % (kinds["original"], "" if kinds["original"] == 1 else "s"))
        if kinds.get("mp4"):
            parts.append("%d MP4 cop%s at up to %dp" % (kinds["mp4"], "y" if kinds["mp4"] == 1 else "ies", MAX_MP4_HEIGHT))
        say("  " + ", ".join(parts))
        if fell_back:
            # Seen on a real library (8 Nov 2026): with Early-Play off, Bunny answers /original with a 403 even when
            # every other switch is off, while the MP4 copies download fine. Say so once, plainly.
            say("  %d of them came as an MP4 copy because Bunny refused the original file. Bunny hands out originals only" % fell_back)
            say("  while Early-Play is on (Encoding settings), and says that makes the original files public. If a copy of")
            say("  up to 1080p is enough for you, there is nothing to do. If you need the full-quality originals, email")
            say("  video@streamvault.one before you change that switch: in our test an original was still being served")
            say("  more than half an hour after we switched it off again.")
        if not check:
            say("  Links were not checked (--no-check). Run again without it to catch refused or missing files.")
    if skipped:
        counts = {}
        for _g, _t, problem in skipped:
            counts[problem.code] = counts.get(problem.code, 0) + 1
        say("Left out %d, listed with the reason in %s:" % (len(skipped), skipped_file))
        labels = {
            "unfinished": "not finished encoding at Bunny",
            "nofile": "no original file and no MP4 copy at Bunny",
            "forbidden": "refused by Bunny (HTTP 403)",
            "missing": "file not found at Bunny (HTTP 404)",
            "webpage": "Bunny answered with a web page, not a video",
            "network": "could not connect",
            "http": "Bunny answered with an error",
        }
        for code, n in sorted(counts.items(), key=lambda kv: -kv[1]):
            say("  %d  %s" % (n, labels.get(code, code)))
        shown = skipped[:8]
        for _g, title, problem in shown:
            say("    - %s  (%s)" % (title[:70], problem.text))
        if len(skipped) > len(shown):
            say("    ... and %d more in %s" % (len(skipped) - len(shown), skipped_file))
        if counts.get("forbidden"):
            say()
            say("StreamVault fetches each file without a referrer or a token. For the length of the move, turn off")
            say("Token Authentication, Allowed domains and Block Direct URL File Access (Bunny switches that one on")
            say("for a new library) on this library in Bunny, then run this script again.")
    if rows:
        say()
        say("Next: open StreamVault > Bring videos over and drop %s onto the page." % os.path.basename(out))
        say("Keep the file private: anyone who has a link in it can download that video from Bunny until you delete it there.")


def parse_args(argv):
    parser = argparse.ArgumentParser(
        prog="bunny-export.py",
        description="Write a CSV of a Bunny Stream library for StreamVault's Bring videos over tool.",
    )
    parser.add_argument("--library", help="Bunny Stream library ID (a number)")
    parser.add_argument("--cdn", help="the library's CDN hostname, like vz-1a2b3c4d-5e6.b-cdn.net")
    parser.add_argument("--out", help="CSV file to write (default bunny-export.csv)")
    parser.add_argument("--prefer", choices=("original", "mp4"), default="original",
                        help="original (default): the original file when Bunny hands it out, else its MP4 copy. "
                             "mp4: Bunny's MP4 copy (1080p at most), smaller and faster to move.")
    parser.add_argument("--no-check", action="store_true", help="do not check that each link downloads")
    parser.add_argument("--limit", type=int, help="only the first N videos, for a trial run")
    parser.add_argument("--selftest", action="store_true", help="test this script against a built-in fake Bunny")
    parser.add_argument("--version", action="version", version=USER_AGENT)
    return parser, parser.parse_args(argv)


def main(argv=None):
    parser, args = parse_args(argv)
    try:
        if args.selftest:
            return selftest(args.out)
        if not args.library or not args.cdn:
            parser.error("--library and --cdn are required (run with --help to see where to find them)")
        if not re.fullmatch(r"\d{1,12}", args.library.strip()):
            raise ExportError("The library ID is a number, like 12345. You gave '%s'." % args.library)
        if args.limit is not None and args.limit < 1:
            raise ExportError("--limit must be 1 or more.")
        cdn = normalise_host(args.cdn)
        key = read_key()
        rows, _ = run_export(args.library.strip(), cdn, key, args.out or "bunny-export.csv",
                             prefer=args.prefer, check=not args.no_check, limit=args.limit)
        if not rows:
            say()
            say("Nothing could be exported. The reasons are above and in the -skipped file.")
            return 3
        return 0
    except ExportError as err:
        print("Error: %s" % err, file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("Stopped.", file=sys.stderr)
        return 130


# --------------------------------------------------------------------------------------
# Self-test: the whole script against a fake Bunny that follows Bunny's published API.
# It proves the logic and the CSV. It does not prove how a real library behaves.
# --------------------------------------------------------------------------------------

def selftest(out_path=None):
    import contextlib
    import io
    import tempfile
    import threading
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

    global API_BASE, CDN_SCHEME, _SELFTEST, _SELFTEST_HOST, SLEEP
    key = "selftest-key-5f3a9c1e-not-a-real-key"
    library = "777"
    coll_a, coll_b = "c0000000-0000-4000-8000-00000000000a", "c0000000-0000-4000-8000-00000000000b"

    def guid(n):
        return "00000000-0000-4000-8000-%012d" % n

    videos = []

    def add(n, **kw):
        v = {"guid": guid(n), "title": "Lesson %d.mp4" % n, "status": 4, "hasOriginal": True,
             "hasMP4Fallback": True, "availableResolutions": "240p,360p,480p,720p,1080p",
             "collectionId": "", "captions": [], "length": 600, "storageSize": 123}
        v.update(kw)
        videos.append(v)

    for n in range(200):
        add(n, collectionId=coll_a if n % 3 == 0 else (coll_b if n % 3 == 1 else ""))
    add(200, hasOriginal=False, availableResolutions="360p,720p,1080p,1440p,2160p")      # MP4 at 1080p
    add(201, hasOriginal=False, hasMP4Fallback=False)                                     # nothing to fetch
    add(202, status=3)                                                                    # still encoding
    add(203, status=5)                                                                    # failed at Bunny
    add(204, title='Cafe "Intro", part 2; ünï\tnext\nline', description="One\ntwo; three, four")
    add(205, title="=HYPERLINK(\"http://example.com\",\"x\")")                           # formula-looking title
    add(206)
    del videos[-1]["hasOriginal"]                                                         # not reported: try it, fall back
    add(207)                                                                              # CDN refuses: 403
    add(208, hasMP4Fallback=False)                                                        # CDN answers with a web page
    add(209)                                                                              # CDN refuses HEAD
    add(210, captions=[{"srclang": "en_US", "label": "x"}, {"srclang": "pt-br", "label": "Portugues"}, {"srclang": "pt-BR", "label": "dup"}])
    add(211, title="")                                                                    # no title: use the ID
    add(212, captions=[{"srclang": "en", "label": "English"}, {"srclang": "fr", "label": "French"}])
    add(213, hasOriginal=False, hasMP4Fallback=False, availableResolutions="2160p")
    add(214, status=8)                                                                    # just-in-time library: original only
    add(215, hasOriginal=False, availableResolutions="240p,360p,480p")                    # largest MP4 is 480p: Bunny does make that one
    add(216, hasOriginal=None)                                                            # null, as the real API sends it; original refused (403)
    add(217, hasOriginal=False, availableResolutions="480p,720p,240p,360p")               # sizes in no order, as the real API lists them

    rules = {
        guid(206): {"original": 404},
        guid(207): {"original": 403, "mp4": 403},
        guid(208): {"original": "html"},
        guid(209): {"original": "nohead"},
        guid(216): {"original": 403},
    }
    state = {"api_calls": [], "cdn_calls": [], "leaks": [], "bad_key": 0, "fail_once": 0, "fail_once_armed": False}

    class Handler(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, *_a):
            pass

        def _send(self, code, body=b"", headers=None, content_length=None):
            self.send_response(code)
            for k, v in (headers or {}).items():
                self.send_header(k, v)
            self.send_header("Content-Length", str(len(body) if content_length is None else content_length))
            self.end_headers()
            if body and self.command != "HEAD":
                self.wfile.write(body)

        def _json(self, code, payload):
            self._send(code, json.dumps(payload).encode(), {"Content-Type": "application/json"})

        def _handle(self):
            parsed = urllib.parse.urlparse(self.path)
            parts = [p for p in parsed.path.split("/") if p]
            if parts and parts[0] == "library":
                state["api_calls"].append(self.path)
                if self.headers.get("AccessKey") != key:
                    state["bad_key"] += 1
                    return self._json(401, {"message": "Unauthorized"})
                if state["fail_once_armed"] and state["fail_once"] == 0:
                    state["fail_once"] += 1
                    return self._send(503, b"", {"Retry-After": "0"})
                if len(parts) >= 2 and parts[1] == "500":
                    return self._send(500)
                if len(parts) < 3 or parts[1] != library:
                    return self._json(404, {"message": "not found"})
                q = urllib.parse.parse_qs(parsed.query)
                page, per = int(q.get("page", ["1"])[0]), int(q.get("itemsPerPage", ["100"])[0])
                data = videos if parts[2] == "videos" else [
                    {"guid": coll_a, "name": "Course A", "videoCount": 67},
                    {"guid": coll_b, "name": "Course B; advanced", "videoCount": 67},
                ]
                chunk = data[(page - 1) * per: page * per]
                return self._json(200, {"totalItems": len(data), "currentPage": page, "itemsPerPage": per, "items": chunk})
            # CDN side
            state["cdn_calls"].append((self.command, self.path))
            if self.headers.get("AccessKey") is not None or self.headers.get("Authorization") is not None:
                state["leaks"].append("credential header sent to the CDN: " + self.path)
            if self.headers.get("Referer") is not None:
                state["leaks"].append("referrer sent to the CDN: " + self.path)
            if len(parts) == 2 and (parts[1] == "original" or re.fullmatch(r"play_\d+p\.mp4", parts[1])):
                kind = "original" if parts[1] == "original" else "mp4"
                rule = rules.get(parts[0], {}).get(kind)
                n = 1_000_000 + int(parts[0][-6:]) if kind == "original" else 500_000 + int(parts[0][-6:])
                if rule == 403:
                    return self._send(403)
                if rule == 404:
                    return self._send(404)
                if rule == "html":
                    return self._send(200, b"", {"Content-Type": "text/html"}, content_length=4096)
                if rule == "nohead" and self.command == "HEAD":
                    return self._send(405)
                if self.command == "GET" and self.headers.get("Range"):
                    return self._send(206, b"x", {"Content-Type": "video/mp4", "Content-Range": "bytes 0-0/%d" % n})
                return self._send(200, b"", {"Content-Type": "video/mp4"}, content_length=n)
            return self._send(404)

        do_GET = do_HEAD = _handle

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    port = server.server_address[1]
    threading.Thread(target=server.serve_forever, daemon=True).start()
    host = "127.0.0.1:%d" % port

    failures, total = [], [0]

    def check(condition, message):
        total[0] += 1
        if not condition:
            failures.append(message)

    def run(args, use_key=key):
        out, err = io.StringIO(), io.StringIO()
        old = os.environ.get("BUNNY_KEY")
        os.environ["BUNNY_KEY"] = use_key
        try:
            with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
                code = main(args)
        finally:
            if old is None:
                os.environ.pop("BUNNY_KEY", None)
            else:
                os.environ["BUNNY_KEY"] = old
        return code, out.getvalue(), err.getvalue()

    def read_csv(path):
        with open(path, newline="", encoding="utf-8-sig") as handle:
            return list(csv.reader(handle))

    saved = (API_BASE, CDN_SCHEME, _SELFTEST, _SELFTEST_HOST, SLEEP)
    API_BASE, CDN_SCHEME, _SELFTEST, _SELFTEST_HOST, SLEEP = "http://" + host, "http://", True, host, (lambda _s: None)
    workdir = tempfile.mkdtemp(prefix="bunny-selftest-")
    try:
        main_out = out_path or os.path.join(workdir, "bunny-export.csv")
        base_args = ["--library", library, "--cdn", host]

        # 1. the whole export, with a 503 on the very first call to prove the retry
        state["fail_once_armed"] = True
        code, out, err = run(base_args + ["--out", main_out])
        state["fail_once_armed"] = False
        check(code == 0, "export exits 0 (got %s: %s %s)" % (code, out[-300:], err))
        check(state["fail_once"] == 1, "a 503 is retried and then succeeds")
        rows = read_csv(main_out)
        with open(main_out, "rb") as handle:
            raw_start = handle.read(40)
        check(raw_start.startswith(b"\xef\xbb\xbf" + b'"Video Title","Download URL"'), "UTF-8 with a BOM, every cell quoted: %r" % raw_start)
        check(rows[0] == COLUMNS, "header row is exactly the documented columns: %r" % (rows[0],))
        body = {r[6]: r for r in rows[1:]}
        check(guid(0) in body and body[guid(0)][1] == "http://%s/%s/original" % (host, guid(0)), "original link built")
        check(body[guid(0)][0] == "Lesson 0.mp4", "title is passed through unchanged (the importer strips .mp4)")
        check(body[guid(0)][2] == str(1_000_000), "size comes from the checked link: %r" % body[guid(0)][2])
        check(body[guid(0)][3] == "Course A", "collection becomes Folder Name")
        check(body[guid(2)][3] == "", "no collection leaves Folder Name empty")
        check(body[guid(1)][3] == "Course B; advanced", "a semicolon in a folder name survives")
        check(body[guid(200)][1].endswith("/play_1080p.mp4"), "no original: MP4 at the best height up to 1080p")
        check(guid(201) not in body and guid(213) not in body, "no original and no MP4: left out")
        check(guid(215) in body and body[guid(215)][1].endswith("/play_480p.mp4"), "a video whose largest MP4 is 480p uses it")
        check(guid(216) in body and body[guid(216)][1].endswith("/play_1080p.mp4"), "original refused (403), hasOriginal null: the MP4 copy is used")
        check(guid(217) in body and body[guid(217)][1].endswith("/play_720p.mp4"), "sizes listed in any order: the best one wins")
        check("1 of them came as an MP4 copy because Bunny refused the original file" in out and "Early-Play" in out,
              "a refused original that fell back to an MP4 is explained, with Early-Play named")
        check(guid(202) not in body and guid(203) not in body, "unfinished and failed videos are left out")
        check(guid(214) in body and body[guid(214)][1].endswith("/original"), "a just-in-time video (status 8) is exported")
        t204 = body[guid(204)][0]
        check("\n" not in t204 and "\t" not in t204 and '"' in t204 and "," in t204 and ";" in t204, "awkward title kept, line breaks and tabs flattened: %r" % t204)
        check(body[guid(204)][5] == "One two; three, four", "description flattened to one line")
        check(body[guid(205)][0].startswith(" ="), "formula-looking title is neutralised for Excel")
        check(body[guid(206)][1].endswith("/play_1080p.mp4"), "missing original (404) falls back to the MP4 copy")
        check(guid(207) not in body and guid(208) not in body, "403 and web-page answers are left out")
        check(guid(209) in body and body[guid(209)][1].endswith("/original"), "a server that refuses HEAD is checked with a one-byte GET")
        check(body[guid(209)][2] == str(1_000_000 + 209), "size read from Content-Range: %r" % body[guid(209)][2])
        check(body[guid(210)][4] == "[pt-br] http://%s/%s/captions/pt-br.vtt" % (host, guid(210)), "bad language tag dropped, duplicate dropped: %r" % body[guid(210)][4])
        check(body[guid(211)][0] == guid(211), "empty title falls back to the video ID")
        check(body[guid(212)][4].count("[") == 2 and "[en] " in body[guid(212)][4] and "[fr] " in body[guid(212)][4], "two captions listed")
        expected_rows = 200 + len([200, 204, 205, 206, 209, 210, 211, 212, 214, 215, 216, 217])
        check(len(rows) - 1 == expected_rows, "exported %d rows, expected %d" % (len(rows) - 1, expected_rows))
        skipped = read_csv(skipped_path(main_out))
        check(skipped[0] == SKIPPED_COLUMNS, "skipped file header")
        reasons = {r[0]: r[2] for r in skipped[1:]}
        check(set(reasons) == {guid(i) for i in (201, 202, 203, 207, 208, 213)}, "skipped set: %r" % sorted(reasons))
        check("403" in reasons.get(guid(207), ""), "403 reason names the HTTP code")
        check("web page" in reasons.get(guid(208), ""), "web page reason")
        check("Token Authentication" in out and "Allowed domains" in out and "Block Direct URL File Access" in out, "403 summary tells the person what to switch off")
        pages = [c for c in state["api_calls"] if "/videos?" in c]
        check(len(pages) >= 3 and all("itemsPerPage=100" in c for c in pages), "videos are read 100 at a time over several pages: %d calls" % len(pages))
        check(not state["leaks"], "no credential or referrer ever reached the CDN: %r" % state["leaks"][:3])
        everything = out + err + open(main_out, encoding="utf-8-sig").read() + open(skipped_path(main_out), encoding="utf-8-sig").read()
        check(key not in everything, "the API key appears in no output and no file")

        # 2. --no-check: no CDN traffic, no sizes, MP4-less original links for everything finished
        state["cdn_calls"].clear()
        nocheck_out = os.path.join(workdir, "nocheck.csv")
        code, out, err = run(base_args + ["--out", nocheck_out, "--no-check"])
        check(code == 0 and not state["cdn_calls"], "--no-check makes no CDN request")
        nc = read_csv(nocheck_out)
        check(all(r[2] == "" for r in nc[1:]), "--no-check leaves Size empty")
        check(any(r[6] == guid(207) for r in nc[1:]), "--no-check exports unchecked links")

        # 3. --prefer mp4 and --limit
        mp4_out = os.path.join(workdir, "mp4.csv")
        code, out, err = run(base_args + ["--out", mp4_out, "--prefer", "mp4", "--limit", "5"])
        m = read_csv(mp4_out)
        check(code == 0 and len(m) - 1 == 5 and all(r[1].endswith("/play_1080p.mp4") for r in m[1:]), "--prefer mp4 --limit 5 gives five MP4 links")
        check("Early-Play" not in out, "no Early-Play advice when no original was refused")

        # 4. wrong key: a clear message, exit 1, key never echoed
        state["cdn_calls"].clear()
        code, out, err = run(base_args + ["--out", os.path.join(workdir, "bad.csv")], use_key="wrong-key-abcdef123456")
        check(code == 1 and "401" in err and "wrong-key-abcdef123456" not in out + err, "a rejected key is explained and never echoed")
        check(not os.path.exists(os.path.join(workdir, "bad.csv")), "nothing is written when the key is rejected")

        # 5. unknown library
        code, out, err = run(["--library", "999", "--cdn", host, "--out", os.path.join(workdir, "nolib.csv")])
        check(code == 1 and "library" in err.lower(), "an unknown library is explained")

        # 6. bad inputs
        code, _o, err = run(["--library", "12x", "--cdn", host])
        check(code == 1 and "number" in err, "library ID must be a number")
        code, _o, err = run(["--library", "1", "--cdn", "video.bunnycdn.com"])
        check(code == 1 and "API address" in err, "the API host is refused as a CDN hostname")
        check(normalise_host("https://VZ-abc123-456.b-cdn.net/") == "vz-abc123-456.b-cdn.net", "scheme, path and case are tidied")
        check(safe_cell("plain") == "plain" and safe_cell("-x") == " -x", "safe_cell")

        # 7. Bunny failing: five attempts, no more, then a plain message
        before = len(state["api_calls"])
        code, out, err = run(["--library", "500", "--cdn", host, "--out", os.path.join(workdir, "down.csv")])
        check(code == 1 and "HTTP 500" in err and key not in out + err, "a server error is reported after retries")
        check(len(state["api_calls"]) - before == 5, "exactly 5 attempts, got %d" % (len(state["api_calls"]) - before))

        # 8. no connection at all: a plain message, and the key is not in it
        closed = socket.socket()
        closed.bind(("127.0.0.1", 0))
        closed_port = closed.getsockname()[1]
        closed.close()
        API_BASE = "http://127.0.0.1:%d" % closed_port
        code, out, err = run(base_args + ["--out", os.path.join(workdir, "dead.csv")])
        API_BASE = "http://" + host
        check(code == 1 and "Could not reach Bunny" in err and key not in out + err, "no connection: a plain message, key not echoed")
    except Exception as exc:  # a crash in the script itself is a failed self-test, not a traceback
        import traceback
        failures.append("self-test crashed: " + traceback.format_exc()[-700:])
    finally:
        server.shutdown()
        API_BASE, CDN_SCHEME, _SELFTEST, _SELFTEST_HOST, SLEEP = saved

    if failures:
        for message in failures:
            print("FAIL: " + message, file=sys.stderr)
        print("selftest: %d checks, %d failed" % (total[0], len(failures)), file=sys.stderr)
        return 1
    say("selftest: %d checks passed" % total[0])
    return 0


if __name__ == "__main__":
    sys.exit(main())