From acdc636b1fd4bde0be2d334d2511f15a02e66b25 Mon Sep 17 00:00:00 2001 From: Fabio Manganiello Date: Sun, 7 Nov 2021 01:00:29 +0100 Subject: [PATCH] Resolve "Mastodon integration" --- CHANGELOG.md | 4 + docs/source/platypush/plugins/mastodon.rst | 5 + docs/source/plugins.rst | 1 + platypush/backend/http/app/routes/auth.py | 3 +- platypush/common/ngrok/__init__.py | 46 + platypush/plugins/mastodon/__init__.py | 1502 ++++++++++++++++++++ platypush/plugins/mastodon/manifest.yaml | 4 + platypush/schemas/mastodon.py | 201 +++ 8 files changed, 1764 insertions(+), 2 deletions(-) create mode 100644 docs/source/platypush/plugins/mastodon.rst create mode 100644 platypush/common/ngrok/__init__.py create mode 100644 platypush/plugins/mastodon/__init__.py create mode 100644 platypush/plugins/mastodon/manifest.yaml create mode 100644 platypush/schemas/mastodon.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 57b1d7bed..c7a3d288a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ Given the high speed of development in the first phase, changes are being report ## [Unreleased] +### Added + +- Added Mastodon integration. + ### Fixed - Fixed `switchbot.status` method in case of virtual devices. diff --git a/docs/source/platypush/plugins/mastodon.rst b/docs/source/platypush/plugins/mastodon.rst new file mode 100644 index 000000000..3d126678d --- /dev/null +++ b/docs/source/platypush/plugins/mastodon.rst @@ -0,0 +1,5 @@ +``mastodon`` +============ + +.. automodule:: platypush.plugins.mastodon + :members: diff --git a/docs/source/plugins.rst b/docs/source/plugins.rst index 45f77764e..99bdc2bdd 100644 --- a/docs/source/plugins.rst +++ b/docs/source/plugins.rst @@ -72,6 +72,7 @@ Plugins platypush/plugins/luma.oled.rst platypush/plugins/mail.imap.rst platypush/plugins/mail.smtp.rst + platypush/plugins/mastodon.rst platypush/plugins/media.chromecast.rst platypush/plugins/media.gstreamer.rst platypush/plugins/media.kodi.rst diff --git a/platypush/backend/http/app/routes/auth.py b/platypush/backend/http/app/routes/auth.py index afdc2f23d..b59561d06 100644 --- a/platypush/backend/http/app/routes/auth.py +++ b/platypush/backend/http/app/routes/auth.py @@ -1,7 +1,6 @@ import datetime import json import logging -from typing import Dict, Optional from flask import Blueprint, request, abort, jsonify @@ -18,7 +17,7 @@ __routes__ = [ @auth.route('/auth', methods=['POST']) -def auth_endpoint() -> Dict[str, Optional[str]]: +def auth_endpoint(): """ Authentication endpoint. It validates the user credentials provided over a JSON payload with the following structure: diff --git a/platypush/common/ngrok/__init__.py b/platypush/common/ngrok/__init__.py new file mode 100644 index 000000000..ed885e749 --- /dev/null +++ b/platypush/common/ngrok/__init__.py @@ -0,0 +1,46 @@ +from threading import RLock +from typing import Optional + +from platypush.config import Config +from platypush.context import get_backend, get_plugin + +_app_tunnel_lock = RLock() +_app_tunnel_url: Optional[str] = None + + +def _get_http_port() -> int: + http = None + if Config.get('backend.http'): + http = get_backend('http') + + assert http, 'The http backend is required in order to subscribe to notifications' + return http.port + + +def create_ngrok_tunnel() -> str: + """ + This method creates an ngrok tunnel for the local web application, + useful to register public HTTPS callback URLs on the fly from plugins + and backends. + """ + global _app_tunnel_url + with _app_tunnel_lock: + if _app_tunnel_url: + return _app_tunnel_url + + local_port = _get_http_port() + + ngrok = None + if Config.get('ngrok'): + ngrok = get_plugin('ngrok') + + assert ngrok, 'The ngrok plugin is required in order to subscribe to notifications' + tunnel_response = ngrok.create_tunnel( + resource=local_port, + protocol='http', + bind_tls=True, + ).output + + _app_tunnel_url = tunnel_response.get('url') + assert _app_tunnel_url, 'Unable to create an ngrok tunnel' + return _app_tunnel_url diff --git a/platypush/plugins/mastodon/__init__.py b/platypush/plugins/mastodon/__init__.py new file mode 100644 index 000000000..1a3d936b9 --- /dev/null +++ b/platypush/plugins/mastodon/__init__.py @@ -0,0 +1,1502 @@ +import os +import requests +from typing import Optional, Union, Iterable, Mapping, Sequence + +from platypush.plugins import Plugin, action +from platypush.schemas.mastodon import MastodonSchema, MastodonSearchSchema, MastodonAccountCreationSchema, \ + MastodonAccountSchema, MastodonStatusSchema, MastodonFeaturedHashtagSchema, MastodonAccountListSchema, \ + MastodonFilterSchema, MastodonMediaSchema, MastodonConversationSchema, MastodonListSchema, \ + MastodonNotificationSchema +from platypush.utils import get_mime_type + + +class MastodonPlugin(Plugin): + """ + Plugin to interact with `Mastodon `_ instances. + + It requires an active API token associated to an app registered on the instance. + In order to get one: + + - Open ``https:///settings/applications/`` + - Create a new application + - Select the scopes relevant for your specific usage. + - Take note of the token reported on the *Your access token* row. + + The notifications subscription service requires the ``ngrok`` plugin and the + `http` backend to be enabled, since we need to expose an external URL that + the Mastodon instance can call when new events occur. + """ + + class SubscriptionConfig: + tunnel_url: str + local_port: int + auth_secret: str + private_key: str + public_key: str + server_key: str + + def __init__(self, base_url: str, access_token: Optional[str] = None, **kwargs): + """ + :param base_url: Base URL of the Mastodon web server, in the form of ``https://``. + :param access_token: Access token as reported on ``https:///settings/applications/``. + """ + super().__init__(**kwargs) + self._base_url = base_url + self._access_token = access_token + self._subscription_config = self.SubscriptionConfig() + + def base_url(self, version: str, base_url: Optional[str] = None) -> str: + return f'{base_url or self._base_url}/api/{version}' + + def _run( + self, path: str, method: str = 'get', version: str = 'v2', headers: Optional[dict] = None, + base_url: Optional[str] = None, access_token: Optional[str] = None, + schema: Optional[MastodonSchema] = None, **kwargs + ) -> Optional[Union[dict, list]]: + headers = { + 'Authorization': f'Bearer {access_token or self._access_token}', + 'Accept': 'application/json', + **(headers or {}), + } + + method = getattr(requests, method.lower()) + rs = method(self.base_url(base_url=base_url, version=version) + '/' + path, headers=headers, **kwargs) + rs.raise_for_status() + rs = rs.json() + if schema: + rs = schema.dump(rs) + return rs + + # noinspection PyShadowingBuiltins + @action + def search( + self, query: str, type: Optional[str] = None, min_id: Optional[str] = None, + max_id: Optional[str] = None, limit: int = 20, offset: int = 0, following: bool = False, + **kwargs + ) -> Mapping[str, Iterable[dict]]: + """ + Perform a search. + + :param query: Search query. + :param type: Filter by type. Supported types: + + - ``accounts`` + - ``hashtags`` + - ``statuses`` + + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param following: Only return results from accounts followed by the user (default: False). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonSearchSchema + """ + return self._run( + 'search', + version='v2', + schema=MastodonSearchSchema(), + params={ + 'q': query, + **({'type': type} if type else {}), + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + **({'following': following} if following else {}), + }, **kwargs + ) + + @action + def register_account( + self, username: str, email: str, password: str, locale: str = 'en', + reason: Optional[str] = None, **kwargs + ) -> dict: + """ + Register a new account. + + It requires the specified API token to have ``write:accounts`` permissions. + + :param username: User name. + :param email: User's email address (must be a valid address). + :param password: The password used for the first login. + :param locale: Language/encoding for the confirmation email. + :param reason: Text that will be reviewed by moderators if registrations require manual approval. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountCreationSchema + """ + return self._run( + 'accounts', + method='post', + version='v1', + schema=MastodonAccountCreationSchema(), + json={ + 'username': username, + 'email': email, + 'password': password, + 'locale': locale, + 'reason': reason, + 'agreement': True, + }, **kwargs + ) + + @action + def update_account( + self, discoverable: Optional[bool] = None, bot: Optional[bool] = None, + display_name: Optional[str] = None, note: Optional[str] = None, + avatar: Optional[str] = None, header: Optional[str] = None, + locked: Optional[bool] = None, privacy: Optional[str] = None, + sensitive: Optional[bool] = None, language: Optional[str] = None, + metadata: Optional[Iterable[Mapping]] = None, **kwargs + ) -> dict: + """ + Updates the properties of the account associated to the access token. + + It requires the specified API token to have ``write:accounts`` permissions. + + :param discoverable: Whether the account should be shown in the profile directory. + :param bot: Whether the account is a bot. + :param display_name: The display name to use for the profile. + :param note: The account bio (HTML is supported). + :param avatar: Path to an avatar image. + :param header: Path to a header image. + :param locked: Whether manual approval of follow requests is required. + :param privacy: Default post privacy for authored statuses. + :param sensitive: Whether to mark authored statuses as sensitive by default. + :param language: Default language to use for authored statuses (ISO 6391 code). + :param metadata: Profile metadata items with ``name`` and ``value``. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema + """ + avatar = os.path.expanduser(avatar) if avatar else None + header = os.path.expanduser(header) if header else None + return self._run( + 'accounts/update_credentials', + method='patch', + version='v1', + schema=MastodonAccountSchema(), + data={ + **({'discoverable': discoverable} if discoverable is not None else {}), + **({'bot': bot} if bot is not None else {}), + **({'display_name': display_name} if display_name is not None else {}), + **({'note': note} if note is not None else {}), + **({'locked': locked} if locked is not None else {}), + **({'source[privacy]': privacy} if privacy is not None else {}), + **({'source[sensitive]': sensitive} if sensitive is not None else {}), + **({'source[language]': language} if language is not None else {}), + **({'fields_attributes': metadata} if metadata is not None else {}), + }, + files={ + **({'avatar': ( + os.path.basename(avatar), open(avatar, 'rb'), get_mime_type(avatar) + )} if avatar is not None else {}), + **({'header': ( + os.path.basename(header), open(header, 'rb'), get_mime_type(header) + )} if header is not None else {}), + }, + **kwargs + ) + + @action + def get_account(self, account_id: str, **kwargs) -> dict: + """ + Retrieve an account by ID. + + It requires the specified API token to have ``read:accounts`` permissions. + + :param account_id: Account ID to retrieve. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema + """ + return self._run( + f'accounts/{account_id}', + version='v1', + schema=MastodonAccountSchema(), + **kwargs + ) + + @action + def get_statuses(self, account_id: str, min_id: Optional[str] = None, max_id: Optional[str] = None, + limit: int = 20, offset: int = 0, **kwargs) -> Iterable[dict]: + """ + Retrieve statuses by account ID. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param account_id: Account ID. + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + f'accounts/{account_id}/statuses', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, + **kwargs + ) + + @action + def get_followers(self, account_id: str, max_id: Optional[str] = None, + since_id: Optional[str] = None, limit: int = 20, offset: int = 0, + **kwargs) -> Iterable[dict]: + """ + Retrieve the list of followers of an account. + + It requires the specified API token to have the ``read:accounts`` permission. + + :param account_id: Account ID. + :param max_id: Return results older than this ID. + :param since_id: Return results newer than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema(many=True) + """ + return self._run( + f'accounts/{account_id}/followers', + version='v1', + schema=MastodonAccountSchema(many=True), + params={ + **({'since_id': since_id} if since_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, + **kwargs + ) + + @action + def get_following(self, account_id: str, max_id: Optional[str] = None, + since_id: Optional[str] = None, limit: int = 20, offset: int = 0, + **kwargs) -> Iterable[dict]: + """ + Retrieve the list of accounts followed by a specified account. + + It requires the specified API token to have the ``read:accounts`` permission. + + :param account_id: Account ID. + :param max_id: Return results older than this ID. + :param since_id: Return results newer than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema(many=True) + """ + return self._run( + f'accounts/{account_id}/following', + version='v1', + schema=MastodonAccountSchema(many=True), + params={ + **({'since_id': since_id} if since_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, + **kwargs + ) + + @action + def get_featured_tags(self, account_id: Optional[str] = None, max_id: Optional[str] = None, + since_id: Optional[str] = None, limit: int = 20, offset: int = 0, + **kwargs) -> Iterable[dict]: + """ + Retrieve the list of featured hashtags of an account. + + It requires the specified API token to have the ``read:accounts`` permission. + + :param account_id: Account ID (if not specified then retrieve the featured tags of the current account). + :param max_id: Return results older than this ID. + :param since_id: Return results newer than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonFeaturedHashtagSchema(many=True) + """ + return self._run( + f'accounts/{account_id}/featured_tags' if account_id else 'featured_tags', + version='v1', + schema=MastodonFeaturedHashtagSchema(many=True), + params={ + **({'since_id': since_id} if since_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, + **kwargs + ) + + @action + def get_featured_lists(self, account_id: str, max_id: Optional[str] = None, + since_id: Optional[str] = None, limit: int = 20, offset: int = 0, + **kwargs) -> Iterable[dict]: + """ + Retrieve the list that you have added a certain account to. + + It requires the specified API token to have the ``read:lists`` permission. + + :param account_id: Account ID. + :param max_id: Return results older than this ID. + :param since_id: Return results newer than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountListSchema(many=True) + """ + return self._run( + f'accounts/{account_id}/lists', + version='v1', + schema=MastodonAccountListSchema(many=True), + params={ + **({'since_id': since_id} if since_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, + **kwargs + ) + + @action + def follow_account(self, account_id: str, notify: bool = False, reblogs: bool = True, **kwargs): + """ + Follow a given account ID. + + It requires the specified API token to have the ``write:follows`` permission. + + :param account_id: Account ID. + :param notify: Receive notifications when this account posts a new status (default: False). + :param reblogs: Receive this account's reblogs on your timeline (default: True). + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/follow', + version='v1', + method='post', + json={'notify': notify, 'reblogs': reblogs}, + **kwargs + ) + + @action + def unfollow_account(self, account_id: str, **kwargs): + """ + Unfollow a given account ID. + + It requires the specified API token to have the ``write:follows`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/unfollow', + version='v1', + method='post', + **kwargs + ) + + @action + def block_account(self, account_id: str, **kwargs): + """ + Block a given account ID. + + It requires the specified API token to have the ``write:blocks`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/block', + version='v1', + method='post', + **kwargs + ) + + @action + def unblock_account(self, account_id: str, **kwargs): + """ + Unblock a given account ID. + + It requires the specified API token to have the ``write:blocks`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/unblock', + version='v1', + method='post', + **kwargs + ) + + @action + def mute_account(self, account_id: str, **kwargs): + """ + Mute a given account ID. + + It requires the specified API token to have the ``write:mutes`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/mute', + version='v1', + method='post', + **kwargs + ) + + @action + def unmute_account(self, account_id: str, **kwargs): + """ + Unmute a given account ID. + + It requires the specified API token to have the ``write:mutes`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/unmute', + version='v1', + method='post', + **kwargs + ) + + @action + def pin_account(self, account_id: str, **kwargs): + """ + Pin a given account ID to your profile. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/pin', + version='v1', + method='post', + **kwargs + ) + + @action + def unpin_account(self, account_id: str, **kwargs): + """ + Unpin a given account ID from your profile. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param account_id: Account ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/unpin', + version='v1', + method='post', + **kwargs + ) + + @action + def set_account_note(self, account_id: str, note: str, **kwargs): + """ + Set a private note for an account. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param account_id: Account ID. + :param note: Note content (HTML is supported). + :param kwargs: ``base_url``/``access_token`` override. + """ + self._run( + f'accounts/{account_id}/note', + version='v1', + method='post', + json={'comment': note}, + **kwargs + ) + + @action + def get_bookmarked_statuses(self, min_id: Optional[str] = None, + max_id: Optional[str] = None, limit: int = 20, **kwargs) -> Iterable[dict]: + """ + Retrieve the list of statuses bookmarked by the user. + + It requires the specified API token to have the ``read:bookmarks`` permission. + + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + 'bookmarks', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, + **kwargs + ) + + @action + def get_favourited_statuses(self, min_id: Optional[str] = None, + max_id: Optional[str] = None, limit: int = 20, **kwargs) -> Iterable[dict]: + """ + Retrieve the list of statuses favourited by the account. + + It requires the specified API token to have the ``read:favourites`` permission. + + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + 'favourites', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, + **kwargs + ) + + @action + def get_muted_accounts(self, max_id: Optional[str] = None, + since_id: Optional[str] = None, limit: int = 20, + **kwargs) -> Iterable[dict]: + """ + Retrieve the list of muted accounts. + + It requires the specified API token to have the ``read:mutes`` permission. + + :param max_id: Return results older than this ID. + :param since_id: Return results newer than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema(many=True) + """ + return self._run( + 'mutes', + version='v1', + schema=MastodonAccountSchema(many=True), + params={ + **({'since_id': since_id} if since_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, + **kwargs + ) + + @action + def get_blocked_accounts(self, max_id: Optional[str] = None, + since_id: Optional[str] = None, limit: int = 20, + **kwargs) -> Iterable[dict]: + """ + Retrieve the list of blocked accounts. + + It requires the specified API token to have the ``read:blocks`` permission. + + :param max_id: Return results older than this ID. + :param since_id: Return results newer than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema(many=True) + """ + return self._run( + 'blocks', + version='v1', + schema=MastodonAccountSchema(many=True), + params={ + **({'since_id': since_id} if since_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, + **kwargs + ) + + @action + def get_filters(self, **kwargs) -> Iterable[dict]: + """ + Retrieve the list of filters created by the account. + + It requires the specified API token to have the ``read:filters`` permission. + + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonFilterSchema(many=True) + """ + return self._run( + 'filters', + version='v1', + schema=MastodonFilterSchema(many=True), + **kwargs + ) + + @action + def create_filter(self, phrase: str, context: Iterable[str], + irreversible: Optional[bool] = None, + whole_word: Optional[bool] = None, + expires_in: Optional[int] = None, + **kwargs) -> dict: + """ + Create a new filter. + + It requires the specified API token to have the ``write:filters`` permission. + + :param phrase: Text to be filtered. + :param context: Array of enumerable strings: ``home``, ``notifications``, ``public``, ``thread``. + At least one context must be specified. + :param irreversible: Should the server irreversibly drop matching entities from home and notifications? + :param whole_word: Consider word boundaries? + :param expires_in: Expires in the specified number of seconds. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonFilterSchema + """ + return self._run( + 'filters', + version='v1', + method='post', + schema=MastodonFilterSchema(), + json={ + 'phrase': phrase, + 'context': context, + **({'irreversible': irreversible} if irreversible is not None else {}), + **({'whole_word': whole_word} if whole_word is not None else {}), + **({'expires_in': expires_in} if expires_in is not None else {}), + }, + **kwargs + ) + + @action + def update_filter(self, filter_id: int, + phrase: Optional[str] = None, + context: Optional[Iterable[str]] = None, + irreversible: Optional[bool] = None, + whole_word: Optional[bool] = None, + expires_in: Optional[int] = None, + **kwargs) -> dict: + """ + Update a filter. + + It requires the specified API token to have the ``write:filters`` permission. + + :param filter_id: Filter ID. + :param phrase: Text to be filtered. + :param context: Array of enumerable strings: ``home``, ``notifications``, ``public``, ``thread``. + At least one context must be specified. + :param irreversible: Should the server irreversibly drop matching entities from home and notifications? + :param whole_word: Consider word boundaries? + :param expires_in: Expires in the specified number of seconds. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonFilterSchema + """ + return self._run( + f'filters/{filter_id}', + version='v1', + method='put', + schema=MastodonFilterSchema(), + json={ + **({'phrase': phrase} if phrase is not None else {}), + **({'context': context} if context is not None else {}), + **({'irreversible': irreversible} if irreversible is not None else {}), + **({'whole_word': whole_word} if whole_word is not None else {}), + **({'expires_in': expires_in} if expires_in is not None else {}), + }, + **kwargs + ) + + @action + def remove_filter(self, filter_id: int, **kwargs): + """ + Remove a filter. + + It requires the specified API token to have the ``write:filters`` permission. + + :param filter_id: Filter ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'filters/{filter_id}', + version='v1', + method='delete', + **kwargs + ) + + @action + def add_featured_tag(self, name: str, **kwargs) -> dict: + """ + Add a featured tag to the current account. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param name: Hashtag name. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonFeaturedHashtagSchema + """ + return self._run( + 'featured_tags', + version='v1', + method='post', + schema=MastodonFeaturedHashtagSchema(), + json={'name': name}, + **kwargs + ) + + @action + def remove_featured_tag(self, tag_id: int, **kwargs): + """ + Remove a featured tag from the current account. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param tag_id: Hashtag ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'featured_tags/{tag_id}', + version='v1', + method='delete', + **kwargs + ) + + @action + def publish_status(self, status: str, in_reply_to_id: Optional[str] = None, + media_ids: Optional[Iterable[str]] = None, + sensitive: Optional[bool] = None, spoiler_text: Optional[str] = None, + visibility: Optional[str] = None, **kwargs) -> dict: + """ + Publish a new status. + + It requires the specified API token to have the ``write:statuses`` permission. + + :param status: Content of the status to publish. + :param in_reply_to_id: Post the status in reply to this status ID. + :param media_ids: Optional list of media IDs to add as attachments. + :param sensitive: Set to true if sensitive. + :param spoiler_text: Set for optional spoiler text. + :param visibility: Supported values: ``public``, ``unlisted``, ``private`` and ``direct``. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema + """ + return self._run( + 'statuses', + version='v1', + method='post', + schema=MastodonStatusSchema(), + json={ + 'status': status, + **({'in_reply_to_id': in_reply_to_id} if in_reply_to_id is None else {}), + **({'media_ids': media_ids} if media_ids else {}), + **({'sensitive': sensitive} if sensitive is None else {}), + **({'spoiler_text': spoiler_text} if spoiler_text is None else {}), + **({'visibility': visibility} if visibility is None else {}), + }, + **kwargs + ) + + @action + def get_status(self, status_id: str, **kwargs) -> dict: + """ + Get a status by ID. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema + """ + return self._run( + f'statuses/{status_id}', + version='v1', + schema=MastodonStatusSchema(), + **kwargs + ) + + @action + def remove_status(self, status_id: str, **kwargs): + """ + Remove a status by ID. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}', + version='v1', + method='delete', + **kwargs + ) + + @action + def add_favourite_status(self, status_id: str, **kwargs): + """ + Favourite a status. + + It requires the specified API token to have the ``write:favourites`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/favourite', + version='v1', + method='post', + **kwargs + ) + + @action + def remove_favourite_status(self, status_id: str, **kwargs): + """ + Undo a status favourite action. + + It requires the specified API token to have the ``write:favourites`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/favourite', + version='v1', + method='delete', + **kwargs + ) + + @action + def reblog_status(self, status_id: str, **kwargs): + """ + Reblog (a.k.a. reshare/boost) a status. + + It requires the specified API token to have the ``write:statuses`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/reblog', + version='v1', + method='post', + **kwargs + ) + + @action + def undo_reblog_status(self, status_id: str, **kwargs): + """ + Undo a status reblog. + + It requires the specified API token to have the ``write:statuses`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/unreblog', + version='v1', + method='post', + **kwargs + ) + + @action + def bookmark_status(self, status_id: str, **kwargs): + """ + Add a status to the bookmarks. + + It requires the specified API token to have the ``write:bookmarks`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/bookmark', + version='v1', + method='post', + **kwargs + ) + + @action + def undo_bookmark_status(self, status_id: str, **kwargs): + """ + Remove a status from the bookmarks. + + It requires the specified API token to have the ``write:bookmarks`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/unbookmark', + version='v1', + method='post', + **kwargs + ) + + @action + def mute_status(self, status_id: str, **kwargs): + """ + Mute updates on a status. + + It requires the specified API token to have the ``write:mutes`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/mute', + version='v1', + method='post', + **kwargs + ) + + @action + def unmute_status(self, status_id: str, **kwargs): + """ + Restore updates on a status. + + It requires the specified API token to have the ``write:mutes`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/unmute', + version='v1', + method='post', + **kwargs + ) + + @action + def pin_status(self, status_id: str, **kwargs): + """ + Pin a status to the profile. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/pin', + version='v1', + method='post', + **kwargs + ) + + @action + def unpin_status(self, status_id: str, **kwargs): + """ + Remove a pinned status. + + It requires the specified API token to have the ``write:accounts`` permission. + + :param status_id: Status ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'statuses/{status_id}/unpin', + version='v1', + method='post', + **kwargs + ) + + @action + def upload_media(self, file: str, description: Optional[str] = None, + thumbnail: Optional[str] = None, **kwargs) -> dict: + """ + Upload media that can be used as attachments. + + It requires the specified API token to have the ``write:media`` permission. + + :param file: Path to the file to upload. + :param thumbnail: Path to the file thumbnail. + :param description: Optional attachment description. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonMediaSchema + """ + file_path = os.path.expanduser(file) + thumbnail_path = os.path.expanduser(thumbnail) if thumbnail else None + return self._run( + 'media', + version='v1', + method='post', + schema=MastodonMediaSchema(), + data={ + **({'description': description} if description else {}), + }, + files={ + 'file': ( + os.path.basename(file_path), open(file_path, 'rb'), get_mime_type(file_path) + ), + **( + { + 'thumbnail': ( + os.path.basename(thumbnail_path), + open(os.path.expanduser(thumbnail_path), 'rb'), + get_mime_type(thumbnail_path) + ), + } if thumbnail_path else {} + ), + }, + **kwargs + ) + + @action + def update_media(self, media_id: str, file: Optional[str] = None, description: Optional[str] = None, + thumbnail: Optional[str] = None, **kwargs) -> dict: + """ + Update a media attachment. + + It requires the specified API token to have the ``write:media`` permission. + + :param media_id: Media ID to update. + :param file: Path to the new file. + :param description: New description. + :param thumbnail: Path to the new thumbnail. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonMediaSchema + """ + file = os.path.expanduser(file) + thumbnail = os.path.expanduser(thumbnail) + return self._run( + f'media/{media_id}', + version='v1', + method='put', + schema=MastodonMediaSchema(), + data={ + **({'description': description} if description else {}), + }, + files={ + 'file': ( + os.path.basename(file), open(file, 'rb'), get_mime_type(file) + ), + **( + { + 'thumbnail': ( + os.path.basename(thumbnail), + open(os.path.expanduser(thumbnail), 'rb'), + get_mime_type(thumbnail) + ), + } if thumbnail else {} + ), + }, + **kwargs + ) + + @action + def get_public_timeline( + self, local: bool = False, remote: bool = False, only_media: bool = False, + min_id: Optional[str] = None, max_id: Optional[str] = None, limit: int = 20, + offset: int = 0, **kwargs + ) -> Iterable[dict]: + """ + Get a list of statuses from the public timeline. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param local: Retrieve only local statuses (default: ``False``). + :param remote: Retrieve only remote statuses (default: ``False``). + :param only_media: Retrieve only statuses with media attached (default: ``False``). + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + 'timelines/public', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'local': local} if local else {}), + **({'remote': remote} if remote else {}), + **({'only_media': only_media} if only_media else {}), + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, **kwargs + ) + + @action + def get_hashtag_timeline( + self, hashtag: str, local: bool = False, only_media: bool = False, + min_id: Optional[str] = None, max_id: Optional[str] = None, limit: int = 20, + offset: int = 0, **kwargs + ) -> Iterable[dict]: + """ + Get a list of statuses associated to a hashtag. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param hashtag: Hashtag to search. + :param local: Retrieve only local statuses (default: ``False``). + :param only_media: Retrieve only statuses with media attached (default: ``False``). + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + f'timelines/tag/{hashtag}', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'local': local} if local else {}), + **({'only_media': only_media} if only_media else {}), + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, **kwargs + ) + + @action + def get_home_timeline( + self, local: bool = False, only_media: bool = False, + min_id: Optional[str] = None, max_id: Optional[str] = None, limit: int = 20, + offset: int = 0, **kwargs + ) -> Iterable[dict]: + """ + Get a list of statuses from the followed users. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param local: Retrieve only local statuses (default: ``False``). + :param only_media: Retrieve only statuses with media attached (default: ``False``). + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + f'timelines/home', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'local': local} if local else {}), + **({'only_media': only_media} if only_media else {}), + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, **kwargs + ) + + @action + def get_list_timeline( + self, list_id: str, + min_id: Optional[str] = None, max_id: Optional[str] = None, limit: int = 20, + offset: int = 0, **kwargs + ) -> Iterable[dict]: + """ + Get a list of statuses from a list timeline. + + It requires the specified API token to have the ``read:lists`` permission. + + :param list_id: List ID. + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param offset: Return results from this offset (default: 0). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonStatusSchema(many=True) + """ + return self._run( + f'timelines/list/{list_id}', + version='v1', + schema=MastodonStatusSchema(many=True), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + **({'offset': offset} if offset else {}), + }, **kwargs + ) + + @action + def get_conversations( + self, min_id: Optional[str] = None, max_id: Optional[str] = None, + limit: int = 20, **kwargs + ) -> Iterable[dict]: + """ + Get a list of user conversations. + + It requires the specified API token to have the ``read:statuses`` permission. + + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonConversationSchema(many=True) + """ + return self._run( + 'conversations', + version='v1', + schema=MastodonConversationSchema(many=True), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, **kwargs + ) + + @action + def remove_conversation(self, conversation_id: int, **kwargs): + """ + Remove a conversation by ID. + + It requires the specified API token to have the ``write_conversations`` permission. + + :param conversation_id: Conversation ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'conversations/{conversation_id}', + version='v1', + method='delete', + **kwargs + ) + + @action + def mark_conversation_as_read(self, conversation_id: int, **kwargs): + """ + Mark a conversation as read. + + It requires the specified API token to have the ``write_conversations`` permission. + + :param conversation_id: Conversation ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'conversations/{conversation_id}/read', + version='v1', + method='post', + **kwargs + ) + + @action + def get_lists(self, list_id: Optional[int] = None, **kwargs) -> Union[dict, Iterable[dict]]: + """ + Get the lists owned by the logged user. + + It requires the specified API token to have the ``read:lists`` permission. + + :param list_id: Retrieve a specific list ID (default: retrieve all). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonListSchema(many=True) + """ + return self._run( + 'lists' + (f'/{list_id}' if list_id else ''), + version='v1', + method='get', + schema=MastodonListSchema(many=list_id is None), + **kwargs + ) + + @action + def create_list(self, title: str, replies_policy: str = 'list', **kwargs) -> dict: + """ + Create a new list. + + It requires the specified API token to have the ``write:lists`` permission. + + :param title: List title. + :param replies_policy: Possible values: ``none``, ``following`` or ``list``. Default: ``list``. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonListSchema + """ + return self._run( + 'lists', + version='v1', + method='post', + schema=MastodonListSchema(), + data={'title': title, 'replies_policy': replies_policy}, + **kwargs + ) + + @action + def update_list( + self, list_id: int, title: Optional[str], replies_policy: Optional[str] = None, **kwargs + ) -> dict: + """ + Update a list. + + It requires the specified API token to have the ``write:lists`` permission. + + :param list_id: List ID. + :param title: New list title. + :param replies_policy: New replies policy. + Possible values: ``none``, ``following`` or ``list``. Default: ``list``. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonListSchema + """ + return self._run( + f'lists/{list_id}', + version='v1', + method='put', + schema=MastodonListSchema(), + data={ + **({'title': title} if title else {}), + **({'replies_policy': replies_policy} if replies_policy else {}), + }, + **kwargs + ) + + @action + def delete_list(self, list_id: int, **kwargs): + """ + Delete a list. + + It requires the specified API token to have the ``write:lists`` permission. + + :param list_id: List ID. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + f'lists/{list_id}', + version='v1', + method='delete', + **kwargs + ) + + @action + def get_list_accounts( + self, list_id: Optional[int] = None, min_id: Optional[str] = None, + max_id: Optional[str] = None, limit: int = 20, **kwargs + ) -> Iterable[dict]: + """ + Get the accounts in a list. + + It requires the specified API token to have the ``read:lists`` permission. + + :param list_id: List ID. + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonAccountSchema(many=True) + """ + return self._run( + f'lists/{list_id}/accounts', + version='v1', + method='get', + schema=MastodonAccountSchema(many=True), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, + **kwargs + ) + + @action + def add_accounts_to_list(self, list_id: int, account_ids: Sequence[str], **kwargs): + """ + Add accounts to a list. + + It requires the specified API token to have the ``write:lists`` permission. + + :param list_id: List ID. + :param account_ids: Accounts that should be added. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonListSchema + """ + return self._run( + f'lists/{list_id}/accounts', + version='v1', + method='post', + data={'account_ids': account_ids}, + **kwargs + ) + + @action + def remove_accounts_from_list(self, list_id: int, account_ids: Sequence[str], **kwargs): + """ + Remove accounts from a list + + It requires the specified API token to have the ``write:lists`` permission. + + :param list_id: List ID. + :param account_ids: Accounts that should be removed. + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonListSchema + """ + return self._run( + f'lists/{list_id}/accounts', + version='v1', + method='delete', + data={'account_ids': account_ids}, + **kwargs + ) + + @action + def get_notifications( + self, notification_id: Optional[str] = None, min_id: Optional[str] = None, + max_id: Optional[str] = None, limit: int = 20, **kwargs + ) -> Union[dict, Iterable[dict]]: + """ + Get the list of notifications of the user. + + It requires the specified API token to have the ``read:notifications`` permission. + + :param notification_id: If specified then retrieve only the notification associated to this ID. + :param min_id: Return results newer than this ID. + :param max_id: Return results older than this ID. + :param limit: Maximum number of results (default: 20). + :param kwargs: ``base_url``/``access_token`` override. + :return: .. schema:: mastodon.MastodonNotificationSchema(many=True) + """ + rs = self._run( + 'notifications' + (f'/{notification_id}' if notification_id else ''), + version='v1', + method='get', + schema=MastodonNotificationSchema(many=notification_id is None), + params={ + **({'min_id': min_id} if min_id else {}), + **({'max_id': max_id} if max_id else {}), + **({'limit': limit} if limit else {}), + }, + **kwargs + ) + + return rs + + @action + def dismiss_notifications(self, notification_id: Optional[str] = None, **kwargs): + """ + Dismiss notifications. + + It requires the specified API token to have the ``write:notifications`` permission. + + :param notification_id: Dismiss only this notification. + :param kwargs: ``base_url``/``access_token`` override. + """ + return self._run( + 'notifications/' + ( + f'{notification_id}/dismiss' if notification_id else 'clear' + ), + version='v1', + method='post', + **kwargs + ) + + +# vim:sw=4:ts=4:et: diff --git a/platypush/plugins/mastodon/manifest.yaml b/platypush/plugins/mastodon/manifest.yaml new file mode 100644 index 000000000..014ab66eb --- /dev/null +++ b/platypush/plugins/mastodon/manifest.yaml @@ -0,0 +1,4 @@ +manifest: + events: {} + package: platypush.plugins.mastodon + type: plugin diff --git a/platypush/schemas/mastodon.py b/platypush/schemas/mastodon.py new file mode 100644 index 000000000..bd87c851a --- /dev/null +++ b/platypush/schemas/mastodon.py @@ -0,0 +1,201 @@ +from random import randint + +from marshmallow import fields, missing +from marshmallow.schema import Schema +from marshmallow.validate import OneOf + +from platypush.schemas import DateTime, Date, StrippedString + +notification_types = ['follow', 'favourite', 'reblog', 'mention', 'poll', 'follow_request'] +list_reply_policies = ['none', 'followed', 'list'] + + +class MastodonSchema(Schema): + pass + + +class MastodonAccountSchema(MastodonSchema): + id = fields.String( + dump_only=True, + metadata=dict( + example=''.join([f'{randint(1, 9)}' for _ in range(18)]), + ) + ) + + username = fields.String( + metadata=dict( + example='admin', + ) + ) + + url = fields.URL() + avatar = fields.URL() + header = fields.URL() + followers_count = fields.Int(dump_only=True) + following_count = fields.Int(dump_only=True) + note = fields.String() + display_name = StrippedString( + metadata=dict( + example='Name Surname', + ) + ) + + locked = fields.Boolean() + bot = fields.Boolean() + discoverable = fields.Boolean() + group = fields.Boolean() + created_at = DateTime(dump_only=True) + last_status_at = DateTime(dump_only=True) + + +class MastodonFeaturedHashtagSchema(MastodonSchema): + id = fields.Int(dump_only=True) + name = fields.String() + statuses_count = fields.Int(dump_only=True) + last_status = DateTime(dump_only=True) + + +class MastodonHashtagHistorySchema(MastodonSchema): + day = Date() + uses = fields.Int() + accounts = fields.Int() + + +class MastodonHashtagSchema(MastodonSchema): + name = fields.String(metadata=dict(example='hashtag')) + url = fields.URL() + history = fields.Nested( + MastodonHashtagHistorySchema, many=True, default=missing + ) + + +class MastodonMediaSchema(MastodonSchema): + id = fields.String(dump_only=True) + description = StrippedString() + type = fields.String(dump_only=True, metadata={'example': 'image'}) + url = fields.URL(dump_only=True) + preview_url = fields.URL(dump_only=True) + remote_url = fields.URL(dump_only=True) + preview_remote_url = fields.URL(dump_only=True) + meta = fields.Dict() + + +class MastodonStatusSchema(MastodonSchema): + id = fields.String( + dump_only=True, + metadata=dict( + example=''.join([f'{randint(1, 9)}' for _ in range(18)]), + ) + ) + + in_reply_to_id = fields.String( + dump_only=True, + allow_none=True, + metadata=dict( + example=''.join([f'{randint(1, 9)}' for _ in range(18)]), + ) + ) + + in_reply_to_account_id = fields.String( + dump_only=True, + allow_none=True, + metadata=dict( + example=''.join([f'{randint(1, 9)}' for _ in range(18)]), + ) + ) + + url = fields.URL(dump_only=True) + content = fields.String(allow_none=False) + account = fields.Nested(MastodonAccountSchema, dump_only=True) + attachments = fields.Nested( + MastodonMediaSchema, + many=True, + dump_only=True, + attribute='media_attachments', + ) + hashtags = fields.Nested( + MastodonHashtagSchema, many=True, + attribute='tags', dump_only=True + ) + + replies_count = fields.Int(dump_only=True) + reblogs_count = fields.Int(dump_only=True) + favourites_count = fields.Int(dump_only=True) + + sensitive = fields.Boolean() + favourited = fields.Boolean() + reblogged = fields.Boolean() + muted = fields.Boolean() + bookmarked = fields.Boolean() + pinned = fields.Boolean() + + created_at = DateTime(dump_only=True) + + +class MastodonSearchSchema(MastodonSchema): + accounts = fields.Nested(MastodonAccountSchema, many=True) + statuses = fields.Nested(MastodonStatusSchema, many=True) + hashtags = fields.Nested(MastodonHashtagSchema, many=True) + + +class MastodonAccountCreationSchema(MastodonSchema): + access_token = fields.String(dump_only=True) + token_type = fields.String(dump_only=True, metadata={'example': 'Bearer'}) + scope = fields.String(dump_only=True, metadata={'example': 'read write follow push'}) + created_at = DateTime(dump_only=True) + + +class MastodonAccountListSchema(MastodonSchema): + id = fields.Int(dump_only=True) + title = StrippedString() + + +class MastodonFilterSchema(MastodonSchema): + id = fields.Int(dump_only=True) + phrase = StrippedString() + whole_word = fields.Boolean() + irreversible = fields.Boolean() + expires_at = DateTime(allow_none=True) + context = fields.List( + fields.String(validate=OneOf(['home', 'notifications', 'public', 'thread'])), + metadata={ + 'example': 'Which context(s) this filter applies to. ' + 'Possible values: home, notifications, public, thread', + } + ) + + +class MastodonConversationSchema(MastodonSchema): + id = fields.Int(dump_only=True) + unread = fields.Boolean() + accounts = fields.Nested(MastodonAccountSchema, many=True) + last_status = fields.Nested(MastodonStatusSchema) + + +class MastodonListSchema(MastodonSchema): + id = fields.Int(dump_only=True) + title = StrippedString() + replies_policy = fields.String(validate=OneOf(list_reply_policies)) + + +class MastodonMentionSchema(MastodonSchema): + id = fields.Int(dump_only=True) + username = StrippedString(metadata=dict(example='user')) + url = fields.URL(metadata=dict(example='https://mastodon.social/@user')) + + +class MastodonNotificationSchema(MastodonSchema): + id = fields.String(dump_only=True) + type = fields.String(validate=OneOf(notification_types)) + account = fields.Nested(MastodonAccountSchema) + status = fields.Nested(MastodonStatusSchema) + mention = fields.Nested(MastodonMentionSchema) + created_at = DateTime(dump_only=True) + + +class MastodonSubscriptionNotificationTypes(MastodonSchema): + follow = fields.Boolean() + reblog = fields.Boolean() + mention = fields.Boolean() + favourite = fields.Boolean() + poll = fields.Boolean()