gateway-gamers.net/sites/40k-rankings/statcheck-elo-scraper/download_and_parse.py

#!/usr/bin/env python3
import csv
import html
import json
import re
import sys
import time
import zipfile
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from xml.etree.ElementTree import iterparse


NS = {
    "main": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
    "rel": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
    "pkgrel": "http://schemas.openxmlformats.org/package/2006/relationships",
}

# Seconds to wait before each retry of a transient failure.
RETRY_DELAYS = (5, 15, 45)


def fetch(req: Request) -> bytes:
    """Read a URL, retrying 429/5xx responses and network errors."""
    for attempt, delay in enumerate((*RETRY_DELAYS, None), start=1):
        try:
            with urlopen(req, timeout=60) as resp:
                return resp.read()
        except HTTPError as e:
            if delay is None or not (e.code == 429 or e.code >= 500):
                raise
            reason = f"HTTP {e.code}"
        except (URLError, TimeoutError, ConnectionError) as e:
            if delay is None:
                raise
            reason = str(e)
        print(f"  {req.full_url.split('?')[0]}: {reason}; retry {attempt} in {delay}s", file=sys.stderr)
        time.sleep(delay)


def metadata_download_url(capture_path: Path) -> str:
    capture = json.loads(capture_path.read_text())
    for body in capture["documentBodies"]:
        payload = body.get("body", {})
        text = payload.get("body", "") if isinstance(payload, dict) else ""
        if '"@content.downloadUrl"' in text:
            return json.loads(text)["@content.downloadUrl"]
    raise RuntimeError("No @content.downloadUrl found in capture")


def download(url: str, output: Path) -> None:
    req = Request(url, headers={"User-Agent": "Mozilla/5.0"})
    output.write_bytes(fetch(req))


def shared_strings(zf: zipfile.ZipFile) -> list[str]:
    if "xl/sharedStrings.xml" not in zf.namelist():
        return []
    values = []
    with zf.open("xl/sharedStrings.xml") as f:
        for event, elem in iterparse(f, events=("end",)):
            if elem.tag == f"{{{NS['main']}}}si":
                parts = []
                for t in elem.iter(f"{{{NS['main']}}}t"):
                    parts.append(t.text or "")
                values.append("".join(parts))
                elem.clear()
    return values


def first_sheet_path(zf: zipfile.ZipFile) -> str:
    workbook = zf.read("xl/workbook.xml")
    first_sheet = re.search(rb'<sheet[^>]+r:id="([^"]+)"', workbook)
    if not first_sheet:
        return "xl/worksheets/sheet1.xml"
    rel_id = first_sheet.group(1).decode()
    rels = zf.read("xl/_rels/workbook.xml.rels")
    pattern = rb'<Relationship[^>]+Id="' + re.escape(rel_id.encode()) + rb'"[^>]+Target="([^"]+)"'
    target = re.search(pattern, rels)
    if not target:
        return "xl/worksheets/sheet1.xml"
    path = html.unescape(target.group(1).decode())
    if path.startswith("/"):
        return path.lstrip("/")
    return "xl/" + path


def column_number(cell_ref: str) -> int:
    letters = re.match(r"[A-Z]+", cell_ref).group(0)
    n = 0
    for ch in letters:
        n = n * 26 + ord(ch) - ord("A") + 1
    return n


def xlsx_to_csv(xlsx_path: Path, csv_path: Path) -> None:
    with zipfile.ZipFile(xlsx_path) as zf, csv_path.open("w", newline="", encoding="utf-8") as out:
        strings = shared_strings(zf)
        sheet = first_sheet_path(zf)
        writer = csv.writer(out)
        row_values = []
        current_row = None
        current_cell = None
        current_type = None
        value = None
        inline_parts = []

        with zf.open(sheet) as f:
            for event, elem in iterparse(f, events=("start", "end")):
                tag = elem.tag
                if event == "start" and tag == f"{{{NS['main']}}}row":
                    current_row = int(elem.attrib.get("r", "0"))
                    row_values = []
                elif event == "start" and tag == f"{{{NS['main']}}}c":
                    current_cell = elem.attrib.get("r", "")
                    current_type = elem.attrib.get("t")
                    value = None
                    inline_parts = []
                elif event == "end" and tag == f"{{{NS['main']}}}v":
                    value = elem.text or ""
                elif event == "end" and tag == f"{{{NS['main']}}}t" and current_type == "inlineStr":
                    inline_parts.append(elem.text or "")
                elif event == "end" and tag == f"{{{NS['main']}}}c":
                    col = column_number(current_cell)
                    while len(row_values) < col - 1:
                        row_values.append("")
                    if current_type == "s" and value not in (None, ""):
                        row_values.append(strings[int(value)])
                    elif current_type == "inlineStr":
                        row_values.append("".join(inline_parts))
                    else:
                        row_values.append(value or "")
                    elem.clear()
                elif event == "end" and tag == f"{{{NS['main']}}}row":
                    writer.writerow(row_values)
                    elem.clear()


def main() -> None:
    capture_path = Path(sys.argv[1] if len(sys.argv) > 1 else "iframe_capture2.json")
    xlsx_path = Path(sys.argv[2] if len(sys.argv) > 2 else "elo.xlsx")
    csv_path = Path(sys.argv[3] if len(sys.argv) > 3 else "elo.csv")
    url = metadata_download_url(capture_path)
    download(url, xlsx_path)
    xlsx_to_csv(xlsx_path, csv_path)
    print(f"downloaded={xlsx_path.stat().st_size} bytes csv={csv_path.stat().st_size} bytes")


if __name__ == "__main__":
    main()