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())