Agent/gateway/core/channels/loader.py

"""
渠道自动加载器。

``_CHANNEL_REGISTRY`` 登记所有可用渠道（channel_id → 插件类路径）；
``CHANNELS_ENABLED`` 环境变量（逗号分隔）控制运行时实际启动哪些渠道。

新增渠道只需两步：
  1. 在对应渠道模块的 Api 类上实现 ``ChannelPlugin`` Protocol
     （``from_env()`` classmethod + ``build_router()`` 方法）
  2. 在 ``_CHANNEL_REGISTRY`` 中追加一行
"""

from __future__ import annotations

import importlib
import logging
import os

from fastapi import APIRouter

logger = logging.getLogger(__name__)

# channel_id → 插件类的完整模块路径
_CHANNEL_REGISTRY: dict[str, str] = {
    "feishu": "gateway.core.channels.feishu.api.FeishuChannelApi",
    # "wechat": "gateway.core.channels.wechat.api.WeChatChannelApi",
}


def _import_plugin(dotted_path: str) -> type:
    module_path, cls_name = dotted_path.rsplit(".", 1)
    module = importlib.import_module(module_path)
    return getattr(module, cls_name)


def load_enabled_channels() -> list[APIRouter]:
    """读取 ``CHANNELS_ENABLED`` 并返回已启用渠道的路由列表。

    未配置时默认启用 ``feishu``。未知渠道 ID 记录警告后跳过，不中断启动。
    """
    enabled_raw = os.getenv("CHANNELS_ENABLED", "feishu")
    enabled = [c.strip() for c in enabled_raw.split(",") if c.strip()]

    routers: list[APIRouter] = []
    for channel_id in enabled:
        if channel_id not in _CHANNEL_REGISTRY:
            logger.warning("Channel '%s' is not in the registry, skipping", channel_id)
            continue
        plugin_cls = _import_plugin(_CHANNEL_REGISTRY[channel_id])
        channel = plugin_cls.from_env()
        routers.append(channel.build_router())
        logger.info("Channel '%s' loaded", channel_id)

    return routers