social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import sqlite3
2from pathlib import Path
3from typing import Callable, cast
4
5from cross.post import Post
6from database.connection import get_conn
7from util.util import LOGGER
8
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
36
37class OutputService(Service):
38 def accept_post(self, post: Post):
39 LOGGER.warning("NOT IMPLEMENTED (%s), accept_post %s", self.url, post.id)
40
41 def delete_post(self, post_id: str):
42 LOGGER.warning("NOT IMPLEMENTED (%s), delete_post %s", self.url, post_id)
43
44 def accept_repost(self, repost_id: str, reposted_id: str):
45 LOGGER.warning(
46 "NOT IMPLEMENTED (%s), accept_repost %s of %s",
47 self.url,
48 repost_id,
49 reposted_id,
50 )
51
52 def delete_repost(self, repost_id: str):
53 LOGGER.warning("NOT IMPLEMENTED (%s), delete_repost %s", self.url, repost_id)
54
55
56class InputService(Service):
57 async def listen(
58 self,
59 outputs: list[OutputService],
60 submitter: Callable[[Callable[[], None]], None],
61 ): # pyright: ignore[reportUnusedParameter]
62 pass