154 lines
4.4 KiB
Python
154 lines
4.4 KiB
Python
"""Wikipedia OAuth."""
|
|
|
|
import sys
|
|
import typing
|
|
import urllib
|
|
from typing import cast
|
|
|
|
import requests
|
|
from flask import current_app, session
|
|
from requests_oauthlib import OAuth1Session
|
|
|
|
from .api import ua, wikimedia_log_config
|
|
from .wikimedia_api_logging import logged_request
|
|
|
|
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], timeout: int = 4) -> 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"],
|
|
)
|
|
oauth.headers.update({"User-Agent": ua})
|
|
proxies = get_edit_proxy()
|
|
return logged_request(
|
|
oauth,
|
|
wikimedia_log_config,
|
|
"POST",
|
|
api_url,
|
|
data=params,
|
|
timeout=timeout,
|
|
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"],
|
|
)
|
|
oauth.headers.update({"User-Agent": ua})
|
|
proxies = get_edit_proxy()
|
|
return logged_request(
|
|
oauth,
|
|
wikimedia_log_config,
|
|
"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(f"API request failed: HTTP {r.status_code}", file=sys.stderr)
|
|
print(f"Response body: {r.text!r}", file=sys.stderr)
|
|
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_oauth_session() -> OAuth1Session | None:
|
|
"""Return an OAuth1Session for the current user, or None if not logged in."""
|
|
if "owner_key" not in session or "owner_secret" not in session:
|
|
return None
|
|
app = current_app
|
|
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"],
|
|
)
|
|
oauth.headers.update({"User-Agent": ua})
|
|
oauth.params = typing.cast(
|
|
dict[str, str | int],
|
|
{"format": "json", "action": "query", "formatversion": 2},
|
|
)
|
|
return oauth
|
|
|
|
|
|
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:
|
|
try:
|
|
reply = userinfo_call()
|
|
except Exception as e:
|
|
print(f"get_username failed, clearing session: {e}", file=sys.stderr)
|
|
session.pop("owner_key", None)
|
|
session.pop("owner_secret", None)
|
|
return None
|
|
if "query" not in reply:
|
|
return None
|
|
session["username"] = reply["query"]["userinfo"]["name"]
|
|
|
|
return cast(str, session["username"])
|