social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import asyncio
2import json
3import re
4import uuid
5from dataclasses import dataclass, field
6from typing import Any, Callable, override
7
8import websockets
9
10from cross.service import InputService, OutputService
11from database.connection import DatabasePool
12from misskey.info import MisskeyService
13from util.util import LOGGER, normalize_service_url
14
15ALLOWED_VISIBILITY = ["public", "home"]
16
17
18@dataclass
19class MisskeyInputOptions:
20 token: str
21 instance: str
22 allowed_visibility: list[str] = field(
23 default_factory=lambda: ALLOWED_VISIBILITY.copy()
24 )
25 filters: list[re.Pattern[str]] = field(default_factory=lambda: [])
26
27 @classmethod
28 def from_dict(cls, data: dict[str, Any]) -> "MisskeyInputOptions":
29 data["instance"] = normalize_service_url(data["instance"])
30
31 if "allowed_visibility" in data:
32 for vis in data.get("allowed_visibility", []):
33 if vis not in ALLOWED_VISIBILITY:
34 raise ValueError(f"Invalid visibility option {vis}!")
35
36 if "filters" in data:
37 data["filters"] = [re.compile(r) for r in data["filters"]]
38
39 return MisskeyInputOptions(**data)
40
41
42class MisskeyInputService(MisskeyService, InputService):
43 def __init__(self, db: DatabasePool, options: MisskeyInputOptions) -> None:
44 super().__init__(options.instance, db)
45 self.options: MisskeyInputOptions = options
46
47 LOGGER.info("Verifying %s credentails...", self.url)
48 responce = self.verify_credentials()
49 self.user_id: str = responce["id"]
50
51 @override
52 def _get_token(self) -> str:
53 return self.options.token
54
55 async def _subscribe_to_home(self, ws: websockets.ClientConnection) -> None:
56 await ws.send(
57 json.dumps(
58 {
59 "type": "connect",
60 "body": {"channel": "homeTimeline", "id": str(uuid.uuid4())},
61 }
62 )
63 )
64 LOGGER.info("Subscribed to 'homeTimeline' channel...")
65
66 @override
67 async def listen(
68 self,
69 outputs: list[OutputService],
70 submitter: Callable[[Callable[[], None]], None],
71 ):
72 streaming: str = f"{'wss' if self.url.startswith('https') else 'ws'}://{self.url.split('://', 1)[1]}"
73 url: str = f"{streaming}/streaming?i={self.options.token}"
74
75 async for ws in websockets.connect(url):
76 try:
77 LOGGER.info("Listening to %s...", streaming)
78 await self._subscribe_to_home(ws)
79
80 async def listen_for_messages():
81 async for msg in ws:
82 LOGGER.info(msg) # TODO
83
84 listen = asyncio.create_task(listen_for_messages())
85
86 _ = await asyncio.gather(listen)
87 except websockets.ConnectionClosedError as e:
88 LOGGER.error(e, stack_info=True, exc_info=True)
89 LOGGER.info("Reconnecting to %s...", streaming)
90 continue