#!/usr/bin/env python3 """A complete HedgeHog partner client: connect a user, send a match, read the analysis. Standard library only. The partner API documentation (docs/PARTNER_API.md in https://gitlab.com/eranlambooij/hedgehog-public) explains every step; the section numbers below point into it. python3 hedgehog_partner_client.py --client-id https://example.com/oauth/client.json match.mat The first run opens the browser to connect the user; later runs reuse the tokens saved in --tokens (keep them in the OS keychain in a real application). The match may be a .mat file, which HedgeHog converts, or OGXM JSON. Your metadata document (section 3) must list "http://127.0.0.1/callback" in redirect_uris; the port is chosen at run time and is not matched (section 3.2). """ import argparse import base64 import hashlib import http.server import json import secrets import sys import time import urllib.error import urllib.parse import urllib.request import webbrowser BASE = "https://hedgehog-bg.com" class HedgehogError(Exception): """A refusal: `code` is stable API, `message` is written for your user (section 11).""" def __init__(self, status: int, body: dict): self.status, self.body = status, body self.code = body.get("error") super().__init__(body.get("message") or body.get("error_description") or self.code) class HedgehogClient: def __init__(self, client_id: str, token_file: str, base: str = BASE): self.client_id, self.token_file, self.base = client_id, token_file, base self.open_browser = webbrowser.open self.poll_seconds = 10 try: with open(token_file, encoding="utf-8") as f: self.tokens = json.load(f) except (OSError, ValueError): self.tokens = {} # --- HTTP --------------------------------------------------------------- def send(self, method: str, path: str, headers: dict, data: bytes | None) -> tuple[int, bytes]: """One HTTP exchange. Replace this to route the client through anything else.""" request = urllib.request.Request(self.base + path, data=data, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=60) as response: return response.status, response.read() except urllib.error.HTTPError as err: return err.code, err.read() def call(self, method: str, path: str, *, json_body=None, raw: bytes | None = None, form: dict | None = None, retry: bool = True) -> dict: headers, data = {}, None if form is not None: headers["Content-Type"] = "application/x-www-form-urlencoded" data = urllib.parse.urlencode(form).encode() else: headers["Authorization"] = "Bearer " + self.tokens.get("access_token", "") if json_body is not None: headers["Content-Type"], data = "application/json", json.dumps(json_body).encode() elif raw is not None: headers["Content-Type"], data = "application/octet-stream", raw status, body = self.send(method, path, headers, data) answer = json.loads(body or b"{}") # An expired access token: refresh once and repeat (section 6.1). if status == 401 and answer.get("error") == "invalid_token" and retry and form is None: self.refresh() return self.call(method, path, json_body=json_body, raw=raw, retry=False) if status >= 400: raise HedgehogError(status, answer) return answer # --- Connecting (sections 5 and 6) --------------------------------------- def token(self, **form) -> None: self.tokens = self.call("POST", "/api/v1/oauth/token", form=dict(form, client_id=self.client_id)) with open(self.token_file, "w", encoding="utf-8") as f: json.dump(self.tokens, f) def refresh(self) -> None: """Rotate the pair. Never run two refreshes at once: the second revokes the grant.""" self.token(grant_type="refresh_token", refresh_token=self.tokens["refresh_token"]) def connect(self) -> None: verifier = secrets.token_urlsafe(64) challenge = base64.urlsafe_b64encode( hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() state, reply = secrets.token_urlsafe(16), {} class Callback(http.server.BaseHTTPRequestHandler): def do_GET(self): reply.update(urllib.parse.parse_qsl(urllib.parse.urlsplit(self.path).query)) self.send_response(200) self.end_headers() self.wfile.write(b"Connected to HedgeHog. You can close this tab.") def log_message(self, *args): pass server = http.server.HTTPServer(("127.0.0.1", 0), Callback) redirect_uri = f"http://127.0.0.1:{server.server_port}/callback" self.open_browser(self.base + "/oauth/authorize?" + urllib.parse.urlencode({ "response_type": "code", "client_id": self.client_id, "redirect_uri": redirect_uri, "scope": "analyze", "state": state, "code_challenge": challenge, "code_challenge_method": "S256"})) while "code" not in reply and "error" not in reply: server.handle_request() server.server_close() # Check the state (it is your request) and the issuer (it is HedgeHog), section 5.2. if reply.get("state") != state or reply.get("iss") != self.base: raise HedgehogError(400, {"error": "invalid_request", "message": "The reply was not for this request."}) if "error" in reply: raise HedgehogError(400, reply) self.token(grant_type="authorization_code", code=reply["code"], redirect_uri=redirect_uri, code_verifier=verifier) # --- Work (sections 8 and 9) --------------------------------------------- def me(self) -> dict: return self.call("GET", "/api/v1/oauth/me") def load_match(self, path: str) -> dict: """OGXM JSON from a .mat (converted by HedgeHog, stored nowhere) or a .json file.""" with open(path, "rb") as f: content = f.read() if path.lower().endswith(".mat"): converted = self.call("POST", "/api/v1/matches/import/mat", raw=content) for note in converted["report"]["warnings"]: print("note:", note) return converted["data"] return json.loads(content) def save(self, match: dict, color: str | None = None) -> str: if color: match = dict(match, playerColor=color) return self.call("POST", "/api/v1/matches/save", json_body=match)["match_id"] def analyze(self, match_id: str, preset: str) -> str: job = self.call("POST", "/api/v1/analyze", json_body={"match_id": match_id, "preset": preset, "notify": False}) return job["analysis_id"] def wait(self, analysis_id: str) -> dict: while True: analysis = self.call("GET", f"/api/v1/analysis/{analysis_id}") if analysis["status"] == "completed": return analysis if analysis["status"] in ("failed", "cancelled"): raise HedgehogError(200, analysis.get("error") or {"message": analysis["status"]}) time.sleep(self.poll_seconds) def best_preset(me: dict, ceiling: str = "2ply") -> str: """The deepest rung up to `ceiling` the user's plan allows (section 8).""" allowed = me["presets"]["match"] ladder = ["1ply", "2ply", "3ply"] usable = [p for p in ladder[:ladder.index(ceiling) + 1] if p in allowed] return usable[-1] if usable else allowed[0] def main(argv=None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("match", help="a .mat file or OGXM JSON") parser.add_argument("--client-id", required=True, help="the URL of your metadata document") parser.add_argument("--tokens", default="hedgehog_tokens.json") parser.add_argument("--color", choices=["white", "black"], help="the side your user played") parser.add_argument("--base", default=BASE) args = parser.parse_args(argv) client = HedgehogClient(args.client_id, args.tokens, args.base) try: if not client.tokens: client.connect() me = client.me() allowance = (me.get("allowance") or {}).get("match") if allowance: left = allowance["limit"] - allowance["used"] + allowance["credits"] print(f"{me['username']} has {left} free match analysis left today.") match_id = client.save(client.load_match(args.match), args.color) analysis = client.wait(client.analyze(match_id, best_preset(me))) except HedgehogError as err: print(err, file=sys.stderr) # written for the user: show it as is return 1 print(json.dumps({key: analysis[key] for key in ("preset", "er_overall", "er_player", "total_decisions", "total_blunders")})) return 0 if __name__ == "__main__": sys.exit(main())