SocialPub/tools/pasture/town/dialects/misskey_api.py

"""Misskey and its forks (Sharkey, later CherryPick, Iceshrimp's Misskey API): every call is POST /api/<endpoint> with
the token as `i`. Accounts are made by the admin made at setup (`admin/accounts/create`); objects are read from the
`note` table. A local note has no `uri` there: its id is the last part of https://<host>/notes/<id>."""
from core import media, podman
from core.http import HttpError
from dialects.base import Driver, Made, Session, Stored, Unsupported
import os

VISIBILITY = {"public": "public", "unlisted": "home", "followers": "followers", "direct": "specified"}
VIS = {"public": "public", "home": "unlisted", "followers": "followers", "specified": "direct"}
LIKE = "❤"


class MisskeyApi(Driver):
	platform = "misskey"
	caps = frozenset({"post", "reply", "cw", "media", "poll", "like", "react", "boost", "bookmark", "follow", "block",
					  "mute", "report", "dm", "delete", "vote", "quote", "profile"})
	db = "misskey"

	def __init__(self, host, db=None):
		super().__init__(host)
		self.db = db or host.split(".")[0]
		self._ids = {}

	def mk(self, s_or_token, endpoint, body=None, ok=None):
		body = dict(body or {})
		token = s_or_token.token if isinstance(s_or_token, Session) else s_or_token
		if token:
			body["i"] = token
		r = self.http.post(f"{self.base}/api/{endpoint}", json=body, template=f"/api/{endpoint}")
		if ok is not False and r.status not in (200, 204):
			raise HttpError("POST", endpoint, r.status, r.text)
		return r.json() if r.body else None

	def admin_token(self):
		here = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
		with open(os.path.join(here, ".state", f"{self.db}.token")) as f:
			return f.read().strip()

	def provision(self, accounts):
		admin = self.admin_token()
		sessions = []
		existing = {r["username"] for r in podman.psql(self.db, 'select username from "user" where host is null')}
		for a in accounts:
			if a.username in existing:
				r = self.http.post(f"{self.base}/api/signin-flow", json={"username": a.username, "password": a.password})
				token = (r.json() or {}).get("i")
				if not token:
					raise RuntimeError(f"{self.host}: cannot sign {a.username} in ({r.status} {r.text[:200]})")
			else:
				token = self.mk(admin, "admin/accounts/create", {"username": a.username, "password": a.password})["token"]
			me = self.mk(token, "i")
			sessions.append(Session(a, token, me["id"], f"{self.base}/users/{me['id']}"))
		return sessions

	def update_profile(self, s, account):
		body = {"name": account.name, "description": account.bio, "isLocked": account.locked, "isBot": account.bot,
				"fields": [{"name": k, "value": v} for k, v in account.fields[:4]]}
		if account.avatar_seed:
			body["avatarId"] = self._upload(s, {"kind": "image", "seed": account.avatar_seed, "alt": None})
		if account.header_seed:
			body["bannerId"] = self._upload(s, {"kind": "image", "seed": account.header_seed, "alt": None})
		self.mk(s, "i/update", body)

	def _upload(self, s, item):
		filename, content, ctype = media.upload(item["kind"], item["seed"])
		form = {"i": s.token, "isSensitive": bool(item.get("sensitive"))}
		if item.get("alt"):
			form["comment"] = item["alt"]
		r = self.http.post(f"{self.base}/api/drive/files/create", files={"file": (filename, content, ctype)}, form=form, ok={200})
		return r.json()["id"]

	# -- accounts
	def lookup(self, s, acct):
		key = (s.token, acct)
		if key not in self._ids:
			name, _, host = acct.partition("@")
			body = {"username": name}
			if host and host != self.host:
				body["host"] = host
			self._ids[key] = self.mk(s, "users/show", body)["id"]
		return self._ids[key]

	def follow(self, s, acct):
		self.mk(s, "following/create", {"userId": self.lookup(s, acct)}, ok=False)
		rel = self.relationship(s, acct)
		return "accepted" if rel["following"] else "requested"

	def unfollow(self, s, acct):
		self.mk(s, "following/delete", {"userId": self.lookup(s, acct)})

	def pending(self, s):
		out = []
		for req in self.mk(s, "following/requests/list", {"limit": 100}) or []:
			u = req["follower"]
			out.append(f"{u['username']}@{u.get('host') or self.host}")
		return out

	def accept(self, s, acct):
		self.mk(s, "following/requests/accept", {"userId": self.lookup(s, acct)})

	def reject(self, s, acct):
		self.mk(s, "following/requests/reject", {"userId": self.lookup(s, acct)})

	def relationship(self, s, acct):
		r = self.mk(s, "users/relation", {"userId": self.lookup(s, acct)})
		r = r[0] if isinstance(r, list) else r
		return {"following": r.get("isFollowing"), "followed_by": r.get("isFollowed"),
				"requested": r.get("hasPendingFollowRequestFromYou"), "blocking": r.get("isBlocking"),
				"muting": r.get("isMuted"), "blocked_by": r.get("isBlocked")}

	def block(self, s, acct):
		self.mk(s, "blocking/create", {"userId": self.lookup(s, acct)})

	def unblock(self, s, acct):
		self.mk(s, "blocking/delete", {"userId": self.lookup(s, acct)})

	def mute(self, s, acct):
		self.mk(s, "mute/create", {"userId": self.lookup(s, acct)})

	def report(self, s, acct, uris, comment):
		self.mk(s, "users/report-abuse", {"userId": self.lookup(s, acct), "comment": comment})

	# -- content
	def post(self, s, spec):
		if spec.kind not in ("note", "image", "video", "audio"):
			raise Unsupported(self.platform, f"post a {spec.kind}")
		visibility = VISIBILITY.get(spec.visibility)
		if visibility is None:
			raise Unsupported(self.platform, f"post with visibility {spec.visibility}")
		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}"
		body = {"text": text, "visibility": visibility}
		if spec.cw:
			body["cw"] = spec.cw
		if visibility == "specified":
			body["visibleUserIds"] = [self.lookup(s, a) for a in spec.mentions]
		if spec.media:
			body["fileIds"] = [self._upload(s, m) for m in spec.media[:16]]
		if spec.poll:
			body["poll"] = {"choices": spec.poll["options"], "multiple": bool(spec.poll.get("multiple")),
							"expiredAfter": spec.poll.get("expires_in", 86400) * 1000}
		if spec.reply_to_uri:
			body["replyId"] = self.local_status_id(s, spec.reply_to_uri) or self.resolve(s, spec.reply_to_uri)
		if spec.quote_uri:
			body["renoteId"] = self.local_status_id(s, spec.quote_uri) or self.resolve(s, spec.quote_uri)
		note = self.mk(s, "notes/create", body)["createdNote"]
		return Made(note.get("uri") or f"{self.base}/notes/{note['id']}", note["id"], note.get("url"))

	def edit(self, s, uri, spec):
		if "edit" not in self.caps:
			raise Unsupported(self.platform, "edit")
		text = spec.text
		for acct in spec.mentions:
			if f"@{acct}" not in text:
				text = f"@{acct} {text}"
		body = {"editId": self._own(uri), "text": text, "cw": spec.cw}
		self.mk(s, "notes/edit", body)

	def delete(self, s, uri):
		self.mk(s, "notes/delete", {"noteId": self._own(uri)})

	def _note(self, s, uri):
		return self.local_status_id(s, uri) or self.resolve(s, uri)

	def like(self, s, uri):
		self.mk(s, "notes/reactions/create", {"noteId": self._note(s, uri), "reaction": LIKE})

	def unlike(self, s, uri):
		self.mk(s, "notes/reactions/delete", {"noteId": self._note(s, uri)})

	def react(self, s, uri, emoji):
		self.mk(s, "notes/reactions/create", {"noteId": self._note(s, uri), "reaction": emoji})

	def boost(self, s, uri):
		self.mk(s, "notes/create", {"renoteId": self._note(s, uri), "visibility": "public"})

	def unboost(self, s, uri):
		self.mk(s, "notes/unrenote", {"noteId": self._note(s, uri)})

	def bookmark(self, s, uri):
		self.mk(s, "notes/favorites/create", {"noteId": self._note(s, uri)})

	def vote(self, s, uri, choices):
		nid = self._note(s, uri)
		for c in choices:
			self.mk(s, "notes/polls/vote", {"noteId": nid, "choice": c})

	def _own(self, uri):
		nid = self.local_status_id(None, uri)
		if nid is None:
			raise LookupError(f"{self.host} does not hold {uri}")
		return nid

	# -- reading back
	def resolve(self, s, uri):
		r = self.mk(s, "ap/show", {"uri": uri}, ok=False)
		if r and r.get("type") == "Note":
			return r["object"]["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 else None

	def stored(self, uris):
		rows = self._rows(uris)
		return {u: rows.get(u) or Stored(False) for u in uris}

	def seen(self, s, uris):
		out = {}
		for uri in uris:
			nid = self.local_status_id(s, uri)
			if nid is None:
				out[uri] = False
				continue
			r = self.http.post(f"{self.base}/api/notes/show", json={"i": s.token, "noteId": nid})
			note = r.json() if r.ok else None
			out[uri] = bool(note) and not note.get("isHidden")
		return out

	def notifications(self, s):
		kinds = {"reaction": "favourite", "renote": "reblog", "reply": "mention", "mention": "mention",
				 "follow": "follow", "receiveFollowRequest": "follow_request", "quote": "quote", "pollEnded": "poll"}
		out = []
		for n in self.mk(s, "i/notifications", {"limit": 100}) or []:
			u = n.get("user") or {}
			note = n.get("note") or {}
			out.append({"type": kinds.get(n["type"], n["type"]), "acct": f"{u.get('username')}@{u.get('host') or self.host}",
						"uri": note.get("uri") or (f"{self.base}/notes/{note['id']}" if note.get("id") else None)})
		return out

	def actor_uri(self, username):
		row = podman.psql(self.db, 'select id from "user" where host is null and username = :\'p1\'', username)
		return f"{self.base}/users/{row[0]['id']}" if row else None

	def _rows(self, uris):
		if not uris:
			return {}
		local = {u: u.rsplit("/", 1)[1] for u in uris if u.startswith(f"{self.base}/notes/")}
		edited = "n.\"updatedAt\" is not null" if self.platform == "sharkey" else "false"
		rows = podman.psql(self.db, f"""
			select n.id, n.uri, n.visibility, n.text, n.cw, {edited} as edited, n."renoteCount" as boosts,
				n."repliesCount" as replies, n.reactions, p.votes, coalesce(r.uri, case when r.id is not null
					then 'https://{self.host}/notes/' || r.id end) as parent_uri
			from note n left join poll p on p."noteId" = n.id left join note r on r.id = n."replyId"
			where (n."renoteId" is null or n.text is not null)
				and (n.uri = any(string_to_array(:'p1', ' ')) or n.id = any(string_to_array(:'p2', ' ')))""",
						   " ".join(uris), " ".join(local.values()) or "-")
		out = {}
		for r in rows:
			uri = r["uri"] or f"{self.base}/notes/{r['id']}"
			if uri not in uris:
				continue
			reactions = r["reactions"] or {}
			out[uri] = Stored(True, False, r["id"], VIS.get(r["visibility"], r["visibility"]), r["text"], r["cw"],
							  bool(r["edited"]), r["parent_uri"], sum(reactions.values()), r["boosts"], r["replies"],
							  r["votes"], reactions, r)
		return out


class Misskey(MisskeyApi):
	platform = "misskey"


class Sharkey(MisskeyApi):
	platform = "sharkey"
	caps = MisskeyApi.caps | {"edit"}