This commit is contained in:
Daniel Langbein 2022-12-10 16:53:57 +01:00
commit a853c647d6
8 changed files with 275 additions and 0 deletions

2
.gitignore vendored Normal file
View File

@ -0,0 +1,2 @@
/.idea/
/venv/

1
requirements.txt Normal file
View File

@ -0,0 +1 @@
Mastodon.py

View File

View File

@ -0,0 +1,55 @@
from pathlib import Path
from mastodon import Mastodon
from mastodon_toot_follower import path_util
class App:
def __init__(self,
name: str = 'MastodonTootFollower',
url: str = 'https://git.privacy1st.de/langfingaz/MastodonTootFollower',
config_dir: Path = None,
):
if len(name) <= 0:
raise ValueError(f'name is too short: {name}')
if config_dir is None:
config_dir = Path.home().joinpath('.' + name)
self.name = name
self.url = url
self.config_dir = config_dir
self.mastodon_dict: dict[str, Mastodon] = dict()
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)
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.
self.mastodon_dict[mastodon_instance] = Mastodon(client_id=self._credential_file(mastodon_instance))
return self.mastodon_dict[mastodon_instance]
def _create_or_load_credentials(self, mastodon_instance: str) -> None:
cf = self._credential_file(mastodon_instance)
# Create application credentials only once and save them.
if cf.exists():
return
Mastodon.create_app(
client_name=self.name,
website=self.url,
api_base_url=mastodon_instance,
to_file=cf,
scopes=['read']
)
def _credential_file(self, mastodon_instance: str) -> Path:
return self._credential_dir().joinpath(path_util.escape(mastodon_instance))
def _credential_dir(self) -> Path:
return self.config_dir.joinpath('credentials')
def _toot_cache_dir(self) -> Path:
return self.config_dir.joinpath('toots')

View File

@ -0,0 +1,99 @@
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

View File

@ -0,0 +1,45 @@
import re
import sys
from mastodon_toot_follower.app import App
from mastodon_toot_follower.conversation import Conversation
def main():
if len(sys.argv) != 2:
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)
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()}')
def usage():
print(f'\n'
f'usage: {sys.argv[0]} MASTODON_TOOT_URL\n'
f'example: {sys.argv[0]} https://mastodon.instance/@user@another.instance/123456', file=sys.stderr)
exit(1)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,71 @@
from pathlib import Path
from mastodon import Mastodon
def main():
app_name = 'MastodonTootFollower'
git_url = 'https://git.privacy1st.de/langfingaz/MastodonTootFollower'
app_credentials = Path(app_name + '.secret')
instance_url = 'https://replace-with-existing-instance'
# Create application credentials only once.
if not app_credentials.exists():
client_id, client_secret = Mastodon.create_app(
client_name=app_name,
website=git_url,
api_base_url=instance_url,
to_file=app_credentials,
scopes=['read']
)
# Create Mastdon API wrapper object.
# Optionally, log in.
#
# Copied from __init__:
# Create a new API wrapper instance based on the given `client_secret` and `client_id` on the
# instance given by `api_base_url`. If you give a `client_id` and it is not a file, you must
# also give a secret. If you specify an `access_token` then you don't need to specify a `client_id`.
# It is allowed to specify neither - in this case, you will be restricted to only using endpoints
# that do not require authentication. If a file is given as `client_id`, client ID, secret and
# base url are read from that file.
mastodon = Mastodon(client_id=app_credentials)
# mastodon.log_in(
# username='my_login_email@example.com',
# password='incrediblygoodpassword',
# to_file='pytooter_usercred.secret'
# )
test_status_info(mastodon, 'https://replace-with-existing-toot-url')
def test_status_info(mastodon: Mastodon, status_url: str, cache_dir: Path = Path('cache')) -> None:
# Example: https://mastodon.instance/@user@another.instance/123456
prefix = 'https://'
if not status_url.startswith(prefix):
raise ValueError(status_url)
status_url = status_url[len(prefix):]
status_url_parts = status_url.split('/')
if len(status_url_parts) != 3:
raise ValueError(status_url)
instance_url, username, status_id = status_url_parts
# Returns information about the given status.
# - number of replies
# result = mastodon.status(id=status_id)
# Returns the following dictionary:
# {
# 'ancestors': # A list of toot dicts
# 'descendants': # A list of toot dicts (including fields `id`, `created_at`, 'edited_at')
# }
result = mastodon.status_context(id=status_id)
print(result)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,2 @@
def escape(filename: str) -> str:
return filename.replace('/', '_')