summaryrefslogtreecommitdiff
path: root/ATRI/exceptions.py
blob: e4d18c7434f59b7d7e2853fc9365e45431fa0125 (plain)
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import time
import json
from pathlib import Path
from typing import Optional
from traceback import format_exc
from pydantic.main import BaseModel

from nonebot.matcher import Matcher
from nonebot.adapters.onebot.v11 import ActionFailed
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent
from nonebot.message import run_postprocessor

from .log import log
from .message import MessageBuilder
from .utils import Limiter, gen_random_str


ERROR_DIR = Path(".") / "data" / "errors"
ERROR_DIR.mkdir(parents=True, exist_ok=True)


class ErrorInfo(BaseModel):
    track_id: str
    prompt: str
    time: str
    content: str


def _save_error(prompt: str, content: str) -> str:
    track_id = gen_random_str(8)
    data = ErrorInfo(
        track_id=track_id,
        prompt=prompt,
        time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        content=content,
    )
    path = ERROR_DIR / f"{track_id}.json"
    with open(path, "w", encoding="utf-8") as r:
        r.write(json.dumps(data.dict(), indent=4))
    return track_id


def load_error(track_id: str) -> ErrorInfo:
    path = ERROR_DIR / f"{track_id}.json"
    return ErrorInfo.parse_file(path)


class BaseBotException(Exception):
    prompt: Optional[str] = "ignore"

    def __init__(self, prompt: Optional[str]) -> None:
        self.prompt = prompt or self.__class__.prompt or self.__class__.__name__
        self.track_id = _save_error(self.prompt, format_exc())
        super().__init__(self.prompt)


class NotConfigured(BaseBotException):
    prompt = "缺少配置"


class InvalidConfigured(BaseBotException):
    prompt = "无效配置"


class WriteFileError(BaseBotException):
    prompt = "写入错误"


class ReadFileError(BaseBotException):
    prompt = "读取文件失败"


class RequestError(BaseBotException):
    prompt = "网页/接口请求错误"


class GetStatusError(BaseBotException):
    prompt = "获取状态失败"


class FormatError(BaseBotException):
    prompt = "格式错误"


class ServiceRegisterError(BaseBotException):
    prompt = "服务注册错误"


class BilibiliDynamicError(BaseBotException):
    prompt = "b站动态订阅错误"


class TwitterDynamicError(BaseBotException):
    prompt = "Twitter动态订阅错误"


class ThesaurusError(BaseBotException):
    prompt = "词库相关错误"


class RssError(BaseBotException):
    prompt = "RSS订阅错误"


limiter = Limiter(3, 600)


@run_postprocessor
async def _(bot: Bot, event, matcher: Matcher, exception: Optional[Exception]):
    if not exception:
        return

    try:
        prompt = exception.__class__.__name__
        track_id = _save_error(prompt, format_exc())
        log.warning(f"Ignore Exception: {prompt}")
    except BaseBotException as err:
        prompt = err.prompt or err.__class__.__name__
        track_id = _save_error(prompt, format_exc())
        log.warning(f"BotException: {prompt}")
    except ActionFailed as err:
        prompt = "请参考协议端输出"
        track_id = _save_error(prompt, format_exc())
        log.warning(f"ActionFailed: {prompt}")
    except Exception as err:
        prompt = "UnkErr " + err.__class__.__name__
        track_id = _save_error(prompt, format_exc())
        log.warning(f"Exception: {prompt}")

    log.error(f"Error Track ID: {track_id}")

    msg = (
        MessageBuilder("呜——出错了...请反馈维护者")
        .text(f"来自: {matcher.module_name}")
        .text(f"信息: {prompt}")
        .text(f"追踪ID: {track_id}")
    )
    if isinstance(event, GroupMessageEvent):
        group_id = str(event.group_id)
        if not limiter.check(group_id):
            msg = MessageBuilder("该群报错提示已达限制, 将冷却10min").text("如需反馈请: 来杯红茶")
        else:
            limiter.increase(group_id)

        if limiter.get_times(group_id) > 3:
            return

    try:
        await bot.send(event, msg)
    except Exception:
        return