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
|
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, PrivateMessageEvent, GroupMessageEvent
from nonebot.message import run_postprocessor
from .log import logger as log
from .config import BotSelfConfig
from .utils import 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) -> dict:
path = ERROR_DIR / f"{track_id}.json"
return json.loads(path.read_bytes())
class BaseBotException(BaseException):
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 = "词库相关错误"
@run_postprocessor
async def _track_error(
bot: Bot, event, matcher: Matcher, exception: Optional[Exception]
) -> None:
if not exception:
return
try:
raise exception
except BaseBotException as err:
prompt = err.prompt or err.__class__.__name__
track_id = err.track_id
except ActionFailed as err:
prompt = "请参考协议端输出"
track_id = _save_error(prompt, format_exc())
except Exception as err:
prompt = "Unknown ERROR->" + err.__class__.__name__
track_id = _save_error(prompt, format_exc())
if isinstance(event, PrivateMessageEvent):
_id = "用户" + event.get_user_id()
elif isinstance(event, GroupMessageEvent):
_id = "群" + str(event.group_id)
else:
_id = "unknown"
log.error(f"Error Track ID: {track_id}")
msg = f"呜——出错了...追踪: {track_id}\n来自: {_id}"
for superusers in BotSelfConfig.superusers:
try:
await bot.send_private_msg(user_id=superusers, message=msg)
except BaseBotException:
return
|
