demand_server/app/strategies/impl/__init__.py

import importlib
import inspect
import pkgutil

from app.strategies.base import BaseStrategy


def _iter_concrete_strategies() -> list[type[BaseStrategy]]:
    import app.strategies.impl as impl_package

    strategies: list[type[BaseStrategy]] = []
    seen_ids: set[str] = set()

    for module_info in pkgutil.iter_modules(impl_package.__path__):
        if module_info.name.startswith("_"):
            continue

        module = importlib.import_module(f"{impl_package.__name__}.{module_info.name}")
        for _, cls in inspect.getmembers(module, inspect.isclass):
            if cls.__module__ != module.__name__:
                continue
            if not issubclass(cls, BaseStrategy) or cls is BaseStrategy:
                continue
            if inspect.isabstract(cls):
                continue

            strategy_id = getattr(cls, "strategy_id", "").strip()
            if not strategy_id:
                continue
            if strategy_id in seen_ids:
                raise ValueError(f"duplicate strategy_id in impl package: {strategy_id}")

            seen_ids.add(strategy_id)
            strategies.append(cls)

    return strategies


def register_all() -> None:
    """自动扫描 impl/ 下策略模块并注册（跳过 _ 开头的模块/基类）。"""
    from app.strategies.registry import StrategyRegistry

    for strategy_cls in _iter_concrete_strategies():
        StrategyRegistry.register(strategy_cls())