climperor/web/cloudflare/_probe_account_d1.py

"""Deep-probe D1 + OpenDota for one account (no secret echo)."""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request

ACCOUNT = "510534f7f6284344aadaf2f5a0794d48"
DB = "9eeb24ba-acc5-4520-b4e7-754ea776394e"
AID = int(os.environ.get("CLIMPEROR_SYNC_ACCOUNT_ID", "143712136"))


def _creds() -> tuple[str, str]:
    email = os.environ.get("CLOUDFLARE_EMAIL") or os.environ.get(
        "KEYZOO_ASSET_META_USERNAME"
    )
    key = os.environ.get("CLOUDFLARE_API_KEY") or os.environ.get(
        "KEYZOO_ASSET_SECRET_GLOBAL_API_KEY"
    )
    if not email or not key:
        raise SystemExit("missing Cloudflare credentials")
    return email, key


def d1(sql: str) -> list:
    email, key = _creds()
    body = json.dumps({"sql": sql}).encode()
    req = urllib.request.Request(
        f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT}/d1/database/{DB}/query",
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "X-Auth-Email": email,
            "X-Auth-Key": key,
        },
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        out = json.loads(r.read().decode())
    results = out.get("result") or []
    if results:
        return results[0].get("results") or []
    return []


def od(path: str):
    req = urllib.request.Request(
        f"https://api.opendota.com/api{path}",
        headers={"User-Agent": "climperor-probe"},
    )
    try:
        with urllib.request.urlopen(req, timeout=25) as r:
            return json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        return {"_error": e.code, "_body": e.read()[:200].decode("utf-8", "replace")}


def main() -> int:
    print("users", json.dumps(d1(f"SELECT * FROM users WHERE account_id={AID}"), ensure_ascii=False))
    print(
        "stats",
        json.dumps(
            d1(
                f"SELECT scope, sample, wins, losses, winrate, length(payload_json) AS plen "
                f"FROM player_stats WHERE account_id={AID}"
            ),
            ensure_ascii=False,
        ),
    )
    print(
        "matches",
        json.dumps(
            d1(f"SELECT COUNT(*) AS n FROM player_matches WHERE account_id={AID}"),
            ensure_ascii=False,
        ),
    )
    print(
        "heroes",
        json.dumps(
            d1(f"SELECT COUNT(*) AS n FROM player_heroes WHERE account_id={AID}"),
            ensure_ascii=False,
        ),
    )
    print(
        "peers",
        json.dumps(
            d1(f"SELECT COUNT(*) AS n FROM player_peers WHERE account_id={AID}"),
            ensure_ascii=False,
        ),
    )
    print(
        "profile",
        json.dumps(
            d1(
                f"SELECT account_id, rank_tier, leaderboard_rank, "
                f"availability_status, availability_note, "
                f"availability_complete, source, fetched_at, enriched_at "
                f"FROM player_profiles WHERE account_id={AID}"
            ),
            ensure_ascii=False,
        ),
    )
    print(
        "stats_all",
        json.dumps(
            d1(
                f"SELECT scope, sample, wins, losses, kda, avg_gpm "
                f"FROM player_stats WHERE account_id={AID} ORDER BY scope"
            ),
            ensure_ascii=False,
        ),
    )

    player = od(f"/players/{AID}")
    if isinstance(player, dict) and "profile" in player:
        p = player.get("profile") or {}
        print(
            "OD player",
            json.dumps(
                {
                    "personaname": p.get("personaname"),
                    "rank_tier": player.get("rank_tier"),
                    "fh_unavailable": player.get("fh_unavailable"),
                },
                ensure_ascii=False,
            ),
        )
    else:
        print("OD player", player)

    wl = od(f"/players/{AID}/wl")
    print("OD wl", wl)
    recent = od(f"/players/{AID}/recentMatches")
    if isinstance(recent, list):
        print("OD recentMatches", len(recent))
        if recent:
            print("OD recent[0].match_id", recent[0].get("match_id"))
    else:
        print("OD recentMatches", recent)
    return 0


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