social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import asyncio
2import json
3import re
4from dataclasses import dataclass, field
5from typing import Any, cast, override
6
7import websockets
8
9from cross.attachments import (
10 LabelsAttachment,
11 LanguagesAttachment,
12 MediaAttachment,
13 QuoteAttachment,
14 RemoteUrlAttachment,
15 SensitiveAttachment,
16)
17from cross.media import Blob, download_blob
18from cross.post import Post
19from cross.service import InputService
20from database.connection import DatabasePool
21from mastodon.info import MastodonService, validate_and_transform
22from mastodon.parser import StatusParser
23
24ALLOWED_VISIBILITY: list[str] = ["public", "unlisted"]
25
26
27@dataclass(kw_only=True)
28class MastodonInputOptions:
29 token: str
30 instance: str
31 allowed_visibility: list[str] = field(
32 default_factory=lambda: ALLOWED_VISIBILITY.copy()
33 )
34 filters: list[re.Pattern[str]] = field(default_factory=lambda: [])
35
36 @classmethod
37 def from_dict(cls, data: dict[str, Any]) -> "MastodonInputOptions":
38 validate_and_transform(data)
39
40 if "allowed_visibility" in data:
41 for vis in data.get("allowed_visibility", []):
42 if vis not in ALLOWED_VISIBILITY:
43 raise ValueError(f"Invalid visibility option {vis}!")
44
45 if "filters" in data:
46 data["filters"] = [re.compile(r) for r in data["filters"]]
47
48 return MastodonInputOptions(**data)
49
50
51class MastodonInputService(MastodonService, InputService):
52 def __init__(self, db: DatabasePool, options: MastodonInputOptions) -> None:
53 super().__init__(options.instance, db)
54 self.options: MastodonInputOptions = options
55
56 self.log.info("Verifying %s credentails...", self.url)
57 responce = self.verify_credentials()
58 self.user_id: str = responce["id"]
59
60 self.log.info("Getting %s configuration...", self.url)
61 responce = self.fetch_instance_info()
62 self.streaming_url: str = responce["urls"]["streaming_api"]
63
64 @override
65 def _get_token(self) -> str:
66 return self.options.token
67
68 def _on_create_post(self, status: dict[str, Any]):
69 if status["account"]["id"] != self.user_id:
70 return
71
72 if status["visibility"] not in self.options.allowed_visibility:
73 return
74
75 reblog: dict[str, Any] | None = status.get("reblog")
76 if reblog:
77 if reblog["account"]["id"] != self.user_id:
78 return
79 self._on_reblog(status, reblog)
80 return
81
82 if status.get("poll"):
83 self.log.info("Skipping '%s'! Contains a poll..", status["id"])
84 return
85
86 quote: dict[str, Any] | None = status.get("quote")
87 if quote:
88 quote = quote['quoted_status'] if quote.get('quoted_status') else quote
89 if not quote or quote["account"]["id"] != self.user_id:
90 return
91
92 in_reply: str | None = status.get("in_reply_to_id")
93 in_reply_to: str | None = status.get("in_reply_to_account_id")
94 if in_reply_to and in_reply_to != self.user_id:
95 return
96
97 parent = None
98 if in_reply:
99 parent = self._get_post(self.url, self.user_id, in_reply)
100 if not parent:
101 self.log.info(
102 "Skipping %s, parent %s not found in db", status["id"], in_reply
103 )
104 return
105 parser = StatusParser()
106 parser.feed(status["content"])
107 text, fragments = parser.get_result()
108
109 post = Post(id=status["id"], parent_id=in_reply, text=text)
110 post.fragments.extend(fragments)
111
112 if quote:
113 post.attachments.put(QuoteAttachment(quoted_id=quote['id'], quoted_user=self.user_id))
114 if status.get("url"):
115 post.attachments.put(RemoteUrlAttachment(url=status["url"]))
116 if status.get("sensitive"):
117 post.attachments.put(SensitiveAttachment(sensitive=True))
118 if status.get("language"):
119 post.attachments.put(LanguagesAttachment(langs=[status["language"]]))
120 if status.get("spoiler"):
121 post.attachments.put(LabelsAttachment(labels=[status["spoiler"]]))
122
123 blobs: list[Blob] = []
124 for media in status.get("media_attachments", []):
125 self.log.info("Downloading %s...", media["url"])
126 blob: Blob | None = download_blob(media["url"], media.get("alt"))
127 if not blob:
128 self.log.error(
129 "Skipping %s! Failed to download media %s.",
130 status["id"],
131 media["url"],
132 )
133 return
134 blobs.append(blob)
135
136 if blobs:
137 post.attachments.put(MediaAttachment(blobs=blobs))
138
139 if parent:
140 self._insert_post(
141 {
142 "user": self.user_id,
143 "service": self.url,
144 "identifier": status["id"],
145 "parent": parent["id"],
146 "root": parent["id"] if not parent["root"] else parent["root"],
147 }
148 )
149 else:
150 self._insert_post(
151 {
152 "user": self.user_id,
153 "service": self.url,
154 "identifier": status["id"],
155 }
156 )
157
158 for out in self.outputs:
159 self.submitter(lambda: out.accept_post(post))
160
161 def _on_reblog(self, status: dict[str, Any], reblog: dict[str, Any]):
162 reposted = self._get_post(self.url, self.user_id, reblog["id"])
163 if not reposted:
164 self.log.info(
165 "Skipping repost '%s' as reposted post '%s' was not found in the db.",
166 status["id"],
167 reblog["id"],
168 )
169 return
170
171 self._insert_post(
172 {
173 "user": self.user_id,
174 "service": self.url,
175 "identifier": status["id"],
176 "reposted": reposted["id"],
177 }
178 )
179
180 for out in self.outputs:
181 self.submitter(lambda: out.accept_repost(status["id"], reblog["id"]))
182
183 def _on_delete_post(self, status_id: str):
184 post = self._get_post(self.url, self.user_id, status_id)
185 if not post:
186 return
187
188 if post["reposted_id"]:
189 for output in self.outputs:
190 self.submitter(lambda: output.delete_repost(status_id))
191 else:
192 for output in self.outputs:
193 self.submitter(lambda: output.delete_post(status_id))
194 self._delete_post_by_id(post["id"])
195
196 def _accept_msg(self, msg: websockets.Data) -> None:
197 data: dict[str, Any] = cast(dict[str, Any], json.loads(msg))
198 event: str = cast(str, data["event"])
199 payload: str = cast(str, data["payload"])
200
201 if event == "update":
202 self._on_create_post(json.loads(payload))
203 elif event == "delete":
204 self._on_delete_post(payload)
205
206 @override
207 async def listen(self):
208 url = f"{self.streaming_url}/api/v1/streaming?stream=user"
209
210 async for ws in websockets.connect(
211 url, additional_headers={"Authorization": f"Bearer {self.options.token}"}
212 ):
213 try:
214 self.log.info("Listening to %s...", self.streaming_url)
215
216 async def listen_for_messages():
217 async for msg in ws:
218 self.submitter(lambda: self._accept_msg(msg))
219
220 listen = asyncio.create_task(listen_for_messages())
221
222 _ = await asyncio.gather(listen)
223 except websockets.ConnectionClosedError as e:
224 self.log.error(e, stack_info=True, exc_info=True)
225 self.log.info("Reconnecting to %s...", self.streaming_url)
226 continue