pkgutil.iter_modules(path=None, prefix='')

# jobs/base.py
from typing import Protocol, runtime_checkable

@runtime_checkable
class JobProtocol(Protocol):
    """插件協(xié)議定義"""
    
    def enabled(self) -> bool:
        """判斷插件是否啟用"""
        ...
    
    def run(self) -> bool:
        """執(zhí)行插件任務(wù)"""
        ...

# main.py
import logging
import pkgutil
import importlib
from types import ModuleType
from jobs.base import JobProtocol

logger = logging.getLogger(__name__)

def load_jobs(package: str = "jobs") -> list[JobProtocol]:
    """動(dòng)態(tài)加載指定包下的所有任務(wù)插件"""
    loaded_jobs: list[JobProtocol] = []

    # 導(dǎo)入目標(biāo)包
    pkg = importlib.import_module(package)
    
    # 遍歷子包
    for finder, name, ispkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
        # 只處理子包
        if not ispkg:
            continue
        
        # 動(dòng)態(tài)導(dǎo)入模塊
        module = importlib.import_module(name)
        
        # 獲取工廠函數(shù)并創(chuàng)建實(shí)例
        if hasattr(module, "job_factory"):
            job = module.job_factory()
            
            # 協(xié)議驗(yàn)證
            if isinstance(job, JobProtocol) and job.enabled():
                loaded_jobs.append(job)
    
    return loaded_jobs

# jobs/jobs1/__init__.py
from jobs.base import JobProtocol
from .job import MyJob1

def job_factory() -> JobProtocol:
    return MyJob1()

# jobs/jobs1/job.py
class MyJob1:
    def enabled(self) -> bool:
        return True

    def run(self) -> bool:
        print(f"{self.__class__.__name__} is running")
        return True

try:
    pkg = importlib.import_module(package)
except ImportError as e:
    logger.error(f"導(dǎo)入包失敗: {e}")
    return []

for finder, name, ispkg in pkgutil.iter_modules(pkg.__path__, prefix):
    try:
        module = importlib.import_module(name)
    except Exception as e:
        logger.error(f"加載模塊 {name} 失敗: {e}")
        continue
    
    if not hasattr(module, "job_factory"):
        continue
    
    try:
        job = module.job_factory()
    except Exception as e:
        logger.error(f"實(shí)例化插件 {name} 失敗: {e}")
        continue

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

from typing import Protocol, runtime_checkable

@runtime_checkable
class JobProtocol(Protocol):
    def enabled(self) -> bool: ...
    def run(self) -> bool: ...

# 檢查實(shí)例是否滿足協(xié)議
if isinstance(job, JobProtocol):
    job.run()

def run_jobs(jobs: list[JobProtocol]) -> None:
    for job in jobs:
        try:
            job.run()
        except Exception as e:
            logger.error(f"任務(wù)執(zhí)行失敗: {e}")
            # 繼續(xù)執(zhí)行其他任務(wù)

"""動(dòng)態(tài)任務(wù)加載器

使用 pkgutil 模塊動(dòng)態(tài)發(fā)現(xiàn)和加載 jobs 包下的所有任務(wù)插件。
"""

import logging
import pkgutil
import importlib
from types import ModuleType
from jobs.base import JobProtocol

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)


def load_jobs(package: str = "jobs") -> list[JobProtocol]:
    """動(dòng)態(tài)加載指定包下的所有任務(wù)插件。

    遍歷 package 下的所有子包，嘗試導(dǎo)入每個(gè)子包并調(diào)用其 job_factory 函數(shù)
    創(chuàng)建任務(wù)實(shí)例。只有實(shí)現(xiàn)了 JobProtocol 協(xié)議且 enabled() 返回 True 的
    任務(wù)才會(huì)被執(zhí)行。

    Args:
        package: 要掃描的包名，默認(rèn)為 "jobs"。

    Returns:
        成功加載的任務(wù)實(shí)例列表。
    """
    loaded_jobs: list[JobProtocol] = []

    try:
        pkg: ModuleType = importlib.import_module(package)
    except ImportError as e:
        logger.error(f"導(dǎo)入包 {package} 失敗: {e}")
        return loaded_jobs

    # pkg.__path__ 可能是 None（當(dāng) package 是命名空間包但沒有子包時(shí)）
    if not hasattr(pkg, "__path__") or pkg.__path__ is None:
        logger.warning(f"包 {package} 沒有 __path__ 屬性，無法遍歷子模塊")
        return loaded_jobs

    for finder, name, ispkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
        # 只處理子包，跳過模塊文件
        if not ispkg:
            logger.debug(f"跳過模塊 {name}（只加載子包）")
            continue

        try:
            module: ModuleType = importlib.import_module(name)
        except Exception as e:
            logger.error(f"加載任務(wù)模塊 {name} 失敗: {e}")
            continue

        if not hasattr(module, "job_factory"):
            logger.warning(f"模塊 {name} 沒有 job_factory 函數(shù)，跳過")
            continue

        try:
            job: JobProtocol = module.job_factory()

            # 使用 Protocol 的運(yùn)行時(shí)檢查功能驗(yàn)證協(xié)議實(shí)現(xiàn)
            if not isinstance(job, JobProtocol):
                logger.warning(
                    f"任務(wù) {name} 未實(shí)現(xiàn) JobProtocol 協(xié)議（缺少 enabled 或 run 方法）"
                )
                continue

            if not job.enabled():
                logger.info(f"任務(wù) {name} 已禁用，跳過")
                continue

            loaded_jobs.append(job)
            logger.info(f"成功加載任務(wù): {name}")

        except Exception as e:
            logger.error(f"創(chuàng)建任務(wù)實(shí)例 {name} 失敗: {e}")
            continue

    return loaded_jobs


def run_jobs(jobs: list[JobProtocol]) -> None:
    """執(zhí)行所有任務(wù)。

    Args:
        jobs: 要執(zhí)行的任務(wù)實(shí)例列表。
    """
    for job in jobs:
        try:
            result = job.run()
            logger.info(f"任務(wù) {job.__class__.__name__} 執(zhí)行完成，結(jié)果: {result}")
        except Exception as e:
            logger.error(f"任務(wù) {job.__class__.__name__} 執(zhí)行失敗: {e}")


def main() -> None:
    """程序入口函數(shù)。"""
    logger.info("開始加載任務(wù)...")
    jobs = load_jobs()
    logger.info(f"共加載 {len(jobs)} 個(gè)任務(wù)")

    logger.info("開始執(zhí)行任務(wù)...")
    run_jobs(jobs)
    logger.info("所有任務(wù)執(zhí)行完畢")


if __name__ == "__main__":
    main()