From e483d160145b1086629c1ba5a2964c5550c5c126 Mon Sep 17 00:00:00 2001 From: halcy Date: Thu, 1 Dec 2022 01:39:08 +0200 Subject: Finish moving code out into smaller files --- mastodon/Mastodon.py | 2852 +-------------------------------------- mastodon/accounts.py | 395 +++++- mastodon/admin.py | 572 ++++++++ mastodon/authentication.py | 22 +- mastodon/conversations.py | 43 + mastodon/endorsements.py | 19 + mastodon/error.py | 90 -- mastodon/errors.py | 90 ++ mastodon/favourites.py | 52 + mastodon/filters.py | 119 ++ mastodon/hashtags.py | 52 + mastodon/instance.py | 97 +- mastodon/internals.py | 4 +- mastodon/lists.py | 112 ++ mastodon/media.py | 114 ++ mastodon/notifications.py | 92 ++ mastodon/polls.py | 61 + mastodon/preferences.py | 71 + mastodon/push.py | 201 +++ mastodon/relationships.py | 136 ++ mastodon/reports.py | 62 + mastodon/search.py | 91 ++ mastodon/statuses.py | 424 +++++- mastodon/streaming_endpoints.py | 75 + mastodon/suggestions.py | 32 + mastodon/timeline.py | 22 +- mastodon/trends.py | 72 + mastodon/utility.py | 116 +- 28 files changed, 3139 insertions(+), 2949 deletions(-) create mode 100644 mastodon/admin.py create mode 100644 mastodon/conversations.py create mode 100644 mastodon/endorsements.py delete mode 100644 mastodon/error.py create mode 100644 mastodon/errors.py create mode 100644 mastodon/favourites.py create mode 100644 mastodon/filters.py create mode 100644 mastodon/hashtags.py create mode 100644 mastodon/lists.py create mode 100644 mastodon/media.py create mode 100644 mastodon/notifications.py create mode 100644 mastodon/polls.py create mode 100644 mastodon/preferences.py create mode 100644 mastodon/push.py create mode 100644 mastodon/relationships.py create mode 100644 mastodon/reports.py create mode 100644 mastodon/search.py create mode 100644 mastodon/streaming_endpoints.py create mode 100644 mastodon/suggestions.py create mode 100644 mastodon/trends.py (limited to 'mastodon') diff --git a/mastodon/Mastodon.py b/mastodon/Mastodon.py index 0ded1cf..23e4d9e 100644 --- a/mastodon/Mastodon.py +++ b/mastodon/Mastodon.py @@ -26,7 +26,7 @@ from .utility import parse_version_string, max_version, api_version from .utility import AttribAccessDict, AttribAccessDict from .utility import Mastodon as Utility -from .error import * +from .errors import * from .versions import _DICT_VERSION_APPLICATION, _DICT_VERSION_MENTION, _DICT_VERSION_MEDIA, _DICT_VERSION_ACCOUNT, _DICT_VERSION_POLL, \ _DICT_VERSION_STATUS, _DICT_VERSION_INSTANCE, _DICT_VERSION_HASHTAG, _DICT_VERSION_EMOJI, _DICT_VERSION_RELATIONSHIP, \ _DICT_VERSION_NOTIFICATION, _DICT_VERSION_CONTEXT, _DICT_VERSION_LIST, _DICT_VERSION_CARD, _DICT_VERSION_SEARCHRESULT, \ @@ -45,11 +45,33 @@ from .accounts import Mastodon as Accounts from .instance import Mastodon as Instance from .timeline import Mastodon as Timeline from .statuses import Mastodon as Statuses +from .media import Mastodon as Media +from .polls import Mastodon as Polls +from .notifications import Mastodon as Notifications +from .conversations import Mastodon as Conversations +from .hashtags import Mastodon as Hashtags +from .filters import Mastodon as Filters +from .suggestions import Mastodon as Suggestions +from .endorsements import Mastodon as Endorsements +from .relationships import Mastodon as Relationships +from .lists import Mastodon as Lists +from .trends import Mastodon as Trends +from .search import Mastodon as Search +from .favourites import Mastodon as Favourites +from .reports import Mastodon as Reports +from .preferences import Mastodon as Preferences +from .push import Mastodon as Push +from .admin import Mastodon as Admin +from .streaming_endpoints import Mastodon as Streaming -## +### # The actual Mastodon class +# +# Almost all code is now imported from smaller files to make editing a bit more pleasant ### -class Mastodon(Utility, Authentication, Accounts, Instance, Timeline, Statuses): +class Mastodon(Utility, Authentication, Accounts, Instance, Timeline, Statuses, Polls, Notifications, Hashtags, + Filters, Suggestions, Endorsements, Relationships, Lists, Trends, Search, Favourites, Reports, + Preferences, Push, Admin, Conversations, Media, Streaming): """ Thorough and easy to use Mastodon API wrapper in Python. @@ -65,2827 +87,3 @@ class Mastodon(Utility, Authentication, Accounts, Instance, Timeline, Statuses): Retrieve the maximum version of Mastodon supported by this version of Mastodon.py """ return Mastodon.__SUPPORTED_MASTODON_VERSION - - ### - # Reading data: Polls - ### - @api_version("2.8.0", "2.8.0", _DICT_VERSION_POLL) - def poll(self, id): - """ - Fetch information about the poll with the given id - - Returns a :ref:`poll dict `. - """ - id = self.__unpack_id(id) - url = '/api/v1/polls/{0}'.format(str(id)) - return self.__api_request('GET', url) - - ### - # Reading data: Notifications - ### - @api_version("1.0.0", "3.5.0", _DICT_VERSION_NOTIFICATION) - def notifications(self, id=None, account_id=None, max_id=None, min_id=None, since_id=None, limit=None, exclude_types=None, types=None, mentions_only=None): - """ - Fetch notifications (mentions, favourites, reblogs, follows) for the logged-in - user. Pass `account_id` to get only notifications originating from the given account. - - There are different types of notifications: - * `follow` - A user followed the logged in user - * `follow_request` - A user has requested to follow the logged in user (for locked accounts) - * `favourite` - A user favourited a post by the logged in user - * `reblog` - A user reblogged a post by the logged in user - * `mention` - A user mentioned the logged in user - * `poll` - A poll the logged in user created or voted in has ended - * `update` - A status the logged in user has reblogged (and only those, as of 4.0.0) has been edited - * `status` - A user that the logged in user has enabned notifications for has enabled `notify` (see :ref:`account_follow() `) - * `admin.sign_up` - For accounts with appropriate permissions (TODO: document which those are when adding the permission API): A new user has signed up - * `admin.report` - For accounts with appropriate permissions (TODO: document which those are when adding the permission API): A new report has been received - Parameters `exclude_types` and `types` are array of these types, specifying them will in- or exclude the - types of notifications given. It is legal to give both parameters at the same tine, the result will then - be the intersection of the results of both filters. Specifying `mentions_only` is a deprecated way to set - `exclude_types` to all but mentions. - - Can be passed an `id` to fetch a single notification. - - Returns a list of :ref:`notification dicts `. - """ - if mentions_only is not None: - if exclude_types is None and types is None: - if mentions_only: - if self.verify_minimum_version("3.5.0", cached=True): - types = ["mention"] - else: - exclude_types = ["follow", "favourite", "reblog", "poll", "follow_request"] - else: - raise MastodonIllegalArgumentError('Cannot specify exclude_types/types when mentions_only is present') - del mentions_only - - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - if account_id is not None: - account_id = self.__unpack_id(account_id) - - if id is None: - params = self.__generate_params(locals(), ['id']) - return self.__api_request('GET', '/api/v1/notifications', params) - else: - id = self.__unpack_id(id) - url = '/api/v1/notifications/{0}'.format(str(id)) - return self.__api_request('GET', url) - - ### - # Reading data: Accounts - ### - @api_version("1.0.0", "1.0.0", _DICT_VERSION_ACCOUNT) - def account(self, id): - """ - Fetch account information by user `id`. - - Does not require authentication for publicly visible accounts. - - Returns a :ref:`account dict `. - """ - id = self.__unpack_id(id) - url = '/api/v1/accounts/{0}'.format(str(id)) - return self.__api_request('GET', url) - - @api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT) - def account_verify_credentials(self): - """ - Fetch logged-in user's account information. - - Returns a :ref:`account dict ` (Starting from 2.1.0, with an additional "source" field). - """ - return self.__api_request('GET', '/api/v1/accounts/verify_credentials') - - @api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT) - def me(self): - """ - Get this user's account. Synonym for `account_verify_credentials()`, does exactly - the same thing, just exists becase `account_verify_credentials()` has a confusing - name. - """ - return self.account_verify_credentials() - - @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) - def account_statuses(self, id, only_media=False, pinned=False, exclude_replies=False, exclude_reblogs=False, tagged=None, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch statuses by user `id`. Same options as :ref:`timeline() ` are permitted. - Returned toots are from the perspective of the logged-in user, i.e. - all statuses visible to the logged-in user (including DMs) are - included. - - If `only_media` is set, return only statuses with media attachments. - If `pinned` is set, return only statuses that have been pinned. Note that - as of Mastodon 2.1.0, this only works properly for instance-local users. - If `exclude_replies` is set, filter out all statuses that are replies. - If `exclude_reblogs` is set, filter out all statuses that are reblogs. - If `tagged` is set, return only statuses that are tagged with `tagged`. Only a single tag without a '#' is valid. - - Does not require authentication for Mastodon versions after 2.7.0 (returns - publicly visible statuses in that case), for publicly visible accounts. - - Returns a list of :ref:`status dicts `. - """ - id = self.__unpack_id(id) - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals(), ['id']) - if not pinned: - del params["pinned"] - if not only_media: - del params["only_media"] - if not exclude_replies: - del params["exclude_replies"] - if not exclude_reblogs: - del params["exclude_reblogs"] - - url = '/api/v1/accounts/{0}/statuses'.format(str(id)) - return self.__api_request('GET', url, params) - - @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) - def account_following(self, id, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch users the given user is following. - - Returns a list of :ref:`account dicts `. - """ - id = self.__unpack_id(id) - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals(), ['id']) - url = '/api/v1/accounts/{0}/following'.format(str(id)) - return self.__api_request('GET', url, params) - - @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) - def account_followers(self, id, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch users the given user is followed by. - - Returns a list of :ref:`account dicts `. - """ - id = self.__unpack_id(id) - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals(), ['id']) - url = '/api/v1/accounts/{0}/followers'.format(str(id)) - return self.__api_request('GET', url, params) - - @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) - def account_relationships(self, id): - """ - Fetch relationship (following, followed_by, blocking, follow requested) of - the logged in user to a given account. `id` can be a list. - - Returns a list of :ref:`relationship dicts `. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/accounts/relationships', - params) - - @api_version("1.0.0", "2.3.0", _DICT_VERSION_ACCOUNT) - def account_search(self, q, limit=None, following=False): - """ - Fetch matching accounts. Will lookup an account remotely if the search term is - in the username@domain format and not yet in the database. Set `following` to - True to limit the search to users the logged-in user follows. - - Returns a list of :ref:`account dicts `. - """ - params = self.__generate_params(locals()) - - if params["following"] == False: - del params["following"] - - return self.__api_request('GET', '/api/v1/accounts/search', params) - - @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) - def account_lists(self, id): - """ - Get all of the logged-in user's lists which the specified user is - a member of. - - Returns a list of :ref:`list dicts `. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ['id']) - url = '/api/v1/accounts/{0}/lists'.format(str(id)) - return self.__api_request('GET', url, params) - - @api_version("3.4.0", "3.4.0", _DICT_VERSION_ACCOUNT) - def account_lookup(self, acct): - """ - Look up an account from user@instance form (@instance allowed but not required for - local accounts). Will only return accounts that the instance already knows about, - and not do any webfinger requests. Use `account_search` if you need to resolve users - through webfinger from remote. - - Returns an :ref:`account dict `. - """ - return self.__api_request('GET', '/api/v1/accounts/lookup', self.__generate_params(locals())) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_FAMILIAR_FOLLOWERS) - def account_familiar_followers(self, id): - """ - Find followers for the account given by id (can be a list) that also follow the - logged in account. - - Returns a list of :ref:`familiar follower dicts ` - """ - if not isinstance(id, list): - id = [id] - for i in range(len(id)): - id[i] = self.__unpack_id(id[i]) - return self.__api_request('GET', '/api/v1/accounts/familiar_followers', {'id': id}, use_json=True) - - ### - # Reading data: Featured hashtags - ### - @api_version("3.0.0", "3.0.0", _DICT_VERSION_FEATURED_TAG) - def featured_tags(self): - """ - Return the hashtags the logged-in user has set to be featured on - their profile as a list of :ref:`featured tag dicts `. - - Returns a list of :ref:`featured tag dicts `. - """ - return self.__api_request('GET', '/api/v1/featured_tags') - - @api_version("3.0.0", "3.0.0", _DICT_VERSION_HASHTAG) - def featured_tag_suggestions(self): - """ - Returns the logged-in user's 10 most commonly-used hashtags. - - Returns a list of :ref:`hashtag dicts `. - """ - return self.__api_request('GET', '/api/v1/featured_tags/suggestions') - - ### - # Reading data: Keyword filters - ### - @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) - def filters(self): - """ - Fetch all of the logged-in user's filters. - - Returns a list of :ref:`filter dicts `. Not paginated. - """ - return self.__api_request('GET', '/api/v1/filters') - - @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) - def filter(self, id): - """ - Fetches information about the filter with the specified `id`. - - Returns a :ref:`filter dict `. - """ - id = self.__unpack_id(id) - url = '/api/v1/filters/{0}'.format(str(id)) - return self.__api_request('GET', url) - - @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) - def filters_apply(self, objects, filters, context): - """ - Helper function: Applies a list of filters to a list of either statuses - or notifications and returns only those matched by none. This function will - apply all filters that match the context provided in `context`, i.e. - if you want to apply only notification-relevant filters, specify - 'notifications'. Valid contexts are 'home', 'notifications', 'public' and 'thread'. - """ - - # Build filter regex - filter_strings = [] - for keyword_filter in filters: - if not context in keyword_filter["context"]: - continue - - filter_string = re.escape(keyword_filter["phrase"]) - if keyword_filter["whole_word"]: - filter_string = "\\b" + filter_string + "\\b" - filter_strings.append(filter_string) - filter_re = re.compile("|".join(filter_strings), flags=re.IGNORECASE) - - # Apply - filter_results = [] - for filter_object in objects: - filter_status = filter_object - if "status" in filter_object: - filter_status = filter_object["status"] - filter_text = filter_status["content"] - filter_text = re.sub(r"<.*?>", " ", filter_text) - filter_text = re.sub(r"\s+", " ", filter_text).strip() - if not filter_re.search(filter_text): - filter_results.append(filter_object) - return filter_results - - ### - # Reading data: Follow suggestions - ### - @api_version("2.4.3", "2.4.3", _DICT_VERSION_ACCOUNT) - def suggestions(self): - """ - Fetch follow suggestions for the logged-in user. - - Returns a list of :ref:`account dicts `. - - """ - return self.__api_request('GET', '/api/v1/suggestions') - - ### - # Reading data: Follow suggestions - ### - @api_version("3.0.0", "3.0.0", _DICT_VERSION_ACCOUNT) - def directory(self, offset=None, limit=None, order=None, local=None): - """ - Fetch the contents of the profile directory, if enabled on the server. - - `offset` how many accounts to skip before returning results. Default 0. - - `limit` how many accounts to load. Default 40. - - `order` "active" to sort by most recently posted statuses (default) or - "new" to sort by most recently created profiles. - - `local` True to return only local accounts. - - Returns a list of :ref:`account dicts `. - - """ - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/directory', params) - - ### - # Reading data: Endorsements - ### - @api_version("2.5.0", "2.5.0", _DICT_VERSION_ACCOUNT) - def endorsements(self): - """ - Fetch list of users endorsed by the logged-in user. - - Returns a list of :ref:`account dicts `. - - """ - return self.__api_request('GET', '/api/v1/endorsements') - - ### - # Reading data: Searching - ### - - def __ensure_search_params_acceptable(self, account_id, offset, min_id, max_id): - """ - Internal Helper: Throw a MastodonVersionError if version is < 2.8.0 but parameters - for search that are available only starting with 2.8.0 are specified. - """ - if any(item is not None for item in (account_id, offset, min_id, max_id)): - if not self.verify_minimum_version("2.8.0", cached=True): - raise MastodonVersionError("Advanced search parameters require Mastodon 2.8.0+") - - @api_version("1.1.0", "2.8.0", _DICT_VERSION_SEARCHRESULT) - def search(self, q, resolve=True, result_type=None, account_id=None, offset=None, min_id=None, max_id=None, exclude_unreviewed=True): - """ - Fetch matching hashtags, accounts and statuses. Will perform webfinger - lookups if resolve is True. Full-text search is only enabled if - the instance supports it, and is restricted to statuses the logged-in - user wrote or was mentioned in. - - `result_type` can be one of "accounts", "hashtags" or "statuses", to only - search for that type of object. - - Specify `account_id` to only get results from the account with that id. - - `offset`, `min_id` and `max_id` can be used to paginate. - - `exclude_unreviewed` can be used to restrict search results for hashtags to only - those that have been reviewed by moderators. It is on by default. When using the - v1 search API (pre 2.4.1), it is ignored. - - Will use search_v1 (no tag dicts in return values) on Mastodon versions before - 2.4.1), search_v2 otherwise. Parameters other than resolve are only available - on Mastodon 2.8.0 or above - this function will throw a MastodonVersionError - if you try to use them on versions before that. Note that the cached version - number will be used for this to avoid uneccesary requests. - - Returns a :ref:`search result dict `, with tags as `hashtag dicts`_. - """ - if self.verify_minimum_version("2.4.1", cached=True): - return self.search_v2(q, resolve=resolve, result_type=result_type, account_id=account_id, offset=offset, min_id=min_id, max_id=max_id, exclude_unreviewed=exclude_unreviewed) - else: - self.__ensure_search_params_acceptable( - account_id, offset, min_id, max_id) - return self.search_v1(q, resolve=resolve) - - @api_version("1.1.0", "2.1.0", "2.1.0") - def search_v1(self, q, resolve=False): - """ - Identical to `search_v2()`, except in that it does not return - tags as :ref:`hashtag dicts `. - - Returns a :ref:`search result dict `. - """ - params = self.__generate_params(locals()) - if not resolve: - del params['resolve'] - return self.__api_request('GET', '/api/v1/search', params) - - @api_version("2.4.1", "2.8.0", _DICT_VERSION_SEARCHRESULT) - def search_v2(self, q, resolve=True, result_type=None, account_id=None, offset=None, min_id=None, max_id=None, exclude_unreviewed=True): - """ - Identical to `search_v1()`, except in that it returns tags as - :ref:`hashtag dicts `, has more parameters, and resolves by default. - - For more details documentation, please see `search()` - - Returns a :ref:`search result dict `. - """ - self.__ensure_search_params_acceptable( - account_id, offset, min_id, max_id) - params = self.__generate_params(locals()) - - if not resolve: - del params["resolve"] - - if not exclude_unreviewed or not self.verify_minimum_version("3.0.0", cached=True): - del params["exclude_unreviewed"] - - if "result_type" in params: - params["type"] = params["result_type"] - del params["result_type"] - - return self.__api_request('GET', '/api/v2/search', params) - - ### - # Reading data: Trends - ### - @api_version("2.4.3", "3.5.0", _DICT_VERSION_HASHTAG) - def trends(self, limit=None): - """ - Alias for :ref:`trending_tags() ` - """ - return self.trending_tags(limit=limit) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_HASHTAG) - def trending_tags(self, limit=None, lang=None): - """ - Fetch trending-hashtag information, if the instance provides such information. - - Specify `limit` to limit how many results are returned (the maximum number - of results is 10, the endpoint is not paginated). - - Does not require authentication unless locked down by the administrator. - - Important versioning note: This endpoint does not exist for Mastodon versions - between 2.8.0 (inclusive) and 3.0.0 (exclusive). - - Pass `lang` to override the global locale parameter, which may affect trend ordering. - - Returns a list of :ref:`hashtag dicts `, sorted by the instance's trending algorithm, - descending. - """ - params = self.__generate_params(locals()) - if self.verify_minimum_version("3.5.0", cached=True): - # Starting 3.5.0, old version is deprecated - return self.__api_request('GET', '/api/v1/trends/tags', params) - else: - return self.__api_request('GET', '/api/v1/trends', params) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS) - def trending_statuses(self): - """ - Fetch trending-status information, if the instance provides such information. - - Specify `limit` to limit how many results are returned (the maximum number - of results is 10, the endpoint is not paginated). - - Pass `lang` to override the global locale parameter, which may affect trend ordering. - - Returns a list of :ref:`status dicts `, sorted by the instances's trending algorithm, - descending. - """ - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/trends/statuses', params) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_CARD) - def trending_links(self): - """ - Fetch trending-link information, if the instance provides such information. - - Specify `limit` to limit how many results are returned (the maximum number - of results is 10, the endpoint is not paginated). - - Returns a list of :ref:`card dicts `, sorted by the instances's trending algorithm, - descending. - """ - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/trends/links', params) - - ### - # Reading data: Lists - ### - @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) - def lists(self): - """ - Fetch a list of all the Lists by the logged-in user. - - Returns a list of :ref:`list dicts `. - """ - return self.__api_request('GET', '/api/v1/lists') - - @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) - def list(self, id): - """ - Fetch info about a specific list. - - Returns a :ref:`list dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('GET', '/api/v1/lists/{0}'.format(id)) - - @api_version("2.1.0", "2.6.0", _DICT_VERSION_ACCOUNT) - def list_accounts(self, id, max_id=None, min_id=None, since_id=None, limit=None): - """ - Get the accounts that are on the given list. - - Returns a list of :ref:`account dicts `. - """ - id = self.__unpack_id(id) - - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals(), ['id']) - return self.__api_request('GET', '/api/v1/lists/{0}/accounts'.format(id)) - - ### - # Reading data: Mutes and Blocks - ### - @api_version("1.1.0", "2.6.0", _DICT_VERSION_ACCOUNT) - def mutes(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch a list of users muted by the logged-in user. - - Returns a list of :ref:`account dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/mutes', params) - - @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) - def blocks(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch a list of users blocked by the logged-in user. - - Returns a list of :ref:`account dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/blocks', params) - - ### - # Reading data: Reports - ### - @api_version("1.1.0", "1.1.0", _DICT_VERSION_REPORT) - def reports(self): - """ - Fetch a list of reports made by the logged-in user. - - Returns a list of :ref:`report dicts `. - - Warning: This method has now finally been removed, and will not - work on Mastodon versions 2.5.0 and above. - """ - if self.verify_minimum_version("2.5.0", cached=True): - raise MastodonVersionError("API removed in Mastodon 2.5.0") - return self.__api_request('GET', '/api/v1/reports') - - ### - # Reading data: Favourites - ### - @api_version("1.0.0", "2.6.0", _DICT_VERSION_STATUS) - def favourites(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch the logged-in user's favourited statuses. - - Returns a list of :ref:`status dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/favourites', params) - - ### - # Reading data: Follow requests - ### - @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) - def follow_requests(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch the logged-in user's incoming follow requests. - - Returns a list of :ref:`account dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/follow_requests', params) - - ### - # Reading data: Domain blocks - ### - @api_version("1.4.0", "2.6.0", "1.4.0") - def domain_blocks(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetch the logged-in user's blocked domains. - - Returns a list of blocked domain URLs (as strings, without protocol specifier). - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/domain_blocks', params) - - ### - # Reading data: Emoji - ### - @api_version("2.1.0", "2.1.0", _DICT_VERSION_EMOJI) - def custom_emojis(self): - """ - Fetch the list of custom emoji the instance has installed. - - Does not require authentication unless locked down by the administrator. - - Returns a list of :ref:`emoji dicts `. - """ - return self.__api_request('GET', '/api/v1/custom_emojis') - - ### - # Reading data: Apps - ### - @api_version("2.0.0", "2.7.2", _DICT_VERSION_APPLICATION) - def app_verify_credentials(self): - """ - Fetch information about the current application. - - Returns an :ref:`application dict `. - """ - return self.__api_request('GET', '/api/v1/apps/verify_credentials') - - ### - # Reading data: Webpush subscriptions - ### - @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH) - def push_subscription(self): - """ - Fetch the current push subscription the logged-in user has for this app. - - Returns a :ref:`push subscription dict `. - """ - return self.__api_request('GET', '/api/v1/push/subscription') - - ### - # Reading data: Preferences - ### - @api_version("2.8.0", "2.8.0", _DICT_VERSION_PREFERENCES) - def preferences(self): - """ - Fetch the user's preferences, which can be used to set some default options. - As of 2.8.0, apps can only fetch, not update preferences. - - Returns a :ref:`preference dict `. - """ - return self.__api_request('GET', '/api/v1/preferences') - - ## - # Reading data: Announcements - ## - - # /api/v1/announcements - @api_version("3.1.0", "3.1.0", _DICT_VERSION_ANNOUNCEMENT) - def announcements(self): - """ - Fetch currently active announcements. - - Returns a list of :ref:`announcement dicts `. - """ - return self.__api_request('GET', '/api/v1/announcements') - - ## - # Reading data: Read markers - ## - @api_version("3.0.0", "3.0.0", _DICT_VERSION_MARKER) - def markers_get(self, timeline=["home"]): - """ - Get the last-read-location markers for the specified timelines. Valid timelines - are the same as in :ref:`timeline() ` - - Note that despite the singular name, `timeline` can be a list. - - Returns a dict of :ref:`read marker dicts `, keyed by timeline name. - """ - if not isinstance(timeline, (list, tuple)): - timeline = [timeline] - params = self.__generate_params(locals()) - - return self.__api_request('GET', '/api/v1/markers', params) - - ### - # Reading data: Bookmarks - ### - @api_version("3.1.0", "3.1.0", _DICT_VERSION_STATUS) - def bookmarks(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Get a list of statuses bookmarked by the logged-in user. - - Returns a list of :ref:`status dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/bookmarks', params) - - ### - # Writing data: Statuses - ### - def __status_internal(self, status, in_reply_to_id=None, media_ids=None, - sensitive=False, visibility=None, spoiler_text=None, - language=None, idempotency_key=None, content_type=None, - scheduled_at=None, poll=None, quote_id=None, edit=False): - if quote_id is not None: - if self.feature_set != "fedibird": - raise MastodonIllegalArgumentError('quote_id is only available with feature set fedibird') - quote_id = self.__unpack_id(quote_id) - - if content_type is not None: - if self.feature_set != "pleroma": - raise MastodonIllegalArgumentError('content_type is only available with feature set pleroma') - # It would be better to read this from nodeinfo and cache, but this is easier - if not content_type in ["text/plain", "text/html", "text/markdown", "text/bbcode"]: - raise MastodonIllegalArgumentError('Invalid content type specified') - - if in_reply_to_id is not None: - in_reply_to_id = self.__unpack_id(in_reply_to_id) - - if scheduled_at is not None: - scheduled_at = self.__consistent_isoformat_utc(scheduled_at) - - params_initial = locals() - - # Validate poll/media exclusivity - if poll is not None: - if media_ids is not None and len(media_ids) != 0: - raise ValueError( - 'Status can have media or poll attached - not both.') - - # Validate visibility parameter - valid_visibilities = ['private', 'public', 'unlisted', 'direct'] - if params_initial['visibility'] is None: - del params_initial['visibility'] - else: - params_initial['visibility'] = params_initial['visibility'].lower() - if params_initial['visibility'] not in valid_visibilities: - raise ValueError('Invalid visibility value! Acceptable values are %s' % valid_visibilities) - - if params_initial['language'] is None: - del params_initial['language'] - - if params_initial['sensitive'] is False: - del [params_initial['sensitive']] - - headers = {} - if idempotency_key is not None: - headers['Idempotency-Key'] = idempotency_key - - if media_ids is not None: - try: - media_ids_proper = [] - if not isinstance(media_ids, (list, tuple)): - media_ids = [media_ids] - for media_id in media_ids: - media_ids_proper.append(self.__unpack_id(media_id)) - except Exception as e: - raise MastodonIllegalArgumentError("Invalid media dict: %s" % e) - - params_initial["media_ids"] = media_ids_proper - - if params_initial['content_type'] is None: - del params_initial['content_type'] - - use_json = False - if poll is not None: - use_json = True - - params = self.__generate_params(params_initial, ['idempotency_key', 'edit']) - if edit is None: - # Post - return self.__api_request('POST', '/api/v1/statuses', params, headers=headers, use_json=use_json) - else: - # Edit - return self.__api_request('PUT', '/api/v1/statuses/{0}'.format(str(self.__unpack_id(edit))), params, headers=headers, use_json=use_json) - - @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) - def status_post(self, status, in_reply_to_id=None, media_ids=None, - sensitive=False, visibility=None, spoiler_text=None, - language=None, idempotency_key=None, content_type=None, - scheduled_at=None, poll=None, quote_id=None): - """ - Post a status. Can optionally be in reply to another status and contain - media. - - `media_ids` should be a list. (If it's not, the function will turn it - into one.) It can contain up to four pieces of media (uploaded via - :ref:`media_post() `). `media_ids` can also be the `media dicts`_ returned - by :ref:`media_post() ` - they are unpacked automatically. - - The `sensitive` boolean decides whether or not media attached to the post - should be marked as sensitive, which hides it by default on the Mastodon - web front-end. - - The visibility parameter is a string value and accepts any of: - 'direct' - post will be visible only to mentioned users - 'private' - post will be visible only to followers - 'unlisted' - post will be public but not appear on the public timeline - 'public' - post will be public - - If not passed in, visibility defaults to match the current account's - default-privacy setting (starting with Mastodon version 1.6) or its - locked setting - private if the account is locked, public otherwise - (for Mastodon versions lower than 1.6). - - The `spoiler_text` parameter is a string to be shown as a warning before - the text of the status. If no text is passed in, no warning will be - displayed. - - Specify `language` to override automatic language detection. The parameter - accepts all valid ISO 639-1 (2-letter) or for languages where that do not - have one, 639-3 (three letter) language codes. - - You can set `idempotency_key` to a value to uniquely identify an attempt - at posting a status. Even if you call this function more than once, - if you call it with the same `idempotency_key`, only one status will - be created. - - Pass a datetime as `scheduled_at` to schedule the toot for a specific time - (the time must be at least 5 minutes into the future). If this is passed, - status_post returns a :ref:`scheduled status dict ` instead. - - Pass `poll` to attach a poll to the status. An appropriate object can be - constructed using :ref:`make_poll() ` . Note that as of Mastodon version - 2.8.2, you can only have either media or a poll attached, not both at - the same time. - - **Specific to "pleroma" feature set:**: Specify `content_type` to set - the content type of your post on Pleroma. It accepts 'text/plain' (default), - 'text/markdown', 'text/html' and 'text/bbcode'. This parameter is not - supported on Mastodon servers, but will be safely ignored if set. - - **Specific to "fedibird" feature set:**: The `quote_id` parameter is - a non-standard extension that specifies the id of a quoted status. - - Returns a :ref:`status dict ` with the new status. - """ - return self.__status_internal( - status, - in_reply_to_id, - media_ids, - sensitive, - visibility, - spoiler_text, - language, - idempotency_key, - content_type, - scheduled_at, - poll, - quote_id, - edit=None - ) - - @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) - def toot(self, status): - """ - Synonym for :ref:`status_post() ` that only takes the status text as input. - - Usage in production code is not recommended. - - Returns a :ref:`status dict ` with the new status. - """ - return self.status_post(status) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS) - def status_update(self, id, status = None, spoiler_text = None, sensitive = None, media_ids = None, poll = None): - """ - Edit a status. The meanings of the fields are largely the same as in :ref:`status_post() `, - though not every field can be edited. - - Note that editing a poll will reset the votes. - """ - return self.__status_internal( - status = status, - media_ids = media_ids, - sensitive = sensitive, - spoiler_text = spoiler_text, - poll = poll, - edit = id - ) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS_EDIT) - def status_history(self, id): - """ - Returns the edit history of a status as a list of :ref:`status edit dicts `, starting - from the original form. Note that this means that a status that has been edited - once will have *two* entries in this list, a status that has been edited twice - will have three, and so on. - """ - id = self.__unpack_id(id) - return self.__api_request('GET', "/api/v1/statuses/{0}/history".format(str(id))) - - def status_source(self, id): - """ - Returns the source of a status for editing. - - Return value is a dictionary containing exactly the parameters you could pass to - :ref:`status_update() ` to change nothing about the status, except `status` is `text` - instead. - """ - id = self.__unpack_id(id) - return self.__api_request('GET', "/api/v1/statuses/{0}/source".format(str(id))) - - @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) - def status_reply(self, to_status, status, in_reply_to_id=None, media_ids=None, - sensitive=False, visibility=None, spoiler_text=None, - language=None, idempotency_key=None, content_type=None, - scheduled_at=None, poll=None, untag=False): - """ - Helper function - acts like status_post, but prepends the name of all - the users that are being replied to to the status text and retains - CW and visibility if not explicitly overridden. - - Set `untag` to True if you want the reply to only go to the user you - are replying to, removing every other mentioned user from the - conversation. - """ - keyword_args = locals() - del keyword_args["self"] - del keyword_args["to_status"] - del keyword_args["untag"] - - user_id = self.__get_logged_in_id() - - # Determine users to mention - mentioned_accounts = collections.OrderedDict() - mentioned_accounts[to_status.account.id] = to_status.account.acct - - if not untag: - for account in to_status.mentions: - if account.id != user_id and not account.id in mentioned_accounts.keys(): - mentioned_accounts[account.id] = account.acct - - # Join into one piece of text. The space is added inside because of self-replies. - status = "".join(map(lambda x: "@" + x + " ", - mentioned_accounts.values())) + status - - # Retain visibility / cw - if visibility is None and 'visibility' in to_status: - visibility = to_status.visibility - if spoiler_text is None and 'spoiler_text' in to_status: - spoiler_text = to_status.spoiler_text - - keyword_args["status"] = status - keyword_args["visibility"] = visibility - keyword_args["spoiler_text"] = spoiler_text - keyword_args["in_reply_to_id"] = to_status.id - return self.status_post(**keyword_args) - - @api_version("2.8.0", "2.8.0", _DICT_VERSION_POLL) - def make_poll(self, options, expires_in, multiple=False, hide_totals=False): - """ - Generate a poll object that can be passed as the `poll` option when posting a status. - - options is an array of strings with the poll options (Maximum, by default: 4), - expires_in is the time in seconds for which the poll should be open. - Set multiple to True to allow people to choose more than one answer. Set - hide_totals to True to hide the results of the poll until it has expired. - """ - poll_params = locals() - del poll_params["self"] - return poll_params - - @api_version("1.0.0", "1.0.0", "1.0.0") - def status_delete(self, id): - """ - Delete a status - - Returns the now-deleted status, with an added "source" attribute that contains - the text that was used to compose this status (this can be used to power - "delete and redraft" functionality) - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}'.format(str(id)) - return self.__api_request('DELETE', url) - - @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) - def status_reblog(self, id, visibility=None): - """ - Reblog / boost a status. - - The visibility parameter functions the same as in :ref:`status_post() ` and - allows you to reduce the visibility of a reblogged status. - - Returns a :ref:`status dict ` with a new status that wraps around the reblogged one. - """ - params = self.__generate_params(locals(), ['id']) - valid_visibilities = ['private', 'public', 'unlisted', 'direct'] - if 'visibility' in params: - params['visibility'] = params['visibility'].lower() - if params['visibility'] not in valid_visibilities: - raise ValueError('Invalid visibility value! Acceptable ' - 'values are %s' % valid_visibilities) - - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/reblog'.format(str(id)) - return self.__api_request('POST', url, params) - - @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) - def status_unreblog(self, id): - """ - Un-reblog a status. - - Returns a :ref:`status dict ` with the status that used to be reblogged. - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/unreblog'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) - def status_favourite(self, id): - """ - Favourite a status. - - Returns a :ref:`status dict ` with the favourited status. - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/favourite'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) - def status_unfavourite(self, id): - """ - Un-favourite a status. - - Returns a :ref:`status dict ` with the un-favourited status. - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/unfavourite'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.4.0", "2.0.0", _DICT_VERSION_STATUS) - def status_mute(self, id): - """ - Mute notifications for a status. - - Returns a :ref:`status dict ` with the now muted status - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/mute'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.4.0", "2.0.0", _DICT_VERSION_STATUS) - def status_unmute(self, id): - """ - Unmute notifications for a status. - - Returns a :ref:`status dict ` with the status that used to be muted. - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/unmute'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("2.1.0", "2.1.0", _DICT_VERSION_STATUS) - def status_pin(self, id): - """ - Pin a status for the logged-in user. - - Returns a :ref:`status dict ` with the now pinned status - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/pin'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("2.1.0", "2.1.0", _DICT_VERSION_STATUS) - def status_unpin(self, id): - """ - Unpin a pinned status for the logged-in user. - - Returns a :ref:`status dict ` with the status that used to be pinned. - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/unpin'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("3.1.0", "3.1.0", _DICT_VERSION_STATUS) - def status_bookmark(self, id): - """ - Bookmark a status as the logged-in user. - - Returns a :ref:`status dict ` with the now bookmarked status - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/bookmark'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("3.1.0", "3.1.0", _DICT_VERSION_STATUS) - def status_unbookmark(self, id): - """ - Unbookmark a bookmarked status for the logged-in user. - - Returns a :ref:`status dict ` with the status that used to be bookmarked. - """ - id = self.__unpack_id(id) - url = '/api/v1/statuses/{0}/unbookmark'.format(str(id)) - return self.__api_request('POST', url) - - ### - # Writing data: Scheduled statuses - ### - @api_version("2.7.0", "2.7.0", _DICT_VERSION_SCHEDULED_STATUS) - def scheduled_status_update(self, id, scheduled_at): - """ - Update the scheduled time of a scheduled status. - - New time must be at least 5 minutes into the future. - - Returns a :ref:`scheduled status dict ` - """ - scheduled_at = self.__consistent_isoformat_utc(scheduled_at) - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ['id']) - url = '/api/v1/scheduled_statuses/{0}'.format(str(id)) - return self.__api_request('PUT', url, params) - - @api_version("2.7.0", "2.7.0", "2.7.0") - def scheduled_status_delete(self, id): - """ - Deletes a scheduled status. - """ - id = self.__unpack_id(id) - url = '/api/v1/scheduled_statuses/{0}'.format(str(id)) - self.__api_request('DELETE', url) - - ### - # Writing data: Polls - ### - @api_version("2.8.0", "2.8.0", _DICT_VERSION_POLL) - def poll_vote(self, id, choices): - """ - Vote in the given poll. - - `choices` is the index of the choice you wish to register a vote for - (i.e. its index in the corresponding polls `options` field. In case - of a poll that allows selection of more than one option, a list of - indices can be passed. - - You can only submit choices for any given poll once in case of - single-option polls, or only once per option in case of multi-option - polls. - - Returns the updated :ref:`poll dict ` - """ - id = self.__unpack_id(id) - if not isinstance(choices, list): - choices = [choices] - params = self.__generate_params(locals(), ['id']) - - url = '/api/v1/polls/{0}/votes'.format(id) - self.__api_request('POST', url, params) - - ### - # Writing data: Notifications - ### - - @api_version("1.0.0", "1.0.0", "1.0.0") - def notifications_clear(self): - """ - Clear out a user's notifications - """ - self.__api_request('POST', '/api/v1/notifications/clear') - - @api_version("1.3.0", "2.9.2", "2.9.2") - def notifications_dismiss(self, id): - """ - Deletes a single notification - """ - id = self.__unpack_id(id) - - if self.verify_minimum_version("2.9.2", cached=True): - url = '/api/v1/notifications/{0}/dismiss'.format(str(id)) - self.__api_request('POST', url) - else: - params = self.__generate_params(locals()) - self.__api_request('POST', '/api/v1/notifications/dismiss', params) - - ### - # Writing data: Conversations - ### - @api_version("2.6.0", "2.6.0", _DICT_VERSION_CONVERSATION) - def conversations_read(self, id): - """ - Marks a single conversation as read. - - Returns the updated :ref:`conversation dict `. - """ - id = self.__unpack_id(id) - url = '/api/v1/conversations/{0}/read'.format(str(id)) - return self.__api_request('POST', url) - - ### - # Writing data: Accounts - ### - @api_version("1.0.0", "3.3.0", _DICT_VERSION_RELATIONSHIP) - def account_follow(self, id, reblogs=True, notify=False): - """ - Follow a user. - - Set `reblogs` to False to hide boosts by the followed user. - Set `notify` to True to get a notification every time the followed user posts. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ["id"]) - - if params["reblogs"] is None: - del params["reblogs"] - - url = '/api/v1/accounts/{0}/follow'.format(str(id)) - return self.__api_request('POST', url, params) - - @api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT) - def follows(self, uri): - """ - Follow a remote user by uri (username@domain). - - Returns a :ref:`account dict `. - """ - params = self.__generate_params(locals()) - return self.__api_request('POST', '/api/v1/follows', params) - - @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) - def account_unfollow(self, id): - """ - Unfollow a user. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/accounts/{0}/unfollow'.format(str(id))) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_RELATIONSHIP) - def account_remove_from_followers(self, id): - """ - Remove a user from the logged in users followers (i.e. make them unfollow the logged in - user / "softblock" them). - - Returns a :ref:`relationship dict ` reflecting the updated following status. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/accounts/{0}/remove_from_followers'.format(str(id))) - - - @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) - def account_block(self, id): - """ - Block a user. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - url = '/api/v1/accounts/{0}/block'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) - def account_unblock(self, id): - """ - Unblock a user. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - url = '/api/v1/accounts/{0}/unblock'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.1.0", "2.4.3", _DICT_VERSION_RELATIONSHIP) - def account_mute(self, id, notifications=True, duration=None): - """ - Mute a user. - - Set `notifications` to False to receive notifications even though the user is - muted from timelines. Pass a `duration` in seconds to have Mastodon automatically - lift the mute after that many seconds. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ['id']) - url = '/api/v1/accounts/{0}/mute'.format(str(id)) - return self.__api_request('POST', url, params) - - @api_version("1.1.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) - def account_unmute(self, id): - """ - Unmute a user. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - url = '/api/v1/accounts/{0}/unmute'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.1.1", "3.1.0", _DICT_VERSION_ACCOUNT) - def account_update_credentials(self, display_name=None, note=None, - avatar=None, avatar_mime_type=None, - header=None, header_mime_type=None, - locked=None, bot=None, - discoverable=None, fields=None): - """ - Update the profile for the currently logged-in user. - - `note` is the user's bio. - - `avatar` and 'header' are images. As with media uploads, it is possible to either - pass image data and a mime type, or a filename of an image file, for either. - - `locked` specifies whether the user needs to manually approve follow requests. - - `bot` specifies whether the user should be set to a bot. - - `discoverable` specifies whether the user should appear in the user directory. - - `fields` can be a list of up to four name-value pairs (specified as tuples) to - appear as semi-structured information in the user's profile. - - Returns the updated `account dict` of the logged-in user. - """ - params_initial = collections.OrderedDict(locals()) - - # Convert fields - if fields is not None: - if len(fields) > 4: - raise MastodonIllegalArgumentError( - 'A maximum of four fields are allowed.') - - fields_attributes = [] - for idx, (field_name, field_value) in enumerate(fields): - params_initial['fields_attributes[' + - str(idx) + '][name]'] = field_name - params_initial['fields_attributes[' + - str(idx) + '][value]'] = field_value - - # Clean up params - for param in ["avatar", "avatar_mime_type", "header", "header_mime_type", "fields"]: - if param in params_initial: - del params_initial[param] - - # Create file info - files = {} - if avatar is not None: - files["avatar"] = self.__load_media_file(avatar, avatar_mime_type) - if header is not None: - files["header"] = self.__load_media_file(header, header_mime_type) - - params = self.__generate_params(params_initial) - return self.__api_request('PATCH', '/api/v1/accounts/update_credentials', params, files=files) - - @api_version("2.5.0", "2.5.0", _DICT_VERSION_RELATIONSHIP) - def account_pin(self, id): - """ - Pin / endorse a user. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - url = '/api/v1/accounts/{0}/pin'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("2.5.0", "2.5.0", _DICT_VERSION_RELATIONSHIP) - def account_unpin(self, id): - """ - Unpin / un-endorse a user. - - Returns a :ref:`relationship dict ` containing the updated relationship to the user. - """ - id = self.__unpack_id(id) - url = '/api/v1/accounts/{0}/unpin'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("3.2.0", "3.2.0", _DICT_VERSION_RELATIONSHIP) - def account_note_set(self, id, comment): - """ - Set a note (visible to the logged in user only) for the given account. - - Returns a :ref:`status dict ` with the `note` updated. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ["id"]) - return self.__api_request('POST', '/api/v1/accounts/{0}/note'.format(str(id)), params) - - @api_version("3.3.0", "3.3.0", _DICT_VERSION_HASHTAG) - def account_featured_tags(self, id): - """ - Get an account's featured hashtags. - - Returns a list of :ref:`hashtag dicts ` (NOT `featured tag dicts`_). - """ - id = self.__unpack_id(id) - return self.__api_request('GET', '/api/v1/accounts/{0}/featured_tags'.format(str(id))) - - ### - # Writing data: Featured hashtags - ### - @api_version("3.0.0", "3.0.0", _DICT_VERSION_FEATURED_TAG) - def featured_tag_create(self, name): - """ - Creates a new featured hashtag displayed on the logged-in user's profile. - - Returns a :ref:`featured tag dict ` with the newly featured tag. - """ - params = self.__generate_params(locals()) - return self.__api_request('POST', '/api/v1/featured_tags', params) - - @api_version("3.0.0", "3.0.0", _DICT_VERSION_FEATURED_TAG) - def featured_tag_delete(self, id): - """ - Deletes one of the logged-in user's featured hashtags. - """ - id = self.__unpack_id(id) - url = '/api/v1/featured_tags/{0}'.format(str(id)) - self.__api_request('DELETE', url) - - ### - # Writing data: Keyword filters - ### - @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) - def filter_create(self, phrase, context, irreversible=False, whole_word=True, expires_in=None): - """ - Creates a new keyword filter. `phrase` is the phrase that should be - filtered out, `context` specifies from where to filter the keywords. - Valid contexts are 'home', 'notifications', 'public' and 'thread'. - - Set `irreversible` to True if you want the filter to just delete statuses - server side. This works only for the 'home' and 'notifications' contexts. - - Set `whole_word` to False if you want to allow filter matches to - start or end within a word, not only at word boundaries. - - Set `expires_in` to specify for how many seconds the filter should be - kept around. - - Returns the :ref:`filter dict ` of the newly created filter. - """ - params = self.__generate_params(locals()) - - for context_val in context: - if not context_val in ['home', 'notifications', 'public', 'thread']: - raise MastodonIllegalArgumentError('Invalid filter context.') - - return self.__api_request('POST', '/api/v1/filters', params) - - @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) - def filter_update(self, id, phrase=None, context=None, irreversible=None, whole_word=None, expires_in=None): - """ - Updates the filter with the given `id`. Parameters are the same - as in `filter_create()`. - - Returns the :ref:`filter dict ` of the updated filter. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ['id']) - url = '/api/v1/filters/{0}'.format(str(id)) - return self.__api_request('PUT', url, params) - - @api_version("2.4.3", "2.4.3", "2.4.3") - def filter_delete(self, id): - """ - Deletes the filter with the given `id`. - """ - id = self.__unpack_id(id) - url = '/api/v1/filters/{0}'.format(str(id)) - self.__api_request('DELETE', url) - - ### - # Writing data: Follow suggestions - ### - @api_version("2.4.3", "2.4.3", _DICT_VERSION_ACCOUNT) - def suggestion_delete(self, account_id): - """ - Remove the user with the given `account_id` from the follow suggestions. - """ - account_id = self.__unpack_id(account_id) - url = '/api/v1/suggestions/{0}'.format(str(account_id)) - self.__api_request('DELETE', url) - - ### - # Writing data: Lists - ### - @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) - def list_create(self, title): - """ - Create a new list with the given `title`. - - Returns the :ref:`list dict ` of the created list. - """ - params = self.__generate_params(locals()) - return self.__api_request('POST', '/api/v1/lists', params) - - @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) - def list_update(self, id, title): - """ - Update info about a list, where "info" is really the lists `title`. - - Returns the :ref:`list dict ` of the modified list. - """ - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ['id']) - return self.__api_request('PUT', '/api/v1/lists/{0}'.format(id), params) - - @api_version("2.1.0", "2.1.0", "2.1.0") - def list_delete(self, id): - """ - Delete a list. - """ - id = self.__unpack_id(id) - self.__api_request('DELETE', '/api/v1/lists/{0}'.format(id)) - - @api_version("2.1.0", "2.1.0", "2.1.0") - def list_accounts_add(self, id, account_ids): - """ - Add the account(s) given in `account_ids` to the list. - """ - id = self.__unpack_id(id) - - if not isinstance(account_ids, list): - account_ids = [account_ids] - account_ids = list(map(lambda x: self.__unpack_id(x), account_ids)) - - params = self.__generate_params(locals(), ['id']) - self.__api_request( - 'POST', '/api/v1/lists/{0}/accounts'.format(id), params) - - @api_version("2.1.0", "2.1.0", "2.1.0") - def list_accounts_delete(self, id, account_ids): - """ - Remove the account(s) given in `account_ids` from the list. - """ - id = self.__unpack_id(id) - - if not isinstance(account_ids, list): - account_ids = [account_ids] - account_ids = list(map(lambda x: self.__unpack_id(x), account_ids)) - - params = self.__generate_params(locals(), ['id']) - self.__api_request( - 'DELETE', '/api/v1/lists/{0}/accounts'.format(id), params) - - ### - # Writing data: Reports - ### - @api_version("1.1.0", "3.5.0", _DICT_VERSION_REPORT) - def report(self, account_id, status_ids=None, comment=None, forward=False, category=None, rule_ids=None): - """ - Report statuses to the instances administrators. - - Accepts a list of toot IDs associated with the report, and a comment. - - Starting with Mastodon 3.5.0, you can also pass a `category` (one out of - "spam", "violation" or "other") and `rule_ids` (a list of rule IDs corresponding - to the rules returned by the :ref:`instance() ` API). - - Set `forward` to True to forward a report of a remote user to that users - instance as well as sending it to the instance local administrators. - - Returns a :ref:`report dict `. - """ - if category is not None and not category in ["spam", "violation", "other"]: - raise MastodonIllegalArgumentError("Invalid report category (must be spam, violation or other)") - - account_id = self.__unpack_id(account_id) - - if status_ids is not None: - if not isinstance(status_ids, list): - status_ids = [status_ids] - status_ids = list(map(lambda x: self.__unpack_id(x), status_ids)) - - params_initial = locals() - if not forward: - del params_initial['forward'] - - params = self.__generate_params(params_initial) - return self.__api_request('POST', '/api/v1/reports/', params) - - ### - # Writing data: Follow requests - ### - @api_version("1.0.0", "3.0.0", _DICT_VERSION_RELATIONSHIP) - def follow_request_authorize(self, id): - """ - Accept an incoming follow request. - - Returns the updated :ref:`relationship dict ` for the requesting account. - """ - id = self.__unpack_id(id) - url = '/api/v1/follow_requests/{0}/authorize'.format(str(id)) - return self.__api_request('POST', url) - - @api_version("1.0.0", "3.0.0", _DICT_VERSION_RELATIONSHIP) - def follow_request_reject(self, id): - """ - Reject an incoming follow request. - - Returns the updated :ref:`relationship dict ` for the requesting account. - """ - id = self.__unpack_id(id) - url = '/api/v1/follow_requests/{0}/reject'.format(str(id)) - return self.__api_request('POST', url) - - ### - # Writing data: Media - ### - @api_version("1.0.0", "3.2.0", _DICT_VERSION_MEDIA) - def media_post(self, media_file, mime_type=None, description=None, focus=None, file_name=None, thumbnail=None, thumbnail_mime_type=None, synchronous=False): - """ - Post an image, video or audio file. `media_file` can either be data or - a file name. If data is passed directly, the mime type has to be specified - manually, otherwise, it is determined from the file name. `focus` should be a tuple - of floats between -1 and 1, giving the x and y coordinates of the images - focus point for cropping (with the origin being the images center). - - Throws a `MastodonIllegalArgumentError` if the mime type of the - passed data or file can not be determined properly. - - `file_name` can be specified to upload a file with the given name, - which is ignored by Mastodon, but some other Fediverse server software - will display it. If no name is specified, a random name will be generated. - The filename of a file specified in media_file will be ignored. - - Starting with Mastodon 3.2.0, `thumbnail` can be specified in the same way as `media_file` - to upload a custom thumbnail image for audio and video files. - - Returns a :ref:`media dict `. This contains the id that can be used in - status_post to attach the media file to a toot. - - When using the v2 API (post Mastodon version 3.1.4), the `url` in the - returned dict will be `null`, since attachments are processed - asynchronously. You can fetch an updated dict using `media`. Pass - "synchronous" to emulate the old behaviour. Not recommended, inefficient - and deprecated, you know the deal. - """ - files = {'file': self.__load_media_file( - media_file, mime_type, file_name)} - - if focus is not None: - focus = str(focus[0]) + "," + str(focus[1]) - - if thumbnail is not None: - if not self.verify_minimum_version("3.2.0", cached=True): - raise MastodonVersionError( - 'Thumbnail requires version > 3.2.0') - files["thumbnail"] = self.__load_media_file( - thumbnail, thumbnail_mime_type) - - # Disambiguate URL by version - if self.verify_minimum_version("3.1.4", cached=True): - ret_dict = self.__api_request( - 'POST', '/api/v2/media', files=files, params={'description': description, 'focus': focus}) - else: - ret_dict = self.__api_request( - 'POST', '/api/v1/media', files=files, params={'description': description, 'focus': focus}) - - # Wait for processing? - if synchronous: - if self.verify_minimum_version("3.1.4"): - while not "url" in ret_dict or ret_dict.url is None: - try: - ret_dict = self.media(ret_dict) - time.sleep(1.0) - except: - raise MastodonAPIError( - "Attachment could not be processed") - else: - # Old version always waits - return ret_dict - - return ret_dict - - @api_version("2.3.0", "3.2.0", _DICT_VERSION_MEDIA) - def media_update(self, id, description=None, focus=None, thumbnail=None, thumbnail_mime_type=None): - """ - Update the metadata of the media file with the given `id`. `description` and - `focus` and `thumbnail` are as in :ref:`media_post() ` . - - Returns the updated :ref:`media dict `. - """ - id = self.__unpack_id(id) - - if focus is not None: - focus = str(focus[0]) + "," + str(focus[1]) - - params = self.__generate_params( - locals(), ['id', 'thumbnail', 'thumbnail_mime_type']) - - if thumbnail is not None: - if not self.verify_minimum_version("3.2.0", cached=True): - raise MastodonVersionError( - 'Thumbnail requires version > 3.2.0') - files = {"thumbnail": self.__load_media_file( - thumbnail, thumbnail_mime_type)} - return self.__api_request('PUT', '/api/v1/media/{0}'.format(str(id)), params, files=files) - else: - return self.__api_request('PUT', '/api/v1/media/{0}'.format(str(id)), params) - - @api_version("3.1.4", "3.1.4", _DICT_VERSION_MEDIA) - def media(self, id): - """ - Get the updated JSON for one non-attached / in progress media upload belonging - to the logged-in user. - """ - id = self.__unpack_id(id) - return self.__api_request('GET', '/api/v1/media/{0}'.format(str(id))) - - ### - # Writing data: Domain blocks - ### - @api_version("1.4.0", "1.4.0", "1.4.0") - def domain_block(self, domain=None): - """ - Add a block for all statuses originating from the specified domain for the logged-in user. - """ - params = self.__generate_params(locals()) - self.__api_request('POST', '/api/v1/domain_blocks', params) - - @api_version("1.4.0", "1.4.0", "1.4.0") - def domain_unblock(self, domain=None): - """ - Remove a domain block for the logged-in user. - """ - params = self.__generate_params(locals()) - self.__api_request('DELETE', '/api/v1/domain_blocks', params) - - ## - # Writing data: Read markers - ## - @api_version("3.0.0", "3.0.0", _DICT_VERSION_MARKER) - def markers_set(self, timelines, last_read_ids): - """ - Set the "last read" marker(s) for the given timeline(s) to the given id(s) - - Note that if you give an invalid timeline name, this will silently do nothing. - - Returns a dict with the updated :ref:`read marker dicts `, keyed by timeline name. - """ - if not isinstance(timelines, (list, tuple)): - timelines = [timelines] - - if not isinstance(last_read_ids, (list, tuple)): - last_read_ids = [last_read_ids] - - if len(last_read_ids) != len(timelines): - raise MastodonIllegalArgumentError( - "Number of specified timelines and ids must be the same") - - params = collections.OrderedDict() - for timeline, last_read_id in zip(timelines, last_read_ids): - params[timeline] = collections.OrderedDict() - params[timeline]["last_read_id"] = self.__unpack_id(last_read_id) - - return self.__api_request('POST', '/api/v1/markers', params, use_json=True) - - ### - # Writing data: Push subscriptions - ### - @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH) - def push_subscription_set(self, endpoint, encrypt_params, follow_events=None, - favourite_events=None, reblog_events=None, - mention_events=None, poll_events=None, - follow_request_events=None, status_events=None, policy='all'): - """ - Sets up or modifies the push subscription the logged-in user has for this app. - - `endpoint` is the endpoint URL mastodon should call for pushes. Note that mastodon - requires https for this URL. `encrypt_params` is a dict with key parameters that allow - the server to encrypt data for you: A public key `pubkey` and a shared secret `auth`. - You can generate this as well as the corresponding private key using the - :ref:`push_subscription_generate_keys() ` function. - - `policy` controls what sources will generate webpush events. Valid values are - `all`, `none`, `follower` and `followed`. - - The rest of the parameters controls what kind of events you wish to subscribe to. - - Returns a :ref:`push subscription dict `. - """ - if not policy in ['all', 'none', 'follower', 'followed']: - raise MastodonIllegalArgumentError("Valid values for policy are 'all', 'none', 'follower' or 'followed'.") - - endpoint = Mastodon.__protocolize(endpoint) - - push_pubkey_b64 = base64.b64encode(encrypt_params['pubkey']) - push_auth_b64 = base64.b64encode(encrypt_params['auth']) - - params = { - 'subscription[endpoint]': endpoint, - 'subscription[keys][p256dh]': push_pubkey_b64, - 'subscription[keys][auth]': push_auth_b64, - 'policy': policy - } - - if follow_events is not None: - params['data[alerts][follow]'] = follow_events - - if favourite_events is not None: - params['data[alerts][favourite]'] = favourite_events - - if reblog_events is not None: - params['data[alerts][reblog]'] = reblog_events - - if mention_events is not None: - params['data[alerts][mention]'] = mention_events - - if poll_events is not None: - params['data[alerts][poll]'] = poll_events - - if follow_request_events is not None: - params['data[alerts][follow_request]'] = follow_request_events - - if follow_request_events is not None: - params['data[alerts][status]'] = status_events - - # Canonicalize booleans - params = self.__generate_params(params) - - return self.__api_request('POST', '/api/v1/push/subscription', params) - - @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH) - def push_subscription_update(self, follow_events=None, - favourite_events=None, reblog_events=None, - mention_events=None, poll_events=None, - follow_request_events=None): - """ - Modifies what kind of events the app wishes to subscribe to. - - Returns the updated :ref:`push subscription dict `. - """ - params = {} - - if follow_events is not None: - params['data[alerts][follow]'] = follow_events - - if favourite_events is not None: - params['data[alerts][favourite]'] = favourite_events - - if reblog_events is not None: - params['data[alerts][reblog]'] = reblog_events - - if mention_events is not None: - params['data[alerts][mention]'] = mention_events - - if poll_events is not None: - params['data[alerts][poll]'] = poll_events - - if follow_request_events is not None: - params['data[alerts][follow_request]'] = follow_request_events - - # Canonicalize booleans - params = self.__generate_params(params) - - return self.__api_request('PUT', '/api/v1/push/subscription', params) - - @api_version("2.4.0", "2.4.0", "2.4.0") - def push_subscription_delete(self): - """ - Remove the current push subscription the logged-in user has for this app. - """ - self.__api_request('DELETE', '/api/v1/push/subscription') - - ### - # Writing data: Annoucements - ### - @api_version("3.1.0", "3.1.0", "3.1.0") - def announcement_dismiss(self, id): - """ - Set the given annoucement to read. - """ - id = self.__unpack_id(id) - - url = '/api/v1/announcements/{0}/dismiss'.format(str(id)) - self.__api_request('POST', url) - - @api_version("3.1.0", "3.1.0", "3.1.0") - def announcement_reaction_create(self, id, reaction): - """ - Add a reaction to an announcement. `reaction` can either be a unicode emoji - or the name of one of the instances custom emoji. - - Will throw an API error if the reaction name is not one of the allowed things - or when trying to add a reaction that the user has already added (adding a - reaction that a different user added is legal and increments the count). - """ - id = self.__unpack_id(id) - - url = '/api/v1/announcements/{0}/reactions/{1}'.format( - str(id), reaction) - self.__api_request('PUT', url) - - @api_version("3.1.0", "3.1.0", "3.1.0") - def announcement_reaction_delete(self, id, reaction): - """ - Remove a reaction to an announcement. - - Will throw an API error if the reaction does not exist. - """ - id = self.__unpack_id(id) - - url = '/api/v1/announcements/{0}/reactions/{1}'.format( - str(id), reaction) - self.__api_request('DELETE', url) - - ### - # Moderation API - ### - @api_version("2.9.1", "4.0.0", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_accounts_v2(self, origin=None, by_domain=None, status=None, username=None, display_name=None, email=None, ip=None, - permissions=None, invited_by=None, role_ids=None, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetches a list of accounts that match given criteria. By default, local accounts are returned. - - * Set `origin` to "local" or "remote" to get only local or remote accounts. - * Set `by_domain` to a domain to get only accounts from that domain. - * Set `status` to one of "active", "pending", "disabled", "silenced" or "suspended" to get only accounts with that moderation status (default: active) - * Set `username` to a string to get only accounts whose username contains this string. - * Set `display_name` to a string to get only accounts whose display name contains this string. - * Set `email` to an email to get only accounts with that email (this only works on local accounts). - * Set `ip` to an ip (as a string, standard v4/v6 notation) to get only accounts whose last active ip is that ip (this only works on local accounts). - * Set `permissions` to "staff" to only get accounts with staff permissions. - * Set `invited_by` to an account id to get only accounts invited by this user. - * Set `role_ids` to a list of role IDs to get only accounts with those roles. - - Returns a list of :ref:`admin account dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - if role_ids is not None: - if not isinstance(role_ids, list): - role_ids = [role_ids] - role_ids = list(map(self.__unpack_id, role_ids)) - - if invited_by is not None: - invited_by = self.__unpack_id(invited_by) - - if permissions is not None and not permissions in ["staff"]: - raise MastodonIllegalArgumentError("Permissions must be staff if passed") - - if origin is not None and not origin in ["local", "remote"]: - raise MastodonIllegalArgumentError("Origin must be local or remote") - - if status is not None and not status in ["active", "pending", "disabled", "silenced", "suspended"]: - raise MastodonIllegalArgumentError("Status must be local or active, pending, disabled, silenced or suspended") - - if not by_domain is None: - by_domain = self.__deprotocolize(by_domain) - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v2/admin/accounts', params) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_accounts(self, remote=False, by_domain=None, status='active', username=None, display_name=None, email=None, ip=None, staff_only=False, max_id=None, min_id=None, since_id=None, limit=None): - """ - Currently a synonym for admin_accounts_v1, now deprecated. You are strongly encouraged to use admin_accounts_v2 instead, since this one is kind of bad. - - !!!!! This function may be switched to calling the v2 API in the future. This is your warning. If you want to keep using v1, use it explicitly. !!!!! - """ - return self.admin_accounts_v1( - remote=remote, - by_domain=by_domain, - status=status, - username=username, - display_name=display_name, - email=email, - ip=ip, - staff_only=staff_only, - max_id=max_id, - min_id=min_id, - since_id=since_id - ) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_accounts_v1(self, remote=False, by_domain=None, status='active', username=None, display_name=None, email=None, ip=None, staff_only=False, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetches a list of accounts that match given criteria. By default, local accounts are returned. - - * Set `remote` to True to get remote accounts, otherwise local accounts are returned (default: local accounts) - * Set `by_domain` to a domain to get only accounts from that domain. - * Set `status` to one of "active", "pending", "disabled", "silenced" or "suspended" to get only accounts with that moderation status (default: active) - * Set `username` to a string to get only accounts whose username contains this string. - * Set `display_name` to a string to get only accounts whose display name contains this string. - * Set `email` to an email to get only accounts with that email (this only works on local accounts). - * Set `ip` to an ip (as a string, standard v4/v6 notation) to get only accounts whose last active ip is that ip (this only works on local accounts). - * Set `staff_only` to True to only get staff accounts (this only works on local accounts). - - Note that setting the boolean parameters to False does not mean "give me users to which this does not apply" but - instead means "I do not care if users have this attribute". - - Deprecated in Mastodon version 3.5.0. - - Returns a list of :ref:`admin account dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals(), ['remote', 'status', 'staff_only']) - - if remote: - params["remote"] = True - - mod_statuses = ["active", "pending", "disabled", "silenced", "suspended"] - if not status in mod_statuses: - raise ValueError("Invalid moderation status requested.") - - if staff_only: - params["staff"] = True - - for mod_status in mod_statuses: - if status == mod_status: - params[status] = True - - if not by_domain is None: - by_domain = self.__deprotocolize(by_domain) - - return self.__api_request('GET', '/api/v1/admin/accounts', params) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account(self, id): - """ - Fetches a single :ref:`admin account dict ` for the user with the given id. - - Returns that dict. - """ - id = self.__unpack_id(id) - return self.__api_request('GET', '/api/v1/admin/accounts/{0}'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_enable(self, id): - """ - Reenables login for a local account for which login has been disabled. - - Returns the updated :ref:`admin account dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/accounts/{0}/enable'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_approve(self, id): - """ - Approves a pending account. - - Returns the updated :ref:`admin account dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/accounts/{0}/approve'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_reject(self, id): - """ - Rejects and deletes a pending account. - - Returns the updated :ref:`admin account dict ` for the account that is now gone. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/accounts/{0}/reject'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_unsilence(self, id): - """ - Unsilences an account. - - Returns the updated :ref:`admin account dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/accounts/{0}/unsilence'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_unsuspend(self, id): - """ - Unsuspends an account. - - Returns the updated :ref:`admin account dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/accounts/{0}/unsuspend'.format(id)) - - @api_version("3.3.0", "3.3.0", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_delete(self, id): - """ - Delete a local user account. - - The deleted accounts :ref:`admin account dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('DELETE', '/api/v1/admin/accounts/{0}'.format(id)) - - @api_version("3.3.0", "3.3.0", _DICT_VERSION_ADMIN_ACCOUNT) - def admin_account_unsensitive(self, id): - """ - Unmark an account as force-sensitive. - - Returns the updated :ref:`admin account dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/accounts/{0}/unsensitive'.format(id)) - - @api_version("2.9.1", "2.9.1", "2.9.1") - def admin_account_moderate(self, id, action=None, report_id=None, warning_preset_id=None, text=None, send_email_notification=True): - """ - Perform a moderation action on an account. - - Valid actions are: - * "disable" - for a local user, disable login. - * "silence" - hide the users posts from all public timelines. - * "suspend" - irreversibly delete all the user's posts, past and future. - * "sensitive" - forcce an accounts media visibility to always be sensitive. - - If no action is specified, the user is only issued a warning. - - Specify the id of a report as `report_id` to close the report with this moderation action as the resolution. - Specify `warning_preset_id` to use a warning preset as the notification text to the user, or `text` to specify text directly. - If both are specified, they are concatenated (preset first). Note that there is currently no API to retrieve or create - warning presets. - - Set `send_email_notification` to False to not send the user an email notification informing them of the moderation action. - """ - if action is None: - action = "none" - - if not send_email_notification: - send_email_notification = None - - id = self.__unpack_id(id) - if report_id is not None: - report_id = self.__unpack_id(report_id) - - params = self.__generate_params(locals(), ['id', 'action']) - - params["type"] = action - - self.__api_request( - 'POST', '/api/v1/admin/accounts/{0}/action'.format(id), params) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) - def admin_reports(self, resolved=False, account_id=None, target_account_id=None, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetches the list of reports. - - Set `resolved` to True to search for resolved reports. `account_id` and `target_account_id` - can be used to get reports filed by or about a specific user. - - Returns a list of :ref:`report dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - if account_id is not None: - account_id = self.__unpack_id(account_id) - - if target_account_id is not None: - target_account_id = self.__unpack_id(target_account_id) - - if not resolved: - resolved = None - - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/admin/reports', params) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) - def admin_report(self, id): - """ - Fetches the report with the given id. - - Returns a :ref:`report dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('GET', '/api/v1/admin/reports/{0}'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) - def admin_report_assign(self, id): - """ - Assigns the given report to the logged-in user. - - Returns the updated :ref:`report dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/reports/{0}/assign_to_self'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) - def admin_report_unassign(self, id): - """ - Unassigns the given report from the logged-in user. - - Returns the updated :ref:`report dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/reports/{0}/unassign'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) - def admin_report_reopen(self, id): - """ - Reopens a closed report. - - Returns the updated :ref:`report dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/reports/{0}/reopen'.format(id)) - - @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) - def admin_report_resolve(self, id): - """ - Marks a report as resolved (without taking any action). - - Returns the updated :ref:`report dict `. - """ - id = self.__unpack_id(id) - return self.__api_request('POST', '/api/v1/admin/reports/{0}/resolve'.format(id)) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_HASHTAG) - def admin_trending_tags(self, limit=None): - """ - Admin version of :ref:`trending_tags() `. Includes unapproved tags. - - Returns a list of :ref:`hashtag dicts `, sorted by the instance's trending algorithm, - descending. - """ - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/admin/trends/tags', params) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS) - def admin_trending_statuses(self): - """ - Admin version of :ref:`trending_statuses() `. Includes unapproved tags. - - Returns a list of :ref:`status dicts `, sorted by the instance's trending algorithm, - descending. - """ - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/admin/trends/statuses', params) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_CARD) - def admin_trending_links(self): - """ - Admin version of :ref:`trending_links() `. Includes unapproved tags. - - Returns a list of :ref:`card dicts `, sorted by the instance's trending algorithm, - descending. - """ - params = self.__generate_params(locals()) - return self.__api_request('GET', '/api/v1/admin/trends/links', params) - - @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) - def admin_domain_blocks(self, id=None, limit:int=None): - """ - Fetches a list of blocked domains. Requires scope `admin:read:domain_blocks`. - - Provide an `id` to fetch a specific domain block based on its database id. - - Returns a list of :ref:`admin domain block dicts `, raises a `MastodonAPIError` if the specified block does not exist. - """ - if id is not None: - id = self.__unpack_id(id) - return self.__api_request('GET', '/api/v1/admin/domain_blocks/{0}'.format(id)) - else: - params = self.__generate_params(locals(),['limit']) - return self.__api_request('GET', '/api/v1/admin/domain_blocks/', params) - - @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) - def admin_create_domain_block(self, domain:str, severity:str=None, reject_media:bool=None, reject_reports:bool=None, private_comment:str=None, public_comment:str=None, obfuscate:bool=None): - """ - Perform a moderation action on a domain. Requires scope `admin:write:domain_blocks`. - - Valid severities are: - * "silence" - hide all posts from federated timelines and do not show notifications to local users from the remote instance's users unless they are following the remote user. - * "suspend" - deny interactions with this instance going forward. This action is reversible. - * "limit" - generally used with reject_media=true to force reject media from an instance without silencing or suspending.. - - If no action is specified, the domain is only silenced. - `domain` is the domain to block. Note that using the top level domain will also imapct all subdomains. ie, example.com will also impact subdomain.example.com. - `reject_media` will not download remote media on to your local instance media storage. - `reject_reports` ignores all reports from the remote instance. - `private_comment` sets a private admin comment for the domain. - `public_comment` sets a publicly available comment for this domain, which will be available to local users and may be available to everyone depending on your settings. - `obfuscate` censors some part of the domain name. Useful if the domain name contains unwanted words like slurs. - - Returns the new domain block as an :ref:`admin domain block dict `. - """ - if domain is None: - raise AttributeError("Must provide a domain to block a domain") - params = self.__generate_params(locals()) - return self.__api_request('POST', '/api/v1/admin/domain_blocks/', params) - - @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) - def admin_update_domain_block(self, id, severity:str=None, reject_media:bool=None, reject_reports:bool=None, private_comment:str=None, public_comment:str=None, obfuscate:bool=None): - """ - Modify existing moderation action on a domain. Requires scope `admin:write:domain_blocks`. - - Valid severities are: - * "silence" - hide all posts from federated timelines and do not show notifications to local users from the remote instance's users unless they are following the remote user. - * "suspend" - deny interactions with this instance going forward. This action is reversible. - * "limit" - generally used with reject_media=true to force reject media from an instance without silencing or suspending. - - If no action is specified, the domain is only silenced. - `domain` is the domain to block. Note that using the top level domain will also imapct all subdomains. ie, example.com will also impact subdomain.example.com. - `reject_media` will not download remote media on to your local instance media storage. - `reject_reports` ignores all reports from the remote instance. - `private_comment` sets a private admin comment for the domain. - `public_comment` sets a publicly available comment for this domain, which will be available to local users and may be available to everyone depending on your settings. - `obfuscate` censors some part of the domain name. Useful if the domain name contains unwanted words like slurs. - - Returns the modified domain block as an :ref:`admin domain block dict `, raises a `MastodonAPIError` if the specified block does not exist. - """ - if id is None: - raise AttributeError("Must provide an id to modify the existing moderation actions on a given domain.") - id = self.__unpack_id(id) - params = self.__generate_params(locals(), ["id"]) - return self.__api_request('PUT', '/api/v1/admin/domain_blocks/{0}'.format(id), params) - - @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) - def admin_delete_domain_block(self, id=None): - """ - Removes moderation action against a given domain. Requires scope `admin:write:domain_blocks`. - - Provide an `id` to remove a specific domain block based on its database id. - - Raises a `MastodonAPIError` if the specified block does not exist. - """ - if id is not None: - id = self.__unpack_id(id) - self.__api_request('DELETE', '/api/v1/admin/domain_blocks/{0}'.format(id)) - else: - raise AttributeError("You must provide an id of an existing domain block to remove it.") - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_ADMIN_MEASURE) - def admin_measures(self, start_at, end_at, active_users=False, new_users=False, interactions=False, opened_reports = False, resolved_reports=False, - tag_accounts=None, tag_uses=None, tag_servers=None, instance_accounts=None, instance_media_attachments=None, instance_reports=None, - instance_statuses=None, instance_follows=None, instance_followers=None): - """ - Retrieves numerical instance information for the time period (at day granularity) between `start_at` and `end_at`. - - * `active_users`: Pass true to retrieve the number of active users on your instance within the time period - * `new_users`: Pass true to retrieve the number of users who joined your instance within the time period - * `interactions`: Pass true to retrieve the number of interactions (favourites, boosts, replies) on local statuses within the time period - * `opened_reports`: Pass true to retrieve the number of reports filed within the time period - * `resolved_reports` = Pass true to retrieve the number of reports resolved within the time period - * `tag_accounts`: Pass a tag ID to get the number of accounts which used that tag in at least one status within the time period - * `tag_uses`: Pass a tag ID to get the number of statuses which used that tag within the time period - * `tag_servers`: Pass a tag ID to to get the number of remote origin servers for statuses which used that tag within the time period - * `instance_accounts`: Pass a domain to get the number of accounts originating from that remote domain within the time period - * `instance_media_attachments`: Pass a domain to get the amount of space used by media attachments from that remote domain within the time period - * `instance_reports`: Pass a domain to get the number of reports filed against accounts from that remote domain within the time period - * `instance_statuses`: Pass a domain to get the number of statuses originating from that remote domain within the time period - * `instance_follows`: Pass a domain to get the number of accounts from a remote domain followed by that local user within the time period - * `instance_followers`: Pass a domain to get the number of local accounts followed by accounts from that remote domain within the time period - - This API call is relatively expensive - watch your servers load if you want to get a lot of statistical data. Especially the instance_statuses stats - might take a long time to compute and, in fact, time out. - - There is currently no way to get tag IDs implemented in Mastodon.py, because the Mastodon public API does not implement one. This will be fixed in a future - release. - - Returns a list of :ref:`admin measure dicts `. - """ - params_init = locals() - keys = [] - for key in ["active_users", "new_users", "interactions", "opened_reports", "resolved_reports"]: - if params_init[key] == True: - keys.append(key) - - params = {} - for key in ["tag_accounts", "tag_uses", "tag_servers"]: - if params_init[key] is not None: - keys.append(key) - params[key] = {"id": self.__unpack_id(params_init[key])} - for key in ["instance_accounts", "instance_media_attachments", "instance_reports", "instance_statuses", "instance_follows", "instance_followers"]: - if params_init[key] is not None: - keys.append(key) - params[key] = {"domain": Mastodon.__deprotocolize(params_init[key]).split("/")[0]} - - if len(keys) == 0: - raise MastodonIllegalArgumentError("Must request at least one metric.") - - params["keys"] = keys - params["start_at"] = self.__consistent_isoformat_utc(start_at) - params["end_at"] = self.__consistent_isoformat_utc(end_at) - - return self.__api_request('POST', '/api/v1/admin/measures', params, use_json=True) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_ADMIN_DIMENSION) - def admin_dimensions(self, start_at, end_at, limit=None, languages=False, sources=False, servers=False, space_usage=False, software_versions=False, - tag_servers=None, tag_languages=None, instance_accounts=None, instance_languages=None): - """ - Retrieves primarily categorical instance information for the time period (at day granularity) between `start_at` and `end_at`. - - * `languages`: Pass true to get the most-used languages on this server - * `sources`: Pass true to get the most-used client apps on this server - * `servers`: Pass true to get the remote servers with the most statuses - * `space_usage`: Pass true to get the how much space is used by different components your software stack - * `software_versions`: Pass true to get the version numbers for your software stack - * `tag_servers`: Pass a tag ID to get the most-common servers for statuses including a trending tag - * `tag_languages`: Pass a tag ID to get the most-used languages for statuses including a trending tag - * `instance_accounts`: Pass a domain to get the most-followed accounts from a remote server - * `instance_languages`: Pass a domain to get the most-used languages from a remote server - - Pass `limit` to set how many results you want on queries where that makes sense. - - This API call is relatively expensive - watch your servers load if you want to get a lot of statistical data. - - There is currently no way to get tag IDs implemented in Mastodon.py, because the Mastodon public API does not implement one. This will be fixed in a future - release. - - Returns a list of :ref:`admin dimension dicts `. - """ - params_init = locals() - keys = [] - for key in ["languages", "sources", "servers", "space_usage", "software_versions"]: - if params_init[key] == True: - keys.append(key) - - params = {} - for key in ["tag_servers", "tag_languages"]: - if params_init[key] is not None: - keys.append(key) - params[key] = {"id": self.__unpack_id(params_init[key])} - for key in ["instance_accounts", "instance_languages"]: - if params_init[key] is not None: - keys.append(key) - params[key] = {"domain": Mastodon.__deprotocolize(params_init[key]).split("/")[0]} - - if len(keys) == 0: - raise MastodonIllegalArgumentError("Must request at least one dimension.") - - params["keys"] = keys - if limit is not None: - params["limit"] = limit - params["start_at"] = self.__consistent_isoformat_utc(start_at) - params["end_at"] = self.__consistent_isoformat_utc(end_at) - - return self.__api_request('POST', '/api/v1/admin/dimensions', params, use_json=True) - - @api_version("3.5.0", "3.5.0", _DICT_VERSION_ADMIN_RETENTION) - def admin_retention(self, start_at, end_at, frequency="day"): - """ - Gets user retention statistics (at `frequency` - "day" or "month" - granularity) between `start_at` and `end_at`. - - Returns a list of :ref:`admin retention dicts ` - """ - if not frequency in ["day", "month"]: - raise MastodonIllegalArgumentError("Frequency must be day or month") - - params = { - "start_at": self.__consistent_isoformat_utc(start_at), - "end_at": self.__consistent_isoformat_utc(end_at), - "frequency": frequency - } - return self.__api_request('POST', '/api/v1/admin/retention', params) - - ### - # Push subscription crypto utilities - ### - def push_subscription_generate_keys(self): - """ - Generates a private key, public key and shared secret for use in webpush subscriptions. - - Returns two dicts: One with the private key and shared secret and another with the - public key and shared secret. - """ - if not IMPL_HAS_CRYPTO: - raise NotImplementedError( - 'To use the crypto tools, please install the webpush feature dependencies.') - - push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend()) - push_key_priv = push_key_pair.private_numbers().private_value - try: - push_key_pub = push_key_pair.public_key().public_bytes( - serialization.Encoding.X962, - serialization.PublicFormat.UncompressedPoint, - ) - except: - push_key_pub = push_key_pair.public_key().public_numbers().encode_point() - - push_shared_secret = os.urandom(16) - - priv_dict = { - 'privkey': push_key_priv, - 'auth': push_shared_secret - } - - pub_dict = { - 'pubkey': push_key_pub, - 'auth': push_shared_secret - } - - return priv_dict, pub_dict - - @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH_NOTIF) - def push_subscription_decrypt_push(self, data, decrypt_params, encryption_header, crypto_key_header): - """ - Decrypts `data` received in a webpush request. Requires the private key dict - from :ref:`push_subscription_generate_keys() ` (`decrypt_params`) as well as the - Encryption and server Crypto-Key headers from the received webpush - - Returns the decoded webpush as a :ref:`push notification dict `. - """ - if (not IMPL_HAS_ECE) or (not IMPL_HAS_CRYPTO): - raise NotImplementedError( - 'To use the crypto tools, please install the webpush feature dependencies.') - - salt = self.__decode_webpush_b64(encryption_header.split("salt=")[1].strip()) - dhparams = self.__decode_webpush_b64(crypto_key_header.split("dh=")[1].split(";")[0].strip()) - p256ecdsa = self.__decode_webpush_b64(crypto_key_header.split("p256ecdsa=")[1].strip()) - dec_key = ec.derive_private_key(decrypt_params['privkey'], ec.SECP256R1(), default_backend()) - decrypted = http_ece.decrypt( - data, - salt=salt, - key=p256ecdsa, - private_key=dec_key, - dh=dhparams, - auth_secret=decrypt_params['auth'], - keylabel="P-256", - version="aesgcm" - ) - - return json.loads(decrypted.decode('utf-8'), object_hook=Mastodon.__json_hooks) - - ### - # Blurhash utilities - ### - def decode_blurhash(self, media_dict, out_size=(16, 16), size_per_component=True, return_linear=True): - """ - Basic media-dict blurhash decoding. - - out_size is the desired result size in pixels, either absolute or per blurhash - component (this is the default). - - By default, this function will return the image as linear RGB, ready for further - scaling operations. If you want to display the image directly, set return_linear - to False. - - Returns the decoded blurhash image as a three-dimensional list: [height][width][3], - with the last dimension being RGB colours. - - For further info and tips for advanced usage, refer to the documentation for the - blurhash module: https://github.com/halcy/blurhash-python - """ - if not IMPL_HAS_BLURHASH: - raise NotImplementedError( - 'To use the blurhash functions, please install the blurhash Python module.') - - # Figure out what size to decode to - decode_components_x, decode_components_y = blurhash.components(media_dict["blurhash"]) - if size_per_component: - decode_size_x = decode_components_x * out_size[0] - decode_size_y = decode_components_y * out_size[1] - else: - decode_size_x = out_size[0] - decode_size_y = out_size[1] - - # Decode - decoded_image = blurhash.decode(media_dict["blurhash"], decode_size_x, decode_size_y, linear=return_linear) - - # And that's pretty much it. - return decoded_image - - ### - # Pagination - ### - def fetch_next(self, previous_page): - """ - Fetches the next page of results of a paginated request. Pass in the - previous page in its entirety, or the pagination information dict - returned as a part of that pages last status ('_pagination_next'). - - Returns the next page or None if no further data is available. - """ - if isinstance(previous_page, list) and len(previous_page) != 0: - if hasattr(previous_page, '_pagination_next'): - params = copy.deepcopy(previous_page._pagination_next) - else: - return None - else: - params = copy.deepcopy(previous_page) - - method = params['_pagination_method'] - del params['_pagination_method'] - - endpoint = params['_pagination_endpoint'] - del params['_pagination_endpoint'] - - return self.__api_request(method, endpoint, params) - - def fetch_previous(self, next_page): - """ - Fetches the previous page of results of a paginated request. Pass in the - previous page in its entirety, or the pagination information dict - returned as a part of that pages first status ('_pagination_prev'). - - Returns the previous page or None if no further data is available. - """ - if isinstance(next_page, list) and len(next_page) != 0: - if hasattr(next_page, '_pagination_prev'): - params = copy.deepcopy(next_page._pagination_prev) - else: - return None - else: - params = copy.deepcopy(next_page) - - method = params['_pagination_method'] - del params['_pagination_method'] - - endpoint = params['_pagination_endpoint'] - del params['_pagination_endpoint'] - - return self.__api_request(method, endpoint, params) - - def fetch_remaining(self, first_page): - """ - Fetches all the remaining pages of a paginated request starting from a - first page and returns the entire set of results (including the first page - that was passed in) as a big list. - - Be careful, as this might generate a lot of requests, depending on what you are - fetching, and might cause you to run into rate limits very quickly. - """ - first_page = copy.deepcopy(first_page) - - all_pages = [] - current_page = first_page - while current_page is not None and len(current_page) > 0: - all_pages.extend(current_page) - current_page = self.fetch_next(current_page) - - return all_pages - - ### - # Streaming - ### - @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) - def stream_user(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): - """ - Streams events that are relevant to the authorized user, i.e. home - timeline and notifications. - """ - return self.__stream('/api/v1/streaming/user', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) - - @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) - def stream_public(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): - """ - Streams public events. - """ - return self.__stream('/api/v1/streaming/public', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) - - @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) - def stream_local(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): - """ - Streams local public events. - """ - return self.__stream('/api/v1/streaming/public/local', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) - - @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) - def stream_hashtag(self, tag, listener, local=False, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): - """ - Stream for all public statuses for the hashtag 'tag' seen by the connected - instance. - - Set local to True to only get local statuses. - """ - if tag.startswith("#"): - raise MastodonIllegalArgumentError( - "Tag parameter should omit leading #") - base = '/api/v1/streaming/hashtag' - if local: - base += '/local' - return self.__stream("{}?tag={}".format(base, tag), listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) - - @api_version("2.1.0", "2.1.0", _DICT_VERSION_STATUS) - def stream_list(self, id, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): - """ - Stream events for the current user, restricted to accounts on the given - list. - """ - id = self.__unpack_id(id) - return self.__stream("/api/v1/streaming/list?list={}".format(id), listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) - - @api_version("2.6.0", "2.6.0", _DICT_VERSION_STATUS) - def stream_direct(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): - """ - Streams direct message events for the logged-in user, as conversation events. - """ - return self.__stream('/api/v1/streaming/direct', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) - - @api_version("2.5.0", "2.5.0", "2.5.0") - def stream_healthy(self): - """ - Returns without True if streaming API is okay, False or raises an error otherwise. - """ - api_okay = self.__api_request('GET', '/api/v1/streaming/health', base_url_override=self.__get_streaming_base(), parse=False) - if api_okay in [b'OK', b'success']: - return True - return False diff --git a/mastodon/accounts.py b/mastodon/accounts.py index 5ecdf93..219edac 100644 --- a/mastodon/accounts.py +++ b/mastodon/accounts.py @@ -1,5 +1,10 @@ +# accounts.py - account related endpoints + +import collections + +from .versions import _DICT_VERSION_ACCOUNT, _DICT_VERSION_STATUS, _DICT_VERSION_RELATIONSHIP, _DICT_VERSION_LIST, _DICT_VERSION_FAMILIAR_FOLLOWERS, _DICT_VERSION_HASHTAG from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS -from .error import MastodonIllegalArgumentError, MastodonAPIError +from .errors import MastodonIllegalArgumentError, MastodonAPIError from .utility import api_version from .internals import Mastodon as Internals @@ -105,3 +110,391 @@ class Mastodon(Internals): Only available to the app that the user originally signed up with. """ self.__api_request('POST', '/api/v1/emails/confirmations') + + ### + # Reading data: Accounts + ### + @api_version("1.0.0", "1.0.0", _DICT_VERSION_ACCOUNT) + def account(self, id): + """ + Fetch account information by user `id`. + + Does not require authentication for publicly visible accounts. + + Returns a :ref:`account dict `. + """ + id = self.__unpack_id(id) + url = '/api/v1/accounts/{0}'.format(str(id)) + return self.__api_request('GET', url) + + @api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT) + def account_verify_credentials(self): + """ + Fetch logged-in user's account information. + + Returns a :ref:`account dict ` (Starting from 2.1.0, with an additional "source" field). + """ + return self.__api_request('GET', '/api/v1/accounts/verify_credentials') + + @api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT) + def me(self): + """ + Get this user's account. Synonym for `account_verify_credentials()`, does exactly + the same thing, just exists becase `account_verify_credentials()` has a confusing + name. + """ + return self.account_verify_credentials() + + @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) + def account_statuses(self, id, only_media=False, pinned=False, exclude_replies=False, exclude_reblogs=False, tagged=None, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch statuses by user `id`. Same options as :ref:`timeline() ` are permitted. + Returned toots are from the perspective of the logged-in user, i.e. + all statuses visible to the logged-in user (including DMs) are + included. + + If `only_media` is set, return only statuses with media attachments. + If `pinned` is set, return only statuses that have been pinned. Note that + as of Mastodon 2.1.0, this only works properly for instance-local users. + If `exclude_replies` is set, filter out all statuses that are replies. + If `exclude_reblogs` is set, filter out all statuses that are reblogs. + If `tagged` is set, return only statuses that are tagged with `tagged`. Only a single tag without a '#' is valid. + + Does not require authentication for Mastodon versions after 2.7.0 (returns + publicly visible statuses in that case), for publicly visible accounts. + + Returns a list of :ref:`status dicts `. + """ + id = self.__unpack_id(id) + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals(), ['id']) + if not pinned: + del params["pinned"] + if not only_media: + del params["only_media"] + if not exclude_replies: + del params["exclude_replies"] + if not exclude_reblogs: + del params["exclude_reblogs"] + + url = '/api/v1/accounts/{0}/statuses'.format(str(id)) + return self.__api_request('GET', url, params) + + @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) + def account_following(self, id, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch users the given user is following. + + Returns a list of :ref:`account dicts `. + """ + id = self.__unpack_id(id) + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals(), ['id']) + url = '/api/v1/accounts/{0}/following'.format(str(id)) + return self.__api_request('GET', url, params) + + @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) + def account_followers(self, id, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch users the given user is followed by. + + Returns a list of :ref:`account dicts `. + """ + id = self.__unpack_id(id) + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals(), ['id']) + url = '/api/v1/accounts/{0}/followers'.format(str(id)) + return self.__api_request('GET', url, params) + + @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) + def account_relationships(self, id): + """ + Fetch relationship (following, followed_by, blocking, follow requested) of + the logged in user to a given account. `id` can be a list. + + Returns a list of :ref:`relationship dicts `. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/accounts/relationships', + params) + + @api_version("1.0.0", "2.3.0", _DICT_VERSION_ACCOUNT) + def account_search(self, q, limit=None, following=False): + """ + Fetch matching accounts. Will lookup an account remotely if the search term is + in the username@domain format and not yet in the database. Set `following` to + True to limit the search to users the logged-in user follows. + + Returns a list of :ref:`account dicts `. + """ + params = self.__generate_params(locals()) + + if params["following"] == False: + del params["following"] + + return self.__api_request('GET', '/api/v1/accounts/search', params) + + @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) + def account_lists(self, id): + """ + Get all of the logged-in user's lists which the specified user is + a member of. + + Returns a list of :ref:`list dicts `. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ['id']) + url = '/api/v1/accounts/{0}/lists'.format(str(id)) + return self.__api_request('GET', url, params) + + @api_version("3.4.0", "3.4.0", _DICT_VERSION_ACCOUNT) + def account_lookup(self, acct): + """ + Look up an account from user@instance form (@instance allowed but not required for + local accounts). Will only return accounts that the instance already knows about, + and not do any webfinger requests. Use `account_search` if you need to resolve users + through webfinger from remote. + + Returns an :ref:`account dict `. + """ + return self.__api_request('GET', '/api/v1/accounts/lookup', self.__generate_params(locals())) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_FAMILIAR_FOLLOWERS) + def account_familiar_followers(self, id): + """ + Find followers for the account given by id (can be a list) that also follow the + logged in account. + + Returns a list of :ref:`familiar follower dicts ` + """ + if not isinstance(id, list): + id = [id] + for i in range(len(id)): + id[i] = self.__unpack_id(id[i]) + return self.__api_request('GET', '/api/v1/accounts/familiar_followers', {'id': id}, use_json=True) + + ### + # Writing data: Accounts + ### + @api_version("1.0.0", "3.3.0", _DICT_VERSION_RELATIONSHIP) + def account_follow(self, id, reblogs=True, notify=False): + """ + Follow a user. + + Set `reblogs` to False to hide boosts by the followed user. + Set `notify` to True to get a notification every time the followed user posts. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ["id"]) + + if params["reblogs"] is None: + del params["reblogs"] + + url = '/api/v1/accounts/{0}/follow'.format(str(id)) + return self.__api_request('POST', url, params) + + @api_version("1.0.0", "2.1.0", _DICT_VERSION_ACCOUNT) + def follows(self, uri): + """ + Follow a remote user by uri (username@domain). + + Returns a :ref:`account dict `. + """ + params = self.__generate_params(locals()) + return self.__api_request('POST', '/api/v1/follows', params) + + @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) + def account_unfollow(self, id): + """ + Unfollow a user. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/accounts/{0}/unfollow'.format(str(id))) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_RELATIONSHIP) + def account_remove_from_followers(self, id): + """ + Remove a user from the logged in users followers (i.e. make them unfollow the logged in + user / "softblock" them). + + Returns a :ref:`relationship dict ` reflecting the updated following status. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/accounts/{0}/remove_from_followers'.format(str(id))) + + + @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) + def account_block(self, id): + """ + Block a user. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + url = '/api/v1/accounts/{0}/block'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.0.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) + def account_unblock(self, id): + """ + Unblock a user. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + url = '/api/v1/accounts/{0}/unblock'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.1.0", "2.4.3", _DICT_VERSION_RELATIONSHIP) + def account_mute(self, id, notifications=True, duration=None): + """ + Mute a user. + + Set `notifications` to False to receive notifications even though the user is + muted from timelines. Pass a `duration` in seconds to have Mastodon automatically + lift the mute after that many seconds. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ['id']) + url = '/api/v1/accounts/{0}/mute'.format(str(id)) + return self.__api_request('POST', url, params) + + @api_version("1.1.0", "1.4.0", _DICT_VERSION_RELATIONSHIP) + def account_unmute(self, id): + """ + Unmute a user. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + url = '/api/v1/accounts/{0}/unmute'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.1.1", "3.1.0", _DICT_VERSION_ACCOUNT) + def account_update_credentials(self, display_name=None, note=None, + avatar=None, avatar_mime_type=None, + header=None, header_mime_type=None, + locked=None, bot=None, + discoverable=None, fields=None): + """ + Update the profile for the currently logged-in user. + + `note` is the user's bio. + + `avatar` and 'header' are images. As with media uploads, it is possible to either + pass image data and a mime type, or a filename of an image file, for either. + + `locked` specifies whether the user needs to manually approve follow requests. + + `bot` specifies whether the user should be set to a bot. + + `discoverable` specifies whether the user should appear in the user directory. + + `fields` can be a list of up to four name-value pairs (specified as tuples) to + appear as semi-structured information in the user's profile. + + Returns the updated `account dict` of the logged-in user. + """ + params_initial = collections.OrderedDict(locals()) + + # Convert fields + if fields is not None: + if len(fields) > 4: + raise MastodonIllegalArgumentError( + 'A maximum of four fields are allowed.') + + fields_attributes = [] + for idx, (field_name, field_value) in enumerate(fields): + params_initial['fields_attributes[' + + str(idx) + '][name]'] = field_name + params_initial['fields_attributes[' + + str(idx) + '][value]'] = field_value + + # Clean up params + for param in ["avatar", "avatar_mime_type", "header", "header_mime_type", "fields"]: + if param in params_initial: + del params_initial[param] + + # Create file info + files = {} + if avatar is not None: + files["avatar"] = self.__load_media_file(avatar, avatar_mime_type) + if header is not None: + files["header"] = self.__load_media_file(header, header_mime_type) + + params = self.__generate_params(params_initial) + return self.__api_request('PATCH', '/api/v1/accounts/update_credentials', params, files=files) + + @api_version("2.5.0", "2.5.0", _DICT_VERSION_RELATIONSHIP) + def account_pin(self, id): + """ + Pin / endorse a user. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + url = '/api/v1/accounts/{0}/pin'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("2.5.0", "2.5.0", _DICT_VERSION_RELATIONSHIP) + def account_unpin(self, id): + """ + Unpin / un-endorse a user. + + Returns a :ref:`relationship dict ` containing the updated relationship to the user. + """ + id = self.__unpack_id(id) + url = '/api/v1/accounts/{0}/unpin'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("3.2.0", "3.2.0", _DICT_VERSION_RELATIONSHIP) + def account_note_set(self, id, comment): + """ + Set a note (visible to the logged in user only) for the given account. + + Returns a :ref:`status dict ` with the `note` updated. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ["id"]) + return self.__api_request('POST', '/api/v1/accounts/{0}/note'.format(str(id)), params) + + @api_version("3.3.0", "3.3.0", _DICT_VERSION_HASHTAG) + def account_featured_tags(self, id): + """ + Get an account's featured hashtags. + + Returns a list of :ref:`hashtag dicts ` (NOT `featured tag dicts`_). + """ + id = self.__unpack_id(id) + return self.__api_request('GET', '/api/v1/accounts/{0}/featured_tags'.format(str(id))) diff --git a/mastodon/admin.py b/mastodon/admin.py new file mode 100644 index 0000000..e2f5f20 --- /dev/null +++ b/mastodon/admin.py @@ -0,0 +1,572 @@ +# admin.py - admin / moderation endpoints + +from .versions import _DICT_VERSION_ADMIN_ACCOUNT, _DICT_VERSION_REPORT, _DICT_VERSION_HASHTAG, _DICT_VERSION_STATUS, _DICT_VERSION_CARD, \ + _DICT_VERSION_ADMIN_DOMAIN_BLOCK, _DICT_VERSION_ADMIN_MEASURE, _DICT_VERSION_ADMIN_DIMENSION, _DICT_VERSION_ADMIN_RETENTION +from .errors import MastodonIllegalArgumentError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Moderation API + ### + @api_version("2.9.1", "4.0.0", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_accounts_v2(self, origin=None, by_domain=None, status=None, username=None, display_name=None, email=None, ip=None, + permissions=None, invited_by=None, role_ids=None, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetches a list of accounts that match given criteria. By default, local accounts are returned. + + * Set `origin` to "local" or "remote" to get only local or remote accounts. + * Set `by_domain` to a domain to get only accounts from that domain. + * Set `status` to one of "active", "pending", "disabled", "silenced" or "suspended" to get only accounts with that moderation status (default: active) + * Set `username` to a string to get only accounts whose username contains this string. + * Set `display_name` to a string to get only accounts whose display name contains this string. + * Set `email` to an email to get only accounts with that email (this only works on local accounts). + * Set `ip` to an ip (as a string, standard v4/v6 notation) to get only accounts whose last active ip is that ip (this only works on local accounts). + * Set `permissions` to "staff" to only get accounts with staff permissions. + * Set `invited_by` to an account id to get only accounts invited by this user. + * Set `role_ids` to a list of role IDs to get only accounts with those roles. + + Returns a list of :ref:`admin account dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + if role_ids is not None: + if not isinstance(role_ids, list): + role_ids = [role_ids] + role_ids = list(map(self.__unpack_id, role_ids)) + + if invited_by is not None: + invited_by = self.__unpack_id(invited_by) + + if permissions is not None and not permissions in ["staff"]: + raise MastodonIllegalArgumentError("Permissions must be staff if passed") + + if origin is not None and not origin in ["local", "remote"]: + raise MastodonIllegalArgumentError("Origin must be local or remote") + + if status is not None and not status in ["active", "pending", "disabled", "silenced", "suspended"]: + raise MastodonIllegalArgumentError("Status must be local or active, pending, disabled, silenced or suspended") + + if not by_domain is None: + by_domain = self.__deprotocolize(by_domain) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v2/admin/accounts', params) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_accounts(self, remote=False, by_domain=None, status='active', username=None, display_name=None, email=None, ip=None, staff_only=False, max_id=None, min_id=None, since_id=None, limit=None): + """ + Currently a synonym for admin_accounts_v1, now deprecated. You are strongly encouraged to use admin_accounts_v2 instead, since this one is kind of bad. + + !!!!! This function may be switched to calling the v2 API in the future. This is your warning. If you want to keep using v1, use it explicitly. !!!!! + """ + return self.admin_accounts_v1( + remote=remote, + by_domain=by_domain, + status=status, + username=username, + display_name=display_name, + email=email, + ip=ip, + staff_only=staff_only, + max_id=max_id, + min_id=min_id, + since_id=since_id + ) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_accounts_v1(self, remote=False, by_domain=None, status='active', username=None, display_name=None, email=None, ip=None, staff_only=False, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetches a list of accounts that match given criteria. By default, local accounts are returned. + + * Set `remote` to True to get remote accounts, otherwise local accounts are returned (default: local accounts) + * Set `by_domain` to a domain to get only accounts from that domain. + * Set `status` to one of "active", "pending", "disabled", "silenced" or "suspended" to get only accounts with that moderation status (default: active) + * Set `username` to a string to get only accounts whose username contains this string. + * Set `display_name` to a string to get only accounts whose display name contains this string. + * Set `email` to an email to get only accounts with that email (this only works on local accounts). + * Set `ip` to an ip (as a string, standard v4/v6 notation) to get only accounts whose last active ip is that ip (this only works on local accounts). + * Set `staff_only` to True to only get staff accounts (this only works on local accounts). + + Note that setting the boolean parameters to False does not mean "give me users to which this does not apply" but + instead means "I do not care if users have this attribute". + + Deprecated in Mastodon version 3.5.0. + + Returns a list of :ref:`admin account dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals(), ['remote', 'status', 'staff_only']) + + if remote: + params["remote"] = True + + mod_statuses = ["active", "pending", "disabled", "silenced", "suspended"] + if not status in mod_statuses: + raise ValueError("Invalid moderation status requested.") + + if staff_only: + params["staff"] = True + + for mod_status in mod_statuses: + if status == mod_status: + params[status] = True + + if not by_domain is None: + by_domain = self.__deprotocolize(by_domain) + + return self.__api_request('GET', '/api/v1/admin/accounts', params) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account(self, id): + """ + Fetches a single :ref:`admin account dict ` for the user with the given id. + + Returns that dict. + """ + id = self.__unpack_id(id) + return self.__api_request('GET', '/api/v1/admin/accounts/{0}'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_enable(self, id): + """ + Reenables login for a local account for which login has been disabled. + + Returns the updated :ref:`admin account dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/accounts/{0}/enable'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_approve(self, id): + """ + Approves a pending account. + + Returns the updated :ref:`admin account dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/accounts/{0}/approve'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_reject(self, id): + """ + Rejects and deletes a pending account. + + Returns the updated :ref:`admin account dict ` for the account that is now gone. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/accounts/{0}/reject'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_unsilence(self, id): + """ + Unsilences an account. + + Returns the updated :ref:`admin account dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/accounts/{0}/unsilence'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_unsuspend(self, id): + """ + Unsuspends an account. + + Returns the updated :ref:`admin account dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/accounts/{0}/unsuspend'.format(id)) + + @api_version("3.3.0", "3.3.0", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_delete(self, id): + """ + Delete a local user account. + + The deleted accounts :ref:`admin account dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('DELETE', '/api/v1/admin/accounts/{0}'.format(id)) + + @api_version("3.3.0", "3.3.0", _DICT_VERSION_ADMIN_ACCOUNT) + def admin_account_unsensitive(self, id): + """ + Unmark an account as force-sensitive. + + Returns the updated :ref:`admin account dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/accounts/{0}/unsensitive'.format(id)) + + @api_version("2.9.1", "2.9.1", "2.9.1") + def admin_account_moderate(self, id, action=None, report_id=None, warning_preset_id=None, text=None, send_email_notification=True): + """ + Perform a moderation action on an account. + + Valid actions are: + * "disable" - for a local user, disable login. + * "silence" - hide the users posts from all public timelines. + * "suspend" - irreversibly delete all the user's posts, past and future. + * "sensitive" - forcce an accounts media visibility to always be sensitive. + + If no action is specified, the user is only issued a warning. + + Specify the id of a report as `report_id` to close the report with this moderation action as the resolution. + Specify `warning_preset_id` to use a warning preset as the notification text to the user, or `text` to specify text directly. + If both are specified, they are concatenated (preset first). Note that there is currently no API to retrieve or create + warning presets. + + Set `send_email_notification` to False to not send the user an email notification informing them of the moderation action. + """ + if action is None: + action = "none" + + if not send_email_notification: + send_email_notification = None + + id = self.__unpack_id(id) + if report_id is not None: + report_id = self.__unpack_id(report_id) + + params = self.__generate_params(locals(), ['id', 'action']) + + params["type"] = action + + self.__api_request( + 'POST', '/api/v1/admin/accounts/{0}/action'.format(id), params) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) + def admin_reports(self, resolved=False, account_id=None, target_account_id=None, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetches the list of reports. + + Set `resolved` to True to search for resolved reports. `account_id` and `target_account_id` + can be used to get reports filed by or about a specific user. + + Returns a list of :ref:`report dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + if account_id is not None: + account_id = self.__unpack_id(account_id) + + if target_account_id is not None: + target_account_id = self.__unpack_id(target_account_id) + + if not resolved: + resolved = None + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/admin/reports', params) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) + def admin_report(self, id): + """ + Fetches the report with the given id. + + Returns a :ref:`report dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('GET', '/api/v1/admin/reports/{0}'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) + def admin_report_assign(self, id): + """ + Assigns the given report to the logged-in user. + + Returns the updated :ref:`report dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/reports/{0}/assign_to_self'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) + def admin_report_unassign(self, id): + """ + Unassigns the given report from the logged-in user. + + Returns the updated :ref:`report dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/reports/{0}/unassign'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) + def admin_report_reopen(self, id): + """ + Reopens a closed report. + + Returns the updated :ref:`report dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/reports/{0}/reopen'.format(id)) + + @api_version("2.9.1", "2.9.1", _DICT_VERSION_REPORT) + def admin_report_resolve(self, id): + """ + Marks a report as resolved (without taking any action). + + Returns the updated :ref:`report dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('POST', '/api/v1/admin/reports/{0}/resolve'.format(id)) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_HASHTAG) + def admin_trending_tags(self, limit=None): + """ + Admin version of :ref:`trending_tags() `. Includes unapproved tags. + + Returns a list of :ref:`hashtag dicts `, sorted by the instance's trending algorithm, + descending. + """ + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/admin/trends/tags', params) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS) + def admin_trending_statuses(self): + """ + Admin version of :ref:`trending_statuses() `. Includes unapproved tags. + + Returns a list of :ref:`status dicts `, sorted by the instance's trending algorithm, + descending. + """ + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/admin/trends/statuses', params) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_CARD) + def admin_trending_links(self): + """ + Admin version of :ref:`trending_links() `. Includes unapproved tags. + + Returns a list of :ref:`card dicts `, sorted by the instance's trending algorithm, + descending. + """ + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/admin/trends/links', params) + + @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) + def admin_domain_blocks(self, id=None, limit:int=None): + """ + Fetches a list of blocked domains. Requires scope `admin:read:domain_blocks`. + + Provide an `id` to fetch a specific domain block based on its database id. + + Returns a list of :ref:`admin domain block dicts `, raises a `MastodonAPIError` if the specified block does not exist. + """ + if id is not None: + id = self.__unpack_id(id) + return self.__api_request('GET', '/api/v1/admin/domain_blocks/{0}'.format(id)) + else: + params = self.__generate_params(locals(),['limit']) + return self.__api_request('GET', '/api/v1/admin/domain_blocks/', params) + + @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) + def admin_create_domain_block(self, domain:str, severity:str=None, reject_media:bool=None, reject_reports:bool=None, private_comment:str=None, public_comment:str=None, obfuscate:bool=None): + """ + Perform a moderation action on a domain. Requires scope `admin:write:domain_blocks`. + + Valid severities are: + * "silence" - hide all posts from federated timelines and do not show notifications to local users from the remote instance's users unless they are following the remote user. + * "suspend" - deny interactions with this instance going forward. This action is reversible. + * "limit" - generally used with reject_media=true to force reject media from an instance without silencing or suspending.. + + If no action is specified, the domain is only silenced. + `domain` is the domain to block. Note that using the top level domain will also imapct all subdomains. ie, example.com will also impact subdomain.example.com. + `reject_media` will not download remote media on to your local instance media storage. + `reject_reports` ignores all reports from the remote instance. + `private_comment` sets a private admin comment for the domain. + `public_comment` sets a publicly available comment for this domain, which will be available to local users and may be available to everyone depending on your settings. + `obfuscate` censors some part of the domain name. Useful if the domain name contains unwanted words like slurs. + + Returns the new domain block as an :ref:`admin domain block dict `. + """ + if domain is None: + raise AttributeError("Must provide a domain to block a domain") + params = self.__generate_params(locals()) + return self.__api_request('POST', '/api/v1/admin/domain_blocks/', params) + + @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) + def admin_update_domain_block(self, id, severity:str=None, reject_media:bool=None, reject_reports:bool=None, private_comment:str=None, public_comment:str=None, obfuscate:bool=None): + """ + Modify existing moderation action on a domain. Requires scope `admin:write:domain_blocks`. + + Valid severities are: + * "silence" - hide all posts from federated timelines and do not show notifications to local users from the remote instance's users unless they are following the remote user. + * "suspend" - deny interactions with this instance going forward. This action is reversible. + * "limit" - generally used with reject_media=true to force reject media from an instance without silencing or suspending. + + If no action is specified, the domain is only silenced. + `domain` is the domain to block. Note that using the top level domain will also imapct all subdomains. ie, example.com will also impact subdomain.example.com. + `reject_media` will not download remote media on to your local instance media storage. + `reject_reports` ignores all reports from the remote instance. + `private_comment` sets a private admin comment for the domain. + `public_comment` sets a publicly available comment for this domain, which will be available to local users and may be available to everyone depending on your settings. + `obfuscate` censors some part of the domain name. Useful if the domain name contains unwanted words like slurs. + + Returns the modified domain block as an :ref:`admin domain block dict `, raises a `MastodonAPIError` if the specified block does not exist. + """ + if id is None: + raise AttributeError("Must provide an id to modify the existing moderation actions on a given domain.") + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ["id"]) + return self.__api_request('PUT', '/api/v1/admin/domain_blocks/{0}'.format(id), params) + + @api_version("4.0.0", "4.0.0", _DICT_VERSION_ADMIN_DOMAIN_BLOCK) + def admin_delete_domain_block(self, id=None): + """ + Removes moderation action against a given domain. Requires scope `admin:write:domain_blocks`. + + Provide an `id` to remove a specific domain block based on its database id. + + Raises a `MastodonAPIError` if the specified block does not exist. + """ + if id is not None: + id = self.__unpack_id(id) + self.__api_request('DELETE', '/api/v1/admin/domain_blocks/{0}'.format(id)) + else: + raise AttributeError("You must provide an id of an existing domain block to remove it.") + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_ADMIN_MEASURE) + def admin_measures(self, start_at, end_at, active_users=False, new_users=False, interactions=False, opened_reports = False, resolved_reports=False, + tag_accounts=None, tag_uses=None, tag_servers=None, instance_accounts=None, instance_media_attachments=None, instance_reports=None, + instance_statuses=None, instance_follows=None, instance_followers=None): + """ + Retrieves numerical instance information for the time period (at day granularity) between `start_at` and `end_at`. + + * `active_users`: Pass true to retrieve the number of active users on your instance within the time period + * `new_users`: Pass true to retrieve the number of users who joined your instance within the time period + * `interactions`: Pass true to retrieve the number of interactions (favourites, boosts, replies) on local statuses within the time period + * `opened_reports`: Pass true to retrieve the number of reports filed within the time period + * `resolved_reports` = Pass true to retrieve the number of reports resolved within the time period + * `tag_accounts`: Pass a tag ID to get the number of accounts which used that tag in at least one status within the time period + * `tag_uses`: Pass a tag ID to get the number of statuses which used that tag within the time period + * `tag_servers`: Pass a tag ID to to get the number of remote origin servers for statuses which used that tag within the time period + * `instance_accounts`: Pass a domain to get the number of accounts originating from that remote domain within the time period + * `instance_media_attachments`: Pass a domain to get the amount of space used by media attachments from that remote domain within the time period + * `instance_reports`: Pass a domain to get the number of reports filed against accounts from that remote domain within the time period + * `instance_statuses`: Pass a domain to get the number of statuses originating from that remote domain within the time period + * `instance_follows`: Pass a domain to get the number of accounts from a remote domain followed by that local user within the time period + * `instance_followers`: Pass a domain to get the number of local accounts followed by accounts from that remote domain within the time period + + This API call is relatively expensive - watch your servers load if you want to get a lot of statistical data. Especially the instance_statuses stats + might take a long time to compute and, in fact, time out. + + There is currently no way to get tag IDs implemented in Mastodon.py, because the Mastodon public API does not implement one. This will be fixed in a future + release. + + Returns a list of :ref:`admin measure dicts `. + """ + params_init = locals() + keys = [] + for key in ["active_users", "new_users", "interactions", "opened_reports", "resolved_reports"]: + if params_init[key] == True: + keys.append(key) + + params = {} + for key in ["tag_accounts", "tag_uses", "tag_servers"]: + if params_init[key] is not None: + keys.append(key) + params[key] = {"id": self.__unpack_id(params_init[key])} + for key in ["instance_accounts", "instance_media_attachments", "instance_reports", "instance_statuses", "instance_follows", "instance_followers"]: + if params_init[key] is not None: + keys.append(key) + params[key] = {"domain": Mastodon.__deprotocolize(params_init[key]).split("/")[0]} + + if len(keys) == 0: + raise MastodonIllegalArgumentError("Must request at least one metric.") + + params["keys"] = keys + params["start_at"] = self.__consistent_isoformat_utc(start_at) + params["end_at"] = self.__consistent_isoformat_utc(end_at) + + return self.__api_request('POST', '/api/v1/admin/measures', params, use_json=True) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_ADMIN_DIMENSION) + def admin_dimensions(self, start_at, end_at, limit=None, languages=False, sources=False, servers=False, space_usage=False, software_versions=False, + tag_servers=None, tag_languages=None, instance_accounts=None, instance_languages=None): + """ + Retrieves primarily categorical instance information for the time period (at day granularity) between `start_at` and `end_at`. + + * `languages`: Pass true to get the most-used languages on this server + * `sources`: Pass true to get the most-used client apps on this server + * `servers`: Pass true to get the remote servers with the most statuses + * `space_usage`: Pass true to get the how much space is used by different components your software stack + * `software_versions`: Pass true to get the version numbers for your software stack + * `tag_servers`: Pass a tag ID to get the most-common servers for statuses including a trending tag + * `tag_languages`: Pass a tag ID to get the most-used languages for statuses including a trending tag + * `instance_accounts`: Pass a domain to get the most-followed accounts from a remote server + * `instance_languages`: Pass a domain to get the most-used languages from a remote server + + Pass `limit` to set how many results you want on queries where that makes sense. + + This API call is relatively expensive - watch your servers load if you want to get a lot of statistical data. + + There is currently no way to get tag IDs implemented in Mastodon.py, because the Mastodon public API does not implement one. This will be fixed in a future + release. + + Returns a list of :ref:`admin dimension dicts `. + """ + params_init = locals() + keys = [] + for key in ["languages", "sources", "servers", "space_usage", "software_versions"]: + if params_init[key] == True: + keys.append(key) + + params = {} + for key in ["tag_servers", "tag_languages"]: + if params_init[key] is not None: + keys.append(key) + params[key] = {"id": self.__unpack_id(params_init[key])} + for key in ["instance_accounts", "instance_languages"]: + if params_init[key] is not None: + keys.append(key) + params[key] = {"domain": Mastodon.__deprotocolize(params_init[key]).split("/")[0]} + + if len(keys) == 0: + raise MastodonIllegalArgumentError("Must request at least one dimension.") + + params["keys"] = keys + if limit is not None: + params["limit"] = limit + params["start_at"] = self.__consistent_isoformat_utc(start_at) + params["end_at"] = self.__consistent_isoformat_utc(end_at) + + return self.__api_request('POST', '/api/v1/admin/dimensions', params, use_json=True) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_ADMIN_RETENTION) + def admin_retention(self, start_at, end_at, frequency="day"): + """ + Gets user retention statistics (at `frequency` - "day" or "month" - granularity) between `start_at` and `end_at`. + + Returns a list of :ref:`admin retention dicts ` + """ + if not frequency in ["day", "month"]: + raise MastodonIllegalArgumentError("Frequency must be day or month") + + params = { + "start_at": self.__consistent_isoformat_utc(start_at), + "end_at": self.__consistent_isoformat_utc(end_at), + "frequency": frequency + } + return self.__api_request('POST', '/api/v1/admin/retention', params) \ No newline at end of file diff --git a/mastodon/authentication.py b/mastodon/authentication.py index b7f15c9..d4cb283 100644 --- a/mastodon/authentication.py +++ b/mastodon/authentication.py @@ -1,3 +1,5 @@ +# authentication.py - app and user creation, login, oauth, getting app info, and the constructor + import requests from requests.models import urlencode import datetime @@ -5,9 +7,11 @@ import os import time import collections -from .error import MastodonIllegalArgumentError, MastodonNetworkError, MastodonVersionError, MastodonAPIError +from .errors import MastodonIllegalArgumentError, MastodonNetworkError, MastodonVersionError, MastodonAPIError +from .versions import _DICT_VERSION_APPLICATION from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS, _DEFAULT_TIMEOUT -from .utility import parse_version_string +from .utility import parse_version_string, api_version + from .internals import Mastodon as Internals class Mastodon(Internals): @@ -367,4 +371,16 @@ class Mastodon(Internals): # We are now logged out, clear token and logged in id self.access_token = None - self.__logged_in_id = None \ No newline at end of file + self.__logged_in_id = None + + ### + # Reading data: Apps + ### + @api_version("2.0.0", "2.7.2", _DICT_VERSION_APPLICATION) + def app_verify_credentials(self): + """ + Fetch information about the current application. + + Returns an :ref:`application dict `. + """ + return self.__api_request('GET', '/api/v1/apps/verify_credentials') diff --git a/mastodon/conversations.py b/mastodon/conversations.py new file mode 100644 index 0000000..ba3ee61 --- /dev/null +++ b/mastodon/conversations.py @@ -0,0 +1,43 @@ +# conversations.py - conversation endpoints + +from .versions import _DICT_VERSION_CONVERSATION +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Conversations + ### + @api_version("2.6.0", "2.6.0", _DICT_VERSION_CONVERSATION) + def conversations(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetches a user's conversations. + + Returns a list of :ref:`conversation dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', "/api/v1/conversations/", params) + + ### + # Writing data: Conversations + ### + @api_version("2.6.0", "2.6.0", _DICT_VERSION_CONVERSATION) + def conversations_read(self, id): + """ + Marks a single conversation as read. + + Returns the updated :ref:`conversation dict `. + """ + id = self.__unpack_id(id) + url = '/api/v1/conversations/{0}/read'.format(str(id)) + return self.__api_request('POST', url) diff --git a/mastodon/endorsements.py b/mastodon/endorsements.py new file mode 100644 index 0000000..b6c27a0 --- /dev/null +++ b/mastodon/endorsements.py @@ -0,0 +1,19 @@ +# notifications.py - endorsement endpoints + +from .versions import _DICT_VERSION_ACCOUNT +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Endorsements + ### + @api_version("2.5.0", "2.5.0", _DICT_VERSION_ACCOUNT) + def endorsements(self): + """ + Fetch list of users endorsed by the logged-in user. + + Returns a list of :ref:`account dicts `. + """ + return self.__api_request('GET', '/api/v1/endorsements') diff --git a/mastodon/error.py b/mastodon/error.py deleted file mode 100644 index 85cc313..0000000 --- a/mastodon/error.py +++ /dev/null @@ -1,90 +0,0 @@ -# error.py - error classes - -## -# Exceptions -## -class MastodonError(Exception): - """Base class for Mastodon.py exceptions""" - - -class MastodonVersionError(MastodonError): - """Raised when a function is called that the version of Mastodon for which - Mastodon.py was instantiated does not support""" - - -class MastodonIllegalArgumentError(ValueError, MastodonError): - """Raised when an incorrect parameter is passed to a function""" - pass - - -class MastodonIOError(IOError, MastodonError): - """Base class for Mastodon.py I/O errors""" - - -class MastodonFileNotFoundError(MastodonIOError): - """Raised when a file requested to be loaded can not be opened""" - pass - - -class MastodonNetworkError(MastodonIOError): - """Raised when network communication with the server fails""" - pass - - -class MastodonReadTimeout(MastodonNetworkError): - """Raised when a stream times out""" - pass - - -class MastodonAPIError(MastodonError): - """Raised when the mastodon API generates a response that cannot be handled""" - pass - - -class MastodonServerError(MastodonAPIError): - """Raised if the Server is malconfigured and returns a 5xx error code""" - pass - - -class MastodonInternalServerError(MastodonServerError): - """Raised if the Server returns a 500 error""" - pass - - -class MastodonBadGatewayError(MastodonServerError): - """Raised if the Server returns a 502 error""" - pass - - -class MastodonServiceUnavailableError(MastodonServerError): - """Raised if the Server returns a 503 error""" - pass - - -class MastodonGatewayTimeoutError(MastodonServerError): - """Raised if the Server returns a 504 error""" - pass - - -class MastodonNotFoundError(MastodonAPIError): - """Raised when the Mastodon API returns a 404 Not Found error""" - pass - - -class MastodonUnauthorizedError(MastodonAPIError): - """Raised when the Mastodon API returns a 401 Unauthorized error - - This happens when an OAuth token is invalid or has been revoked, - or when trying to access an endpoint that can't be used without - authentication without providing credentials.""" - pass - - -class MastodonRatelimitError(MastodonError): - """Raised when rate limiting is set to manual mode and the rate limit is exceeded""" - pass - - -class MastodonMalformedEventError(MastodonError): - """Raised when the server-sent event stream is malformed""" - pass diff --git a/mastodon/errors.py b/mastodon/errors.py new file mode 100644 index 0000000..85cc313 --- /dev/null +++ b/mastodon/errors.py @@ -0,0 +1,90 @@ +# error.py - error classes + +## +# Exceptions +## +class MastodonError(Exception): + """Base class for Mastodon.py exceptions""" + + +class MastodonVersionError(MastodonError): + """Raised when a function is called that the version of Mastodon for which + Mastodon.py was instantiated does not support""" + + +class MastodonIllegalArgumentError(ValueError, MastodonError): + """Raised when an incorrect parameter is passed to a function""" + pass + + +class MastodonIOError(IOError, MastodonError): + """Base class for Mastodon.py I/O errors""" + + +class MastodonFileNotFoundError(MastodonIOError): + """Raised when a file requested to be loaded can not be opened""" + pass + + +class MastodonNetworkError(MastodonIOError): + """Raised when network communication with the server fails""" + pass + + +class MastodonReadTimeout(MastodonNetworkError): + """Raised when a stream times out""" + pass + + +class MastodonAPIError(MastodonError): + """Raised when the mastodon API generates a response that cannot be handled""" + pass + + +class MastodonServerError(MastodonAPIError): + """Raised if the Server is malconfigured and returns a 5xx error code""" + pass + + +class MastodonInternalServerError(MastodonServerError): + """Raised if the Server returns a 500 error""" + pass + + +class MastodonBadGatewayError(MastodonServerError): + """Raised if the Server returns a 502 error""" + pass + + +class MastodonServiceUnavailableError(MastodonServerError): + """Raised if the Server returns a 503 error""" + pass + + +class MastodonGatewayTimeoutError(MastodonServerError): + """Raised if the Server returns a 504 error""" + pass + + +class MastodonNotFoundError(MastodonAPIError): + """Raised when the Mastodon API returns a 404 Not Found error""" + pass + + +class MastodonUnauthorizedError(MastodonAPIError): + """Raised when the Mastodon API returns a 401 Unauthorized error + + This happens when an OAuth token is invalid or has been revoked, + or when trying to access an endpoint that can't be used without + authentication without providing credentials.""" + pass + + +class MastodonRatelimitError(MastodonError): + """Raised when rate limiting is set to manual mode and the rate limit is exceeded""" + pass + + +class MastodonMalformedEventError(MastodonError): + """Raised when the server-sent event stream is malformed""" + pass diff --git a/mastodon/favourites.py b/mastodon/favourites.py new file mode 100644 index 0000000..9d86424 --- /dev/null +++ b/mastodon/favourites.py @@ -0,0 +1,52 @@ +# favourites.py - favourites and also bookmarks + +from .versions import _DICT_VERSION_STATUS +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Favourites + ### + @api_version("1.0.0", "2.6.0", _DICT_VERSION_STATUS) + def favourites(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch the logged-in user's favourited statuses. + + Returns a list of :ref:`status dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/favourites', params) + + ### + # Reading data: Bookmarks + ### + @api_version("3.1.0", "3.1.0", _DICT_VERSION_STATUS) + def bookmarks(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Get a list of statuses bookmarked by the logged-in user. + + Returns a list of :ref:`status dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/bookmarks', params) + \ No newline at end of file diff --git a/mastodon/filters.py b/mastodon/filters.py new file mode 100644 index 0000000..5f373d4 --- /dev/null +++ b/mastodon/filters.py @@ -0,0 +1,119 @@ +# filters.py - Filter-related endpoints + +import re + +from .versions import _DICT_VERSION_FILTER +from .errors import MastodonIllegalArgumentError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Keyword filters + ### + @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) + def filters(self): + """ + Fetch all of the logged-in user's filters. + + Returns a list of :ref:`filter dicts `. Not paginated. + """ + return self.__api_request('GET', '/api/v1/filters') + + @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) + def filter(self, id): + """ + Fetches information about the filter with the specified `id`. + + Returns a :ref:`filter dict `. + """ + id = self.__unpack_id(id) + url = '/api/v1/filters/{0}'.format(str(id)) + return self.__api_request('GET', url) + + @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) + def filters_apply(self, objects, filters, context): + """ + Helper function: Applies a list of filters to a list of either statuses + or notifications and returns only those matched by none. This function will + apply all filters that match the context provided in `context`, i.e. + if you want to apply only notification-relevant filters, specify + 'notifications'. Valid contexts are 'home', 'notifications', 'public' and 'thread'. + """ + + # Build filter regex + filter_strings = [] + for keyword_filter in filters: + if not context in keyword_filter["context"]: + continue + + filter_string = re.escape(keyword_filter["phrase"]) + if keyword_filter["whole_word"]: + filter_string = "\\b" + filter_string + "\\b" + filter_strings.append(filter_string) + filter_re = re.compile("|".join(filter_strings), flags=re.IGNORECASE) + + # Apply + filter_results = [] + for filter_object in objects: + filter_status = filter_object + if "status" in filter_object: + filter_status = filter_object["status"] + filter_text = filter_status["content"] + filter_text = re.sub(r"<.*?>", " ", filter_text) + filter_text = re.sub(r"\s+", " ", filter_text).strip() + if not filter_re.search(filter_text): + filter_results.append(filter_object) + return filter_results + + ### + # Writing data: Keyword filters + ### + @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) + def filter_create(self, phrase, context, irreversible=False, whole_word=True, expires_in=None): + """ + Creates a new keyword filter. `phrase` is the phrase that should be + filtered out, `context` specifies from where to filter the keywords. + Valid contexts are 'home', 'notifications', 'public' and 'thread'. + + Set `irreversible` to True if you want the filter to just delete statuses + server side. This works only for the 'home' and 'notifications' contexts. + + Set `whole_word` to False if you want to allow filter matches to + start or end within a word, not only at word boundaries. + + Set `expires_in` to specify for how many seconds the filter should be + kept around. + + Returns the :ref:`filter dict ` of the newly created filter. + """ + params = self.__generate_params(locals()) + + for context_val in context: + if not context_val in ['home', 'notifications', 'public', 'thread']: + raise MastodonIllegalArgumentError('Invalid filter context.') + + return self.__api_request('POST', '/api/v1/filters', params) + + @api_version("2.4.3", "2.4.3", _DICT_VERSION_FILTER) + def filter_update(self, id, phrase=None, context=None, irreversible=None, whole_word=None, expires_in=None): + """ + Updates the filter with the given `id`. Parameters are the same + as in `filter_create()`. + + Returns the :ref:`filter dict ` of the updated filter. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ['id']) + url = '/api/v1/filters/{0}'.format(str(id)) + return self.__api_request('PUT', url, params) + + @api_version("2.4.3", "2.4.3", "2.4.3") + def filter_delete(self, id): + """ + Deletes the filter with the given `id`. + """ + id = self.__unpack_id(id) + url = '/api/v1/filters/{0}'.format(str(id)) + self.__api_request('DELETE', url) \ No newline at end of file diff --git a/mastodon/hashtags.py b/mastodon/hashtags.py new file mode 100644 index 0000000..89e8cac --- /dev/null +++ b/mastodon/hashtags.py @@ -0,0 +1,52 @@ +# hashtags.py - hashtag and featured-hashtag endpoints + +from .versions import _DICT_VERSION_FEATURED_TAG, _DICT_VERSION_HASHTAG +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Featured hashtags + ### + @api_version("3.0.0", "3.0.0", _DICT_VERSION_FEATURED_TAG) + def featured_tags(self): + """ + Return the hashtags the logged-in user has set to be featured on + their profile as a list of :ref:`featured tag dicts `. + + Returns a list of :ref:`featured tag dicts `. + """ + return self.__api_request('GET', '/api/v1/featured_tags') + + @api_version("3.0.0", "3.0.0", _DICT_VERSION_HASHTAG) + def featured_tag_suggestions(self): + """ + Returns the logged-in user's 10 most commonly-used hashtags. + + Returns a list of :ref:`hashtag dicts `. + """ + return self.__api_request('GET', '/api/v1/featured_tags/suggestions') + + ### + # Writing data: Featured hashtags + ### + @api_version("3.0.0", "3.0.0", _DICT_VERSION_FEATURED_TAG) + def featured_tag_create(self, name): + """ + Creates a new featured hashtag displayed on the logged-in user's profile. + + Returns a :ref:`featured tag dict ` with the newly featured tag. + """ + params = self.__generate_params(locals()) + return self.__api_request('POST', '/api/v1/featured_tags', params) + + @api_version("3.0.0", "3.0.0", _DICT_VERSION_FEATURED_TAG) + def featured_tag_delete(self, id): + """ + Deletes one of the logged-in user's featured hashtags. + """ + id = self.__unpack_id(id) + url = '/api/v1/featured_tags/{0}'.format(str(id)) + self.__api_request('DELETE', url) + \ No newline at end of file diff --git a/mastodon/instance.py b/mastodon/instance.py index 88445d1..dfbbefb 100644 --- a/mastodon/instance.py +++ b/mastodon/instance.py @@ -1,5 +1,7 @@ -from .versions import _DICT_VERSION_INSTANCE, _DICT_VERSION_ACTIVITY -from .error import MastodonIllegalArgumentError, MastodonNotFoundError +# instance.py - instance-level endpoints, directory, emoji, announcements + +from .versions import _DICT_VERSION_INSTANCE, _DICT_VERSION_ACTIVITY, _DICT_VERSION_ACCOUNT, _DICT_VERSION_EMOJI, _DICT_VERSION_ANNOUNCEMENT +from .errors import MastodonIllegalArgumentError, MastodonNotFoundError from .utility import api_version from .compat import urlparse @@ -94,3 +96,94 @@ class Mastodon(Internals): Returns a list of `id` + `text` dicts, same as the `rules` field in the :ref:`instance dicts `. """ return self.__api_request('GET', '/api/v1/instance/rules') + + ### + # Reading data: Directory + ### + @api_version("3.0.0", "3.0.0", _DICT_VERSION_ACCOUNT) + def directory(self, offset=None, limit=None, order=None, local=None): + """ + Fetch the contents of the profile directory, if enabled on the server. + + `offset` how many accounts to skip before returning results. Default 0. + + `limit` how many accounts to load. Default 40. + + `order` "active" to sort by most recently posted statuses (default) or + "new" to sort by most recently created profiles. + + `local` True to return only local accounts. + + Returns a list of :ref:`account dicts `. + + """ + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/directory', params) + + ### + # Reading data: Emoji + ### + @api_version("2.1.0", "2.1.0", _DICT_VERSION_EMOJI) + def custom_emojis(self): + """ + Fetch the list of custom emoji the instance has installed. + + Does not require authentication unless locked down by the administrator. + + Returns a list of :ref:`emoji dicts `. + """ + return self.__api_request('GET', '/api/v1/custom_emojis') + + ## + # Reading data: Announcements + ## + @api_version("3.1.0", "3.1.0", _DICT_VERSION_ANNOUNCEMENT) + def announcements(self): + """ + Fetch currently active announcements. + + Returns a list of :ref:`announcement dicts `. + """ + return self.__api_request('GET', '/api/v1/announcements') + + ### + # Writing data: Annoucements + ### + @api_version("3.1.0", "3.1.0", "3.1.0") + def announcement_dismiss(self, id): + """ + Set the given annoucement to read. + """ + id = self.__unpack_id(id) + + url = '/api/v1/announcements/{0}/dismiss'.format(str(id)) + self.__api_request('POST', url) + + @api_version("3.1.0", "3.1.0", "3.1.0") + def announcement_reaction_create(self, id, reaction): + """ + Add a reaction to an announcement. `reaction` can either be a unicode emoji + or the name of one of the instances custom emoji. + + Will throw an API error if the reaction name is not one of the allowed things + or when trying to add a reaction that the user has already added (adding a + reaction that a different user added is legal and increments the count). + """ + id = self.__unpack_id(id) + + url = '/api/v1/announcements/{0}/reactions/{1}'.format( + str(id), reaction) + self.__api_request('PUT', url) + + @api_version("3.1.0", "3.1.0", "3.1.0") + def announcement_reaction_delete(self, id, reaction): + """ + Remove a reaction to an announcement. + + Will throw an API error if the reaction does not exist. + """ + id = self.__unpack_id(id) + + url = '/api/v1/announcements/{0}/reactions/{1}'.format( + str(id), reaction) + self.__api_request('DELETE', url) diff --git a/mastodon/internals.py b/mastodon/internals.py index 0e77421..aa82553 100644 --- a/mastodon/internals.py +++ b/mastodon/internals.py @@ -1,3 +1,5 @@ +# internals.py - many internal helpers + import datetime from contextlib import closing import mimetypes @@ -14,7 +16,7 @@ import base64 import os from .utility import AttribAccessDict, AttribAccessList, parse_version_string -from .error import MastodonNetworkError, MastodonIllegalArgumentError, MastodonRatelimitError, MastodonNotFoundError, \ +from .errors import MastodonNetworkError, MastodonIllegalArgumentError, MastodonRatelimitError, MastodonNotFoundError, \ MastodonUnauthorizedError, MastodonInternalServerError, MastodonBadGatewayError, MastodonServiceUnavailableError, \ MastodonGatewayTimeoutError, MastodonServerError, MastodonAPIError, MastodonMalformedEventError from .compat import urlparse, magic, PurePath diff --git a/mastodon/lists.py b/mastodon/lists.py new file mode 100644 index 0000000..0882133 --- /dev/null +++ b/mastodon/lists.py @@ -0,0 +1,112 @@ +# list.py - list endpoints + +from .versions import _DICT_VERSION_LIST, _DICT_VERSION_ACCOUNT +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Lists + ### + @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) + def lists(self): + """ + Fetch a list of all the Lists by the logged-in user. + + Returns a list of :ref:`list dicts `. + """ + return self.__api_request('GET', '/api/v1/lists') + + @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) + def list(self, id): + """ + Fetch info about a specific list. + + Returns a :ref:`list dict `. + """ + id = self.__unpack_id(id) + return self.__api_request('GET', '/api/v1/lists/{0}'.format(id)) + + @api_version("2.1.0", "2.6.0", _DICT_VERSION_ACCOUNT) + def list_accounts(self, id, max_id=None, min_id=None, since_id=None, limit=None): + """ + Get the accounts that are on the given list. + + Returns a list of :ref:`account dicts `. + """ + id = self.__unpack_id(id) + + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals(), ['id']) + return self.__api_request('GET', '/api/v1/lists/{0}/accounts'.format(id)) + + ### + # Writing data: Lists + ### + @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) + def list_create(self, title): + """ + Create a new list with the given `title`. + + Returns the :ref:`list dict ` of the created list. + """ + params = self.__generate_params(locals()) + return self.__api_request('POST', '/api/v1/lists', params) + + @api_version("2.1.0", "2.1.0", _DICT_VERSION_LIST) + def list_update(self, id, title): + """ + Update info about a list, where "info" is really the lists `title`. + + Returns the :ref:`list dict ` of the modified list. + """ + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ['id']) + return self.__api_request('PUT', '/api/v1/lists/{0}'.format(id), params) + + @api_version("2.1.0", "2.1.0", "2.1.0") + def list_delete(self, id): + """ + Delete a list. + """ + id = self.__unpack_id(id) + self.__api_request('DELETE', '/api/v1/lists/{0}'.format(id)) + + @api_version("2.1.0", "2.1.0", "2.1.0") + def list_accounts_add(self, id, account_ids): + """ + Add the account(s) given in `account_ids` to the list. + """ + id = self.__unpack_id(id) + + if not isinstance(account_ids, list): + account_ids = [account_ids] + account_ids = list(map(lambda x: self.__unpack_id(x), account_ids)) + + params = self.__generate_params(locals(), ['id']) + self.__api_request( + 'POST', '/api/v1/lists/{0}/accounts'.format(id), params) + + @api_version("2.1.0", "2.1.0", "2.1.0") + def list_accounts_delete(self, id, account_ids): + """ + Remove the account(s) given in `account_ids` from the list. + """ + id = self.__unpack_id(id) + + if not isinstance(account_ids, list): + account_ids = [account_ids] + account_ids = list(map(lambda x: self.__unpack_id(x), account_ids)) + + params = self.__generate_params(locals(), ['id']) + self.__api_request( + 'DELETE', '/api/v1/lists/{0}/accounts'.format(id), params) \ No newline at end of file diff --git a/mastodon/media.py b/mastodon/media.py new file mode 100644 index 0000000..3c815fb --- /dev/null +++ b/mastodon/media.py @@ -0,0 +1,114 @@ +# admin.py - admin / moderation endpoints + +import time + +from .versions import _DICT_VERSION_MEDIA +from .errors import MastodonVersionError, MastodonAPIError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Media + ### + @api_version("3.1.4", "3.1.4", _DICT_VERSION_MEDIA) + def media(self, id): + """ + Get the updated JSON for one non-attached / in progress media upload belonging + to the logged-in user. + """ + id = self.__unpack_id(id) + return self.__api_request('GET', '/api/v1/media/{0}'.format(str(id))) + + ### + # Writing data: Media + ### + @api_version("1.0.0", "3.2.0", _DICT_VERSION_MEDIA) + def media_post(self, media_file, mime_type=None, description=None, focus=None, file_name=None, thumbnail=None, thumbnail_mime_type=None, synchronous=False): + """ + Post an image, video or audio file. `media_file` can either be data or + a file name. If data is passed directly, the mime type has to be specified + manually, otherwise, it is determined from the file name. `focus` should be a tuple + of floats between -1 and 1, giving the x and y coordinates of the images + focus point for cropping (with the origin being the images center). + + Throws a `MastodonIllegalArgumentError` if the mime type of the + passed data or file can not be determined properly. + + `file_name` can be specified to upload a file with the given name, + which is ignored by Mastodon, but some other Fediverse server software + will display it. If no name is specified, a random name will be generated. + The filename of a file specified in media_file will be ignored. + + Starting with Mastodon 3.2.0, `thumbnail` can be specified in the same way as `media_file` + to upload a custom thumbnail image for audio and video files. + + Returns a :ref:`media dict `. This contains the id that can be used in + status_post to attach the media file to a toot. + + When using the v2 API (post Mastodon version 3.1.4), the `url` in the + returned dict will be `null`, since attachments are processed + asynchronously. You can fetch an updated dict using `media`. Pass + "synchronous" to emulate the old behaviour. Not recommended, inefficient + and deprecated, will eat your API quota, you know the deal. + """ + files = {'file': self.__load_media_file( + media_file, mime_type, file_name)} + + if focus is not None: + focus = str(focus[0]) + "," + str(focus[1]) + + if thumbnail is not None: + if not self.verify_minimum_version("3.2.0", cached=True): + raise MastodonVersionError('Thumbnail requires version > 3.2.0') + files["thumbnail"] = self.__load_media_file( + thumbnail, thumbnail_mime_type) + + # Disambiguate URL by version + if self.verify_minimum_version("3.1.4", cached=True): + ret_dict = self.__api_request( + 'POST', '/api/v2/media', files=files, params={'description': description, 'focus': focus}) + else: + ret_dict = self.__api_request( + 'POST', '/api/v1/media', files=files, params={'description': description, 'focus': focus}) + + # Wait for processing? + if synchronous: + if self.verify_minimum_version("3.1.4"): + while not "url" in ret_dict or ret_dict.url is None: + try: + ret_dict = self.media(ret_dict) + time.sleep(5.0) + except: + raise MastodonAPIError("Attachment could not be processed") + else: + # Old version always waits + return ret_dict + + return ret_dict + + @api_version("2.3.0", "3.2.0", _DICT_VERSION_MEDIA) + def media_update(self, id, description=None, focus=None, thumbnail=None, thumbnail_mime_type=None): + """ + Update the metadata of the media file with the given `id`. `description` and + `focus` and `thumbnail` are as in :ref:`media_post() ` . + + Returns the updated :ref:`media dict `. + """ + id = self.__unpack_id(id) + + if focus is not None: + focus = str(focus[0]) + "," + str(focus[1]) + + params = self.__generate_params( + locals(), ['id', 'thumbnail', 'thumbnail_mime_type']) + + if thumbnail is not None: + if not self.verify_minimum_version("3.2.0", cached=True): + raise MastodonVersionError('Thumbnail requires version > 3.2.0') + files = {"thumbnail": self.__load_media_file( + thumbnail, thumbnail_mime_type)} + return self.__api_request('PUT', '/api/v1/media/{0}'.format(str(id)), params, files=files) + else: + return self.__api_request('PUT', '/api/v1/media/{0}'.format(str(id)), params) diff --git a/mastodon/notifications.py b/mastodon/notifications.py new file mode 100644 index 0000000..f65b3fb --- /dev/null +++ b/mastodon/notifications.py @@ -0,0 +1,92 @@ +# notifications.py - notification endpoints + +from .versions import _DICT_VERSION_NOTIFICATION +from .errors import MastodonIllegalArgumentError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Notifications + ### + @api_version("1.0.0", "3.5.0", _DICT_VERSION_NOTIFICATION) + def notifications(self, id=None, account_id=None, max_id=None, min_id=None, since_id=None, limit=None, exclude_types=None, types=None, mentions_only=None): + """ + Fetch notifications (mentions, favourites, reblogs, follows) for the logged-in + user. Pass `account_id` to get only notifications originating from the given account. + + There are different types of notifications: + * `follow` - A user followed the logged in user + * `follow_request` - A user has requested to follow the logged in user (for locked accounts) + * `favourite` - A user favourited a post by the logged in user + * `reblog` - A user reblogged a post by the logged in user + * `mention` - A user mentioned the logged in user + * `poll` - A poll the logged in user created or voted in has ended + * `update` - A status the logged in user has reblogged (and only those, as of 4.0.0) has been edited + * `status` - A user that the logged in user has enabned notifications for has enabled `notify` (see :ref:`account_follow() `) + * `admin.sign_up` - For accounts with appropriate permissions (TODO: document which those are when adding the permission API): A new user has signed up + * `admin.report` - For accounts with appropriate permissions (TODO: document which those are when adding the permission API): A new report has been received + Parameters `exclude_types` and `types` are array of these types, specifying them will in- or exclude the + types of notifications given. It is legal to give both parameters at the same tine, the result will then + be the intersection of the results of both filters. Specifying `mentions_only` is a deprecated way to set + `exclude_types` to all but mentions. + + Can be passed an `id` to fetch a single notification. + + Returns a list of :ref:`notification dicts `. + """ + if mentions_only is not None: + if exclude_types is None and types is None: + if mentions_only: + if self.verify_minimum_version("3.5.0", cached=True): + types = ["mention"] + else: + exclude_types = ["follow", "favourite", "reblog", "poll", "follow_request"] + else: + raise MastodonIllegalArgumentError('Cannot specify exclude_types/types when mentions_only is present') + del mentions_only + + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + if account_id is not None: + account_id = self.__unpack_id(account_id) + + if id is None: + params = self.__generate_params(locals(), ['id']) + return self.__api_request('GET', '/api/v1/notifications', params) + else: + id = self.__unpack_id(id) + url = '/api/v1/notifications/{0}'.format(str(id)) + return self.__api_request('GET', url) + + ### + # Writing data: Notifications + ### + @api_version("1.0.0", "1.0.0", "1.0.0") + def notifications_clear(self): + """ + Clear out a user's notifications + """ + self.__api_request('POST', '/api/v1/notifications/clear') + + @api_version("1.3.0", "2.9.2", "2.9.2") + def notifications_dismiss(self, id): + """ + Deletes a single notification + """ + id = self.__unpack_id(id) + + if self.verify_minimum_version("2.9.2", cached=True): + url = '/api/v1/notifications/{0}/dismiss'.format(str(id)) + self.__api_request('POST', url) + else: + params = self.__generate_params(locals()) + self.__api_request('POST', '/api/v1/notifications/dismiss', params) diff --git a/mastodon/polls.py b/mastodon/polls.py new file mode 100644 index 0000000..d974531 --- /dev/null +++ b/mastodon/polls.py @@ -0,0 +1,61 @@ +# polls.py - poll related endpoints and tooling + +from .versions import _DICT_VERSION_POLL +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Polls + ### + @api_version("2.8.0", "2.8.0", _DICT_VERSION_POLL) + def poll(self, id): + """ + Fetch information about the poll with the given id + + Returns a :ref:`poll dict `. + """ + id = self.__unpack_id(id) + url = '/api/v1/polls/{0}'.format(str(id)) + return self.__api_request('GET', url) + + ### + # Writing data: Polls + ### + @api_version("2.8.0", "2.8.0", _DICT_VERSION_POLL) + def poll_vote(self, id, choices): + """ + Vote in the given poll. + + `choices` is the index of the choice you wish to register a vote for + (i.e. its index in the corresponding polls `options` field. In case + of a poll that allows selection of more than one option, a list of + indices can be passed. + + You can only submit choices for any given poll once in case of + single-option polls, or only once per option in case of multi-option + polls. + + Returns the updated :ref:`poll dict ` + """ + id = self.__unpack_id(id) + if not isinstance(choices, list): + choices = [choices] + params = self.__generate_params(locals(), ['id']) + + url = '/api/v1/polls/{0}/votes'.format(id) + self.__api_request('POST', url, params) + + def make_poll(self, options, expires_in, multiple=False, hide_totals=False): + """ + Generate a poll object that can be passed as the `poll` option when posting a status. + + options is an array of strings with the poll options (Maximum, by default: 4), + expires_in is the time in seconds for which the poll should be open. + Set multiple to True to allow people to choose more than one answer. Set + hide_totals to True to hide the results of the poll until it has expired. + """ + poll_params = locals() + del poll_params["self"] + return poll_params \ No newline at end of file diff --git a/mastodon/preferences.py b/mastodon/preferences.py new file mode 100644 index 0000000..5e7435f --- /dev/null +++ b/mastodon/preferences.py @@ -0,0 +1,71 @@ +# preferences.py - user preferences, markers + +import collections + +from .versions import _DICT_VERSION_PREFERENCES, _DICT_VERSION_MARKER +from .errors import MastodonIllegalArgumentError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Preferences + ### + @api_version("2.8.0", "2.8.0", _DICT_VERSION_PREFERENCES) + def preferences(self): + """ + Fetch the user's preferences, which can be used to set some default options. + As of 2.8.0, apps can only fetch, not update preferences. + + Returns a :ref:`preference dict `. + """ + return self.__api_request('GET', '/api/v1/preferences') + + ## + # Reading data: Read markers + ## + @api_version("3.0.0", "3.0.0", _DICT_VERSION_MARKER) + def markers_get(self, timeline=["home"]): + """ + Get the last-read-location markers for the specified timelines. Valid timelines + are the same as in :ref:`timeline() ` + + Note that despite the singular name, `timeline` can be a list. + + Returns a dict of :ref:`read marker dicts `, keyed by timeline name. + """ + if not isinstance(timeline, (list, tuple)): + timeline = [timeline] + params = self.__generate_params(locals()) + + return self.__api_request('GET', '/api/v1/markers', params) + + ## + # Writing data: Read markers + ## + @api_version("3.0.0", "3.0.0", _DICT_VERSION_MARKER) + def markers_set(self, timelines, last_read_ids): + """ + Set the "last read" marker(s) for the given timeline(s) to the given id(s) + + Note that if you give an invalid timeline name, this will silently do nothing. + + Returns a dict with the updated :ref:`read marker dicts `, keyed by timeline name. + """ + if not isinstance(timelines, (list, tuple)): + timelines = [timelines] + + if not isinstance(last_read_ids, (list, tuple)): + last_read_ids = [last_read_ids] + + if len(last_read_ids) != len(timelines): + raise MastodonIllegalArgumentError("Number of specified timelines and ids must be the same") + + params = collections.OrderedDict() + for timeline, last_read_id in zip(timelines, last_read_ids): + params[timeline] = collections.OrderedDict() + params[timeline]["last_read_id"] = self.__unpack_id(last_read_id) + + return self.__api_request('POST', '/api/v1/markers', params, use_json=True) + \ No newline at end of file diff --git a/mastodon/push.py b/mastodon/push.py new file mode 100644 index 0000000..def348a --- /dev/null +++ b/mastodon/push.py @@ -0,0 +1,201 @@ +# push.py - webpush endpoints and tooling + +import base64 +import os +import json + +from .versions import _DICT_VERSION_PUSH, _DICT_VERSION_PUSH_NOTIF +from .errors import MastodonIllegalArgumentError +from .utility import api_version +from .compat import IMPL_HAS_CRYPTO, ec, serialization, default_backend +from .compat import IMPL_HAS_ECE, http_ece + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Webpush subscriptions + ### + @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH) + def push_subscription(self): + """ + Fetch the current push subscription the logged-in user has for this app. + + Returns a :ref:`push subscription dict `. + """ + return self.__api_request('GET', '/api/v1/push/subscription') + + ### + # Writing data: Push subscriptions + ### + @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH) + def push_subscription_set(self, endpoint, encrypt_params, follow_events=None, + favourite_events=None, reblog_events=None, + mention_events=None, poll_events=None, + follow_request_events=None, status_events=None, policy='all'): + """ + Sets up or modifies the push subscription the logged-in user has for this app. + + `endpoint` is the endpoint URL mastodon should call for pushes. Note that mastodon + requires https for this URL. `encrypt_params` is a dict with key parameters that allow + the server to encrypt data for you: A public key `pubkey` and a shared secret `auth`. + You can generate this as well as the corresponding private key using the + :ref:`push_subscription_generate_keys() ` function. + + `policy` controls what sources will generate webpush events. Valid values are + `all`, `none`, `follower` and `followed`. + + The rest of the parameters controls what kind of events you wish to subscribe to. + + Returns a :ref:`push subscription dict `. + """ + if not policy in ['all', 'none', 'follower', 'followed']: + raise MastodonIllegalArgumentError("Valid values for policy are 'all', 'none', 'follower' or 'followed'.") + + endpoint = Mastodon.__protocolize(endpoint) + + push_pubkey_b64 = base64.b64encode(encrypt_params['pubkey']) + push_auth_b64 = base64.b64encode(encrypt_params['auth']) + + params = { + 'subscription[endpoint]': endpoint, + 'subscription[keys][p256dh]': push_pubkey_b64, + 'subscription[keys][auth]': push_auth_b64, + 'policy': policy + } + + if follow_events is not None: + params['data[alerts][follow]'] = follow_events + + if favourite_events is not None: + params['data[alerts][favourite]'] = favourite_events + + if reblog_events is not None: + params['data[alerts][reblog]'] = reblog_events + + if mention_events is not None: + params['data[alerts][mention]'] = mention_events + + if poll_events is not None: + params['data[alerts][poll]'] = poll_events + + if follow_request_events is not None: + params['data[alerts][follow_request]'] = follow_request_events + + if follow_request_events is not None: + params['data[alerts][status]'] = status_events + + # Canonicalize booleans + params = self.__generate_params(params) + + return self.__api_request('POST', '/api/v1/push/subscription', params) + + @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH) + def push_subscription_update(self, follow_events=None, + favourite_events=None, reblog_events=None, + mention_events=None, poll_events=None, + follow_request_events=None): + """ + Modifies what kind of events the app wishes to subscribe to. + + Returns the updated :ref:`push subscription dict `. + """ + params = {} + + if follow_events is not None: + params['data[alerts][follow]'] = follow_events + + if favourite_events is not None: + params['data[alerts][favourite]'] = favourite_events + + if reblog_events is not None: + params['data[alerts][reblog]'] = reblog_events + + if mention_events is not None: + params['data[alerts][mention]'] = mention_events + + if poll_events is not None: + params['data[alerts][poll]'] = poll_events + + if follow_request_events is not None: + params['data[alerts][follow_request]'] = follow_request_events + + # Canonicalize booleans + params = self.__generate_params(params) + + return self.__api_request('PUT', '/api/v1/push/subscription', params) + + @api_version("2.4.0", "2.4.0", "2.4.0") + def push_subscription_delete(self): + """ + Remove the current push subscription the logged-in user has for this app. + """ + self.__api_request('DELETE', '/api/v1/push/subscription') + + ### + # Push subscription crypto utilities + ### + def push_subscription_generate_keys(self): + """ + Generates a private key, public key and shared secret for use in webpush subscriptions. + + Returns two dicts: One with the private key and shared secret and another with the + public key and shared secret. + """ + if not IMPL_HAS_CRYPTO: + raise NotImplementedError( + 'To use the crypto tools, please install the webpush feature dependencies.') + + push_key_pair = ec.generate_private_key(ec.SECP256R1(), default_backend()) + push_key_priv = push_key_pair.private_numbers().private_value + try: + push_key_pub = push_key_pair.public_key().public_bytes( + serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint, + ) + except: + push_key_pub = push_key_pair.public_key().public_numbers().encode_point() + + push_shared_secret = os.urandom(16) + + priv_dict = { + 'privkey': push_key_priv, + 'auth': push_shared_secret + } + + pub_dict = { + 'pubkey': push_key_pub, + 'auth': push_shared_secret + } + + return priv_dict, pub_dict + + @api_version("2.4.0", "2.4.0", _DICT_VERSION_PUSH_NOTIF) + def push_subscription_decrypt_push(self, data, decrypt_params, encryption_header, crypto_key_header): + """ + Decrypts `data` received in a webpush request. Requires the private key dict + from :ref:`push_subscription_generate_keys() ` (`decrypt_params`) as well as the + Encryption and server Crypto-Key headers from the received webpush + + Returns the decoded webpush as a :ref:`push notification dict `. + """ + if (not IMPL_HAS_ECE) or (not IMPL_HAS_CRYPTO): + raise NotImplementedError( + 'To use the crypto tools, please install the webpush feature dependencies.') + + salt = self.__decode_webpush_b64(encryption_header.split("salt=")[1].strip()) + dhparams = self.__decode_webpush_b64(crypto_key_header.split("dh=")[1].split(";")[0].strip()) + p256ecdsa = self.__decode_webpush_b64(crypto_key_header.split("p256ecdsa=")[1].strip()) + dec_key = ec.derive_private_key(decrypt_params['privkey'], ec.SECP256R1(), default_backend()) + decrypted = http_ece.decrypt( + data, + salt=salt, + key=p256ecdsa, + private_key=dec_key, + dh=dhparams, + auth_secret=decrypt_params['auth'], + keylabel="P-256", + version="aesgcm" + ) + + return json.loads(decrypted.decode('utf-8'), object_hook=Mastodon.__json_hooks) diff --git a/mastodon/relationships.py b/mastodon/relationships.py new file mode 100644 index 0000000..9cf15db --- /dev/null +++ b/mastodon/relationships.py @@ -0,0 +1,136 @@ +# relationships.py - endpoints for user and domain blocks and mutes as well as follow requests + +from .versions import _DICT_VERSION_ACCOUNT, _DICT_VERSION_RELATIONSHIP +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Mutes and Blocks + ### + @api_version("1.1.0", "2.6.0", _DICT_VERSION_ACCOUNT) + def mutes(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch a list of users muted by the logged-in user. + + Returns a list of :ref:`account dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/mutes', params) + + @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) + def blocks(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch a list of users blocked by the logged-in user. + + Returns a list of :ref:`account dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/blocks', params) + + ### + # Reading data: Follow requests + ### + @api_version("1.0.0", "2.6.0", _DICT_VERSION_ACCOUNT) + def follow_requests(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch the logged-in user's incoming follow requests. + + Returns a list of :ref:`account dicts `. + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/follow_requests', params) + + ### + # Reading data: Domain blocks + ### + @api_version("1.4.0", "2.6.0", "1.4.0") + def domain_blocks(self, max_id=None, min_id=None, since_id=None, limit=None): + """ + Fetch the logged-in user's blocked domains. + + Returns a list of blocked domain URLs (as strings, without protocol specifier). + """ + if max_id is not None: + max_id = self.__unpack_id(max_id, dateconv=True) + + if min_id is not None: + min_id = self.__unpack_id(min_id, dateconv=True) + + if since_id is not None: + since_id = self.__unpack_id(since_id, dateconv=True) + + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/domain_blocks', params) + + ### + # Writing data: Follow requests + ### + @api_version("1.0.0", "3.0.0", _DICT_VERSION_RELATIONSHIP) + def follow_request_authorize(self, id): + """ + Accept an incoming follow request. + + Returns the updated :ref:`relationship dict ` for the requesting account. + """ + id = self.__unpack_id(id) + url = '/api/v1/follow_requests/{0}/authorize'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.0.0", "3.0.0", _DICT_VERSION_RELATIONSHIP) + def follow_request_reject(self, id): + """ + Reject an incoming follow request. + + Returns the updated :ref:`relationship dict ` for the requesting account. + """ + id = self.__unpack_id(id) + url = '/api/v1/follow_requests/{0}/reject'.format(str(id)) + return self.__api_request('POST', url) + + ### + # Writing data: Domain blocks + ### + @api_version("1.4.0", "1.4.0", "1.4.0") + def domain_block(self, domain=None): + """ + Add a block for all statuses originating from the specified domain for the logged-in user. + """ + params = self.__generate_params(locals()) + self.__api_request('POST', '/api/v1/domain_blocks', params) + + @api_version("1.4.0", "1.4.0", "1.4.0") + def domain_unblock(self, domain=None): + """ + Remove a domain block for the logged-in user. + """ + params = self.__generate_params(locals()) + self.__api_request('DELETE', '/api/v1/domain_blocks', params) \ No newline at end of file diff --git a/mastodon/reports.py b/mastodon/reports.py new file mode 100644 index 0000000..3d6380e --- /dev/null +++ b/mastodon/reports.py @@ -0,0 +1,62 @@ + +# reports.py - report endpoints + +from .versions import _DICT_VERSION_REPORT +from .errors import MastodonVersionError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Reports + ### + @api_version("1.1.0", "1.1.0", _DICT_VERSION_REPORT) + def reports(self): + """ + Fetch a list of reports made by the logged-in user. + + Returns a list of :ref:`report dicts `. + + Warning: This method has now finally been removed, and will not + work on Mastodon versions 2.5.0 and above. + """ + if self.verify_minimum_version("2.5.0", cached=True): + raise MastodonVersionError("API removed in Mastodon 2.5.0") + return self.__api_request('GET', '/api/v1/reports') + + ### + # Writing data: Reports + ### + @api_version("1.1.0", "3.5.0", _DICT_VERSION_REPORT) + def report(self, account_id, status_ids=None, comment=None, forward=False, category=None, rule_ids=None): + """ + Report statuses to the instances administrators. + + Accepts a list of toot IDs associated with the report, and a comment. + + Starting with Mastodon 3.5.0, you can also pass a `category` (one out of + "spam", "violation" or "other") and `rule_ids` (a list of rule IDs corresponding + to the rules returned by the :ref:`instance() ` API). + + Set `forward` to True to forward a report of a remote user to that users + instance as well as sending it to the instance local administrators. + + Returns a :ref:`report dict `. + """ + if category is not None and not category in ["spam", "violation", "other"]: + raise MastodonIllegalArgumentError("Invalid report category (must be spam, violation or other)") + + account_id = self.__unpack_id(account_id) + + if status_ids is not None: + if not isinstance(status_ids, list): + status_ids = [status_ids] + status_ids = list(map(lambda x: self.__unpack_id(x), status_ids)) + + params_initial = locals() + if not forward: + del params_initial['forward'] + + params = self.__generate_params(params_initial) + return self.__api_request('POST', '/api/v1/reports/', params) \ No newline at end of file diff --git a/mastodon/search.py b/mastodon/search.py new file mode 100644 index 0000000..b77002f --- /dev/null +++ b/mastodon/search.py @@ -0,0 +1,91 @@ +# search.py - search endpoints + +from .versions import _DICT_VERSION_SEARCHRESULT +from .errors import MastodonVersionError +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Searching + ### + def __ensure_search_params_acceptable(self, account_id, offset, min_id, max_id): + """ + Internal Helper: Throw a MastodonVersionError if version is < 2.8.0 but parameters + for search that are available only starting with 2.8.0 are specified. + """ + if any(item is not None for item in (account_id, offset, min_id, max_id)): + if not self.verify_minimum_version("2.8.0", cached=True): + raise MastodonVersionError("Advanced search parameters require Mastodon 2.8.0+") + + @api_version("1.1.0", "2.8.0", _DICT_VERSION_SEARCHRESULT) + def search(self, q, resolve=True, result_type=None, account_id=None, offset=None, min_id=None, max_id=None, exclude_unreviewed=True): + """ + Fetch matching hashtags, accounts and statuses. Will perform webfinger + lookups if resolve is True. Full-text search is only enabled if + the instance supports it, and is restricted to statuses the logged-in + user wrote or was mentioned in. + + `result_type` can be one of "accounts", "hashtags" or "statuses", to only + search for that type of object. + + Specify `account_id` to only get results from the account with that id. + + `offset`, `min_id` and `max_id` can be used to paginate. + + `exclude_unreviewed` can be used to restrict search results for hashtags to only + those that have been reviewed by moderators. It is on by default. When using the + v1 search API (pre 2.4.1), it is ignored. + + Will use search_v1 (no tag dicts in return values) on Mastodon versions before + 2.4.1), search_v2 otherwise. Parameters other than resolve are only available + on Mastodon 2.8.0 or above - this function will throw a MastodonVersionError + if you try to use them on versions before that. Note that the cached version + number will be used for this to avoid uneccesary requests. + + Returns a :ref:`search result dict `, with tags as `hashtag dicts`_. + """ + if self.verify_minimum_version("2.4.1", cached=True): + return self.search_v2(q, resolve=resolve, result_type=result_type, account_id=account_id, offset=offset, min_id=min_id, max_id=max_id, exclude_unreviewed=exclude_unreviewed) + else: + self.__ensure_search_params_acceptable(account_id, offset, min_id, max_id) + return self.search_v1(q, resolve=resolve) + + @api_version("1.1.0", "2.1.0", "2.1.0") + def search_v1(self, q, resolve=False): + """ + Identical to `search_v2()`, except in that it does not return + tags as :ref:`hashtag dicts `. + + Returns a :ref:`search result dict `. + """ + params = self.__generate_params(locals()) + if not resolve: + del params['resolve'] + return self.__api_request('GET', '/api/v1/search', params) + + @api_version("2.4.1", "2.8.0", _DICT_VERSION_SEARCHRESULT) + def search_v2(self, q, resolve=True, result_type=None, account_id=None, offset=None, min_id=None, max_id=None, exclude_unreviewed=True): + """ + Identical to `search_v1()`, except in that it returns tags as + :ref:`hashtag dicts `, has more parameters, and resolves by default. + + For more details documentation, please see `search()` + + Returns a :ref:`search result dict `. + """ + self.__ensure_search_params_acceptable(account_id, offset, min_id, max_id) + params = self.__generate_params(locals()) + + if not resolve: + del params["resolve"] + + if not exclude_unreviewed or not self.verify_minimum_version("3.0.0", cached=True): + del params["exclude_unreviewed"] + + if "result_type" in params: + params["type"] = params["result_type"] + del params["result_type"] + + return self.__api_request('GET', '/api/v2/search', params) diff --git a/mastodon/statuses.py b/mastodon/statuses.py index ae891d5..36a7d2b 100644 --- a/mastodon/statuses.py +++ b/mastodon/statuses.py @@ -1,10 +1,14 @@ - -from .versions import _DICT_VERSION_STATUS, _DICT_VERSION_CARD, _DICT_VERSION_CONTEXT, _DICT_VERSION_ACCOUNT, _DICT_VERSION_SCHEDULED_STATUS +# statuses.py - status endpoints (regular and scheduled) + +import collections + +from .versions import _DICT_VERSION_STATUS, _DICT_VERSION_CARD, _DICT_VERSION_CONTEXT, _DICT_VERSION_ACCOUNT, _DICT_VERSION_SCHEDULED_STATUS, \ + _DICT_VERSION_STATUS_EDIT +from .errors import MastodonIllegalArgumentError from .utility import api_version from .internals import Mastodon as Internals - class Mastodon(Internals): ### # Reading data: Statuses @@ -105,4 +109,416 @@ class Mastodon(Internals): id = self.__unpack_id(id) url = '/api/v1/scheduled_statuses/{0}'.format(str(id)) return self.__api_request('GET', url) - \ No newline at end of file + + ### + # Writing data: Statuses + ### + def __status_internal(self, status, in_reply_to_id=None, media_ids=None, + sensitive=False, visibility=None, spoiler_text=None, + language=None, idempotency_key=None, content_type=None, + scheduled_at=None, poll=None, quote_id=None, edit=False): + if quote_id is not None: + if self.feature_set != "fedibird": + raise MastodonIllegalArgumentError('quote_id is only available with feature set fedibird') + quote_id = self.__unpack_id(quote_id) + + if content_type is not None: + if self.feature_set != "pleroma": + raise MastodonIllegalArgumentError('content_type is only available with feature set pleroma') + # It would be better to read this from nodeinfo and cache, but this is easier + if not content_type in ["text/plain", "text/html", "text/markdown", "text/bbcode"]: + raise MastodonIllegalArgumentError('Invalid content type specified') + + if in_reply_to_id is not None: + in_reply_to_id = self.__unpack_id(in_reply_to_id) + + if scheduled_at is not None: + scheduled_at = self.__consistent_isoformat_utc(scheduled_at) + + params_initial = locals() + + # Validate poll/media exclusivity + if poll is not None: + if media_ids is not None and len(media_ids) != 0: + raise ValueError( + 'Status can have media or poll attached - not both.') + + # Validate visibility parameter + valid_visibilities = ['private', 'public', 'unlisted', 'direct'] + if params_initial['visibility'] is None: + del params_initial['visibility'] + else: + params_initial['visibility'] = params_initial['visibility'].lower() + if params_initial['visibility'] not in valid_visibilities: + raise ValueError('Invalid visibility value! Acceptable values are %s' % valid_visibilities) + + if params_initial['language'] is None: + del params_initial['language'] + + if params_initial['sensitive'] is False: + del [params_initial['sensitive']] + + headers = {} + if idempotency_key is not None: + headers['Idempotency-Key'] = idempotency_key + + if media_ids is not None: + try: + media_ids_proper = [] + if not isinstance(media_ids, (list, tuple)): + media_ids = [media_ids] + for media_id in media_ids: + media_ids_proper.append(self.__unpack_id(media_id)) + except Exception as e: + raise MastodonIllegalArgumentError("Invalid media dict: %s" % e) + + params_initial["media_ids"] = media_ids_proper + + if params_initial['content_type'] is None: + del params_initial['content_type'] + + use_json = False + if poll is not None: + use_json = True + + params = self.__generate_params(params_initial, ['idempotency_key', 'edit']) + if edit is None: + # Post + return self.__api_request('POST', '/api/v1/statuses', params, headers=headers, use_json=use_json) + else: + # Edit + return self.__api_request('PUT', '/api/v1/statuses/{0}'.format(str(self.__unpack_id(edit))), params, headers=headers, use_json=use_json) + + @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) + def status_post(self, status, in_reply_to_id=None, media_ids=None, + sensitive=False, visibility=None, spoiler_text=None, + language=None, idempotency_key=None, content_type=None, + scheduled_at=None, poll=None, quote_id=None): + """ + Post a status. Can optionally be in reply to another status and contain + media. + + `media_ids` should be a list. (If it's not, the function will turn it + into one.) It can contain up to four pieces of media (uploaded via + :ref:`media_post() `). `media_ids` can also be the `media dicts`_ returned + by :ref:`media_post() ` - they are unpacked automatically. + + The `sensitive` boolean decides whether or not media attached to the post + should be marked as sensitive, which hides it by default on the Mastodon + web front-end. + + The visibility parameter is a string value and accepts any of: + 'direct' - post will be visible only to mentioned users + 'private' - post will be visible only to followers + 'unlisted' - post will be public but not appear on the public timeline + 'public' - post will be public + + If not passed in, visibility defaults to match the current account's + default-privacy setting (starting with Mastodon version 1.6) or its + locked setting - private if the account is locked, public otherwise + (for Mastodon versions lower than 1.6). + + The `spoiler_text` parameter is a string to be shown as a warning before + the text of the status. If no text is passed in, no warning will be + displayed. + + Specify `language` to override automatic language detection. The parameter + accepts all valid ISO 639-1 (2-letter) or for languages where that do not + have one, 639-3 (three letter) language codes. + + You can set `idempotency_key` to a value to uniquely identify an attempt + at posting a status. Even if you call this function more than once, + if you call it with the same `idempotency_key`, only one status will + be created. + + Pass a datetime as `scheduled_at` to schedule the toot for a specific time + (the time must be at least 5 minutes into the future). If this is passed, + status_post returns a :ref:`scheduled status dict ` instead. + + Pass `poll` to attach a poll to the status. An appropriate object can be + constructed using :ref:`make_poll() ` . Note that as of Mastodon version + 2.8.2, you can only have either media or a poll attached, not both at + the same time. + + **Specific to "pleroma" feature set:**: Specify `content_type` to set + the content type of your post on Pleroma. It accepts 'text/plain' (default), + 'text/markdown', 'text/html' and 'text/bbcode'. This parameter is not + supported on Mastodon servers, but will be safely ignored if set. + + **Specific to "fedibird" feature set:**: The `quote_id` parameter is + a non-standard extension that specifies the id of a quoted status. + + Returns a :ref:`status dict ` with the new status. + """ + return self.__status_internal( + status, + in_reply_to_id, + media_ids, + sensitive, + visibility, + spoiler_text, + language, + idempotency_key, + content_type, + scheduled_at, + poll, + quote_id, + edit=None + ) + + @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) + def toot(self, status): + """ + Synonym for :ref:`status_post() ` that only takes the status text as input. + + Usage in production code is not recommended. + + Returns a :ref:`status dict ` with the new status. + """ + return self.status_post(status) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS) + def status_update(self, id, status = None, spoiler_text = None, sensitive = None, media_ids = None, poll = None): + """ + Edit a status. The meanings of the fields are largely the same as in :ref:`status_post() `, + though not every field can be edited. + + Note that editing a poll will reset the votes. + """ + return self.__status_internal( + status = status, + media_ids = media_ids, + sensitive = sensitive, + spoiler_text = spoiler_text, + poll = poll, + edit = id + ) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS_EDIT) + def status_history(self, id): + """ + Returns the edit history of a status as a list of :ref:`status edit dicts `, starting + from the original form. Note that this means that a status that has been edited + once will have *two* entries in this list, a status that has been edited twice + will have three, and so on. + """ + id = self.__unpack_id(id) + return self.__api_request('GET', "/api/v1/statuses/{0}/history".format(str(id))) + + def status_source(self, id): + """ + Returns the source of a status for editing. + + Return value is a dictionary containing exactly the parameters you could pass to + :ref:`status_update() ` to change nothing about the status, except `status` is `text` + instead. + """ + id = self.__unpack_id(id) + return self.__api_request('GET', "/api/v1/statuses/{0}/source".format(str(id))) + + @api_version("1.0.0", "2.8.0", _DICT_VERSION_STATUS) + def status_reply(self, to_status, status, in_reply_to_id=None, media_ids=None, + sensitive=False, visibility=None, spoiler_text=None, + language=None, idempotency_key=None, content_type=None, + scheduled_at=None, poll=None, untag=False): + """ + Helper function - acts like status_post, but prepends the name of all + the users that are being replied to to the status text and retains + CW and visibility if not explicitly overridden. + + Set `untag` to True if you want the reply to only go to the user you + are replying to, removing every other mentioned user from the + conversation. + """ + keyword_args = locals() + del keyword_args["self"] + del keyword_args["to_status"] + del keyword_args["untag"] + + user_id = self.__get_logged_in_id() + + # Determine users to mention + mentioned_accounts = collections.OrderedDict() + mentioned_accounts[to_status.account.id] = to_status.account.acct + + if not untag: + for account in to_status.mentions: + if account.id != user_id and not account.id in mentioned_accounts.keys(): + mentioned_accounts[account.id] = account.acct + + # Join into one piece of text. The space is added inside because of self-replies. + status = "".join(map(lambda x: "@" + x + " ", + mentioned_accounts.values())) + status + + # Retain visibility / cw + if visibility is None and 'visibility' in to_status: + visibility = to_status.visibility + if spoiler_text is None and 'spoiler_text' in to_status: + spoiler_text = to_status.spoiler_text + + keyword_args["status"] = status + keyword_args["visibility"] = visibility + keyword_args["spoiler_text"] = spoiler_text + keyword_args["in_reply_to_id"] = to_status.id + return self.status_post(**keyword_args) + + @api_version("1.0.0", "1.0.0", "1.0.0") + def status_delete(self, id): + """ + Delete a status + + Returns the now-deleted status, with an added "source" attribute that contains + the text that was used to compose this status (this can be used to power + "delete and redraft" functionality) + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}'.format(str(id)) + return self.__api_request('DELETE', url) + + @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) + def status_reblog(self, id, visibility=None): + """ + Reblog / boost a status. + + The visibility parameter functions the same as in :ref:`status_post() ` and + allows you to reduce the visibility of a reblogged status. + + Returns a :ref:`status dict ` with a new status that wraps around the reblogged one. + """ + params = self.__generate_params(locals(), ['id']) + valid_visibilities = ['private', 'public', 'unlisted', 'direct'] + if 'visibility' in params: + params['visibility'] = params['visibility'].lower() + if params['visibility'] not in valid_visibilities: + raise ValueError('Invalid visibility value! Acceptable ' + 'values are %s' % valid_visibilities) + + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/reblog'.format(str(id)) + return self.__api_request('POST', url, params) + + @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) + def status_unreblog(self, id): + """ + Un-reblog a status. + + Returns a :ref:`status dict ` with the status that used to be reblogged. + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/unreblog'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) + def status_favourite(self, id): + """ + Favourite a status. + + Returns a :ref:`status dict ` with the favourited status. + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/favourite'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.0.0", "2.0.0", _DICT_VERSION_STATUS) + def status_unfavourite(self, id): + """ + Un-favourite a status. + + Returns a :ref:`status dict ` with the un-favourited status. + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/unfavourite'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.4.0", "2.0.0", _DICT_VERSION_STATUS) + def status_mute(self, id): + """ + Mute notifications for a status. + + Returns a :ref:`status dict ` with the now muted status + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/mute'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("1.4.0", "2.0.0", _DICT_VERSION_STATUS) + def status_unmute(self, id): + """ + Unmute notifications for a status. + + Returns a :ref:`status dict ` with the status that used to be muted. + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/unmute'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("2.1.0", "2.1.0", _DICT_VERSION_STATUS) + def status_pin(self, id): + """ + Pin a status for the logged-in user. + + Returns a :ref:`status dict ` with the now pinned status + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/pin'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("2.1.0", "2.1.0", _DICT_VERSION_STATUS) + def status_unpin(self, id): + """ + Unpin a pinned status for the logged-in user. + + Returns a :ref:`status dict ` with the status that used to be pinned. + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/unpin'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("3.1.0", "3.1.0", _DICT_VERSION_STATUS) + def status_bookmark(self, id): + """ + Bookmark a status as the logged-in user. + + Returns a :ref:`status dict ` with the now bookmarked status + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/bookmark'.format(str(id)) + return self.__api_request('POST', url) + + @api_version("3.1.0", "3.1.0", _DICT_VERSION_STATUS) + def status_unbookmark(self, id): + """ + Unbookmark a bookmarked status for the logged-in user. + + Returns a :ref:`status dict ` with the status that used to be bookmarked. + """ + id = self.__unpack_id(id) + url = '/api/v1/statuses/{0}/unbookmark'.format(str(id)) + return self.__api_request('POST', url) + + ### + # Writing data: Scheduled statuses + ### + @api_version("2.7.0", "2.7.0", _DICT_VERSION_SCHEDULED_STATUS) + def scheduled_status_update(self, id, scheduled_at): + """ + Update the scheduled time of a scheduled status. + + New time must be at least 5 minutes into the future. + + Returns a :ref:`scheduled status dict ` + """ + scheduled_at = self.__consistent_isoformat_utc(scheduled_at) + id = self.__unpack_id(id) + params = self.__generate_params(locals(), ['id']) + url = '/api/v1/scheduled_statuses/{0}'.format(str(id)) + return self.__api_request('PUT', url, params) + + @api_version("2.7.0", "2.7.0", "2.7.0") + def scheduled_status_delete(self, id): + """ + Deletes a scheduled status. + """ + id = self.__unpack_id(id) + url = '/api/v1/scheduled_statuses/{0}'.format(str(id)) + self.__api_request('DELETE', url) diff --git a/mastodon/streaming_endpoints.py b/mastodon/streaming_endpoints.py new file mode 100644 index 0000000..9ff72f5 --- /dev/null +++ b/mastodon/streaming_endpoints.py @@ -0,0 +1,75 @@ + # relationships.py - endpoints for user and domain blocks and mutes as well as follow requests + +from .versions import _DICT_VERSION_STATUS +from .errors import MastodonIllegalArgumentError +from .defaults import _DEFAULT_STREAM_TIMEOUT, _DEFAULT_STREAM_RECONNECT_WAIT_SEC +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Streaming + ### + @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) + def stream_user(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): + """ + Streams events that are relevant to the authorized user, i.e. home + timeline and notifications. + """ + return self.__stream('/api/v1/streaming/user', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) + + @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) + def stream_public(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): + """ + Streams public events. + """ + return self.__stream('/api/v1/streaming/public', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) + + @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) + def stream_local(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): + """ + Streams local public events. + """ + return self.__stream('/api/v1/streaming/public/local', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) + + @api_version("1.1.0", "1.4.2", _DICT_VERSION_STATUS) + def stream_hashtag(self, tag, listener, local=False, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): + """ + Stream for all public statuses for the hashtag 'tag' seen by the connected + instance. + + Set local to True to only get local statuses. + """ + if tag.startswith("#"): + raise MastodonIllegalArgumentError("Tag parameter should omit leading #") + base = '/api/v1/streaming/hashtag' + if local: + base += '/local' + return self.__stream("{}?tag={}".format(base, tag), listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) + + @api_version("2.1.0", "2.1.0", _DICT_VERSION_STATUS) + def stream_list(self, id, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): + """ + Stream events for the current user, restricted to accounts on the given + list. + """ + id = self.__unpack_id(id) + return self.__stream("/api/v1/streaming/list?list={}".format(id), listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) + + @api_version("2.6.0", "2.6.0", _DICT_VERSION_STATUS) + def stream_direct(self, listener, run_async=False, timeout=_DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=_DEFAULT_STREAM_RECONNECT_WAIT_SEC): + """ + Streams direct message events for the logged-in user, as conversation events. + """ + return self.__stream('/api/v1/streaming/direct', listener, run_async=run_async, timeout=timeout, reconnect_async=reconnect_async, reconnect_async_wait_sec=reconnect_async_wait_sec) + + @api_version("2.5.0", "2.5.0", "2.5.0") + def stream_healthy(self): + """ + Returns without True if streaming API is okay, False or raises an error otherwise. + """ + api_okay = self.__api_request('GET', '/api/v1/streaming/health', base_url_override=self.__get_streaming_base(), parse=False) + if api_okay in [b'OK', b'success']: + return True + return False diff --git a/mastodon/suggestions.py b/mastodon/suggestions.py new file mode 100644 index 0000000..ab55993 --- /dev/null +++ b/mastodon/suggestions.py @@ -0,0 +1,32 @@ +# suggestions.py - follow suggestion endpoints + +from .versions import _DICT_VERSION_ACCOUNT +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Follow suggestions + ### + @api_version("2.4.3", "2.4.3", _DICT_VERSION_ACCOUNT) + def suggestions(self): + """ + Fetch follow suggestions for the logged-in user. + + Returns a list of :ref:`account dicts `. + + """ + return self.__api_request('GET', '/api/v1/suggestions') + + ### + # Writing data: Follow suggestions + ### + @api_version("2.4.3", "2.4.3", _DICT_VERSION_ACCOUNT) + def suggestion_delete(self, account_id): + """ + Remove the user with the given `account_id` from the follow suggestions. + """ + account_id = self.__unpack_id(account_id) + url = '/api/v1/suggestions/{0}'.format(str(account_id)) + self.__api_request('DELETE', url) \ No newline at end of file diff --git a/mastodon/timeline.py b/mastodon/timeline.py index b5a4068..04406ed 100644 --- a/mastodon/timeline.py +++ b/mastodon/timeline.py @@ -1,5 +1,7 @@ +# timeline.py - endpoints for reading various different timelines + from .versions import _DICT_VERSION_STATUS, _DICT_VERSION_CONVERSATION -from .error import MastodonIllegalArgumentError, MastodonNotFoundError +from .errors import MastodonIllegalArgumentError, MastodonNotFoundError from .utility import api_version from .internals import Mastodon as Internals @@ -101,21 +103,3 @@ class Mastodon(Internals): id = self.__unpack_id(id) return self.timeline('list/{0}'.format(id), max_id=max_id, min_id=min_id, since_id=since_id, limit=limit, only_media=only_media, local=local, remote=remote) - @api_version("2.6.0", "2.6.0", _DICT_VERSION_CONVERSATION) - def conversations(self, max_id=None, min_id=None, since_id=None, limit=None): - """ - Fetches a user's conversations. - - Returns a list of :ref:`conversation dicts `. - """ - if max_id is not None: - max_id = self.__unpack_id(max_id, dateconv=True) - - if min_id is not None: - min_id = self.__unpack_id(min_id, dateconv=True) - - if since_id is not None: - since_id = self.__unpack_id(since_id, dateconv=True) - - params = self.__generate_params(locals()) - return self.__api_request('GET', "/api/v1/conversations/", params) diff --git a/mastodon/trends.py b/mastodon/trends.py new file mode 100644 index 0000000..09a0ae5 --- /dev/null +++ b/mastodon/trends.py @@ -0,0 +1,72 @@ +# trends.py - trend-related endpoints + +from .versions import _DICT_VERSION_HASHTAG, _DICT_VERSION_STATUS, _DICT_VERSION_CARD +from .utility import api_version + +from .internals import Mastodon as Internals + +class Mastodon(Internals): + ### + # Reading data: Trends + ### + @api_version("2.4.3", "3.5.0", _DICT_VERSION_HASHTAG) + def trends(self, limit=None): + """ + Alias for :ref:`trending_tags() ` + """ + return self.trending_tags(limit=limit) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_HASHTAG) + def trending_tags(self, limit=None, lang=None): + """ + Fetch trending-hashtag information, if the instance provides such information. + + Specify `limit` to limit how many results are returned (the maximum number + of results is 10, the endpoint is not paginated). + + Does not require authentication unless locked down by the administrator. + + Important versioning note: This endpoint does not exist for Mastodon versions + between 2.8.0 (inclusive) and 3.0.0 (exclusive). + + Pass `lang` to override the global locale parameter, which may affect trend ordering. + + Returns a list of :ref:`hashtag dicts `, sorted by the instance's trending algorithm, + descending. + """ + params = self.__generate_params(locals()) + if self.verify_minimum_version("3.5.0", cached=True): + # Starting 3.5.0, old version is deprecated + return self.__api_request('GET', '/api/v1/trends/tags', params) + else: + return self.__api_request('GET', '/api/v1/trends', params) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_STATUS) + def trending_statuses(self): + """ + Fetch trending-status information, if the instance provides such information. + + Specify `limit` to limit how many results are returned (the maximum number + of results is 10, the endpoint is not paginated). + + Pass `lang` to override the global locale parameter, which may affect trend ordering. + + Returns a list of :ref:`status dicts `, sorted by the instances's trending algorithm, + descending. + """ + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/trends/statuses', params) + + @api_version("3.5.0", "3.5.0", _DICT_VERSION_CARD) + def trending_links(self): + """ + Fetch trending-link information, if the instance provides such information. + + Specify `limit` to limit how many results are returned (the maximum number + of results is 10, the endpoint is not paginated). + + Returns a list of :ref:`card dicts `, sorted by the instances's trending algorithm, + descending. + """ + params = self.__generate_params(locals()) + return self.__api_request('GET', '/api/v1/trends/links', params) diff --git a/mastodon/utility.py b/mastodon/utility.py index 53980b6..2510dca 100644 --- a/mastodon/utility.py +++ b/mastodon/utility.py @@ -1,10 +1,13 @@ # utility.py - utility functions, externally usable import re -from decorator import decorate -from .error import MastodonVersionError, MastodonAPIError import dateutil import datetime +import copy + +from decorator import decorate +from .errors import MastodonVersionError, MastodonAPIError +from .compat import IMPL_HAS_BLURHASH, blurhash # Module level: @@ -141,3 +144,112 @@ class Mastodon(): return datetime.datetime.fromtimestamp(epoch_time) else: raise MastodonAPIError("No server time in response.") + + ### + # Blurhash utilities + ### + def decode_blurhash(self, media_dict, out_size=(16, 16), size_per_component=True, return_linear=True): + """ + Basic media-dict blurhash decoding. + + out_size is the desired result size in pixels, either absolute or per blurhash + component (this is the default). + + By default, this function will return the image as linear RGB, ready for further + scaling operations. If you want to display the image directly, set return_linear + to False. + + Returns the decoded blurhash image as a three-dimensional list: [height][width][3], + with the last dimension being RGB colours. + + For further info and tips for advanced usage, refer to the documentation for the + blurhash module: https://github.com/halcy/blurhash-python + """ + if not IMPL_HAS_BLURHASH: + raise NotImplementedError( + 'To use the blurhash functions, please install the blurhash Python module.') + + # Figure out what size to decode to + decode_components_x, decode_components_y = blurhash.components(media_dict["blurhash"]) + if size_per_component: + decode_size_x = decode_components_x * out_size[0] + decode_size_y = decode_components_y * out_size[1] + else: + decode_size_x = out_size[0] + decode_size_y = out_size[1] + + # Decode + decoded_image = blurhash.decode(media_dict["blurhash"], decode_size_x, decode_size_y, linear=return_linear) + + # And that's pretty much it. + return decoded_image + + ### + # Pagination + ### + def fetch_next(self, previous_page): + """ + Fetches the next page of results of a paginated request. Pass in the + previous page in its entirety, or the pagination information dict + returned as a part of that pages last status ('_pagination_next'). + + Returns the next page or None if no further data is available. + """ + if isinstance(previous_page, list) and len(previous_page) != 0: + if hasattr(previous_page, '_pagination_next'): + params = copy.deepcopy(previous_page._pagination_next) + else: + return None + else: + params = copy.deepcopy(previous_page) + + method = params['_pagination_method'] + del params['_pagination_method'] + + endpoint = params['_pagination_endpoint'] + del params['_pagination_endpoint'] + + return self.__api_request(method, endpoint, params) + + def fetch_previous(self, next_page): + """ + Fetches the previous page of results of a paginated request. Pass in the + previous page in its entirety, or the pagination information dict + returned as a part of that pages first status ('_pagination_prev'). + + Returns the previous page or None if no further data is available. + """ + if isinstance(next_page, list) and len(next_page) != 0: + if hasattr(next_page, '_pagination_prev'): + params = copy.deepcopy(next_page._pagination_prev) + else: + return None + else: + params = copy.deepcopy(next_page) + + method = params['_pagination_method'] + del params['_pagination_method'] + + endpoint = params['_pagination_endpoint'] + del params['_pagination_endpoint'] + + return self.__api_request(method, endpoint, params) + + def fetch_remaining(self, first_page): + """ + Fetches all the remaining pages of a paginated request starting from a + first page and returns the entire set of results (including the first page + that was passed in) as a big list. + + Be careful, as this might generate a lot of requests, depending on what you are + fetching, and might cause you to run into rate limits very quickly. + """ + first_page = copy.deepcopy(first_page) + + all_pages = [] + current_page = first_page + while current_page is not None and len(current_page) > 0: + all_pages.extend(current_page) + current_page = self.fetch_next(current_page) + + return all_pages -- cgit v1.2.3