Mastodon plugin

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)

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
       {

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
 
 

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}")

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)

pronoun_update.py

@@ -1,3 +1,8 @@
+import re
+
+PRONOUN_REPLACE_REGEX = re.compile(r"\[\S*\]")
+
+
 class PronounUpdate:
     pronouns: str
 
@@ -5,4 +10,4 @@ class PronounUpdate:
         self.pronouns = pronouns
 
     def replace_pronouns_in_bio(self, bio: str) -> str:
-        return self.pronouns
+        return PRONOUN_REPLACE_REGEX.sub(f"[{self.pronouns}]", bio)

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"],
 )