Nixos module: initial commit

flake.nix

@@ -9,15 +9,16 @@
   };
 
   outputs = { self, nixpkgs, flake-utils }:
-    flake-utils.lib.eachDefaultSystem (system:
+    flake-utils.lib.eachDefaultSystem
+      (system:
         let
           pkgs = import nixpkgs { inherit system; };
           pythonPackages = pkgs.python310Packages;
-        pythonDeps = with pythonPackages; [ mypy flask pytest pylint black types-setuptools requests types-requests ];
+          pythonDeps = with pythonPackages; [ gunicorn mypy flask pytest pylint black types-setuptools requests types-requests ];
 
         in
         {
-        packages.pronoun-o-matic = pythonPackages.buildPythonApplication {
+          packages.pronoun-o-matic = pythonPackages.buildPythonPackage {
             pname = "pronoun-o-matic";
             version = "1.0";
             propagatedBuildInputs = pythonDeps;
@@ -27,5 +28,65 @@
             buildInputs = [ pythonPackages.python ] ++ pythonDeps;
           };
           defaultPackage = self.packages.${system}.pronoun-o-matic;
-      });
+        }) // {
+      nixosModule = { config, lib, pkgs, options, ... }:
+        with lib;
+        let
+          system = pkgs.system;
+        in
+        let
+          pkgs = nixpkgs.${system};
+          cfg = config.services.pronoun-o-matic;
+          pronoun-o-matic = self.packages.${system}.pronoun-o-matic;
+          python-pkgs = pronoun-o-matic.pythonModule.pkgs;
+          python-path = python-pkgs.makePythonPath [ pronoun-o-matic ];
+          gunicorn = python-pkgs.gunicorn;
+        in
+        {
+          options.services.pronoun-o-matic = {
+            enable = mkEnableOption "Enables the pronoun-o-matic HTTP service";
+            workers = mkOption {
+              type = types.int;
+              default = 1;
+              example = 4;
+              description = ''
+                The number of gunicorn worker processes for handling requests.
+              '';
+            };
+            package = mkOption {
+              type = types.package;
+              description = "Package to use for the server.";
+              default = pronoun-o-matic;
+            };
+            bind = mkOption {
+              type = types.str;
+              description = "The address and port gunicorn should bind to";
+              default = "127.0.0.1:8080";
+            };
+            plugins = mkOption {
+              type = types.listOf types.attrs;
+              default = [{ name = "log"; }];
+            };
+          };
+
+          config = mkIf cfg.enable {
+            systemd.services.pronoun-o-matic = {
+              wantedBy = [ "multi-user.target" ];
+              environment = {
+                PYTHONPATH = python-path;
+                PRONOUNOMATIC_PLUGINS = builtins.toJSON cfg.plugins;
+              };
+              serviceConfig = {
+                ExecStart = ''
+                  ${gunicorn}/bin/gunicorn pronounomatic.core.app:app \
+                    --name pronoun-o-matic \
+                    --workers ${toString cfg.workers} \
+                    --bind ${cfg.bind}
+                '';
+                Restart = "on-failure";
+              };
+            };
+          };
+        };
+    };
 }

setup.py

@@ -12,8 +12,8 @@ setup(
         "console_scripts": ["pronounomatic=pronounomatic.core.app:main"],
         "pronounomatic.plugins": [
             "log = pronounomatic.plugins.log_plugin:LogPlugin",
-#            "mastodon = pronounomatic.plugins.mastodon:MastodonPlugin",
+            "mastodon = pronounomatic.plugins.mastodon:MastodonPlugin",
-#            "discord = pronounomatic.plugins.discord:DiscordPlugin",
+            "discord = pronounomatic.plugins.discord:DiscordPlugin",
         ],
     },
 )

src/core/app.py

@@ -4,12 +4,13 @@ import logging
 logging.getLogger().setLevel("INFO")
 
 from typing import Tuple
+import json
 from flask import Flask, request
 from .plugin import Plugin, load_plugins_for_config
 from .pronoun_update import PronounUpdate
 
 app = Flask(__name__)
-#app.config.from_envvar("PRONOUNOMATIC_SETTINGS")
+app.config.from_prefixed_env("PRONOUNOMATIC", loads=json.loads)
 
 plugins = load_plugins_for_config(app.config.get("PLUGINS", [{"name": "log"}]))
 

src/plugins/discord.py

@@ -0,0 +1,66 @@
+"""
+Configuration:
+    Options:
+        token - Discord access token
+"""
+
+from requests.models import Response
+from ..core.pronoun_update import PronounUpdate
+from typing import Any, Dict
+from ..core.plugin import InvalidPluginDefinitionException, Plugin
+import requests
+
+
+class DiscordApiException(Exception):
+    def __init__(self, error: str) -> None:
+        super().__init__(f"Discord api returned an error: {error}")
+
+    pass
+
+
+class DiscordApiClient:
+    host: str
+    s = requests.Session()
+
+    def __init__(self, token: str):
+        self.s.headers.update({"Authorization": token})
+
+    def verify_credentials(self) -> Response:
+        user_data = self.s.get("https://discord.com/api/v9/users/@me")
+        if not user_data.ok:
+            raise DiscordApiException(user_data.json()["message"])
+        print(user_data.json())
+        return user_data 
+
+    def updated_data(self, updated_data: Dict[str, Any]) -> Response:
+        resp = self.s.patch(
+            "https://discord.com/api/v9/users/@me", json=updated_data, 
+        )
+        if not resp.ok:
+            raise DiscordApiException(resp.json()["message"])
+        return resp
+
+
+class DiscordPlugin(Plugin):
+    s = requests.Session()
+    api_client:DiscordApiClient 
+
+    @classmethod
+    def name(cls) -> str:
+        return "discord"
+
+    def _load_from_config(self, config: dict[str, Any]) -> None:
+        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(access_token)}"
+            )
+        self.api_client = DiscordApiClient(access_token)
+
+        self.api_client.verify_credentials()
+
+    def update_bio(self, update_pronouns: PronounUpdate) -> None:
+        current_bio = self.api_client.verify_credentials().json()["bio"]
+        bio = update_pronouns.replace_pronouns_in_bio(current_bio)
+        # TODO: discord seems to be getting a pronoun field, maybe use that
+        self.api_client.updated_data({"bio": bio})

src/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 ..core.pronoun_update import PronounUpdate
+from ..core.plugin import InvalidPluginDefinitionException, Plugin
+from typing import Any, Dict
+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)