diff --git a/app.py b/app.py index 2597115..4fd6802 100644 --- a/app.py +++ b/app.py @@ -1,35 +1,35 @@ #!/usr/bin/env python -from plugin import Plugin, load_plugins_for_config -from typing import Tuple -from flask import Flask, request -from pronoun_update import PronounUpdate - -import asyncio - -app = Flask(__name__) - -plugins = load_plugins_for_config(app.config.get("PLUGINS", [{"name": "log"}])) - - -async def trigger_update_pronouns(pronouns: str) -> None: - update = PronounUpdate(pronouns) - await asyncio.gather(*[plugin.update_bio(update) for plugin in plugins]) - - -@app.route("/update_pronouns", methods=["POST"]) -async def update_pronouns() -> Tuple[str, int]: - pronouns = request.values.get("pronouns") - if not pronouns: - return "Please specify a pronouns parameter as a part of your request", 422 - - await trigger_update_pronouns(pronouns) - return pronouns, 200 - - if __name__ == "__main__": import logging logging.getLogger().setLevel("INFO") +from plugin import Plugin, load_plugins_for_config +from typing import Tuple +from flask import Flask, request +from pronoun_update import PronounUpdate + +app = Flask(__name__) +app.config.from_envvar("PRONOUNOMATIC_SETTINGS") + +plugins = load_plugins_for_config(app.config.get("PLUGINS", [{"name": "log"}])) + + +def trigger_update_pronouns(pronouns: str) -> None: + update = PronounUpdate(pronouns) + [plugin.update_bio(update) for plugin in plugins] + + +@app.route("/update_pronouns", methods=["POST"]) +def update_pronouns() -> Tuple[str, int]: + pronouns = request.values.get("pronouns") + if not pronouns: + return "Please specify a pronouns parameter as a part of your request", 422 + + trigger_update_pronouns(pronouns) + return pronouns, 200 + + +if __name__ == "__main__": app.run(host="0.0.0.0", port=8080) diff --git a/config.py b/config.py new file mode 100644 index 0000000..e69de29 diff --git a/flake.nix b/flake.nix index 5e8a063..0a47820 100644 --- a/flake.nix +++ b/flake.nix @@ -13,7 +13,7 @@ let pkgs = import nixpkgs { inherit system; }; pythonPackages = pkgs.python310Packages; - pythonDeps = with pythonPackages; [ mypy flask pytest pylint black types-setuptools ]; + pythonDeps = with pythonPackages; [ mypy flask pytest pylint black types-setuptools requests types-requests ]; in { diff --git a/plugin.py b/plugin.py index b8adc35..3d0ea7e 100644 --- a/plugin.py +++ b/plugin.py @@ -1,22 +1,28 @@ from abc import abstractclassmethod, abstractmethod -from pronoun_update import PronounUpdate -from typing import Any, Callable, List -from importlib.metadata import entry_points from functools import cache +from importlib.metadata import entry_points +import logging +from typing import Any, Callable, List + +from pronoun_update import PronounUpdate class Plugin: + def __init__(self, config: dict[str, str]): + logging.info(f"Loading plugin {self.name()}") + self._load_from_config(config) + + @abstractmethod + def _load_from_config(self, config: dict[str, Any]) -> None: + pass + @classmethod @abstractmethod def name(cls) -> str: pass @abstractmethod - def __init__(self, config: dict[str, str]): - pass - - @abstractmethod - async def update_bio(self, update_pronouns: PronounUpdate) -> None: + def update_bio(self, update_pronouns: PronounUpdate) -> None: pass diff --git a/plugins/log_plugin.py b/plugins/log_plugin.py index e886f66..01345e6 100644 --- a/plugins/log_plugin.py +++ b/plugins/log_plugin.py @@ -5,17 +5,17 @@ from plugin import Plugin class LogPlugin(Plugin): + """A simple plugin that simply logs the pronoun update""" + log_level: int @classmethod def name(cls) -> str: return "log" - def __init__(self, config: dict[str, Any]): + def _load_from_config(self, config: dict[str, Any]) -> None: log_level_str = config.get("level", "INFO") self.log_level = logging.getLevelName(log_level_str) - """ A simple plugin that simply logs the pronoun update """ - - async def update_bio(self, pronoun_update: PronounUpdate) -> None: + def update_bio(self, pronoun_update: PronounUpdate) -> None: logging.log(self.log_level, f"Updating pronouns to: {pronoun_update.pronouns}") diff --git a/plugins/mastodon.py b/plugins/mastodon.py new file mode 100644 index 0000000..453582c --- /dev/null +++ b/plugins/mastodon.py @@ -0,0 +1,121 @@ +""" +Configuration: + Options: + host - the url of a mastodon api compatible server(for example: https://mastodon.example/) + token - Access token obtained by authenticating the user. Needs to have access to at least the read:accounts and write:accounts + + In order to obtain an access token you can either copy it from for example a browser cookie set by logging in with another client(not recommended, but simpler and easier) or registering an app and asking a client for authorization + + First register an application by doing + ``` + curl -X POST \ + -F 'client_name=Pronoun-o-matic' \ + -F 'redirect_uris=urn:ietf:wg:oauth:2.0:oob' \ + -F 'scopes=read:accounts write:accounts' \ + https://mastodon.example/api/v1/apps + ``` + That should return a json object with a client_id and a client_secret values. + + Open an url in your browser: + ``` + https://mastodon.example/oauth/authorize + ?client_id=CLIENT_ID + &scope=read:accounts+write:accounts + &redirect_uri=urn:ietf:wg:oauth:2.0:oob + &response_type=code + ``` + + You should recieve a client secret which you can use get the access token. Simply place them in a curl command like: + ``` + curl -X POST \ + -F 'client_id=your_client_id_here' \ + -F 'client_secret=your_client_secret_here' \ + -F 'redirect_uri=urn:ietf:wg:oauth:2.0:oob' \ + -F 'grant_type=client_credentials' \ + https://mastodon.example/oauth/token + ``` +""" + +from requests.models import Response +from pronoun_update import PronounUpdate +from typing import Any, Dict +from plugin import InvalidPluginDefinitionException, Plugin +import requests + + +class MastodonApiException(Exception): + def __init__(self, error: str) -> None: + super().__init__(f"Mastodon api returned an error: {error}") + + pass + + +class MastodonApiClient: + host: str + s = requests.Session() + + def __init__(self, host: str, token: str): + self.host = host + self.s.headers.update({"Authorization": f"Bearer {token}"}) + + def verify_credentials(self) -> Response: + creds = self.s.get(f"{self.host}/api/v1/accounts/verify_credentials") + if not creds.ok: + error = creds.json().get("error", "the server didn't return an error") + raise MastodonApiException(error) + return creds + + def update_credentials(self, updated_credentials: Dict[str, Any]) -> Response: + creds = self.s.patch( + f"{self.host}/api/v1/accounts/update_credentials", data=updated_credentials + ) + if not creds.ok: + error = creds.json().get("error", "the server didn't return an error") + raise InvalidPluginDefinitionException( + f"Provided token is invalid. {error}" + ) + return creds + + +class MastodonPlugin(Plugin): + s = requests.Session() + api_client: MastodonApiClient + + @classmethod + def name(cls) -> str: + return "mastodon" + + def _load_from_config(self, config: dict[str, Any]) -> None: + host = config.get("host", None) + if not isinstance(host, str): + raise InvalidPluginDefinitionException( + f"Host needs to be set to a string with the url to the mastodon compatible server, got {type(host)}" + ) + self.host = host + + access_token = config.get("token", None) + if not isinstance(access_token, str): + raise InvalidPluginDefinitionException( + f"Access token needs to be set to a string with an access token, got {type(host)}" + ) + self.access_token = access_token + + self.api_client = MastodonApiClient(host, access_token) + + # Verify the credentials + self.api_client.verify_credentials() + + def update_bio(self, update_pronouns: PronounUpdate) -> None: + credentials = self.api_client.verify_credentials().json()["source"] + new_credentials = {} + new_credentials["note"] = update_pronouns.replace_pronouns_in_bio( + credentials["note"] + ) + + for idx, field in enumerate(credentials["fields"]): + name = update_pronouns.replace_pronouns_in_bio(field["name"]) + value = update_pronouns.replace_pronouns_in_bio(field["value"]) + new_credentials[f"fields_attributes[{idx}][name]"] = name + new_credentials[f"fields_attributes[{idx}][value]"] = value + + self.api_client.update_credentials(new_credentials) diff --git a/pronoun_update.py b/pronoun_update.py index dca2a0b..acdf60e 100644 --- a/pronoun_update.py +++ b/pronoun_update.py @@ -1,6 +1,6 @@ import re -PRONOUN_REPLACE_REGEX = re.compile(r"\[S*\]") +PRONOUN_REPLACE_REGEX = re.compile(r"\[\S*\]") class PronounUpdate: @@ -10,4 +10,4 @@ class PronounUpdate: self.pronouns = pronouns def replace_pronouns_in_bio(self, bio: str) -> str: - return PRONOUN_REPLACE_REGEX.sub(bio, f"[{self.pronouns}]") + return PRONOUN_REPLACE_REGEX.sub(f"[{self.pronouns}]", bio) diff --git a/setup.py b/setup.py index 772d17f..df4f4ed 100644 --- a/setup.py +++ b/setup.py @@ -8,7 +8,12 @@ setup( # Modules to import from other scripts: packages=find_packages(), package_dir={"": "./"}, - entry_points={"pronounomatic.plugins": "test = plugins.log_plugin:LogPlugin"}, + entry_points={ + "pronounomatic.plugins": [ + "log = plugins.log_plugin:LogPlugin", + "mastodon = plugins.mastodon:MastodonPlugin", + ], + }, # Executables scripts=["app.py"], )