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

"""PrivaPub: roots signed up on /clientapi, personas made under them, and each persona's Mastodon token exchanged for the
root's JWT exactly as decePubClient does (RFC 8693, client `decepub`). Everything a Mastodon client can say goes
through the Mastodon API; circles, communities and located posts through /clientapi. Objects are read from Mongo."""
from core import podman
from dialects.base import Made, Session, Stored, Unsupported
from dialects.mastodon_api import MastodonApi

VIS = {0: "public", 1: "unlisted", 2: "followers", 3: "direct", 4: "circle", 5: "located",
	   "Public": "public", "Unlisted": "unlisted", "FollowersOnly": "followers", "Direct": "direct", "Circle": "circle",
	   "LocalGeo": "located"}


class PrivaPub(MastodonApi):
	platform = "privapub"
	caps = MastodonApi.caps | {"react", "quote", "circle", "community", "located"}

	def __init__(self, host="privapub.test"):
		super().__init__(host)
		self._jwts = {}

	# -- roots and personas
	def jwt(self, root, password):
		"""The root's JWT: signed up the first time, signed in again by any later command that reuses saved sessions."""
		if root in self._jwts:
			return self._jwts[root]
		body = {"userName": root, "password": password}
		r = self.http.post(self.base + "/clientapi/user/signup", json=body)
		token = (r.json() or {}).get("token") if r.ok else None
		if not token:
			token = self.http.post(self.base + "/clientapi/user/login", json=body, ok={200}).json()["token"]
		self._jwts[root] = token
		return token

	def personas(self, root, password):
		jwt = self.jwt(root, password)
		r = self.http.get(self.base + "/clientapi/avatar/private/list", headers={"Authorization": f"Bearer {jwt}"}, ok={200})
		return {p["userName"]: p for p in r.json() or []}

	def provision(self, accounts):
		"""Each account is a persona; `account.root` names its root, whose password is `account.password`."""
		sessions = []
		for root in dict.fromkeys(a.root for a in accounts):
			mine = [a for a in accounts if a.root == root]
			password = mine[0].password
			jwt = self.jwt(root, password)
			existing = self.personas(root, password)
			for a in mine:
				if a.username not in existing:
					self.http.post(self.base + "/clientapi/avatar/private/insert", ok={200, 201},
								   headers={"Authorization": f"Bearer {jwt}"},
								   json={"userName": a.username, "name": a.name or a.username, "biography": a.bio or a.name or a.username,
										 "fields": dict(a.fields)})
			existing = self.personas(root, password)
			for a in mine:
				token = self.exchange(jwt, existing[a.username]["id"])
				sessions.append(self.session_from_token(a, token))
		by_name = {s.account.username: s for s in sessions}
		return [by_name[a.username] for a in accounts]

	def exchange(self, jwt, avatar_id):
		return self.http.post(self.base + "/oauth/token", ok={200}, form={
			"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", "client_id": "decepub",
			"subject_token": jwt, "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
			"avatar_id": avatar_id, "scope": "read write follow"}).json()["access_token"]

	def actor_uri(self, username):
		return f"https://{self.host}/peasants/{username}"

	# -- groups: communities (FEP-1b12) and circles
	def group(self, s, username, name, community, policy="followers", approve=False):
		jwt = self.jwt(s.account.root, s.account.password)
		r = self.http.post(self.base + "/clientapi/group/insert", headers={"Authorization": f"Bearer {jwt}"}, ok={200, 201},
						   json={"avatarId": s.local_id, "userName": username, "name": name, "description": name,
								 "isCommunity": community, "postingPolicy": policy, "isDiscoverable": community,
								 "manuallyApprovesMembers": approve})
		return r.json()

	# -- content
	def post(self, s, spec):
		if spec.visibility in ("circle", "community", "located"):
			return self._client_post(s, spec)
		return super().post(s, spec)

	def _client_post(self, s, spec):
		jwt = self.jwt(s.account.root, s.account.password)
		body = {"avatarId": s.local_id, "text": spec.text, "title": spec.title, "hasContentWarning": bool(spec.cw),
				"spoilerText": spec.cw}
		if spec.visibility == "located":
			loc = spec.location
			body.update({"latitude": loc["lat"], "longitude": loc["lng"], "rangeKm": loc.get("range_km", 5)})
		else:
			body["groupId"] = spec.group
		if spec.reply_to_uri:
			body["answeringToPostId"] = self.local_status_id(s, spec.reply_to_uri)
		r = self.http.post(self.base + "/clientapi/post/insert", headers={"Authorization": f"Bearer {jwt}"},
						   ok={200, 201}, json=body).json()
		post_id = r.get("id") if isinstance(r, dict) else None
		if not post_id:
			raise RuntimeError(f"/clientapi/post/insert answered {r!r}")
		row = podman.mongo(f"db.Post.findOne({{_id: ObjectId('{post_id}')}}, {{ObjectURI: 1}})")
		return Made(row["ObjectURI"], post_id, None)

	def react(self, s, uri, emoji):
		import urllib.parse
		sid = self.local_status_id(s, uri) or self.resolve(s, uri)
		self.api(s, "PUT", f"/api/v1/pleroma/statuses/{sid}/reactions/{urllib.parse.quote(emoji)}", ok={200})

	# -- reading back
	def _rows(self, uris):
		if not uris:
			return {}
		import json
		docs = podman.mongo(f"db.Post.find({{ObjectURI: {{$in: {json.dumps(list(uris))}}}}}, "
							"{ObjectURI: 1, Visibility: 1, Text: 1, ContentHtml: 1, SpoilerText: 1, EditedAt: 1, "
							"InReplyToURI: 1, FavouritesCount: 1, ReblogsCount: 1, RepliesCount: 1, DownvotesCount: 1, "
							"Poll: 1, DeletedAt: 1, ReblogOfPostId: 1, AuthorGone: 1, GroupId: 1, ConversationId: 1}).toArray()") or []
		ids = [d["_id"]["$oid"] if isinstance(d["_id"], dict) else d["_id"] for d in docs]
		reactions = {}
		if ids:
			for g in podman.mongo(f"db.Reaction.aggregate([{{$match: {{PostId: {{$in: {json.dumps(ids)}}}}}}}, "
								  "{$group: {_id: {p: '$PostId', e: '$Emoji'}, n: {$sum: 1}}}]).toArray()") or []:
				reactions.setdefault(g["_id"]["p"], {})[g["_id"]["e"]] = g["n"]
		out = {}
		for d in docs:
			if d.get("ReblogOfPostId"):
				continue
			pid = d["_id"]["$oid"] if isinstance(d["_id"], dict) else d["_id"]
			poll = d.get("Poll") or {}
			votes = [o.get("Votes", o.get("VotesCount", 0)) for o in poll.get("Options", [])] if poll else None
			out[d["ObjectURI"]] = Stored(True, bool(d.get("DeletedAt")), pid, VIS.get(d.get("Visibility"), str(d.get("Visibility"))),
										 d.get("Text") or d.get("ContentHtml"), d.get("SpoilerText") or None,
										 bool(d.get("EditedAt")), d.get("InReplyToURI"), d.get("FavouritesCount", 0),
										 d.get("ReblogsCount", 0), d.get("RepliesCount", 0), votes, reactions.get(pid, {}), d)
		return out