pronoun-o-matic/plugins/mastodon.py

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