social media crossposting tool. 3rd time's the charm
mastodon misskey crossposting bluesky
4
fork

Configure Feed

Select the types of activity you want to include in your feed.

add settings

zenfyr.dev ed796641 bfd693d6

verified
+108 -12
+30 -12
main.py
··· 1 + import asyncio 2 + import json 1 3 import queue 2 4 import threading 3 5 from pathlib import Path 4 - from time import sleep 5 6 from typing import Callable 6 7 7 8 import env 8 9 from database.migrations import DatabaseMigrator 9 - from util.util import LOGGER 10 + from registry import create_input_service, create_output_service 11 + from registry_bootstrap import bootstrap 12 + from util.util import LOGGER, read_env 10 13 11 14 12 15 def main() -> None: ··· 15 18 if not data.exists(): 16 19 data.mkdir(parents=True) 17 20 18 - settings = data.joinpath("settings.json") 19 - database = data.joinpath("db.sqlite") 21 + settings_path = data.joinpath("settings.json") 22 + database_path = data.joinpath("db.sqlite") 20 23 21 - if not settings.exists(): 22 - LOGGER.info("First launch detected! Creating %s and exiting!", settings) 24 + if not settings_path.exists(): 25 + LOGGER.info("First launch detected! Creating %s and exiting!", settings_path) 23 26 return 24 27 25 - LOGGER.info("Loading settings...") 26 - # TODO 27 - 28 - migrator = DatabaseMigrator(database, Path(env.MIGRATIONS_DIR)) 28 + migrator = DatabaseMigrator(database_path, Path(env.MIGRATIONS_DIR)) 29 29 try: 30 30 migrator.migrate() 31 31 except Exception: ··· 34 34 finally: 35 35 migrator.close() 36 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 + 37 56 LOGGER.info("Starting task worker...") 38 57 39 58 def worker(task_queue: queue.Queue[Callable[[], None] | None]): ··· 55 74 56 75 LOGGER.info("Connecting to %s...", "TODO") # TODO 57 76 try: 58 - task_queue.put(lambda: print("hi")) 59 - sleep(10) # TODO 77 + asyncio.run(input.listen(outputs, lambda c: task_queue.put(c))) 60 78 except KeyboardInterrupt: 61 79 LOGGER.info("Stopping...") 62 80
+31
registry.py
··· 1 + from pathlib import Path 2 + from typing import Any, Callable 3 + 4 + from cross.service import InputService, OutputService 5 + 6 + input_factories: dict[str, Callable[[Path, dict[str, Any]], InputService]] = {} 7 + output_factories: dict[str, Callable[[Path, dict[str, Any]], OutputService]] = {} 8 + 9 + 10 + def create_input_service(db: Path, data: dict[str, Any]) -> InputService: 11 + if "type" not in data: 12 + raise ValueError("No `type` field in input data!") 13 + type: str = str(data["type"]) 14 + del data["type"] 15 + 16 + factory = input_factories.get(type) 17 + if not factory: 18 + raise KeyError(f"No such input service {type}!") 19 + return factory(db, data) 20 + 21 + 22 + def create_output_service(db: Path, data: dict[str, Any]) -> OutputService: 23 + if "type" not in data: 24 + raise ValueError("No `type` field in input data!") 25 + type: str = str(data["type"]) 26 + del data["type"] 27 + 28 + factory = output_factories.get(type) 29 + if not factory: 30 + raise KeyError(f"No such output service {type}!") 31 + return factory(db, data)
+28
registry_bootstrap.py
··· 1 + from pathlib import Path 2 + from typing import Any 3 + 4 + from registry import input_factories, output_factories 5 + 6 + 7 + class LazyFactory: 8 + def __init__(self, module_path: str, class_name: str, options_class_name: str): 9 + self.module_path: str = module_path 10 + self.class_name: str = class_name 11 + self.options_class_name: str = options_class_name 12 + 13 + def __call__(self, db: Path, d: dict[str, Any]): 14 + module = __import__( 15 + self.module_path, fromlist=[self.class_name, self.options_class_name] 16 + ) 17 + service_class = getattr(module, self.class_name) 18 + options_class = getattr(module, self.options_class_name) 19 + return service_class(db, options_class.from_dict(d)) 20 + 21 + 22 + def bootstrap(): 23 + input_factories["mastodon-wss"] = LazyFactory( 24 + "mastodon.input", "MastodonInputService", "MastodonInputOptions" 25 + ) 26 + input_factories["misskey-wss"] = LazyFactory( 27 + "misskey.input", "MisskeyInputService", "MisskeyInputOptions" 28 + )
+19
util/util.py
··· 1 1 import logging 2 2 import sys 3 + import os 4 + from typing import Any 3 5 4 6 logging.basicConfig(stream=sys.stderr, level=logging.INFO) 5 7 LOGGER = logging.getLogger("XPost") 8 + 9 + def read_env(data: dict[str, Any]) -> None: 10 + keys = list(data.keys()) 11 + for key in keys: 12 + val = data[key] 13 + match val: 14 + case str(): 15 + if val.startswith('env:'): 16 + envval = os.environ.get(val[4:]) 17 + if envval is None: 18 + del data[key] 19 + else: 20 + data[key] = envval 21 + case dict(): 22 + read_env(val) 23 + case _: 24 + pass