add-links/add_links/wikidata_oauth.py

"""Wikipedia OAuth."""

import typing
import urllib
from typing import cast

import requests
from flask import current_app, session
from requests_oauthlib import OAuth1Session

wiki_hostname = "en.wikipedia.org"
api_url = f"https://{wiki_hostname}/w/api.php"


class LoginNeeded(Exception):
    """Not logged in."""

    pass


def get_edit_proxy() -> dict[str, str]:
    """Retrieve proxy information from config."""
    edit_proxy = current_app.config.get("EDIT_PROXY")
    if edit_proxy:
        return {"http": edit_proxy, "https": edit_proxy}
    else:
        return {}


def api_post_request(params: dict[str, str | int]) -> requests.Response:
    """HTTP Post using Oauth."""
    app = current_app
    # url = "https://www.wikidata.org/w/api.php"
    client_key = app.config["CLIENT_KEY"]
    client_secret = app.config["CLIENT_SECRET"]
    oauth = OAuth1Session(
        client_key,
        client_secret=client_secret,
        resource_owner_key=session["owner_key"],
        resource_owner_secret=session["owner_secret"],
    )
    proxies = get_edit_proxy()
    return oauth.post(api_url, data=params, timeout=4, proxies=proxies)


def raw_request(params: typing.Mapping[str, str | int]) -> requests.Response:
    """Low-level API request."""
    app = current_app
    # url = "https://www.wikidata.org/w/api.php?" + urlencode(params)
    client_key = app.config["CLIENT_KEY"]
    client_secret = app.config["CLIENT_SECRET"]
    if "owner_key" not in session or "owner_secret" not in session:
        raise LoginNeeded
    oauth = OAuth1Session(
        client_key,
        client_secret=client_secret,
        resource_owner_key=session["owner_key"],
        resource_owner_secret=session["owner_secret"],
    )
    proxies = get_edit_proxy()
    return oauth.get(
        api_url + "?" + urllib.parse.urlencode(params), timeout=4, proxies=proxies
    )


def api_request(params: typing.Mapping[str, str | int]) -> dict[str, typing.Any]:
    """Make an API request with OAuth."""
    r = raw_request(params)
    try:
        return cast(dict[str, typing.Any], r.json())
    except Exception:
        print("text")
        print(r.text)
        print("---")
        raise


def get_token() -> str:
    """Get CSRF tokebn from MediaWiki API."""
    params: dict[str, str | int] = {
        "action": "query",
        "meta": "tokens",
        "format": "json",
        "formatversion": 2,
    }
    reply = api_request(params)
    token: str = reply["query"]["tokens"]["csrftoken"]

    return token


def userinfo_call() -> typing.Mapping[str, typing.Any]:
    """Request user information via OAuth."""
    params = {"action": "query", "meta": "userinfo", "format": "json"}
    return api_request(params)


def get_username() -> None | str:
    """Get the username or None if not logged in."""
    if "owner_key" not in session:
        return None  # not authorized

    if "username" not in session:
        reply = userinfo_call()
        if "query" not in reply:
            return None
        session["username"] = reply["query"]["userinfo"]["name"]

    return cast(str, session["username"])