Rename wikidata_oauth to mediawiki_oauth

add_links/mediawiki_oauth.py

@@ -0,0 +1,109 @@
+"""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"])