diff options
Diffstat (limited to 'de_uvok/activitypub_fuse/status_provider.py')
-rw-r--r-- | de_uvok/activitypub_fuse/status_provider.py | 122 |
1 files changed, 122 insertions, 0 deletions
diff --git a/de_uvok/activitypub_fuse/status_provider.py b/de_uvok/activitypub_fuse/status_provider.py new file mode 100644 index 0000000..b434ad5 --- /dev/null +++ b/de_uvok/activitypub_fuse/status_provider.py @@ -0,0 +1,122 @@ +from typing import Optional + +import logging +import requests +import urllib.parse + +from .status import Status + +logger = logging.Logger(__name__) + +_api_url_ap_template = "https://{server}/users/{user}/outbox?page=true" +_api_url_m_lookup_template = "https://{server}/api/v1/accounts/lookup" +_api_url_m_status_template = "https://{server}/api/v1/accounts/{uid}/statuses" + + +class StatusProvider: + def load_statuses(self, max_id="") -> tuple[list[Status], Optional[str]]: + raise NotImplementedError + + def _fallback_not_found(self): + return [Status("not-found", "User not found", "1970-01-01T00:00:00Z")] + + def _fallback_error(self, error_msg: str): + return [Status("error", error_msg, "1970-01-01T00:00:00Z")] + + +class ActivityPubStatusProvider(StatusProvider): + def __init__(self, server: str, user: str): + self.server = server + self.user = user + + def load_statuses(self, max_id="") -> tuple[list[Status], Optional[str]]: + url = _api_url_ap_template.format(server=self.server, user=self.user) + if max_id: + url += "&" + urllib.parse.urlencode({"max_id": max_id}) + logger.debug("Get AP status from %s", url) + res = requests.get(url) + if res.status_code == 404: + return self._fallback_not_found(), None + + try: + res.raise_for_status() + except requests.exceptions.RequestException as e: + logger.error("Request error: %s", e) + return self._fallback_error(getattr(e, "message", str(e))), None + + stats = res.json() + status_items = stats.get("orderedItems", None) + if not status_items: + return ( + self._fallback_error("Malformed content in querying AP outbox."), + None, + ) + + # consider reposts for getting max_id... + ss = [ + Status( + s["object"]["id"].split("/")[-1], + s["object"]["content"], + s["object"]["published"], + ) + for s in status_items + if s.get("type", None) == "Create" + and "object" in s + and all(key in s["object"] for key in ["id", "content", "published"]) + ] + # ... but don't return it + return [s for s in ss if len(s.content)], ss[-1].id + + +class MastodonStatusProvider(StatusProvider): + def __init__(self, server: str, user: str): + self.server = server + self.user = user + self.userid = 0 + + def load_statuses(self, max_id="") -> tuple[list[Status], Optional[str]]: + url = _api_url_m_lookup_template.format(server=self.server) + url += "?" + urllib.parse.urlencode({"acct": self.user}) + res = requests.get(url) + if res.status_code == 404: + return self._fallback_not_found(), None + try: + res.raise_for_status() + except requests.exceptions.RequestException as e: + logger.error("Request error: %s", e) + return self._fallback_error(getattr(e, "message", str(e))), None + + user = res.json() + self.userid = user.get("id", None) + if not self.userid: + return self._fallback_error("Malformed content in querying user ID."), None + + url = _api_url_m_status_template.format( + server=self.server, uid=urllib.parse.quote(self.userid) + ) + + if max_id: + url += "?" + urllib.parse.urlencode({"max_id": max_id}) + logger.debug("Get Masto status from %s", url) + + res = requests.get(url) + if res.status_code == 404: + return self._fallback_not_found(), None + try: + res.raise_for_status() + except requests.exceptions.RequestException as e: + logger.error("Request error: %s", e) + return self._fallback_error(getattr(e, "message", str(e))), None + statuses = res.json() + # consider reposts for getting max_id... + ss = [ + Status( + s["id"], + s["content"], + s["created_at"], + ) + for s in statuses + if all(key in s for key in ["id", "content", "created_at"]) + ] + # ... but don't return it + return [s for s in ss if len(s.content)], ss[-1].id |