Multiple changes

- rework of "Conversation" class
- "Update" class
- Flask development server
- Waitress production server
- output as RSS/Atom feed
- output as JSON and HTML
- improved CLI output

Makefile

@@ -0,0 +1,40 @@
+.ONESHELL:
+SHELL := bash
+# https://github.com/JordanMartinez/purescript-cookbook/blob/master/makefile
+# set -e = bash immediately exits if any command has a non-zero exit status.
+# set -u = a reference to any shell variable you haven't previously
+#    defined -- with the exceptions of $* and $@ -- is an error, and causes
+#    the program to immediately exit with non-zero code.
+# set -o pipefail = the first non-zero exit code emitted in one part of a
+#    pipeline (e.g. `cat file.txt | grep 'foo'`) will be used as the exit
+#    code for the entire pipeline. If all exit codes of a pipeline are zero,
+#    the pipeline will emit an exit code of 0.
+.SHELLFLAGS := -eu -o pipefail -c
+
+.PHONY: all
+all: prod
+
+.PHONY: prod
+prod: requirements.txt
+	source venv/bin/activate
+	export PYTHONPATH="${PYTHONPATH}:src"
+	python3 src/mastodon_toot_follower/server/waitress_server.py
+
+.PHONY: dev
+dev: requirements.txt
+	source venv/bin/activate
+	export PYTHONPATH="${PYTHONPATH}:src"
+	python3 src/mastodon_toot_follower/server/flask_server.py
+
+
+.PHONY: requirements.txt
+requirements.txt:
+	if ! [ -d venv ]; then
+	  python3 -m venv venv
+	  source venv/bin/activate
+	  python3 -m pip install -r requirements.txt
+	fi
+
+.PHONY: clean
+clean:
+	rm -rf venv

requirements.txt

@@ -1 +1,4 @@
-Mastodon.py
+Mastodon.py
+Flask
+feedgen
+waitress

src/mastodon_toot_follower/app.py

@@ -22,9 +22,9 @@ class App:
 
         config_dir.mkdir(parents=True, exist_ok=True)
         self._credential_dir().mkdir(parents=True, exist_ok=True)
-        self._toot_cache_dir().mkdir(parents=True, exist_ok=True)
+        self.toot_cache_dir().mkdir(parents=True, exist_ok=True)
 
-    def _get_mastodon(self, mastodon_instance: str) -> Mastodon:
+    def get_mastodon(self, mastodon_instance: str) -> Mastodon:
         if mastodon_instance not in self.mastodon_dict:
             self._create_or_load_credentials(mastodon_instance)
             # Create Mastdon API wrapper object.
@@ -51,5 +51,5 @@ class App:
     def _credential_dir(self) -> Path:
         return self.config_dir.joinpath('credentials')
 
-    def _toot_cache_dir(self) -> Path:
+    def toot_cache_dir(self) -> Path:
         return self.config_dir.joinpath('toots')

src/mastodon_toot_follower/conversation.py

@@ -1,99 +0,0 @@
-from mastodon_toot_follower import path_util
-import json
-from mastodon_toot_follower.app import App
-
-
-class Conversation:
-    def __init__(self, app: App, mastodon_instance: str, toot_id: str):
-        self.app = app
-        self.mastodon = self.app._get_mastodon(mastodon_instance)
-
-        self.toot_id = toot_id
-        self.toot = self.mastodon.status(id=self.toot_id)
-        self.toot_file = self.app._toot_cache_dir().joinpath(path_util.escape(self.toot['uri']) + '.json')
-
-        self._replies = None
-        self._changed_toots = None
-        self._saved = False
-
-    def get_conversation_length(self) -> int:
-        return len(self.toots())
-
-    def get_changed_toots_and_update(self):
-        """
-        Compares the old and current versions of this conversation.
-        Saves the current version of the conversation afterwards.
-
-        :return: New and changes toots which are part of this conversation.
-        """
-        if self.is_new():
-            self._changed_toots = []
-            self.save()
-            return self._changed_toots
-        else:
-            changed_toots = self.get_changed_toots()
-            if len(changed_toots) > 0:
-                self.save()
-            return changed_toots
-
-    def get_changed_toots(self) -> list:
-        """
-        Compares the old and current versions of this conversation.
-
-        :return: New and changes toots which are part of this conversation.
-        """
-        if self._changed_toots is None:
-            old_toots = self.load()
-            old_toots_dict = {old_toot['id']: old_toot['edited_at']
-                              for old_toot in old_toots}
-            toots = self.toots()
-
-            self._changed_toots = []
-            for toot in toots:
-                toot_id_not_found = object()
-                old_edited_at = old_toots_dict.get(toot['id'], toot_id_not_found)
-                if old_edited_at is toot_id_not_found or toot['edited_at'] != old_edited_at:
-                    # New reply or has been edited.
-                    self._changed_toots.append(toot)
-
-        return self._changed_toots
-
-    def is_new(self):
-        """
-        :return: True if this is a new conversation of which no old version exists.
-        """
-        return not self.toot_file.exists()
-
-    def save(self):
-        """
-        :return: Saves the current version of the conversation. This overwrites the previously saved version.
-        """
-        if self._saved:
-            return
-        self._saved = True
-        toots = [{'id': entry['id'], 'edited_at': entry['edited_at']}
-                 for entry in self.toots()]
-        with self.toot_file.open('w') as f:
-            f.write(json.dumps(toots))
-
-    def load(self) -> list:
-        """
-        :return: Loads an old version of this conversation.
-        """
-        if self._saved:
-            raise ValueError('The old version of this conversation has been overwritten.')
-        return json.loads(self.toot_file.read_text())
-
-    def toots(self) -> list:
-        """
-        :return: List of initial toot and it's replies.
-        """
-        return [self.toot] + self.replies()
-
-    def replies(self) -> list:
-        """
-        :return: List of replies to initial toot.
-        """
-        if self._replies is None:
-            self._replies = self.mastodon.status_context(id=self.toot_id)['descendants']
-        return self._replies

src/mastodon_toot_follower/conversation/conversation.py

@@ -0,0 +1,80 @@
+from feedgen.feed import FeedGenerator
+from flask import url_for
+
+from mastodon_toot_follower import path_util
+from mastodon_toot_follower.app import App
+from mastodon_toot_follower.conversation.update import Update as ConversationUpdate
+
+
+class Conversation:
+    def __init__(self, app: App, mastodon_instance: str, toot_id: str, seed: str = ''):
+        self.app = app
+        self.mastodon = self.app.get_mastodon(mastodon_instance)
+
+        self.toot_id = toot_id
+        self.toot = self.mastodon.status(id=self.toot_id)
+
+        if len(seed) > 0:
+            self.file = self.app.toot_cache_dir().joinpath(f'{path_util.escape(self.toot["uri"])}_{seed}.json')
+        else:
+            self.file = self.app.toot_cache_dir().joinpath(f'{path_util.escape(self.toot["uri"])}.json')
+
+        # List of replies to initial toot.
+        self.replies = self.mastodon.status_context(id=self.toot_id)['descendants']
+
+        self.previous_updates: list[ConversationUpdate] = ConversationUpdate.load(self.file)
+        self.new_updates = ConversationUpdate.get_new_updates(self.previous_updates, self.toots())
+
+        self._changes_saved = False
+
+    def as_feed(self, feed_url) -> FeedGenerator:
+        """
+        Returns a FeedGenerator object "fg".
+
+        atomfeed = fg.atom_str(pretty=True)  # Get the ATOM feed as string
+        rssfeed = fg.rss_str(pretty=True)  # Get the RSS feed as string
+        """
+
+        fg = FeedGenerator()
+        fg.id(self.toot['uri'])  # Unique feed ID.
+        fg.title(f'MastodonConversation{self.toot_id}')
+        fg.author({'name': self.toot['account']['acct']})
+        fg.link(href=self.toot['uri'], rel='alternate')  # Webpage.
+        fg.logo(url_for('static', filename='favicon.png'))
+        fg.subtitle('RSS feed of a conversation on Mastodon.')
+        fg.link(href=feed_url, rel='self')  # URL of the feed itself.
+        fg.language('en')
+
+        update: ConversationUpdate
+        for update in self.updates():
+            fe = fg.add_entry()
+            fe.id(update.dict['uri'])
+            acct = update.dict['acct']
+            if update.dict['reason'] == 'new':
+                fe.title(f'Reply by {acct}')
+                fe.published(update.dict['created_at'])
+            else:
+                fe.title(f'Edit by {acct}')
+                fe.published(update.dict['edited_at'])
+            fe.link(href=update.dict['uri'])
+            fe.content(update.dict['content'])
+
+        return fg
+
+    def conversation_length(self) -> int:
+        return len(self.toots())
+
+    def toots(self) -> list:
+        """
+        :return: List of initial toot and it's replies.
+        """
+        return [self.toot] + self.replies
+
+    def updates(self) -> list[ConversationUpdate]:
+        return self.previous_updates + self.new_updates
+
+    def save_changes(self) -> None:
+        if self._changes_saved:
+            return
+        ConversationUpdate.save(self.updates(), self.file)
+        self._changes_saved = True

src/mastodon_toot_follower/conversation/update.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import inspect
+from pathlib import Path
+from typing import Literal
+import json
+
+
+class Update:
+    @classmethod
+    def save(cls, updates: list[Update], file: Path) -> None:
+        with file.open('w') as f:
+            dict_list = [update.dict for update in updates]
+            f.write(json.dumps(dict_list))
+
+    @classmethod
+    def load(cls, file: Path) -> list[Update]:
+        if not file.exists():
+            return []
+        dict_list = json.loads(file.read_text())
+        return [cls._from_dict(d) for d in dict_list]
+
+    @classmethod
+    def get_new_updates(cls,
+                        previous_updates: list[Update],
+                        toots: list[dict]) -> list[Update]:
+        previous_updates_dict = {pu.dict['id']: pu.dict['edited_at']
+                                 for pu in previous_updates}
+        new_updates: list[Update] = []
+        for toot in toots:
+            id_not_in_previous_updates = object()
+            old_edited_at = previous_updates_dict.get(toot['id'], id_not_in_previous_updates)
+            if old_edited_at is id_not_in_previous_updates:
+                new_updates.append(Update.new('new', toot))
+            elif str(toot['edited_at']) != old_edited_at:
+                new_updates.append(Update.new('edited', toot))
+        return new_updates
+
+    @classmethod
+    def new(cls, reason: Literal['new', 'edited'], toot: dict) -> Update:
+        update = Update()
+        update.dict = {
+            'reason': reason,
+            #
+            'created_at': str(toot['created_at']),
+            'edited_at': str(toot['edited_at']),
+            #
+            'id': toot['id'],
+            'uri': toot['uri'],
+            #
+            'acct': toot['account']['acct'],
+            'username': toot['account']['username'],
+            'display_name': toot['account']['display_name'],
+            #
+            'content': toot['content'],
+            'language': toot['language'],
+            'media_attachments': toot['media_attachments'],
+        }
+        return update
+
+    @classmethod
+    def _from_dict(cls, d: dict) -> Update:
+        update = Update()
+        update.dict = d
+        return update
+
+    def __init__(self):
+        # Make constructor "private":
+        # Can only be called from classmethod of Update class.
+        stack = inspect.stack()
+        if 'cls' not in stack[1][0].f_locals or stack[1][0].f_locals["cls"] is not Update:
+            raise ValueError('This constructor is private. Please use one of the classemthods.')
+
+        self.dict = {}
+
+    def get_title(self) -> str:
+        acct = self.dict['acct']
+        if self.dict['reason'] == 'new':
+            return f'Reply by {acct}'
+        else:
+            return f'Edit by {acct}'
+
+    def get_date(self):
+        acct = self.dict['acct']
+        if self.dict['reason'] == 'new':
+            return self.dict['created_at']
+        else:
+            return self.dict['edited_at']
+
+    def __str__(self):
+        return f'{self.get_date()}\n' \
+               f'\t{self.get_title()}\n' \
+               f'\t{self.dict["uri"]}\n'

src/mastodon_toot_follower/main.py

@@ -1,8 +1,10 @@
-import re
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
 import sys
 
+from mastodon_toot_follower import mastodon_util
 from mastodon_toot_follower.app import App
-from mastodon_toot_follower.conversation import Conversation
+from mastodon_toot_follower.conversation.conversation import Conversation
 
 
 def main():
@@ -10,28 +12,21 @@ def main():
         usage()
     toot_url = sys.argv[1]
 
-    #
-    # PARSE TOOT URL
-    #
-
-    # Example: https://mastodon.instance/@user@another.instance/123456
-    pattern = re.compile(r'(^https://[^/]+)/([^/]+)/([0-9]+)$')
-    match = pattern.match(toot_url)
-    if match is None:
-        print(f'Could not parse toot url: {toot_url}', file=sys.stderr)
+    try:
+        instance_url, username, toot_id = mastodon_util.parse_toot_url(url=toot_url)
+    except Exception as e:
+        print(e, file=sys.stderr)
         usage()
-    instance_url, username, toot_id = match.group(1), match.group(2), match.group(3)
-
-    #
-    # RUN THE APP
-    #
 
     app = App()
     conversation = Conversation(app=app,
                                 mastodon_instance=instance_url,
                                 toot_id=toot_id)
-    print(f'Conversation length: {conversation.get_conversation_length()}')
-    print(f'Updates: {conversation.get_changed_toots_and_update()}')
+    conversation.save_changes()
+
+    print(f'Conversation length: {conversation.conversation_length()}')
+    print(f'Conversation updates:\n')
+    print('\n'.join([str(update) for update in conversation.new_updates]))
 
 
 def usage():

src/mastodon_toot_follower/mastodon_util.py

@@ -0,0 +1,14 @@
+import re
+
+
+def parse_toot_url(url: str) -> tuple[str, str, str]:
+    """
+    :param url: E.g. https://mastodon.instance/@user@another.instance/123456
+    :return: instance, username, toot_id
+    """
+
+    pattern = re.compile(r'(^https://[^/]+)/([^/]+)/([0-9]+)$')
+    match = pattern.match(url)
+    if match is None:
+        raise ValueError(f'Could not parse toot url: {url}')
+    return match.group(1), match.group(2), match.group(3)

src/mastodon_toot_follower/path_util.py

@@ -1,2 +1,30 @@
+import os
+from pathlib import Path
+
+
 def escape(filename: str) -> str:
     return filename.replace('/', '_')
+
+
+def _get_path_util_py_dir() -> Path:
+    return Path(os.path.dirname(os.path.realpath(__file__)))
+
+
+def get_src_dir() -> Path:
+    return _get_path_util_py_dir().parent
+
+
+def get_project_dir() -> Path:
+    return get_src_dir().parent
+
+
+def get_templates_dir() -> Path:
+    return get_project_dir().joinpath('templates')
+
+
+def get_static_dir() -> Path:
+    return get_project_dir().joinpath('static')
+
+
+if __name__ == '__main__':
+    get_project_dir()

src/mastodon_toot_follower/server/flask_server.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from enum import Enum
+
+from flask import Flask, Response, render_template
+from flask import request
+
+from mastodon_toot_follower import mastodon_util, path_util
+from mastodon_toot_follower.app import App as MastodonApp
+from mastodon_toot_follower.conversation.conversation import Conversation
+
+# Create Flask's `app` object
+app = Flask(
+    __name__,
+    instance_relative_config=False,
+    template_folder=str(path_util.get_templates_dir()),
+    static_folder=str(path_util.get_static_dir()),
+)
+
+mastodon_app = MastodonApp()
+
+
+class Templates(Enum):
+    index = 'index.html'
+    updates = 'updates.html'
+
+
+@app.route('/', methods=['GET'])
+def hello_world():
+    return render_template(Templates.index.value)
+
+
+@app.route('/rss/')
+def rss():
+    url = request.args.get('url', 'None')
+    instance_url, username, toot_id = mastodon_util.parse_toot_url(url=url)
+
+    conversation = Conversation(app=mastodon_app, mastodon_instance=instance_url, toot_id=toot_id)
+    conversation.save_changes()
+
+    fg = conversation.as_feed(request.url)
+    return Response(fg.rss_str(), mimetype='application/rss+xml')
+
+
+@app.route('/html/<string:seed>/')
+def html(seed: str):
+    url = request.args.get('url', 'None')
+    instance_url, username, toot_id = mastodon_util.parse_toot_url(url=url)
+
+    conversation = Conversation(app=mastodon_app, mastodon_instance=instance_url, toot_id=toot_id, seed=seed)
+    conversation.save_changes()
+
+    return render_template(Templates.updates.value, updates=conversation.new_updates)
+
+
+@app.route('/json/<string:seed>/')
+def json(seed: str):
+    url = request.args.get('url', 'None')
+    instance_url, username, toot_id = mastodon_util.parse_toot_url(url=url)
+
+    conversation = Conversation(app=mastodon_app, mastodon_instance=instance_url, toot_id=toot_id, seed=seed)
+    conversation.save_changes()
+
+    # If you return a dict or list from a view, it will be converted to a JSON response.
+    return [update.dict for update in conversation.new_updates]
+
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=5000)

src/mastodon_toot_follower/server/waitress_server.py

@@ -0,0 +1,8 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from waitress import serve
+
+import flask_server
+
+if __name__ == '__main__':
+    serve(flask_server.app, host='0.0.0.0', port=5000)

static/index.js

@@ -0,0 +1,26 @@
+function html_button() {
+    button_result('/html/' + get_seed() + '/?url=' + get_url());
+}
+
+function json_button() {
+    button_result('/json/' + get_seed() + '/?url=' + get_url());
+}
+
+function rss_button() {
+    button_result('/rss/?url=' + get_url());
+}
+
+function button_result(url) {
+    let href =  document.getElementById("href1")
+    href.href = url;
+    href.innerHTML = url;
+    // window.open(url);
+}
+
+function get_url() {
+    return document.getElementById('input1').value
+}
+
+function get_seed() {
+    return document.getElementById('input2').value
+}

static/style.css

@@ -0,0 +1,50 @@
+/* Source: https://dev.to/dcodeyt/creating-beautiful-html-tables-with-css-428l */
+
+/*  Styling the table */
+.styled-table {
+    border-collapse: collapse;
+    margin: 25px 0;
+    font-size: 0.9em;
+    font-family: sans-serif;
+    min-width: 400px;
+    box-shadow: 0 0 20px rgba(0, 0, 0, 0.15);
+}
+
+/* Styling the header */
+.styled-table thead tr {
+    background-color: #009879;
+    color: #ffffff;
+    text-align: left;
+}
+
+/* Add spaces */
+.styled-table th,
+.styled-table td {
+    padding: 12px 15px;
+}
+
+/* Table rows
+
+    Add a bottom border to each row for separation
+    Add a lighter background to every second row to help readability
+    Add a dark border to the very last row to signify the end of the table
+*/
+
+.styled-table tbody tr {
+    /* If you want your borders to look sharp on high resolution displays you can use `thin` instead of `1px` */
+    border-bottom: thin solid #dddddd;
+}
+
+.styled-table tbody tr:nth-of-type(even) {
+    background-color: #f3f3f3;
+}
+
+.styled-table tbody tr:last-of-type {
+    border-bottom: medium solid #009879;
+}
+
+/* Make the active row look different */
+.styled-table tbody tr.active-row {
+    font-weight: bold;
+    color: #009879;
+}

templates/index.html

@@ -0,0 +1,39 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <meta http-equiv="X-UA-Compatible" content="ie=edge">
+  <title>My Website</title>
+  <link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
+  <!-- https://developer.mozilla.org/en-US/docs/Glossary/Favicon -->
+  <link rel="icon" href="{{ url_for('static', filename='favicon.png') }}" type="image/x-icon">
+</head>
+<body>
+<main>
+  <h1>MastodonTootFollower</h1>
+  <p>
+    <label for="input1">URL of toot to follow: </label>
+    <input name="html_url" type="url" maxlength="256" id="input1"/>
+  </p>
+  <p>
+    <label for="input2">Private seed; required for consecutive updates: </label>
+    <input name="html_url" type="text" maxlength="256" id="input2"/>
+  </p>
+  <ul>
+    <li>
+      <button onclick="rss_button()">All updates as RSS feed</button>
+    </li>
+    <li>
+      <button onclick="json_button()">Consecutive updates as JSON file</button>
+    </li>
+    <li>
+      <button onclick="html_button()">Consecutive updates as HTML page</button>
+    </li>
+  </ul>
+  <p>Resulting link: <a href="https://localhost:443" id="href1"></a></p>
+
+</main>
+<script src="{{ url_for('static', filename='index.js') }}"></script>
+</body>
+</html>

templates/updates.html

@@ -0,0 +1,36 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="UTF-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <meta http-equiv="X-UA-Compatible" content="ie=edge">
+  <title>My Website</title>
+  <link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
+  <!-- https://developer.mozilla.org/en-US/docs/Glossary/Favicon -->
+  <link rel="icon" href="{{ url_for('static', filename='favicon.png') }}" type="image/x-icon">
+</head>
+<body>
+
+<h1>Consecutive Toot Updates</h1>
+
+<table class="styled-table">
+  <thead>
+  <tr>
+    <th>Date</th>
+    <th>Account</th>
+    <th>Content</th>
+  </tr>
+  </thead>
+  <tbody>
+  {% for update in updates %}
+    <tr>
+      <td><a href="{{ update.dict[uri] }}">{{ update.get_date() }}</a></td>
+      <td>{{ update.dict['acct'] }}</td>
+      <td>{{ update.dict['content'] | safe }}</td>
+    </tr>
+  {% endfor %}
+  </tbody>
+</table>
+
+</body>
+</html>