71 lines
3.8 KiB
Python
71 lines
3.8 KiB
Python
 
|
"""Pixelfed 0.14: accounts made with `artisan user:create` in its container, tokens as Passport personal access tokens made
|
||
|
|
through tinker (peers/pixelfed.sh numbers its clients past every user, or Passport refuses them), objects read from its
|
||
|
|
Postgres (`pixelfed`). Every post is a photo post: Pixelfed refuses one without media ("Empty statuses are not allowed"),
|
||
|
|
so a post the plan gives no picture gets one; replies are comments and may be words alone. A local status keeps no uri:
|
||
|
|
it is /p/<username>/<id>. A remote post is kept with its page address (its `url`) as `uri` when it has one, and its id
|
||
|
|
as `object_url`."""
|
||
|
|
import dataclasses
|
||
|
|
|
||
|
|
from core import podman
|
||
|
|
from dialects.base import Stored, Unsupported
|
||
|
|
from dialects.mastodon_api import MastodonApi
|
||
|
|
|
||
|
|
# Pixelfed's statuses.visibility, as the town names visibilities
|
||
|
|
VIS = {"public": "public", "unlisted": "unlisted", "private": "followers", "direct": "direct"}
|
||
|
|
|
||
|
|
|
||
|
|
class Pixelfed(MastodonApi):
|
||
|
|
platform = "pixelfed"
|
||
|
|
caps = frozenset({"post", "reply", "cw", "media", "like", "boost", "bookmark", "follow", "block", "mute", "delete",
|
||
|
|
"edit", "profile"})
|
||
|
|
container = "pasture-pixelfed"
|
||
|
|
db = "pixelfed"
|
||
|
|
|
||
|
|
def provision(self, accounts):
|
||
|
|
existing = {r["username"] for r in podman.psql(self.db, "select username from users")}
|
||
|
|
for a in accounts:
|
||
|
|
if a.username in existing:
|
||
|
|
continue
|
||
|
|
podman.exec_(self.container, "php", "artisan", "user:create", f"--name={a.name or a.username}", f"--username={a.username}",
|
||
|
|
f"--email={a.username}@{self.host}", f"--password={a.password}", "--confirm_email=1", "--no-interaction")
|
||
|
|
return [self.session_from_token(a, self._token(a.username)) for a in accounts]
|
||
|
|
|
||
|
|
def _token(self, username):
|
||
|
|
out = podman.exec_(self.container, "php", "artisan", "tinker", "--execute",
|
||
|
|
f"echo App\\Models\\User::where('username', '{username}')->first()"
|
||
|
|
"->createToken('pasture-town', ['read', 'write', 'follow', 'push'])->accessToken;")
|
||
|
|
return out.strip().splitlines()[-1].strip()
|
||
|
|
|
||
|
|
def post(self, s, spec):
|
||
|
|
if spec.kind not in ("note", "image"):
|
||
|
|
raise Unsupported(self.platform, f"post a {spec.kind}")
|
||
|
|
if spec.poll:
|
||
|
|
raise Unsupported(self.platform, "post a poll")
|
||
|
|
if not spec.media and not spec.reply_to_uri:
|
||
|
|
spec = dataclasses.replace(spec, media=[{"kind": "image", "seed": spec.text[:24], "alt": "a picture for the words"}])
|
||
|
|
return super().post(s, spec)
|
||
|
|
|
||
|
|
def _rows(self, uris):
|
||
|
|
if not uris:
|
||
|
|
return {}
|
||
|
|
local = {u: u.rsplit("/", 1)[1] for u in uris if u.startswith(f"{self.base}/p/")}
|
||
|
|
rows = podman.psql(self.db, f"""
|
||
|
|
select s.id::text as local_id, s.uri as stored_uri, s.object_url,
|
||
|
|
'{self.base}/p/' || p.username || '/' || s.id as local_uri,
|
||
|
|
s.visibility, s.caption as text, s.cw_summary as cw, s.edited_at is not null as edited,
|
||
|
|
s.deleted_at is not null as deleted,
|
||
|
|
coalesce(r.object_url, r.uri, case when r.id is not null then '{self.base}/p/' || rp.username || '/' || r.id end) as parent_uri,
|
||
|
|
s.likes_count as likes, s.reblogs_count as boosts, s.reply_count as replies
|
||
|
|
from statuses s join profiles p on p.id = s.profile_id
|
||
|
|
left join statuses r on r.id = s.in_reply_to_id left join profiles rp on rp.id = r.profile_id
|
||
|
|
where s.reblog_of_id is null and (s.uri = any(string_to_array(:'p1', ' ')) or s.object_url = any(string_to_array(:'p1', ' '))
|
||
|
|
or s.id::text = any(string_to_array(:'p2', ' ')))""", " ".join(uris), " ".join(local.values()) or "-")
|
||
|
|
out = {}
|
||
|
|
for r in rows:
|
||
|
|
# the address the post was asked by: its id (object_url), its uri, or a local post's /p/ address
|
||
|
|
uri = next((u for u in (r["object_url"], r["stored_uri"], r["local_uri"]) if u in uris), None)
|
||
|
|
if uri is None:
|
||
|
|
continue
|
||
|
|
out[uri] = Stored(True, bool(r["deleted"]), r["local_id"], VIS.get(r["visibility"], r["visibility"]), r["text"], r["cw"],
|
||
|
|
bool(r["edited"]), r["parent_uri"], r["likes"], r["boosts"], r["replies"], None, {}, r)
|
||
|
|
return out
|