from __future__ import annotations import logging import os from pathlib import Path from yuxi.channel.extensions.viber.types import ViberAccountConfig, ViberTokenSource logger = logging.getLogger(__name__) ENV_VIBER_AUTH_TOKEN = "VIBER_AUTH_TOKEN" class ViberConfigAdapter: def __init__(self): self._raw_config: dict = {} def list_account_ids(self, config: dict) -> list[str]: self._raw_config = config viber_cfg = self._get_viber_config() accounts = viber_cfg.get("accounts", {}) if isinstance(viber_cfg, dict) else {} if accounts: return list(accounts.keys()) return ["default"] async def resolve_account(self, account_id: str) -> dict: raw = self._load_raw_config(account_id) viber_account = self._build_account(account_id, raw) return self._to_dict(viber_account) def _get_viber_config(self) -> dict: channels = self._raw_config.get("channels", {}) return channels.get("viber", {}) if channels else self._raw_config def is_configured(self, account: dict) -> bool: return bool(account.get("auth_token")) def is_enabled(self, account: dict) -> bool: return account.get("enabled", True) def disabled_reason(self, account: dict) -> str: if not self.is_configured(account): return "Viber Auth Token is required" return "" async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None: account = await self.resolve_account(account_id) return account.get("allow_from", []) def describe_account(self, account: dict) -> dict: return { "account_id": account.get("account_id", ""), "name": account.get("name", ""), "configured": self.is_configured(account), "token_source": account.get("token_source", "none"), } def default_account_id(self, config: dict) -> str: channels = config.get("channels", {}) viber_cfg = channels.get("viber", {}) if channels else config if isinstance(viber_cfg, dict): return viber_cfg.get("default_account", "default") return "default" def config_schema(self) -> dict: return { "type": "object", "properties": { "enabled": {"type": "boolean", "default": True}, "auth_token": {"type": "string", "title": "Viber Auth Token"}, "token_file": {"type": "string", "title": "Token File Path"}, "sender_name": {"type": "string", "title": "Bot Display Name", "default": "ForcePilot Bot"}, "sender_avatar": {"type": "string", "title": "Bot Avatar URL", "default": ""}, "webhook_url": {"type": "string", "title": "Webhook Callback URL"}, "dm_policy": { "type": "string", "enum": ["pairing", "allowlist", "open", "disabled"], "default": "pairing", }, "allow_from": {"type": "array", "items": {"type": "string"}}, "streaming_enabled": {"type": "boolean", "default": True}, "text_chunk_limit": {"type": "integer", "default": 7000}, "media_max_mb": {"type": "integer", "default": 50}, "accounts": { "type": "object", "additionalProperties": { "type": "object", "properties": { "name": {"type": "string"}, "auth_token": {"type": "string"}, "token_file": {"type": "string"}, "sender_name": {"type": "string"}, "sender_avatar": {"type": "string"}, "webhook_url": {"type": "string"}, "dm_policy": { "type": "string", "enum": ["pairing", "allowlist", "open", "disabled"], }, "allow_from": {"type": "array", "items": {"type": "string"}}, "streaming_enabled": {"type": "boolean"}, "text_chunk_limit": {"type": "integer"}, "media_max_mb": {"type": "integer"}, "min_api_version": {"type": "integer", "default": 7}, "welcome_message": {"type": "string", "default": ""}, }, }, }, "default_account": {"type": "string", "default": "default"}, "min_api_version": {"type": "integer", "default": 7}, "welcome_message": {"type": "string", "default": ""}, }, } def _load_raw_config(self, account_id: str) -> dict: viber_cfg = self._get_viber_config() if account_id == "__base__": return viber_cfg if isinstance(viber_cfg, dict) else {} accounts = viber_cfg.get("accounts", {}) if isinstance(viber_cfg, dict) else {} return accounts.get(account_id, {}) if isinstance(accounts, dict) else {} def _build_account(self, account_id: str, raw: dict) -> ViberAccountConfig: base_raw = self._load_raw_config("__base__") token, token_source = self._resolve_token(account_id, raw, base_raw) return ViberAccountConfig( account_id=account_id, auth_token=token, token_source=token_source, token_file=raw.get("token_file", base_raw.get("token_file", "")), sender_name=raw.get("sender_name", base_raw.get("sender_name", "ForcePilot Bot")), sender_avatar=raw.get("sender_avatar", base_raw.get("sender_avatar", "")), webhook_url=raw.get("webhook_url", base_raw.get("webhook_url", "")), name=raw.get("name", account_id), dm_policy=raw.get("dm_policy", base_raw.get("dm_policy", "pairing")), allow_from=raw.get("allow_from", base_raw.get("allow_from", [])), streaming_enabled=raw.get("streaming_enabled", base_raw.get("streaming_enabled", True)), text_chunk_limit=raw.get("text_chunk_limit", base_raw.get("text_chunk_limit", 7000)), media_max_mb=raw.get("media_max_mb", base_raw.get("media_max_mb", 50)), min_api_version=raw.get("min_api_version", base_raw.get("min_api_version", 7)), welcome_message=raw.get("welcome_message", base_raw.get("welcome_message", "")), ) @staticmethod def _resolve_token(account_id: str, raw: dict, base_raw: dict) -> tuple[str, ViberTokenSource]: if raw.get("auth_token"): return raw["auth_token"], ViberTokenSource.ACCOUNT if raw.get("token_file"): token = ViberConfigAdapter._read_file(raw["token_file"]) if token: return token, ViberTokenSource.TOKEN_FILE if account_id == "default": if base_raw.get("auth_token"): return base_raw["auth_token"], ViberTokenSource.ACCOUNT if base_raw.get("token_file"): token = ViberConfigAdapter._read_file(base_raw["token_file"]) if token: return token, ViberTokenSource.TOKEN_FILE env_token = os.environ.get(ENV_VIBER_AUTH_TOKEN, "") if env_token: return env_token, ViberTokenSource.ENV return "", ViberTokenSource.NONE @staticmethod def _read_file(filepath: str) -> str | None: try: path = Path(filepath) if path.exists(): return path.read_text().strip() except OSError: pass return None @staticmethod def _to_dict(account: ViberAccountConfig) -> dict: return { "account_id": account.account_id, "auth_token": account.auth_token, "token_source": account.token_source.value, "token_file": account.token_file, "sender_name": account.sender_name, "sender_avatar": account.sender_avatar, "webhook_url": account.webhook_url, "name": account.name, "dm_policy": account.dm_policy, "allow_from": account.allow_from, "streaming_enabled": account.streaming_enabled, "text_chunk_limit": account.text_chunk_limit, "media_max_mb": account.media_max_mb, "min_api_version": account.min_api_version, "welcome_message": account.welcome_message, "enabled": True, }