social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import sqlite3
2from abc import ABC, abstractmethod
3from typing import Callable, cast
4
5from cross.post import Post
6from database.connection import DatabasePool
7from util.util import LOGGER
8
9
10class Service:
11 def __init__(self, url: str, db: DatabasePool) -> None:
12 self.url: str = url
13 self.db: DatabasePool = db
14 # self._lock: threading.Lock = threading.Lock()
15
16 def get_post(self, url: str, user: str, identifier: str) -> sqlite3.Row | None:
17 cursor = self.db.get_conn().cursor()
18 _ = cursor.execute(
19 """
20 SELECT * FROM posts
21 WHERE service = ?
22 AND user_id = ?
23 AND identifier = ?
24 """,
25 (url, user, identifier),
26 )
27 return cast(sqlite3.Row, cursor.fetchone())
28
29 def get_post_by_id(self, id: int) -> sqlite3.Row | None:
30 cursor = self.db.get_conn().cursor()
31 _ = cursor.execute("SELECT * FROM posts WHERE id = ?", (id,))
32 return cast(sqlite3.Row, cursor.fetchone())
33
34
35class OutputService(Service):
36 def accept_post(self, post: Post):
37 LOGGER.warning("NOT IMPLEMENTED (%s), accept_post %s", self.url, post.id)
38
39 def delete_post(self, post_id: str):
40 LOGGER.warning("NOT IMPLEMENTED (%s), delete_post %s", self.url, post_id)
41
42 def accept_repost(self, repost_id: str, reposted_id: str):
43 LOGGER.warning(
44 "NOT IMPLEMENTED (%s), accept_repost %s of %s",
45 self.url,
46 repost_id,
47 reposted_id,
48 )
49
50 def delete_repost(self, repost_id: str):
51 LOGGER.warning("NOT IMPLEMENTED (%s), delete_repost %s", self.url, repost_id)
52
53
54class InputService(ABC, Service):
55 outputs: list[OutputService]
56 submitter: Callable[[Callable[[], None]], None]
57
58 @abstractmethod
59 async def listen(self):
60 pass