import json from dataclasses import dataclass from pathlib import Path from linkki_poster.models import Config from webapp.config import ConfigFileError @dataclass(frozen=True) class TelegramChatTarget: chat_id: str name: str channel_username: str discussion_group_id: str is_default: bool @dataclass(frozen=True) class TelegramBotTarget: bot_id: str name: str token: str chats: list[TelegramChatTarget] is_default: bool @dataclass(frozen=True) class TelegramTargetCatalog: bots: list[TelegramBotTarget] default_bot_id: str def load_telegram_target_catalog(config_dir: Path) -> TelegramTargetCatalog: settings_file = config_dir / "telegram_config.json" if not settings_file.exists(): return TelegramTargetCatalog(bots=[], default_bot_id="") try: payload = json.loads(settings_file.read_text(encoding="utf-8")) except json.JSONDecodeError as error: raise ConfigFileError(settings_file, f"Invalid JSON: {error.msg}") from error if not isinstance(payload, dict) or not isinstance(payload.get("bots"), list): raise ConfigFileError(settings_file, "Expected a top-level 'bots' list.") bots = _load_multi_bot_catalog(payload) default_bot_id = "" for bot in bots: if bot.is_default: default_bot_id = bot.bot_id break return TelegramTargetCatalog(bots=bots, default_bot_id=default_bot_id) def resolve_target_config(catalog: TelegramTargetCatalog, bot_id: str, chat_id: str) -> Config: bot = next((item for item in catalog.bots if item.bot_id == bot_id), None) if bot is None: raise ValueError("Choose a Telegram bot before posting or scheduling.") chat = next((item for item in bot.chats if item.chat_id == chat_id), None) if chat is None: raise ValueError("Choose a Telegram chat before posting or scheduling.") return Config( bot_token=bot.token, chat_id=chat.chat_id, channel_username=chat.channel_username, discussion_group_id=chat.discussion_group_id, ) def resolve_poster_config(config_dir: Path, bot_id: str, chat_id: str) -> Config: return resolve_target_config(load_telegram_target_catalog(config_dir), bot_id, chat_id) def default_target_selection(catalog: TelegramTargetCatalog) -> tuple[str, str]: if not catalog.default_bot_id: return "", "" bot = next((item for item in catalog.bots if item.bot_id == catalog.default_bot_id), None) if bot is None: return "", "" for chat in bot.chats: if chat.is_default: return bot.bot_id, chat.chat_id return bot.bot_id, "" def serialise_catalog_for_client(catalog: TelegramTargetCatalog) -> dict: return { "default_bot_id": catalog.default_bot_id, "bots": [ { "id": bot.bot_id, "name": bot.name, "default": bot.is_default, "default_chat_id": next((chat.chat_id for chat in bot.chats if chat.is_default), ""), "chats": [ { "chat_id": chat.chat_id, "name": chat.name, "has_discussion_group": bool(chat.discussion_group_id), "channel_username": chat.channel_username, "channel_internal_id": extract_internal_chat_id(chat.chat_id), "discussion_group_internal_id": extract_internal_chat_id(chat.discussion_group_id), "default": chat.is_default, } for chat in bot.chats ], } for bot in catalog.bots ], } def resolve_target_labels(catalog: TelegramTargetCatalog, bot_id: str, chat_id: str) -> tuple[str, str]: bot_label = bot_id chat_label = chat_id for bot in catalog.bots: if bot.bot_id != bot_id: continue bot_label = bot.name for chat in bot.chats: if chat.chat_id == chat_id: chat_label = chat.name break break return bot_label, chat_label def extract_internal_chat_id(chat_id: str) -> str: chat_id = str(chat_id).strip() if chat_id.startswith("-100") and len(chat_id) > 4: return chat_id[4:] return chat_id.lstrip("-") def _load_multi_bot_catalog(payload: dict) -> list[TelegramBotTarget]: bots: list[TelegramBotTarget] = [] for bot_payload in payload.get("bots", []): if not isinstance(bot_payload, dict): continue name = str(bot_payload.get("display_name", "")).strip() token = str(bot_payload.get("token", "")).strip() bot_id = str(bot_payload.get("id", "")).strip() or token.split(":", 1)[0].strip() if not name or not token or not bot_id: continue chats = _load_chats(bot_payload.get("channels", [])) if not chats: continue bots.append( TelegramBotTarget( bot_id=bot_id, name=name, token=token, chats=chats, is_default=bool(bot_payload.get("default", False)), ) ) return bots def _load_chats(raw_chats: object) -> list[TelegramChatTarget]: chats: list[TelegramChatTarget] = [] if not isinstance(raw_chats, list): return chats for chat_payload in raw_chats: if not isinstance(chat_payload, dict): continue name = str(chat_payload.get("display_name", "")).strip() chat_id = str(chat_payload.get("channel_id", "")).strip() if not name or not chat_id: continue chats.append( TelegramChatTarget( chat_id=chat_id, name=name, channel_username=str(chat_payload.get("channel_username", "")).strip(), discussion_group_id=str(chat_payload.get("discussion_group_id", "")).strip(), is_default=bool(chat_payload.get("default", False)), ) ) return chats