initial commit

.gitignore

@@ -0,0 +1,8 @@
+# Created by venv; see https://docs.python.org/3/library/venv.html
+*
+!mastolynx.py
+!mastolynx.service
+!mastolynx.env.example
+!requirements.txt
+!.gitignore
+!README.md

README.md

@@ -0,0 +1,5 @@
+# MastoLynx
+
+Mastodon webhook for [Travelynx](https://travelynx.de)
+
+See <mastolynx.service> and <mastolynx.env.example> for usage information.

mastolynx.env.example

@@ -0,0 +1,7 @@
+MASTOLYNX_DEBUG=0
+MASTOLYNX_INSTANCE_URL=https://bark.lgbt
+MASTOLYNX_CLIENT_NAME=MastoLynx
+MASTOLYNX_STATUS_LANGUAGE=en
+MASTOLYNX_APPID_FILENAME=./mastolynx.appid
+MASTOLYNX_TOKEN_FILENAME=./mastolynx.token
+MASTOLYNX_AUTH_TOKEN=YOUR_AUTH_TOKEN_HERE__YOUR_AUTH_TOKEN_HERE__YOUR_AUTH_TOKEN_HERE

mastolynx.py

@@ -0,0 +1,169 @@
+import os
+from pathlib import Path
+from typing import Any, cast
+
+from beeprint import pp
+
+# from urllib.parse import urlparse
+from mastodon import Mastodon
+from flask import Flask, jsonify, request
+
+ENV_PREFIX = "MASTOLYNX_"
+CLIENT_SCOPES = ["profile", "write:statuses"]
+
+
+class Configuration:
+    __env_prefix: str = ENV_PREFIX
+    client_name: str = "MastoLynx (Development)"
+    user_agent: str = "MastoLynx/0.1"
+    appid_filename: str = "./mastolynx.appid"
+    token_filename: str = "./mastolynx.token"
+    instance_url: str = None  # type: ignore
+    auth_token: str = None  # type: ignore
+    status_language: str = "en"  # type: ignore
+    debug: str = ""
+
+    def __init__(self, env_prefix: str = ENV_PREFIX) -> None:
+        self.__env_prefix = env_prefix
+        for key, default in vars(Configuration).items():
+            if key.startswith("__"):
+                continue
+            try:
+                setattr(self, key, os.environ[self.__env_prefix + key.upper()])
+            except KeyError:
+                if default is None:
+                    raise KeyError(f"No default for '{key}'")
+
+        if self.debug not in ["", "0"]:
+            pp(self)
+
+
+def masto_login() -> Mastodon:
+    print("Signing into Mastodon...")
+
+    token_path = Path(config.token_filename)
+    if not token_path.exists():
+        os.umask(int(0o077))
+        mastodon = masto_register()
+    else:
+        mastodon = Mastodon(access_token=token_path)
+
+    print(
+        "Signed in to Mastodon: @{user}@{instance}".format(
+            user=mastodon.me().username, instance=mastodon.instance().domain
+        )
+    )
+
+    return mastodon
+
+
+def masto_register() -> Mastodon:
+    print("Registering new client in Mastodon")
+
+    client_name = config.client_name
+    appid_path = Path(config.appid_filename)
+    if not appid_path.exists():
+        print("Creating new Mastodon Client ID")
+        os.umask(int(0o077))
+        Mastodon.create_app(
+            client_name=client_name,
+            scopes=CLIENT_SCOPES,
+            api_base_url=config.instance_url,
+            to_file=appid_path,
+            user_agent=config.user_agent,
+        )
+
+    mastodon = Mastodon(client_id=appid_path)
+    url = mastodon.auth_request_url(scopes=CLIENT_SCOPES)
+    print(
+        f"Open '{url}' in your browser, authorise '{client_name}', "
+        "and paste the code you get back in here"
+    )
+    if "XDG_CURRENT_DESKTOP" in os.environ:
+        os.system(f"xdg-open '{url}'")
+
+    code = input("> ")
+    token_path = Path(config.token_filename)
+    mastodon.log_in(code=code, to_file=token_path, scopes=CLIENT_SCOPES)
+
+    return mastodon
+
+
+config = Configuration()
+mastodon = masto_login()
+
+
+app = Flask(__name__)
+
+
+def format_status(train: str, destination: str) -> str:
+    hashtags_etc = "\n\n#travelynx"
+    match config.status_language:
+        case "de":
+            status_text = f"Ich bin gerade in {train} nach {destination}!"
+        case _:
+            status_text = f"I'm currently on {train} to {destination}!"
+
+    return status_text + hashtags_etc
+
+
+@app.route("/", methods=["POST"])
+def checkin():
+    # uri = urlparse(os.environ["REQUEST_URI"])
+    # uri_path = Path(uri.path)
+    # match uri_path.parts[0]:
+    #     case _:
+    #         pass
+
+    # sanity checks
+    if not (req_token := request.headers.get("Authorization", "")).startswith(
+        "Bearer "
+    ):
+        return "Missing bearer token", 403
+    elif req_token.removeprefix("Bearer ") != config.auth_token:
+        return "Invalid bearer token", 403
+
+    webhook_data: dict[str, Any] = cast(dict[str, str], request.json)
+    if config.debug not in ["", "0"]:
+        pp(webhook_data)
+
+    try:
+        reason: str = webhook_data["reason"]
+        status: dict[str, Any] = webhook_data["status"]
+
+        train: str = status["train"]["type"].strip() + " " + status["train"]["no"]
+        destination: str = status["toStation"]["name"]
+
+        status_text = format_status(train, destination)
+
+        if reason == "ping":
+            print("Received Ping from Travelynx with following status:")
+            print("-" * 40)
+            print(status_text)
+            print("-" * 40)
+            return jsonify("pong!")
+
+        elif reason == "update":
+            status_visibility = "direct"
+            match status["visibility"]["desc"]:
+                case "followers":
+                    status_visibility = "private"
+                case "travelynx":
+                    status_visibility = "unlisted"
+                case "public":
+                    status_visibility = "public"
+
+            print("Posting check-in with visibility '{status_visibility}:")
+            mastodon.status_post(
+                status=status_text,
+                visibility=status_visibility,
+                language=config.status_language,
+            )
+
+            return jsonify("ok!")
+
+    except KeyError:
+        return "Malformed request body", 400
+
+    # return jsonify(checkin_data)
+    return jsonify(webhook_data)

mastolynx.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=MastoLynx (Travelynx Mastodon Integration)
+After=network-online.target
+
+[Service]
+Type=simple
+User=mastolynx
+Environment=VIRTUAL_ENV=/opt/mastolynx
+Environment=PYTHONUNBUFFERED=TRUE
+EnvironmentFile=/opt/mastolynx/mastolynx.env
+WorkingDirectory=/opt/mastolynx
+ExecStart=/opt/mastolynx/bin/gunicorn -w 1 mastolynx:app -b [::]:8834
+
+[Install]
+WantedBy=multi-user.target

requirements.txt

@@ -0,0 +1,4 @@
+flask
+gunicorn
+mastodon.py
+beeprint