message / plugins /content /main.py
hunian
refactor(plugins): 插件短名并统一 MCP tool 为 {plugin}-{tool}
cc826a1
Raw
History Blame Contribute Delete
1.83 kB
import logging
from typing import Optional
from .core import ContentExtractorCore
logger = logging.getLogger(__name__)
class ContentExtractorPlugin:
"""内容提取插件"""
def __init__(self) -> None:
self.name = "content"
self.version = "1.0.0"
self.enabled = False
self.core: Optional[ContentExtractorCore] = None
def on_enable(self) -> None:
"""启用插件并初始化内容提取核心逻辑"""
self.enabled = True
self.core = ContentExtractorCore()
self._sync_api_core(self.core)
logger.info(f"内容提取插件 {self.version} 已启用")
def on_disable(self) -> None:
"""禁用插件并释放运行时实例"""
self.enabled = False
self._sync_api_core(None)
self.core = None
logger.info(f"内容提取插件 {self.version} 已禁用")
def get_status(self) -> dict:
"""获取插件运行状态"""
return {
"name": self.name,
"version": self.version,
"enabled": self.enabled,
"ocr_loaded": self._is_ocr_loaded(),
"message": "运行正常" if self.enabled else "已禁用",
}
def _is_ocr_loaded(self) -> bool:
"""检查OCR引擎是否已加载,不触发懒加载初始化"""
if self.core is None:
return False
return self.core.ocr_service.is_loaded()
def _sync_api_core(self, core: Optional[ContentExtractorCore]) -> None:
"""同步API模块的插件实例和核心实例"""
try:
from . import api
api.set_plugin_instance(self)
api.set_core_instance(core)
except ImportError as e:
logger.warning(f"同步内容提取插件API实例失败: {e}")
plugin = ContentExtractorPlugin()