core/kernel/registry.py

"""技能注册 ``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"]