"""技能注册 ``SkillRegistry``：自动发现、注册与查询（需求 1.1、1.3）。 启动时扫描 ``skills/`` 包下的每个子包 / 模块，导入并查找实现了 ``PharmaSkill`` 契约的技能实例，按 ``SkillMeta.id`` 注册（需求 1.1）。 **关键容错保证（需求 1.3）**：单个 Skill 的导入或实例化抛出异常时，注册器以 ``try/except`` 捕获、记录失败并**跳过**该 Skill，绝不影响其余 Skill 的注册与 整体应用的运行。失败信息收集到 ``self.failures`` 并经 ``logging`` 输出；若注入了 ``audit`` 服务，则一并落审计。 发现约定（与 design.md「2. 技能注册 Registry」一致），按以下优先级在每个候选 模块中解析技能实例： 1. 模块级变量 ``SKILL``（且为 ``PharmaSkill`` 实例）。 2. 工厂函数 ``get_skill()``（返回 ``PharmaSkill`` 实例）。 3. 在模块内**定义**的 ``PharmaSkill`` 具体子类（非抽象），实例化其首个匹配者。 注册表通过 ``all()`` 按 ``SkillMeta.order`` 排序输出，供导航使用；``get(id)`` 按 id 精确查询。 """ from __future__ import annotations import importlib import inspect import logging import pkgutil import sys from pathlib import Path from types import ModuleType from typing import TYPE_CHECKING, Optional, Union from .skill_base import PharmaSkill if TYPE_CHECKING: # pragma: no cover - 仅类型检查 from .services import Services logger = logging.getLogger(__name__) #: ``discover`` 可接收的来源类型：已导入的包对象、可导入的包名、或目录路径。 DiscoverSource = Union[ModuleType, str, Path] class SkillRegistry: """技能注册表：发现、注册与查询 Skill。""" def __init__(self, svc: "Services" | None = None) -> None: self._skills: dict[str, PharmaSkill] = {} self.svc = svc #: 发现期间的失败记录列表，元素为 ``{"name", "error", "type"}``。 self.failures: list[dict] = [] # ------------------------------------------------------------------ # 发现与注册 # ------------------------------------------------------------------ def discover(self, source: DiscoverSource = "skills") -> None: """扫描 ``source`` 下的所有子模块 / 子包并注册其中的 Skill。 ``source`` 可为： - 已导入的包对象（``ModuleType``）； - 可导入的包名字符串（默认 ``"skills"``）； - 指向技能目录的 ``Path`` / 路径字符串（其父目录会被临时加入 ``sys.path`` 以便按包名导入）。 单个子模块导入或技能实例化失败时被捕获并跳过（需求 1.3）。 """ package = self._resolve_package(source) if package is None: return # 包自身可能没有 __path__（即并非真正的包），此时无可扫描内容。 pkg_path = getattr(package, "__path__", None) if pkg_path is None: logger.warning("发现源 %r 不是包，跳过扫描。", source) return for mod_info in pkgutil.iter_modules(pkg_path): full_name = f"{package.__name__}.{mod_info.name}" self._load_one(full_name) def _resolve_package(self, source: DiscoverSource) -> Optional[ModuleType]: """把 ``source`` 解析为已导入的包对象；解析失败返回 ``None``。""" if isinstance(source, ModuleType): return source # 目录路径：把父目录加入 sys.path，按目录名作为包名导入。 if isinstance(source, Path) or ( isinstance(source, str) and ("/" in source or "\\" in source) ): skills_path = Path(source) if not skills_path.exists(): logger.error("技能目录不存在：%s", skills_path) return None parent = str(skills_path.parent) if parent not in sys.path: sys.path.insert(0, parent) pkg_name = skills_path.name try: return importlib.import_module(pkg_name) except Exception as exc: # noqa: BLE001 - 顶层包导入失败需容错 self._record_failure(pkg_name, exc) return None # 普通包名字符串。 try: return importlib.import_module(str(source)) except Exception as exc: # noqa: BLE001 self._record_failure(str(source), exc) return None def _load_one(self, module_name: str) -> None: """导入单个候选模块、解析其 Skill 实例并注册；失败则跳过（需求 1.3）。""" try: module = importlib.import_module(module_name) skill = self._resolve_skill(module) if skill is None: logger.debug("模块 %s 未发现 Skill，跳过。", module_name) return self.register(skill) except Exception as exc: # noqa: BLE001 - 容错核心：任何异常均跳过 self._record_failure(module_name, exc) def _resolve_skill(self, module: ModuleType) -> Optional[PharmaSkill]: """按约定优先级从模块解析出一个 ``PharmaSkill`` 实例。""" # 1) 模块级 SKILL 实例。 skill = getattr(module, "SKILL", None) if isinstance(skill, PharmaSkill): return skill # 2) get_skill() 工厂。 factory = getattr(module, "get_skill", None) if callable(factory): instance = factory() if isinstance(instance, PharmaSkill): return instance # 3) 模块内定义的具体 PharmaSkill 子类。 for _, obj in inspect.getmembers(module, inspect.isclass): if ( issubclass(obj, PharmaSkill) and obj is not PharmaSkill and not inspect.isabstract(obj) and obj.__module__ == module.__name__ ): return obj() return None def register(self, skill: PharmaSkill) -> None: """按 ``meta.id`` 注册一个 Skill 实例（重复 id 覆盖并告警）。""" meta = getattr(skill, "meta", None) skill_id = getattr(meta, "id", None) if not skill_id: raise ValueError(f"Skill {skill!r} 缺少 meta.id，无法注册。") if skill_id in self._skills: logger.warning("Skill id 冲突：%s 已存在，将被覆盖。", skill_id) self._skills[skill_id] = skill logger.info("已注册 Skill：%s", skill_id) def _record_failure(self, name: str, exc: Exception) -> None: """记录一次发现失败：写入 ``failures``、日志，并尽力落审计。""" self.failures.append( {"name": name, "error": str(exc), "type": type(exc).__name__} ) logger.warning("加载 Skill 失败，已跳过：%s（%s）", name, exc) # 若注入了审计服务则一并记录（当前阶段可能为 None，需容错）。 audit = getattr(self.svc, "audit", None) recorder = getattr(audit, "record_error", None) if callable(recorder): try: recorder("skill_load_failed", name, str(exc)) except Exception: # noqa: BLE001 - 审计失败不得影响发现 logger.debug("审计记录 Skill 加载失败时再次出错。", exc_info=True) # ------------------------------------------------------------------ # 查询 # ------------------------------------------------------------------ def all(self) -> list[PharmaSkill]: """返回全部已注册 Skill，按 ``SkillMeta.order`` 升序排列。""" return sorted( self._skills.values(), key=lambda s: getattr(s.meta, "order", 100), ) def get(self, skill_id: str) -> Optional[PharmaSkill]: """按 id 精确查询；不存在返回 ``None``。""" return self._skills.get(skill_id) def __len__(self) -> int: return len(self._skills) def __contains__(self, skill_id: object) -> bool: return skill_id in self._skills __all__ = ["SkillRegistry"]