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

"""What every peer driver offers. A driver acts as one account of its platform through that platform's own API, and reads
back what the platform holds without ever making it fetch: `stored` asks its database, `seen` its API as one account.
Anything a platform cannot do raises Unsupported, and the planner never asks for it."""
from dataclasses import dataclass, field
from typing import Optional

from core.http import client


class Unsupported(Exception):
	def __init__(self, platform, what):
		super().__init__(f"{platform} cannot {what}")
		self.platform = platform
		self.what = what


@dataclass
class Account:
	platform: str
	username: str
	password: str
	name: str = None
	bio: str = None
	fields: list = field(default_factory=list)  # [(name, value)]
	locked: bool = False
	bot: bool = False
	lang: str = "en"
	avatar_seed: str = None
	header_seed: str = None
	root: str = None  # PrivaPub only: the root login that owns this persona

	def acct(self, host):
		return f"{self.username}@{host}"


@dataclass
class Session:
	account: Account
	token: str
	local_id: str  # the account's id in its own server's API
	actor_uri: str  # its ActivityPub id


@dataclass
class PostSpec:
	text: str
	kind: str = "note"  # note | article | page | event | link | video | audio | image
	visibility: str = "public"  # public | unlisted | followers | direct | circle | community | located
	cw: str = None
	mentions: list = field(default_factory=list)  # accts: "name@host"
	tags: list = field(default_factory=list)
	media: list = field(default_factory=list)  # [{"kind": "image|audio|video", "seed": str, "alt": str, "sensitive": bool}]
	poll: dict = None  # {"options": [...], "multiple": bool, "expires_in": seconds}
	quote_uri: str = None
	reply_to_uri: str = None
	title: str = None
	link: str = None
	event: dict = None  # {"start", "end", "place"}
	language: str = None
	group: str = None  # community or circle reference (driver-specific id)
	location: dict = None  # PrivaPub located posts: {"lat", "lng", "range_km"}


@dataclass
class Made:
	uri: str
	local_id: str
	url: Optional[str] = None


@dataclass
class Stored:
	"""One object as a server holds it, read from its database."""
	exists: bool
	deleted: bool = False
	local_id: str = None
	visibility: str = None
	text: str = None
	cw: str = None
	edited: bool = False
	parent_uri: str = None
	likes: int = None
	boosts: int = None
	replies: int = None
	votes: list = None
	reactions: dict = None
	raw: dict = None


class Driver:
	"""Base for every platform. `host` is its site name (gts.test); `api` its base URL through Caddy."""
	platform = "base"
	caps = frozenset()

	def __init__(self, host):
		self.host = host
		self.base = f"https://{host}"
		self.http = client()

	# -- accounts
	def provision(self, accounts):
		"""Creates the accounts (idempotently) and returns a Session for each, in order."""
		raise Unsupported(self.platform, "provision accounts")

	def update_profile(self, s, account):
		raise Unsupported(self.platform, "update profiles")

	def lookup(self, s, acct):
		"""The local id of account `acct` (name@host) as `s` sees it, resolving it through WebFinger if needed."""
		raise Unsupported(self.platform, "look up accounts")

	# -- the graph
	def follow(self, s, acct):
		"""Follows acct; returns "accepted" or "requested"."""
		raise Unsupported(self.platform, "follow")

	def unfollow(self, s, acct):
		raise Unsupported(self.platform, "unfollow")

	def pending(self, s):
		"""The accts waiting for `s` to accept their follow."""
		raise Unsupported(self.platform, "list follow requests")

	def accept(self, s, acct):
		raise Unsupported(self.platform, "accept follow requests")

	def reject(self, s, acct):
		raise Unsupported(self.platform, "reject follow requests")

	def relationship(self, s, acct):
		"""{following, followed_by, requested, blocking, muting}"""
		raise Unsupported(self.platform, "read relationships")

	def block(self, s, acct):
		raise Unsupported(self.platform, "block")

	def unblock(self, s, acct):
		raise Unsupported(self.platform, "unblock")

	def mute(self, s, acct):
		raise Unsupported(self.platform, "mute")

	def report(self, s, acct, uris, comment):
		raise Unsupported(self.platform, "report")

	# -- content
	def post(self, s, spec):
		raise Unsupported(self.platform, "post")

	def edit(self, s, uri, spec):
		raise Unsupported(self.platform, "edit")

	def delete(self, s, uri):
		raise Unsupported(self.platform, "delete")

	def like(self, s, uri):
		raise Unsupported(self.platform, "like")

	def unlike(self, s, uri):
		raise Unsupported(self.platform, "unlike")

	def boost(self, s, uri):
		raise Unsupported(self.platform, "boost")

	def unboost(self, s, uri):
		raise Unsupported(self.platform, "unboost")

	def react(self, s, uri, emoji):
		raise Unsupported(self.platform, "react")

	def vote(self, s, uri, choices):
		raise Unsupported(self.platform, "vote")

	def bookmark(self, s, uri):
		raise Unsupported(self.platform, "bookmark")

	# -- reading back, never fetching
	def local_status_id(self, s, uri):
		"""This server's API id for the object `uri`, or None if it does not hold it. Never fetches."""
		raise Unsupported(self.platform, "find statuses by URI")

	def resolve(self, s, uri):
		"""This server's API id for `uri`, fetching it if it must (a deliberate fetch, recorded by the caller)."""
		raise Unsupported(self.platform, "resolve statuses")

	def stored(self, uris):
		"""{uri: Stored} for every uri, from the database."""
		raise Unsupported(self.platform, "read its database")

	def seen(self, s, uris):
		"""{uri: bool}: whether `s` can see each object through the API."""
		out = {}
		for uri in uris:
			sid = self.local_status_id(s, uri)
			out[uri] = sid is not None and self.visible(s, sid)
		return out

	def visible(self, s, local_id):
		raise Unsupported(self.platform, "read statuses")

	def notifications(self, s):
		"""[{type, acct, uri}] newest first, normalised to Mastodon's types."""
		raise Unsupported(self.platform, "read notifications")

	def actor_uri(self, username):
		"""The ActivityPub id of a local account, as this server's WebFinger names it."""
		r = self.http.get(f"{self.base}/.well-known/webfinger", params={"resource": f"acct:{username}@{self.host}"},
						  headers={"Accept": "application/jrd+json"})
		for link in (r.json() or {}).get("links", []) if r.ok else []:
			if link.get("rel") == "self" and "json" in (link.get("type") or ""):
				return link["href"]
		return None