astrbot_plugin_office_assistant/internal_hooks.py at master · Clhikari/astrbot_plugin_office_assistant · GitHub

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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from __future__ import annotations

from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass, field
from inspect import isawaitable
from typing import Any

from .domain.document.hooks import (
    AfterExportContext,
    AfterExportHook,
    BeforeExportHook,
    BlockNormalizationContext,
    BlockNormalizeHook,
    ExportPreparationContext,
    run_after_export_hooks,
    run_before_export_hooks,
    run_block_normalize_hooks,
)

__all__ = [
    "AfterExportContext",
    "AfterExportHook",
    "BeforeExportHook",
    "BlockNormalizationContext",
    "BlockNormalizeHook",
    "ExportPreparationContext",
    "NoticeBuildContext",
    "NoticeBuildHook",
    "ToolExposureContext",
    "ToolExposureHook",
    "run_after_export_hooks",
    "run_before_export_hooks",
    "run_block_normalize_hooks",
    "run_notice_hooks",
    "run_tool_exposure_hooks",
]


@dataclass(slots=True)
class NoticeBuildContext:
    event: Any
    request: Any
    should_expose: bool
    can_process_upload: bool
    explicit_tool_name: str | None
    notices: list[str] = field(default_factory=list)
    section_names: list[str] = field(default_factory=list)
    system_notices: list[str] = field(default_factory=list)
    system_section_names: list[str] = field(default_factory=list)


@dataclass(slots=True)
class ToolExposureContext:
    event: Any
    request: Any
    should_expose: bool
    can_process_upload: bool
    explicit_tool_name: str | None


NoticeBuildHook = Callable[
    [NoticeBuildContext],
    NoticeBuildContext | None | Awaitable[NoticeBuildContext | None],
]
ToolExposureHook = Callable[
    [ToolExposureContext],
    ToolExposureContext | None | Awaitable[ToolExposureContext | None],
]


async def run_notice_hooks(
    hooks: Sequence[NoticeBuildHook],
    context: NoticeBuildContext,
) -> NoticeBuildContext:
    current = context
    for hook in hooks:
        result = hook(current)
        if isawaitable(result):
            result = await result
        if result is not None:
            current = result
    return current


async def run_tool_exposure_hooks(
    hooks: Sequence[ToolExposureHook],
    context: ToolExposureContext,
) -> ToolExposureContext:
    current = context
    for hook in hooks:
        result = hook(current)
        if isawaitable(result):
            result = await result
        if result is not None:
            current = result
    return current