social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import queue
2import threading
3from pathlib import Path
4from time import sleep
5from typing import Callable
6
7import env
8from database.migrations import DatabaseMigrator
9from util.util import LOGGER
10
11
12def main() -> None:
13 data = Path(env.DATA_DIR)
14
15 if not data.exists():
16 data.mkdir(parents=True)
17
18 settings = data.joinpath("settings.json")
19 database = data.joinpath("db.sqlite")
20
21 if not settings.exists():
22 LOGGER.info("First launch detected! Creating %s and exiting!", settings)
23 return
24
25 LOGGER.info("Loading settings...")
26 # TODO
27
28 migrator = DatabaseMigrator(database, Path(env.MIGRATIONS_DIR))
29 try:
30 migrator.migrate()
31 except Exception:
32 LOGGER.exception("Failed to migrate database!")
33 return
34 finally:
35 migrator.close()
36
37 LOGGER.info("Starting task worker...")
38
39 def worker(task_queue: queue.Queue[Callable[[], None] | None]):
40 while True:
41 task = task_queue.get()
42 if task is None:
43 break
44
45 try:
46 task()
47 except Exception:
48 LOGGER.exception("Exception in worker thread!")
49 finally:
50 task_queue.task_done()
51
52 task_queue: queue.Queue[Callable[[], None] | None] = queue.Queue()
53 thread = threading.Thread(target=worker, args=(task_queue,), daemon=True)
54 thread.start()
55
56 LOGGER.info("Connecting to %s...", "TODO") # TODO
57 try:
58 task_queue.put(lambda: print("hi"))
59 sleep(10) # TODO
60 except KeyboardInterrupt:
61 LOGGER.info("Stopping...")
62
63 task_queue.join()
64 task_queue.put(None)
65 thread.join()
66
67
68if __name__ == "__main__":
69 main()