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