252 lines
9.4 KiB
Python
252 lines
9.4 KiB
Python
 
|
"""Everything that speaks the Mastodon client API: Mastodon, GoToSocial, Akkoma, and later Pleroma, Iceshrimp.NET, Hollo,
|
||
|
|
snac, Friendica, Pixelfed, Mitra. Subclasses provide accounts and tokens (each platform makes them its own way) and
|
||
|
|
`_rows(uris)`, their database's view of a set of objects."""
|
||
|
|
import time
|
||
|
|
|
||
|
|
from core import media
|
||
|
|
from core.http import HttpError
|
||
|
|
from dialects.base import Driver, Made, Session, Stored, Unsupported
|
||
|
|
|
||
|
|
VISIBILITY = {"public": "public", "unlisted": "unlisted", "followers": "private", "direct": "direct"}
|
||
|
|
|
||
|
|
|
||
|
|
class MastodonApi(Driver):
|
||
|
|
platform = "mastodon-api"
|
||
|
|
caps = frozenset({"post", "reply", "cw", "media", "poll", "like", "boost", "bookmark", "follow", "block", "mute",
|
||
|
|
"report", "dm", "delete", "edit", "vote", "profile"})
|
||
|
|
media_endpoint = "/api/v2/media"
|
||
|
|
max_media = 4
|
||
|
|
|
||
|
|
def __init__(self, host):
|
||
|
|
super().__init__(host)
|
||
|
|
self._ids = {} # (session token, acct) -> local account id
|
||
|
|
|
||
|
|
# -- plumbing
|
||
|
|
def api(self, s, method, path, ok=None, **kw):
|
||
|
|
headers = kw.pop("headers", {})
|
||
|
|
if s is not None:
|
||
|
|
headers["Authorization"] = f"Bearer {s.token}"
|
||
|
|
return self.http.request(method, self.base + path, headers=headers, ok=ok, template=path.split("?")[0], **kw)
|
||
|
|
|
||
|
|
def api_json(self, s, method, path, **kw):
|
||
|
|
return self.api(s, method, path, ok={200, 201, 202}, **kw).json()
|
||
|
|
|
||
|
|
def session_from_token(self, account, token):
|
||
|
|
me = self.api_json(None, "GET", "/api/v1/accounts/verify_credentials", headers={"Authorization": f"Bearer {token}"})
|
||
|
|
actor = self.actor_uri(account.username) or me.get("url")
|
||
|
|
return Session(account, token, me["id"], actor)
|
||
|
|
|
||
|
|
# -- profiles
|
||
|
|
def update_profile(self, s, account):
|
||
|
|
form = {"display_name": account.name, "note": account.bio, "locked": account.locked, "bot": account.bot}
|
||
|
|
for i, (k, v) in enumerate(account.fields[:4]):
|
||
|
|
form[f"fields_attributes[{i}][name]"] = k
|
||
|
|
form[f"fields_attributes[{i}][value]"] = v
|
||
|
|
files = {}
|
||
|
|
if account.avatar_seed:
|
||
|
|
files["avatar"] = media.upload("image", account.avatar_seed)
|
||
|
|
if account.header_seed:
|
||
|
|
files["header"] = media.upload("image", account.header_seed)
|
||
|
|
self.api(s, "PATCH", "/api/v1/accounts/update_credentials", ok={200}, form=form, files=files or None)
|
||
|
|
|
||
|
|
# -- accounts as this server knows them
|
||
|
|
def lookup(self, s, acct):
|
||
|
|
key = (s.token, acct)
|
||
|
|
if key in self._ids:
|
||
|
|
return self._ids[key]
|
||
|
|
found = None
|
||
|
|
r = self.api(s, "GET", "/api/v1/accounts/lookup", params={"acct": acct})
|
||
|
|
if r.ok and r.json():
|
||
|
|
found = r.json()["id"]
|
||
|
|
if found is None:
|
||
|
|
r = self.api(s, "GET", "/api/v2/search", params={"q": f"@{acct}", "resolve": "true", "type": "accounts"})
|
||
|
|
for a in (r.json() or {}).get("accounts", []) if r.ok else []:
|
||
|
|
if _same_acct(a.get("acct"), acct, self.host):
|
||
|
|
found = a["id"]
|
||
|
|
break
|
||
|
|
if found is None:
|
||
|
|
raise LookupError(f"{self.host} cannot find {acct}")
|
||
|
|
self._ids[key] = found
|
||
|
|
return found
|
||
|
|
|
||
|
|
def follow(self, s, acct):
|
||
|
|
rel = self.api_json(s, "POST", f"/api/v1/accounts/{self.lookup(s, acct)}/follow")
|
||
|
|
return "accepted" if rel.get("following") else "requested"
|
||
|
|
|
||
|
|
def unfollow(self, s, acct):
|
||
|
|
self.api(s, "POST", f"/api/v1/accounts/{self.lookup(s, acct)}/unfollow", ok={200})
|
||
|
|
|
||
|
|
def pending(self, s):
|
||
|
|
r = self.api(s, "GET", "/api/v1/follow_requests", params={"limit": 80}, ok={200})
|
||
|
|
return [_full_acct(a["acct"], self.host) for a in r.json() or []]
|
||
|
|
|
||
|
|
def accept(self, s, acct):
|
||
|
|
self.api(s, "POST", f"/api/v1/follow_requests/{self.lookup(s, acct)}/authorize", ok={200})
|
||
|
|
|
||
|
|
def reject(self, s, acct):
|
||
|
|
self.api(s, "POST", f"/api/v1/follow_requests/{self.lookup(s, acct)}/reject", ok={200})
|
||
|
|
|
||
|
|
def relationship(self, s, acct):
|
||
|
|
rels = self.api_json(s, "GET", "/api/v1/accounts/relationships", params={"id": [self.lookup(s, acct)]})
|
||
|
|
r = rels[0] if rels else {}
|
||
|
|
return {k: bool(r.get(k)) for k in ("following", "followed_by", "requested", "blocking", "muting", "blocked_by")}
|
||
|
|
|
||
|
|
def block(self, s, acct):
|
||
|
|
self.api(s, "POST", f"/api/v1/accounts/{self.lookup(s, acct)}/block", ok={200})
|
||
|
|
|
||
|
|
def unblock(self, s, acct):
|
||
|
|
self.api(s, "POST", f"/api/v1/accounts/{self.lookup(s, acct)}/unblock", ok={200})
|
||
|
|
|
||
|
|
def mute(self, s, acct):
|
||
|
|
self.api(s, "POST", f"/api/v1/accounts/{self.lookup(s, acct)}/mute", ok={200})
|
||
|
|
|
||
|
|
def report(self, s, acct, uris, comment):
|
||
|
|
ids = [i for i in (self.local_status_id(s, u) for u in uris) if i]
|
||
|
|
self.api(s, "POST", "/api/v1/reports", ok={200}, form={"account_id": self.lookup(s, acct), "status_ids": ids,
|
||
|
|
"comment": comment, "forward": True})
|
||
|
|
|
||
|
|
# -- content
|
||
|
|
def upload(self, s, item):
|
||
|
|
filename, content, ctype = media.upload(item["kind"], item["seed"])
|
||
|
|
r = self.api(s, "POST", self.media_endpoint, files={"file": (filename, content, ctype)},
|
||
|
|
form={"description": item.get("alt")})
|
||
|
|
if r.status == 404 and self.media_endpoint != "/api/v1/media":
|
||
|
|
r = self.api(s, "POST", "/api/v1/media", files={"file": (filename, content, ctype)},
|
||
|
|
form={"description": item.get("alt")})
|
||
|
|
if not r.ok:
|
||
|
|
raise HttpError("POST", self.media_endpoint, r.status, r.text)
|
||
|
|
attachment = r.json()
|
||
|
|
for _ in range(60):
|
||
|
|
if attachment.get("url"):
|
||
|
|
break
|
||
|
|
time.sleep(1)
|
||
|
|
attachment = self.api_json(s, "GET", f"/api/v1/media/{attachment['id']}")
|
||
|
|
return attachment["id"]
|
||
|
|
|
||
|
|
def status_form(self, s, spec):
|
||
|
|
text = spec.text
|
||
|
|
for acct in spec.mentions:
|
||
|
|
if f"@{acct}" not in text:
|
||
|
|
text = f"@{acct} {text}"
|
||
|
|
for tag in spec.tags:
|
||
|
|
if f"#{tag}" not in text:
|
||
|
|
text = f"{text} #{tag}"
|
||
|
|
visibility = VISIBILITY.get(spec.visibility)
|
||
|
|
if visibility is None:
|
||
|
|
raise Unsupported(self.platform, f"post with visibility {spec.visibility}")
|
||
|
|
form = {"status": text, "visibility": visibility, "language": spec.language}
|
||
|
|
if spec.cw:
|
||
|
|
form["spoiler_text"] = spec.cw
|
||
|
|
form["sensitive"] = True
|
||
|
|
if spec.media:
|
||
|
|
form["media_ids"] = [self.upload(s, m) for m in spec.media[:self.max_media]]
|
||
|
|
if any(m.get("sensitive") for m in spec.media):
|
||
|
|
form["sensitive"] = True
|
||
|
|
if spec.poll:
|
||
|
|
form["poll[options]"] = spec.poll["options"]
|
||
|
|
form["poll[expires_in]"] = spec.poll.get("expires_in", 86400)
|
||
|
|
form["poll[multiple]"] = bool(spec.poll.get("multiple"))
|
||
|
|
if spec.reply_to_uri:
|
||
|
|
form["in_reply_to_id"] = self.local_status_id(s, spec.reply_to_uri) or self.resolve(s, spec.reply_to_uri)
|
||
|
|
if spec.quote_uri:
|
||
|
|
if "quote" not in self.caps:
|
||
|
|
raise Unsupported(self.platform, "quote")
|
||
|
|
form["quoted_status_id"] = self.local_status_id(s, spec.quote_uri) or self.resolve(s, spec.quote_uri)
|
||
|
|
return form
|
||
|
|
|
||
|
|
def post(self, s, spec):
|
||
|
|
if spec.kind not in ("note", "image", "video", "audio"):
|
||
|
|
raise Unsupported(self.platform, f"post a {spec.kind}")
|
||
|
|
status = self.api_json(s, "POST", "/api/v1/statuses", form=self.status_form(s, spec))
|
||
|
|
return Made(status["uri"], status["id"], status.get("url"))
|
||
|
|
|
||
|
|
def edit(self, s, uri, spec):
|
||
|
|
sid = self._own(s, uri)
|
||
|
|
text = spec.text
|
||
|
|
for acct in spec.mentions:
|
||
|
|
if f"@{acct}" not in text:
|
||
|
|
text = f"@{acct} {text}"
|
||
|
|
form = {"status": text}
|
||
|
|
if spec.cw:
|
||
|
|
form["spoiler_text"] = spec.cw
|
||
|
|
self.api(s, "PUT", f"/api/v1/statuses/{sid}", ok={200}, form=form)
|
||
|
|
|
||
|
|
def delete(self, s, uri):
|
||
|
|
self.api(s, "DELETE", f"/api/v1/statuses/{self._own(s, uri)}", ok={200})
|
||
|
|
|
||
|
|
def _act(self, s, uri, action):
|
||
|
|
sid = self.local_status_id(s, uri) or self.resolve(s, uri)
|
||
|
|
self.api(s, "POST", f"/api/v1/statuses/{sid}/{action}", ok={200})
|
||
|
|
|
||
|
|
def like(self, s, uri):
|
||
|
|
self._act(s, uri, "favourite")
|
||
|
|
|
||
|
|
def unlike(self, s, uri):
|
||
|
|
self._act(s, uri, "unfavourite")
|
||
|
|
|
||
|
|
def boost(self, s, uri):
|
||
|
|
self._act(s, uri, "reblog")
|
||
|
|
|
||
|
|
def unboost(self, s, uri):
|
||
|
|
self._act(s, uri, "unreblog")
|
||
|
|
|
||
|
|
def bookmark(self, s, uri):
|
||
|
|
self._act(s, uri, "bookmark")
|
||
|
|
|
||
|
|
def vote(self, s, uri, choices):
|
||
|
|
sid = self.local_status_id(s, uri) or self.resolve(s, uri)
|
||
|
|
status = self.api_json(s, "GET", f"/api/v1/statuses/{sid}")
|
||
|
|
poll = status.get("poll")
|
||
|
|
if not poll:
|
||
|
|
raise LookupError(f"{self.host} shows no poll on {uri}")
|
||
|
|
self.api(s, "POST", f"/api/v1/polls/{poll['id']}/votes", ok={200}, form={"choices": choices})
|
||
|
|
|
||
|
|
def _own(self, s, uri):
|
||
|
|
sid = self.local_status_id(s, uri)
|
||
|
|
if sid is None:
|
||
|
|
raise LookupError(f"{self.host} does not hold {uri}")
|
||
|
|
return sid
|
||
|
|
|
||
|
|
# -- reading back
|
||
|
|
def resolve(self, s, uri):
|
||
|
|
r = self.api(s, "GET", "/api/v2/search", params={"q": uri, "resolve": "true", "type": "statuses"})
|
||
|
|
for st in (r.json() or {}).get("statuses", []) if r.ok else []:
|
||
|
|
if st.get("uri") == uri or st.get("url") == uri:
|
||
|
|
return st["id"]
|
||
|
|
raise LookupError(f"{self.host} cannot resolve {uri}")
|
||
|
|
|
||
|
|
def local_status_id(self, s, uri):
|
||
|
|
row = self._rows([uri]).get(uri)
|
||
|
|
return row.local_id if row and row.exists and not row.deleted else None
|
||
|
|
|
||
|
|
def stored(self, uris):
|
||
|
|
rows = self._rows(uris)
|
||
|
|
return {u: rows.get(u) or Stored(False) for u in uris}
|
||
|
|
|
||
|
|
def visible(self, s, local_id):
|
||
|
|
return self.api(s, "GET", f"/api/v1/statuses/{local_id}").status == 200
|
||
|
|
|
||
|
|
def status(self, s, local_id):
|
||
|
|
r = self.api(s, "GET", f"/api/v1/statuses/{local_id}")
|
||
|
|
return r.json() if r.ok else None
|
||
|
|
|
||
|
|
def notifications(self, s):
|
||
|
|
r = self.api(s, "GET", "/api/v1/notifications", params={"limit": 80}, ok={200})
|
||
|
|
out = []
|
||
|
|
for n in r.json() or []:
|
||
|
|
out.append({"type": n["type"], "acct": _full_acct((n.get("account") or {}).get("acct", ""), self.host),
|
||
|
|
"uri": (n.get("status") or {}).get("uri")})
|
||
|
|
return out
|
||
|
|
|
||
|
|
def _rows(self, uris):
|
||
|
|
raise Unsupported(self.platform, "read its database")
|
||
|
|
|
||
|
|
|
||
|
|
def _full_acct(acct, host):
|
||
|
|
return acct if "@" in acct else f"{acct}@{host}"
|
||
|
|
|
||
|
|
|
||
|
|
def _same_acct(found, wanted, host):
|
||
|
|
if not found:
|
||
|
|
return False
|
||
|
|
return _full_acct(found, host).lower() == wanted.lower()
|