long_articles_agentic_supply/supply/handlers/_discovery.py

"""XXL-JOB handler 自动发现 —— pkgutil + importlib 扫描导入

扫描 supply.handlers 包下的所有非私有模块，逐一 import，
触发 @xxl_job 装饰器将 handler 注册到 _XXL_REGISTRY。
"""

import importlib
import logging
import pkgutil

logger = logging.getLogger(__name__)


def discover_and_register(package: str = "supply.handlers") -> None:
    """自动发现并导入 package 下的所有 handler 模块

    跳过 _ 前缀的私有模块和子包。每个模块被 import 时，
    @xxl_job 装饰器自动执行，将 handler 写入共享的 _XXL_REGISTRY。

    幂等调用：importlib 使用 sys.modules 缓存，重复调用是 no-op。

    Args:
        package: 点分隔的 Python 包路径，默认 "supply.handlers"
    """
    try:
        pkg = importlib.import_module(package)
    except ImportError:
        logger.warning(
            "XXL-JOB handler 包 %r 不存在，无 handler 注册", package
        )
        return

    registered = 0
    for finder, name, is_pkg in pkgutil.iter_modules(
        pkg.__path__, pkg.__name__ + "."
    ):
        if name.split(".")[-1].startswith("_"):
            continue

        try:
            importlib.import_module(name)
            registered += 1
            logger.debug("XXL-JOB 发现 handler 模块: %s", name)
        except Exception:
            logger.exception("XXL-JOB handler 模块 %r 导入失败，跳过", name)

    if registered:
        logger.info(
            "XXL-JOB 自动发现完成: 已导入 %d 个 handler 模块", registered
        )
    else:
        logger.info("XXL-JOB 自动发现: 在 %r 中未找到 handler 模块", package)