| 1 | """Auto-discovery for built-in channel modules and external plugins.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import importlib |
| 6 | import pkgutil |
| 7 | from typing import TYPE_CHECKING |
| 8 | |
| 9 | from loguru import logger |
| 10 | |
| 11 | if TYPE_CHECKING: |
| 12 | from nanobot.channels.base import BaseChannel |
| 13 | |
| 14 | _INTERNAL = frozenset({"base", "manager", "registry"}) |
| 15 | |
| 16 | |
| 17 | def discover_channel_names() -> list[str]: |
| 18 | """Return all built-in channel module names by scanning the package (zero imports).""" |
| 19 | import nanobot.channels as pkg |
| 20 | |
| 21 | return [ |
| 22 | name |
| 23 | for _, name, ispkg in pkgutil.iter_modules(pkg.__path__) |
| 24 | if name not in _INTERNAL and not ispkg |
| 25 | ] |
| 26 | |
| 27 | |
| 28 | def load_channel_class(module_name: str) -> type[BaseChannel]: |
| 29 | """Import *module_name* and return the first BaseChannel subclass found.""" |
| 30 | from nanobot.channels.base import BaseChannel as _Base |
| 31 | |
| 32 | mod = importlib.import_module(f"nanobot.channels.{module_name}") |
| 33 | for attr in dir(mod): |
| 34 | obj = getattr(mod, attr) |
| 35 | if isinstance(obj, type) and issubclass(obj, _Base) and obj is not _Base: |
| 36 | return obj |
| 37 | raise ImportError(f"No BaseChannel subclass in nanobot.channels.{module_name}") |
| 38 | |
| 39 | |
| 40 | def discover_plugins() -> dict[str, type[BaseChannel]]: |
| 41 | """Discover external channel plugins registered via entry_points.""" |
| 42 | from importlib.metadata import entry_points |
| 43 | |
| 44 | plugins: dict[str, type[BaseChannel]] = {} |
| 45 | for ep in entry_points(group="nanobot.channels"): |
| 46 | try: |
| 47 | cls = ep.load() |
| 48 | plugins[ep.name] = cls |
| 49 | except Exception as e: |
| 50 | logger.warning("Failed to load channel plugin '{}': {}", ep.name, e) |
| 51 | return plugins |
| 52 | |
| 53 | |
| 54 | def discover_all() -> dict[str, type[BaseChannel]]: |
| 55 | """Return all channels: built-in (pkgutil) merged with external (entry_points). |
| 56 | |
| 57 | Built-in channels take priority — an external plugin cannot shadow a built-in name. |
| 58 | """ |
| 59 | builtin: dict[str, type[BaseChannel]] = {} |
| 60 | for modname in discover_channel_names(): |
| 61 | try: |
| 62 | builtin[modname] = load_channel_class(modname) |
| 63 | except ImportError as e: |
| 64 | logger.debug("Skipping built-in channel '{}': {}", modname, e) |
| 65 | |
| 66 | external = discover_plugins() |
| 67 | shadowed = set(external) & set(builtin) |
| 68 | if shadowed: |
| 69 | logger.warning("Plugin(s) shadowed by built-in channels (ignored): {}", shadowed) |
| 70 | |
| 71 | return {**external, **builtin} |
| 72 |