[mastodon] add 'favorite', 'list', and 'hashtag' extractors (#5529)

This commit is contained in:
Mike Fährmann
2024-04-30 15:53:32 +02:00
parent 7fd31aaf8e
commit 9b1995dda3
3 changed files with 81 additions and 6 deletions

View File

@@ -136,6 +136,36 @@ class MastodonBookmarkExtractor(MastodonExtractor):
return MastodonAPI(self).account_bookmarks()
class MastodonFavoriteExtractor(MastodonExtractor):
"""Extractor for mastodon favorites"""
subcategory = "favorite"
pattern = BASE_PATTERN + r"/favourites"
example = "https://mastodon.social/favourites"
def statuses(self):
return MastodonAPI(self).account_favorites()
class MastodonListExtractor(MastodonExtractor):
"""Extractor for mastodon lists"""
subcategory = "list"
pattern = BASE_PATTERN + r"/lists/(\w+)"
example = "https://mastodon.social/lists/12345"
def statuses(self):
return MastodonAPI(self).timelines_list(self.item)
class MastodonHashtagExtractor(MastodonExtractor):
"""Extractor for mastodon hashtags"""
subcategory = "hashtag"
pattern = BASE_PATTERN + r"/tags/(\w+)"
example = "https://mastodon.social/tags/NAME"
def statuses(self):
return MastodonAPI(self).timelines_tag(self.item)
class MastodonFollowingExtractor(MastodonExtractor):
"""Extractor for followed mastodon users"""
subcategory = "following"
@@ -205,37 +235,55 @@ class MastodonAPI():
raise exception.NotFoundError("account")
def account_bookmarks(self):
"""Statuses the user has bookmarked"""
endpoint = "/v1/bookmarks"
return self._pagination(endpoint, None)
def account_favorites(self):
"""Statuses the user has favourited"""
endpoint = "/v1/favourites"
return self._pagination(endpoint, None)
def account_following(self, account_id):
"""Accounts which the given account is following"""
endpoint = "/v1/accounts/{}/following".format(account_id)
return self._pagination(endpoint, None)
def account_lookup(self, username):
"""Quickly lookup a username to see if it is available"""
endpoint = "/v1/accounts/lookup"
params = {"acct": username}
return self._call(endpoint, params).json()
def account_search(self, query, limit=40):
"""Search for accounts"""
"""Search for matching accounts by username or display name"""
endpoint = "/v1/accounts/search"
params = {"q": query, "limit": limit}
return self._call(endpoint, params).json()
def account_statuses(self, account_id, only_media=True,
exclude_replies=False):
"""Fetch an account's statuses"""
"""Statuses posted to the given account"""
endpoint = "/v1/accounts/{}/statuses".format(account_id)
params = {"only_media" : "true" if only_media else "false",
"exclude_replies": "true" if exclude_replies else "false"}
return self._pagination(endpoint, params)
def status(self, status_id):
"""Fetch a status"""
"""Obtain information about a status"""
endpoint = "/v1/statuses/" + status_id
return self._call(endpoint).json()
def timelines_list(self, list_id):
"""View statuses in the given list timeline"""
endpoint = "/v1/timelines/list/" + list_id
return self._pagination(endpoint, None)
def timelines_tag(self, hashtag):
"""View public statuses containing the given hashtag"""
endpoint = "/v1/timelines/tag/" + hashtag
return self._pagination(endpoint, None)
def _call(self, endpoint, params=None):
if endpoint.startswith("http"):
url = endpoint