social media crossposting tool. 3rd time's the charm
zenfyr.dev
mastodon
misskey
crossposting
bluesky
1import asyncio
2import json
3import queue
4import threading
5from pathlib import Path
6from typing import Callable
7
8import env
9from database.migrations import DatabaseMigrator
10from registry import create_input_service, create_output_service
11from registry_bootstrap import bootstrap
12from util.util import LOGGER, read_env
13
14
15def main() -> None:
16 data = Path(env.DATA_DIR)
17
18 if not data.exists():
19 data.mkdir(parents=True)
20
21 settings_path = data.joinpath("settings.json")
22 database_path = data.joinpath("db.sqlite")
23
24 if not settings_path.exists():
25 LOGGER.info("First launch detected! Creating %s and exiting!", settings_path)
26 return
27
28 migrator = DatabaseMigrator(database_path, 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("Bootstrapping registries...")
38 bootstrap()
39
40 LOGGER.info("Loading settings...")
41
42 with open(settings_path) as f:
43 settings = json.load(f)
44 read_env(settings)
45
46 if "input" not in settings:
47 raise KeyError("No `input` sepcified in settings!")
48 if "outputs" not in settings:
49 raise KeyError("No `outputs` spicified in settings!")
50
51 input = create_input_service(database_path, settings["input"])
52 outputs = [
53 create_output_service(database_path, data) for data in settings["outputs"]
54 ]
55
56 LOGGER.info("Starting task worker...")
57
58 def worker(task_queue: queue.Queue[Callable[[], None] | None]):
59 while True:
60 task = task_queue.get()
61 if task is None:
62 break
63
64 try:
65 task()
66 except Exception:
67 LOGGER.exception("Exception in worker thread!")
68 finally:
69 task_queue.task_done()
70
71 task_queue: queue.Queue[Callable[[], None] | None] = queue.Queue()
72 thread = threading.Thread(target=worker, args=(task_queue,), daemon=True)
73 thread.start()
74
75 LOGGER.info("Connecting to %s...", "TODO") # TODO
76 try:
77 asyncio.run(input.listen(outputs, lambda c: task_queue.put(c)))
78 except KeyboardInterrupt:
79 LOGGER.info("Stopping...")
80
81 task_queue.join()
82 task_queue.put(None)
83 thread.join()
84
85
86if __name__ == "__main__":
87 main()