File size: 8,242 Bytes
19729e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """技能注册 ``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"]
|