1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
from typing import Optional
import logging
import requests
import urllib.parse
from .types 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
|
