mal.py

#!/usr/bin/env python3

from ctypes import c_char_p
from flask import Flask, Response, request
from multiprocessing import Lock, Manager, Process
from pprint import pprint
from requests import Session, get, post
from requests.adapters import HTTPAdapter
from requests.exceptions import ConnectionError, HTTPError
from secrets import token_urlsafe
from time import sleep
from webbrowser import open


class CallbackServer:
    def __init__(self):
        self.auth_code = Manager().Value(c_char_p, "")
        self.lock = Lock()
        self.lock.acquire()
        self.app = Flask(__name__)
        self.app.add_url_rule("/", "handle", self.handle)
        self.app.add_url_rule("/healthcheck", "healthcheck", self.healthcheck)
        self.server = Process(target=self.app.run, kwargs={"port": 8000})
        self.server.start()
        while True:
            try:
                get("http://localhost:8000/healthcheck")
            except ConnectionError:
                continue
            else:
                break

    def handle(self):
        self.auth_code.value = request.args.get("code")
        self.lock.release()
        return Response(status=200)

    def healthcheck(self):
        return Response(status=204)


class CallbackContext:
    def __enter__(self):
        self.cb = CallbackServer()
        return self

    @property
    def auth_code(self):
        return self.cb.auth_code.value

    def __exit__(self, exc_type, exc_value, exc_tb):
        self.cb.lock.acquire()
        self.cb.server.terminate()
        self.cb.server.join()
        self.cb.lock.release()


class Authorization:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
        self.refresh_token = None

    def is_authorized(self):
        return (
            get(
                "https://api.myanimelist.net/v2/users/@me",
                headers={"Authorization": f"Bearer {self.access_token}"},
            ).status_code
            == 200
        )

    def refresh(self):
        if not self.refresh_token:
            return self.token()
        req = post(
            "https://myanimelist.net/v1/oauth2/token",
            data={
                "client_id": f"{self.client_id}",
                "client_secret": f"{self.client_secret}",
                "grant_type": "refresh_token",
                "refresh_token": f"{self.refresh_token}",
            },
        )
        req.raise_for_status()
        self.access_token = req.json()["access_token"]
        self.refresh_token = req.json()["refresh_token"]
        return self.access_token

    def token(self):
        if self.access_token:
            return self.access_token
        nonce = token_urlsafe(100)[:128]
        with CallbackContext() as cb:
            open(
                f"https://myanimelist.net/v1/oauth2/authorize?response_type=code&client_id={self.client_id}&code_challenge={nonce}"
            )
        req = post(
            "https://myanimelist.net/v1/oauth2/token",
            data={
                "client_id": f"{self.client_id}",
                "client_secret": f"{self.client_secret}",
                "code": f"{cb.auth_code}",
                "code_verifier": f"{nonce}",
                "grant_type": "authorization_code",
            },
        )
        req.raise_for_status()
        self.access_token = req.json()["access_token"]
        self.refresh_token = req.json()["refresh_token"]
        return self.access_token


auth = Authorization(
    client_id="PUT YOUR CLIENT ID HERE",
    client_secret="PUT YOUR CLIENT SECRET HERE",
)


def dorefresh(func):
    global auth

    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except HTTPError:
            auth.refresh()
        return func(*args, **kwargs)

    return wrapper


@dorefresh
def mal_req(reqtype, path, **kwargs):
    global auth
    kwargs["headers"] = kwargs.get("headers", {}) | {
        "Authorization": f"Bearer {auth.token()}"
    }
    req = reqtype(f"https://api.myanimelist.net/v2{path}", timeout=5, **kwargs)
    if req.status_code == 401:
        req.raise_for_status()
    return req


session = Session()
session.mount("https://", HTTPAdapter(max_retries=5))
pprint(mal_req(session.get, "/users/@me").json())