Spaces:
Runtime error
Runtime error
File size: 2,130 Bytes
ec90eae | 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 | """
PregoPal - 插件基类
===================
定义 LoopPlugin 基类、PluginRegistry、LoopStage、LoopContext。
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import List
class LoopStage(Enum):
"""插件可注册的阶段"""
FAMILY_QUIZ = "family_quiz"
SUMMARIZE = "summarize"
ANALYZE = "analyze"
BRIEF = "brief"
THREE_DAY = "three_day"
CONSOLIDATE = "consolidate"
@dataclass
class LoopContext:
"""插件上下文：共享数据容器"""
briefing: dict = field(default_factory=dict)
weight_data: dict = field(default_factory=dict)
diet_records: list = field(default_factory=list)
family_recipes: list = field(default_factory=list)
family_memory: dict = field(default_factory=dict)
analysis_results: dict = field(default_factory=dict)
errors: list = field(default_factory=list)
@dataclass
class PluginResult:
"""插件执行结果"""
success: bool = True
data: dict = field(default_factory=dict)
message: str = ""
class LoopPlugin(ABC):
"""Loop 扩展插件基类 — 所有插件继承此类"""
@abstractmethod
def stage(self) -> LoopStage:
"""此插件注册到哪个阶段"""
@abstractmethod
def name(self) -> str:
"""插件唯一名称"""
@abstractmethod
async def run(self, ctx: LoopContext) -> PluginResult:
"""执行插件逻辑"""
class PluginRegistry:
"""插件注册中心"""
def __init__(self):
self._plugins: List[LoopPlugin] = []
def register(self, plugin: LoopPlugin):
"""注册一个插件"""
self._plugins.append(plugin)
def get_plugins(self, stage: LoopStage) -> List[LoopPlugin]:
"""获取指定阶段的所有插件"""
return [p for p in self._plugins if p.stage() == stage]
def get_all(self) -> List[LoopPlugin]:
"""获取所有插件"""
return self._plugins.copy()
def unregister(self, name: str):
"""根据名称注销插件"""
self._plugins = [p for p in self._plugins if p.name() != name]
|