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
|
