69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
from typing import Callable
|
|
|
|
from ..constants import (ADAPTER_SETTING_KEYS, DEFAULT_TELEGRAM_API_BASE, FLAG_ON, MATRIX_ENABLED,
|
|
MATRIX_HOMESERVER, MATRIX_ROOM_ID, MATRIX_TOKEN, MATRIX_USER_ID, TELEGRAM_API_BASE,
|
|
TELEGRAM_CHAT_ID, TELEGRAM_ENABLED, TELEGRAM_TOKEN)
|
|
from .base import ChatAdapter, Target
|
|
from .matrix import BACKEND_MATRIX, MatrixAdapter
|
|
from .telegram import BACKEND_TELEGRAM, TelegramAdapter
|
|
|
|
|
|
def build_telegram(store, timeout: float | None = None) -> TelegramAdapter | None:
|
|
if store.get(TELEGRAM_ENABLED) != FLAG_ON:
|
|
return None
|
|
token, chat_id = store.get(TELEGRAM_TOKEN), store.get(TELEGRAM_CHAT_ID)
|
|
if not (token and chat_id):
|
|
return None
|
|
options = {} if timeout is None else {"timeout": timeout}
|
|
return TelegramAdapter(token, chat_id, store.get(TELEGRAM_API_BASE) or DEFAULT_TELEGRAM_API_BASE, **options)
|
|
|
|
|
|
def build_matrix(store, timeout: float | None = None) -> MatrixAdapter | None:
|
|
if store.get(MATRIX_ENABLED) != FLAG_ON:
|
|
return None
|
|
homeserver, user_id = store.get(MATRIX_HOMESERVER), store.get(MATRIX_USER_ID)
|
|
token, room_id = store.get(MATRIX_TOKEN), store.get(MATRIX_ROOM_ID)
|
|
if not (homeserver and user_id and token and room_id):
|
|
return None
|
|
options = {} if timeout is None else {"timeout": timeout}
|
|
return MatrixAdapter(homeserver, user_id, token, room_id, **options)
|
|
|
|
|
|
FACTORIES: tuple[Callable[[object], ChatAdapter | None], ...] = (build_telegram, build_matrix)
|
|
|
|
|
|
def build_adapters(store) -> list[ChatAdapter]:
|
|
return [adapter for adapter in (factory(store) for factory in FACTORIES) if adapter is not None]
|
|
|
|
|
|
def default_targets(store) -> list[Target]:
|
|
targets = []
|
|
if store.get(TELEGRAM_ENABLED) == FLAG_ON and store.get(TELEGRAM_CHAT_ID):
|
|
targets.append(Target(BACKEND_TELEGRAM, store.get(TELEGRAM_CHAT_ID)))
|
|
if store.get(MATRIX_ENABLED) == FLAG_ON and store.get(MATRIX_ROOM_ID):
|
|
targets.append(Target(BACKEND_MATRIX, store.get(MATRIX_ROOM_ID)))
|
|
return targets
|
|
|
|
|
|
class AdapterRegistry:
|
|
def __init__(self, store, factory: Callable[[object], list[ChatAdapter]] = build_adapters):
|
|
self._store = store
|
|
self._factory = factory
|
|
self._signature: tuple | None = None
|
|
self._adapters: list[ChatAdapter] = []
|
|
|
|
async def current(self) -> list[ChatAdapter]:
|
|
signature = tuple(self._store.get(key) for key in ADAPTER_SETTING_KEYS)
|
|
if signature != self._signature:
|
|
old = self._adapters
|
|
self._signature = signature
|
|
self._adapters = self._factory(self._store)
|
|
for adapter in old:
|
|
await adapter.aclose()
|
|
return self._adapters
|
|
|
|
async def aclose(self) -> None:
|
|
adapters, self._adapters, self._signature = self._adapters, [], None
|
|
for adapter in adapters:
|
|
await adapter.aclose()
|