chongming-plugins/aibox_sophon_adapter/mqtt_clients.py

"""
AI Box Sophon Adapter — MQTT 客户端能力组件

从插件 __init__.py 抽取的三个外部 MQTT 客户端启动块，统一模式:
  - enabled=False → 仅警告"未启用"并返回 None
  - enabled=True  → 懒加载客户端类，启动失败降级为 None（仅警告，不阻断插件）

内部 MQTT 主题约定（勿改）:
  与盒子侧约定一致（盒子只设根主题 box，子路径固定 aibox_sophon/...）；
  勿用 chongming/{插件}/{操作} 保留命名空间——网关 MQTT Bridge 订阅了
  chongming/+/+，会拦截 3 级主题并按 Worker RPC 路由，而 edge_app_controller
  不是 @rpc_method（Worker 无对应 Queryable），导致 worker_manager 报
  "Worker ... 无回复，尝试重试"。
"""
import logging
import os
from typing import Any

logger = logging.getLogger(__name__)

# 插件包名（作为 importlib 探测的基准，兼容 plugins.xxx 与 xxx 两种导入路径）
_PKG = __package__ or "plugins.aibox_sophon_adapter"


def _import_first(candidates: list) -> Any:
    """按序尝试导入模块，全部失败返回 None。"""
    import importlib

    for name in candidates:
        try:
            return importlib.import_module(name)
        except ImportError:
            continue
    return None


def build_subscribe_routes(topics_cfg: list) -> list:
    """
    根据配置的订阅主题列表构建 MQTTSubscribeClient 路由。

    topics_cfg: [{"topic": "...", "handler": "board_ping", "reply_topic": "..."}, ...]
      - topic:       订阅主题（支持 + / # 通配符）
      - handler:     处理函数名，按序从插件 router 模块、插件包自身查找
      - reply_topic: 可选，回复主题；处理后返回值将发布到该主题（支持 {1}{2} 占位符）

    适配约定（按首个位置参数类型）:
      - Pydantic 模型: 用 payload 构造模型实例后调用（如 board_ping -> BoardPingRequest(**payload)）
      - dict: 直接传整个 payload 和 topic（如 edge_app_controller(payload, topic)）
      - 其他: 将 payload 作为关键字参数展开调用

    返回: [{"topic": ..., "handler": async callable(payload, topic), "reply_topic": ...}]
    """
    from typing import get_origin

    from pydantic import BaseModel

    # 注意: 不能用 "from .. import router" 取子模块，因为插件包属性 router
    # 已被 routes.router（FastAPI APIRouter）遮蔽，故用 importlib 按名导入。
    pkg_base = _PKG
    pkg_leaf = _PKG.rsplit(".", 1)[-1]
    router_mod = _import_first([f"{pkg_base}.router", f"{pkg_leaf}.router"])
    pkg_mod = _import_first([pkg_base, pkg_leaf])

    routes = []
    for t in topics_cfg or []:
        topic = t.get("topic", "")
        handler_name = t.get("handler", "")
        reply_topic = t.get("reply_topic")
        if not topic or not handler_name:
            logger.warning("[MQTTSubscribe] 跳过无效订阅项: %s", t)
            continue

        fn = None
        if router_mod is not None:
            fn = getattr(router_mod, handler_name, None)
        if fn is None and pkg_mod is not None:
            fn = getattr(pkg_mod, handler_name, None)
        if fn is None:
            logger.warning(
                "[MQTTSubscribe] 未知 handler '%s'，跳过 %s", handler_name, topic
            )
            continue

        # 探测首个位置参数的类型注解
        hints = getattr(fn, "__annotations__", {})
        first_type: Any = None
        for pname, ptype in hints.items():
            if pname in ("worker_id", "return"):
                continue
            first_type = ptype
            break

        if (
            isinstance(first_type, type)
            and issubclass(first_type, BaseModel)
        ):
            model_cls = first_type

            async def _cb_model(data, topic, _fn=fn, _model=model_cls):
                return await _fn(_model(**data))
            routes.append({
                "topic": topic,
                "handler": _cb_model,
                "reply_topic": reply_topic,
            })
        elif first_type is dict or get_origin(first_type) is dict:
            # 处理函数期望整个 dict 作为第一个参数（如 edge_app_controller(payload, topic)）
            async def _cb_payload(data, topic, _fn=fn):
                return await _fn(data, topic)
            routes.append({
                "topic": topic,
                "handler": _cb_payload,
                "reply_topic": reply_topic,
            })
        else:
            # 其他签名：将 payload 按关键字参数展开调用
            async def _cb_kwargs(data, topic, _fn=fn):
                if isinstance(data, dict):
                    return await _fn(**data)
                return await _fn()
            routes.append({
                "topic": topic,
                "handler": _cb_kwargs,
                "reply_topic": reply_topic,
            })

    return routes


async def start_mqtt_forward(cfg: dict) -> Any:
    """启动外部 MQTT 转发客户端，返回客户端或 None。"""
    enabled = cfg.get("enabled", False)
    if not enabled:
        logger.warning("[MQTTForward] mqtt_forward 未启用，配置内容: %s", cfg)
        return None

    from chongming_mqtt.mqtt_forward import MQTTForwardClient

    # uri 含凭据，优先配置；留空时从环境变量 AIBOX_MQTT_URI 兜底（避免凭据入库）
    uri = cfg.get("uri") or os.environ.get("AIBOX_MQTT_URI", "")
    topic = cfg.get("topic", "/board_ping")
    logger.warning(
        "[MQTTForward] 配置已加载: enabled=%s, uri=%s, topic=%s", enabled, uri, topic
    )
    if not uri:
        return None

    client = MQTTForwardClient(uri=uri, topic=topic)
    try:
        await client.start()
        logger.warning("[MQTTForward] 外部转发已启用: %s → %s", uri, topic)
        return client
    except Exception as e:
        logger.warning("[MQTTForward] 启动失败，将跳过转发: %s", e)
        return None


async def start_mqtt_subscribe(cfg: dict) -> Any:
    """启动外部 MQTT 订阅客户端（本插件直接订阅外部 Broker，不走网关 Bridge）。"""
    enabled = cfg.get("enabled", False)
    if not enabled:
        logger.warning("[MQTTSubscribe] mqtt_subscribe 未启用，配置内容: %s", cfg)
        return None

    from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient

    # uri 含凭据，优先配置；留空时从环境变量 AIBOX_MQTT_URI 兜底（避免凭据入库）
    uri = cfg.get("uri") or os.environ.get("AIBOX_MQTT_URI", "")
    topics = cfg.get("topics", []) or []
    logger.warning(
        "[MQTTSubscribe] 配置已加载: enabled=%s, uri=%s, topics=%s",
        enabled, uri, topics,
    )
    if not (uri and topics):
        logger.warning("[MQTTSubscribe] 未配置 uri 或 topics，跳过外部订阅")
        return None

    routes = build_subscribe_routes(topics)
    client = MQTTSubscribeClient(uri=uri, routes=routes)
    try:
        await client.start()
        logger.warning("[MQTTSubscribe] 外部订阅已启用: %s", uri)
        return client
    except Exception as e:
        logger.warning("[MQTTSubscribe] 启动失败，将跳过订阅: %s", e)
        return None


async def start_internal_mqtt(cfg: dict) -> tuple[Any, str, str]:
    """
    启动内部 MQTT 请求-响应客户端
    （handler 内对任意应用进程发起请求-响应，主题显式传入，默认连内部 127.0.0.1:1883）。

    返回 (client | None, request_topic, response_topic)：
    主题无论是否启用都按配置解析并返回（由调用方写入 global_var）。
    """
    req_topic = cfg.get(
        "request_topic", "box/aibox_sophon/edge_app_controller"
    )
    resp_topic = cfg.get(
        "response_topic", "box/aibox_sophon/edge_app_controller_reply"
    )
    enabled = cfg.get("enabled", False)
    if not enabled:
        logger.warning("[InternalMQTT] internal_mqtt 未启用，配置内容: %s", cfg)
        return None, req_topic, resp_topic

    from chongming_mqtt.internal_mqtt import InternalMQTTClient

    uri = cfg.get("uri", "mqtt://127.0.0.1:1883")
    logger.warning(
        "[InternalMQTT] 配置已加载: enabled=%s, uri=%s, request_topic=%s, response_topic=%s",
        enabled, uri, req_topic, resp_topic,
    )
    client = InternalMQTTClient(uri=uri)
    try:
        await client.start()
        logger.warning("[InternalMQTT] 内部请求-响应客户端已启用: %s", uri)
        return client, req_topic, resp_topic
    except Exception as e:
        logger.warning("[InternalMQTT] 启动失败，将跳过: %s", e)
        return None, req_topic, resp_topic