diff options
| -rw-r--r-- | src/infini/core.py | 16 | ||||
| -rw-r--r-- | src/infini/exceptions.py | 4 | ||||
| -rw-r--r-- | src/infini/generator.py | 48 | ||||
| -rw-r--r-- | src/infini/loader.py | 26 | ||||
| -rw-r--r-- | src/infini/output.py | 6 | ||||
| -rw-r--r-- | src/infini/register.py | 30 | ||||
| -rw-r--r-- | src/infini/typing.py | 14 |
7 files changed, 115 insertions, 29 deletions
diff --git a/src/infini/core.py b/src/infini/core.py index 08b81f15..7c84a15f 100644 --- a/src/infini/core.py +++ b/src/infini/core.py @@ -1,23 +1,23 @@ from infini.input import Input from infini.interceptor import Interceptor -from infini.generator import TextGenerator +from infini.generator import Generator from infini.handler import Handler from infini.injector import Injector from infini.output import Output -from infini.typing import Any, Generator, Union +from infini.typing import Any, Generator as GeneratorT, Union from infini.exceptions import ValueError class Core: pre_interceptor: Interceptor handler: Handler - generator: TextGenerator + generator: Generator interceptor: Interceptor injector: Injector def input( self, input: Input - ) -> Generator[Union[str, Output], Any, None]: # TODO 支持Workflow + ) -> GeneratorT[Union[str, Output], Any, None]: for pre_intercepted_stream in self.pre_intercept(input): if isinstance(pre_intercepted_stream, Output): if not isinstance(pre_intercepted_stream, Output): @@ -53,10 +53,12 @@ class Core: if handled_stream.block: return - def pre_intercept(self, input: Input) -> Generator[Union[Input, Output], Any, None]: + def pre_intercept( + self, input: Input + ) -> GeneratorT[Union[Input, Output], Any, None]: return self.pre_interceptor.input(input) - def handle(self, input: Input) -> Generator[Output, Any, None]: + def handle(self, input: Input) -> GeneratorT[Output, Any, None]: iterator = self.handler.input(input) for output in iterator: yield output @@ -66,5 +68,5 @@ class Core: def intercept( self, output: Output, output_text: str - ) -> Generator[Union[str, Output], Any, None]: + ) -> GeneratorT[Union[str, Output], Any, None]: return self.interceptor.output(output, output_text) diff --git a/src/infini/exceptions.py b/src/infini/exceptions.py index 37386ce3..2e2b0197 100644 --- a/src/infini/exceptions.py +++ b/src/infini/exceptions.py @@ -10,5 +10,9 @@ class UnknownEvent(InfiniException): """文本事件不存在""" +class UnknownEventType(InfiniException, TypeError): + """文本事件不存在""" + + class ValueError(InfiniException, ValueError): """错误的数据""" diff --git a/src/infini/generator.py b/src/infini/generator.py index 5473dade..254f0cdd 100644 --- a/src/infini/generator.py +++ b/src/infini/generator.py @@ -1,20 +1,37 @@ from infini.output import Output -from infini.typing import Dict, Callable, Union -from infini.exceptions import UnknownEvent +from infini.typing import Type, Dict, Callable, Union, Optional +from infini.exceptions import UnknownEvent, UnknownEventType from infini.injector import Injector from jinja2 import Template +import abc -class TextGenerator: # TODO 兼容多类型事件 + +class BaseGenerator(metaclass=abc.ABCMeta): + type: str events: Dict[str, str] global_variables: Dict[str, Union[str, Callable]] + @abc.abstractmethod + def output(self, output: Output, injector: Injector) -> str: + raise NotImplementedError + + @abc.abstractmethod + def match(self, output: Output) -> Template: + raise NotImplementedError + + +class TextGenerator(BaseGenerator): + type = "text" + def __init__(self) -> None: self.events = {} self.global_variables = {} def output(self, output: Output, injector: Injector) -> str: - assert output.type == "text", "文本生成器应当传入类型为 'text' 的 Output 实例" + assert ( + output.type == self.type + ), f"文本生成器应当传入类型为 '{self.type}' 的 Output 实例" variables = self.global_variables.copy() variables.update(output.variables) for name, variable in variables.items(): @@ -26,3 +43,26 @@ class TextGenerator: # TODO 兼容多类型事件 if context := self.events.get(output.name): return Template(context) raise UnknownEvent(f"事件不存在: {output.name}") + + +class Generator: + generators: Dict[str, BaseGenerator] + events: Dict[str, str] + global_variables: Dict[str, Union[str, Callable]] + + def __init__(self) -> None: + self.generators = {"text": TextGenerator()} + + def output(self, output: Output, injector: Injector) -> str: + assert ( + output.type != "workflow" + ), "生成器应当传入类型为非 'workflow' 的 Output 实例" + if not (generator := self.match(output)): + raise UnknownEventType(f"没有为事件类型 '{output.type}' 注册生成器") + + generator.events = self.events + generator.global_variables = self.global_variables + return generator.output(output, injector) + + def match(self, output: Output) -> Optional[BaseGenerator]: + return self.generators.get(output.type) diff --git a/src/infini/loader.py b/src/infini/loader.py index 061ed8b9..07ebe0a8 100644 --- a/src/infini/loader.py +++ b/src/infini/loader.py @@ -1,6 +1,6 @@ from importlib.util import spec_from_file_location from infini.core import Core -from infini.generator import TextGenerator +from infini.generator import BaseGenerator, Generator, TextGenerator from infini.handler import Handler from infini.injector import Injector from infini.interceptor import Interceptor @@ -81,15 +81,17 @@ class InfiniLoader(importlib.abc.Loader): exec(Path(self.filename).read_text("utf-8"), vars(module)) -def install(): - sys.meta_path.insert(0, InfiniMetaFinder()) +def _install(): + if not sys.meta_path: + raise OSError("Var 'sys.meta_path' is empty, since Python is stop.") + if not isinstance(sys.meta_path[0], InfiniMetaFinder): + sys.meta_path.insert(0, InfiniMetaFinder()) -def uninstall(): +def _uninstall(): for meta_path in sys.meta_path: if isinstance(meta_path, InfiniMetaFinder): sys.meta_path.remove(meta_path) - break class Loader: @@ -98,6 +100,7 @@ class Loader: events: Dict[str, str] global_variables: Dict[str, Union[str, Callable]] interceptors: List[RouterType] + generators: Dict[str, BaseGenerator] def __init__(self) -> None: self.pre_interceptors = [] @@ -105,6 +108,7 @@ class Loader: self.events = {} self.global_variables = {} self.interceptors = [] + self.generators = {} self.prepare() def __enter__(self) -> "Loader": @@ -124,7 +128,7 @@ class Loader: return register_variables def prepare(self) -> None: - install() + _install() def load(self, name: str) -> ModuleType: self.prepare() @@ -168,7 +172,7 @@ class Loader: return list def close(self): - uninstall() + _uninstall() def inject_core(self, core: Core): pre_interceptor = Interceptor() @@ -176,6 +180,7 @@ class Loader: generator = TextGenerator() interceptor = Interceptor() injector = Injector() + generator = Generator() self.inject_pre_interceptor(pre_interceptor) self.inject_handler(handler) @@ -221,12 +226,13 @@ class Loader: self.inject_handler(handler) return handler - def inject_generator(self, generator: TextGenerator): + def inject_generator(self, generator: Generator): generator.events = self.events generator.global_variables = self.global_variables + generator.generators.update(self.generators) - def into_generator(self) -> TextGenerator: - generator = TextGenerator() + def into_generator(self) -> Generator: + generator = Generator() self.inject_generator(generator) return generator diff --git a/src/infini/output.py b/src/infini/output.py index 73aa5213..13f26d47 100644 --- a/src/infini/output.py +++ b/src/infini/output.py @@ -1,8 +1,8 @@ -from infini.typing import Literal, Dict, Any +from infini.typing import Union, Literal, Dict, Any class Output: - type: Literal["null", "text", "workflow"] + type: Union[Literal["null", "text", "workflow"], str] name: str status: int block: bool @@ -11,7 +11,7 @@ class Output: def __init__( self, - type: Literal["null", "text", "workflow"], + type: Union[Literal["null", "text", "workflow"], str], name: str, *, status: int = 0, diff --git a/src/infini/register.py b/src/infini/register.py index 0586ae7e..0d85c3c0 100644 --- a/src/infini/register.py +++ b/src/infini/register.py @@ -1,7 +1,8 @@ +from infini.typing import List, Dict, Any, Callable, RouterType, Optional, Union, Type from infini.input import Input from infini.output import Output from infini.router import Contains, Router -from infini.typing import List, Dict, Any, Callable, RouterType, Optional, Union +from infini.generator import BaseGenerator from functools import wraps @@ -11,6 +12,7 @@ class Register: events: Dict[str, str] global_variables: Dict[str, Union[str, Callable]] interceptors: List[RouterType] + generators: Dict[str, BaseGenerator] def __init__(self) -> None: self.pre_interceptors = [] @@ -18,6 +20,7 @@ class Register: self.events = {} self.global_variables = {} self.interceptors = [] + self.generators = {} def pre_interceptor(self, router: Union[Router, str], priority: int = 0): def decorator(func): @@ -54,9 +57,27 @@ class Register: return decorator def regist_textevent(self, name: str, text: str): + import warnings + + warnings.warn( + "Infini will soon deprecated `regist_textevent`, " + "use `register_textevent` instead." + ) + self.events[name] = text + + def register_textevent(self, name: str, text: str): self.events[name] = text def regist_variable(self, name: str, data: Any): + import warnings + + warnings.warn( + "Infini will soon deprecated `regist_variable`, " + "use `register_variable` instead." + ) + self.global_variables[name] = data + + def register_variable(self, name: str, data: Any): self.global_variables[name] = data def dynamic_variable(self, name: Optional[str] = None): @@ -86,3 +107,10 @@ class Register: return wrapper return decorator + + def register_generator( + self, generator: Union[BaseGenerator, Type[BaseGenerator]] + ) -> None: + if not isinstance(generator, BaseGenerator): + generator = generator() + self.generators[generator.type] = generator diff --git a/src/infini/typing.py b/src/infini/typing.py index 874e8753..14c219fc 100644 --- a/src/infini/typing.py +++ b/src/infini/typing.py @@ -1,7 +1,9 @@ from typing import ( + TYPE_CHECKING, Dict as Dict, List as List, Any as Any, + Type as Type, Optional as Optional, Generic as Generic, Callable as Callable, @@ -14,14 +16,18 @@ from typing import ( Union, ) from types import ModuleType as ModuleType, GeneratorType as GeneratorType -from . import router, input, output + +if TYPE_CHECKING: + from infini.router import Router + from infini.input import Input + from infini.output import Output T = TypeVar("T") -Stream = Union["input.Input[Any]", "output.Output"] -OutputGenerator = Generator["output.Output", Any, None] +Stream = Union["Input[Any]", "Output"] +OutputGenerator = Generator["Output", Any, None] class RouterType(TypedDict): priority: int - router: router.Router + router: "Router" handler: Callable[..., Union[Stream, OutputGenerator]] |