Compare commits

..

No commits in common. "d688121d96215f13ad933be3b4fa5659bb1cefed" and "1cf0531b309059290885ef38bdafd94929a5e097" have entirely different histories.

8 changed files with 25 additions and 162 deletions

20
app.py
View file

@ -1,35 +1,35 @@
#!/usr/bin/env python #!/usr/bin/env python
if __name__ == "__main__":
import logging
logging.getLogger().setLevel("INFO")
from plugin import Plugin, load_plugins_for_config from plugin import Plugin, load_plugins_for_config
from typing import Tuple from typing import Tuple
from flask import Flask, request from flask import Flask, request
from pronoun_update import PronounUpdate from pronoun_update import PronounUpdate
import asyncio
app = Flask(__name__) app = Flask(__name__)
app.config.from_envvar("PRONOUNOMATIC_SETTINGS")
plugins = load_plugins_for_config(app.config.get("PLUGINS", [{"name": "log"}])) plugins = load_plugins_for_config(app.config.get("PLUGINS", [{"name": "log"}]))
def trigger_update_pronouns(pronouns: str) -> None: async def trigger_update_pronouns(pronouns: str) -> None:
update = PronounUpdate(pronouns) update = PronounUpdate(pronouns)
[plugin.update_bio(update) for plugin in plugins] await asyncio.gather(*[plugin.update_bio(update) for plugin in plugins])
@app.route("/update_pronouns", methods=["POST"]) @app.route("/update_pronouns", methods=["POST"])
def update_pronouns() -> Tuple[str, int]: async def update_pronouns() -> Tuple[str, int]:
pronouns = request.values.get("pronouns") pronouns = request.values.get("pronouns")
if not pronouns: if not pronouns:
return "Please specify a pronouns parameter as a part of your request", 422 return "Please specify a pronouns parameter as a part of your request", 422
trigger_update_pronouns(pronouns) await trigger_update_pronouns(pronouns)
return pronouns, 200 return pronouns, 200
if __name__ == "__main__": if __name__ == "__main__":
import logging
logging.getLogger().setLevel("INFO")
app.run(host="0.0.0.0", port=8080) app.run(host="0.0.0.0", port=8080)

View file

View file

@ -13,7 +13,7 @@
let let
pkgs = import nixpkgs { inherit system; }; pkgs = import nixpkgs { inherit system; };
pythonPackages = pkgs.python310Packages; pythonPackages = pkgs.python310Packages;
pythonDeps = with pythonPackages; [ mypy flask pytest pylint black types-setuptools requests types-requests ]; pythonDeps = with pythonPackages; [ mypy flask pytest pylint black types-setuptools ];
in in
{ {

View file

@ -1,28 +1,22 @@
from abc import abstractclassmethod, abstractmethod from abc import abstractclassmethod, abstractmethod
from functools import cache
from importlib.metadata import entry_points
import logging
from typing import Any, Callable, List
from pronoun_update import PronounUpdate from pronoun_update import PronounUpdate
from typing import Any, Callable, List
from importlib.metadata import entry_points
from functools import cache
class Plugin: 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 @classmethod
@abstractmethod @abstractmethod
def name(cls) -> str: def name(cls) -> str:
pass pass
@abstractmethod @abstractmethod
def update_bio(self, update_pronouns: PronounUpdate) -> None: def __init__(self, config: dict[str, str]):
pass
@abstractmethod
async def update_bio(self, update_pronouns: PronounUpdate) -> None:
pass pass

View file

@ -5,17 +5,17 @@ from plugin import Plugin
class LogPlugin(Plugin): class LogPlugin(Plugin):
"""A simple plugin that simply logs the pronoun update"""
log_level: int log_level: int
@classmethod @classmethod
def name(cls) -> str: def name(cls) -> str:
return "log" return "log"
def _load_from_config(self, config: dict[str, Any]) -> None: def __init__(self, config: dict[str, Any]):
log_level_str = config.get("level", "INFO") log_level_str = config.get("level", "INFO")
self.log_level = logging.getLevelName(log_level_str) self.log_level = logging.getLevelName(log_level_str)
def update_bio(self, pronoun_update: PronounUpdate) -> None: """ A simple plugin that simply logs the pronoun update """
async def update_bio(self, pronoun_update: PronounUpdate) -> None:
logging.log(self.log_level, f"Updating pronouns to: {pronoun_update.pronouns}") logging.log(self.log_level, f"Updating pronouns to: {pronoun_update.pronouns}")

View file

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

View file

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

View file

@ -8,12 +8,7 @@ setup(
# Modules to import from other scripts: # Modules to import from other scripts:
packages=find_packages(), packages=find_packages(),
package_dir={"": "./"}, package_dir={"": "./"},
entry_points={ entry_points={"pronounomatic.plugins": "test = plugins.log_plugin:LogPlugin"},
"pronounomatic.plugins": [
"log = plugins.log_plugin:LogPlugin",
"mastodon = plugins.mastodon:MastodonPlugin",
],
},
# Executables # Executables
scripts=["app.py"], scripts=["app.py"],
) )