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