stern-api/collect_play_history.py

#!/usr/bin/env python3
"""Collect authenticated Stern Insider play-history snapshots as JSON Lines.

Uses the legacy JWT token endpoint by default. It never prints or writes the
password, access token, or refresh token.
"""

from __future__ import annotations

import argparse
import getpass
import json
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, urljoin, urlparse
from urllib.request import Request, urlopen

DEFAULT_ENDPOINTS = (
    "/api/v1/portal/my_activity/",
    "/api/v1/portal/game_machine_high_scores/",
    "/api/v1/portal/user_title_stats/",
    "/api/v1/portal/user_stats/",
)
USER_AGENT = "stern-play-history-collector/1.0"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--base-url", required=True, help="API origin, e.g. https://iop-dev.strndev.com"
    )
    parser.add_argument("--username", help="Account username; prompts if omitted")
    parser.add_argument(
        "--password", help="Account password; prompts securely if omitted"
    )
    parser.add_argument(
        "--token-path",
        default="/api/v2/token/",
        help="JWT login path (default: /api/v2/token/)",
    )
    parser.add_argument(
        "--endpoint",
        action="append",
        help="Authenticated GET endpoint to collect; repeat to override the defaults",
    )
    parser.add_argument(
        "--query",
        action="append",
        default=[],
        metavar="KEY=VALUE",
        help="Query parameter applied to every endpoint; repeatable",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("play-history.jsonl"),
        help="JSON Lines snapshot file",
    )
    parser.add_argument(
        "--interval",
        type=float,
        default=0,
        help="Seconds between collection cycles; 0 runs once",
    )
    parser.add_argument(
        "--allow-http",
        action="store_true",
        help="Permit an insecure HTTP API URL (not recommended)",
    )
    return parser.parse_args()


def normalized_base_url(raw_url: str, allow_http: bool) -> str:
    parsed = urlparse(raw_url)
    if parsed.scheme not in {"https", "http"} or not parsed.netloc:
        raise ValueError("--base-url must be an absolute http(s) URL")
    if parsed.scheme != "https" and not allow_http:
        raise ValueError("refusing HTTP; use HTTPS or pass --allow-http explicitly")
    return raw_url.rstrip("/") + "/"


def parse_query(values: list[str]) -> dict[str, str]:
    query: dict[str, str] = {}
    for value in values:
        key, separator, item = value.partition("=")
        if not separator or not key:
            raise ValueError(f"invalid --query {value!r}; use KEY=VALUE")
        query[key] = item
    return query


def request_json(
    url: str,
    method: str = "GET",
    payload: dict[str, Any] | None = None,
    access_token: str | None = None,
) -> tuple[int, Any]:
    body = json.dumps(payload).encode() if payload is not None else None
    headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
    if body is not None:
        headers["Content-Type"] = "application/json"
    if access_token:
        headers["Authorization"] = f"Bearer {access_token}"
    request = Request(url, data=body, headers=headers, method=method)
    try:
        with urlopen(request, timeout=30) as response:
            status, raw = response.status, response.read()
    except HTTPError as error:
        status, raw = error.code, error.read()
    except URLError as error:
        raise RuntimeError(f"network error for {url}: {error.reason}") from error

    try:
        return status, json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return status, {"non_json_response_bytes": len(raw)}


def authenticate(base_url: str, token_path: str, username: str, password: str) -> str:
    status, response = request_json(
        urljoin(base_url, token_path.lstrip("/")),
        method="POST",
        payload={"username": username, "password": password},
    )
    if (
        status != 200
        or not isinstance(response, dict)
        or not isinstance(response.get("access"), str)
    ):
        raise RuntimeError(f"login failed with HTTP {status}: {json.dumps(response)}")
    return response["access"]


def collect_once(
    base_url: str,
    endpoints: tuple[str, ...],
    query: dict[str, str],
    access_token: str,
    output: Path,
) -> None:
    retrieved_at = datetime.now(timezone.utc).isoformat()
    with output.open("a", encoding="utf-8") as destination:
        for endpoint in endpoints:
            url = urljoin(base_url, endpoint.lstrip("/"))
            if query:
                url += "?" + urlencode(query)
            status, data = request_json(url, access_token=access_token)
            snapshot = {
                "retrieved_at": retrieved_at,
                "endpoint": endpoint,
                "status": status,
                "data": data,
            }
            destination.write(
                json.dumps(snapshot, separators=(",", ":"), default=str) + "\n"
            )
            print(f"{endpoint}: HTTP {status}", file=sys.stderr)


def main() -> int:
    args = parse_args()
    try:
        base_url = normalized_base_url(args.base_url, args.allow_http)
        query = parse_query(args.query)
    except ValueError as error:
        print(f"error: {error}", file=sys.stderr)
        return 2

    username = args.username or input("Username: ")
    password = args.password or getpass.getpass("Password: ")
    endpoints = tuple(args.endpoint) if args.endpoint else DEFAULT_ENDPOINTS
    args.output.parent.mkdir(parents=True, exist_ok=True)

    try:
        access_token = authenticate(base_url, args.token_path, username, password)
        while True:
            collect_once(base_url, endpoints, query, access_token, args.output)
            if args.interval <= 0:
                return 0
            time.sleep(args.interval)
    except RuntimeError as error:
        print(f"error: {error}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())