src/infini/handler.py


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

from infini.input import Input
from infini.output import Output
from infini.typing import List, Any, RouterType, Callable, Generator, Union
from infini.queue import EventQueue


class Handler:
    handlers: List[RouterType]

    def input(self, input: Input):
        if (queue := self.match(input.get_plain_text())).is_empty():
            yield Output.empty()
            return
        while not queue.is_empty():
            if isinstance(stream := queue.pop()(input), Generator):
                for output in stream:
                    yield output
                    if output.block:
                        return
            else:
                yield stream
                if stream.block:
                    return

    def match(
        self, text: str
    ) -> EventQueue[Callable[[Input], Union[Output, Generator[Output, Any, None]]]]:
        queue = EventQueue()

        for handler in self.handlers:
            if handler["router"].match(text):
                queue.push(handler["priority"], handler["handler"])

        return queue