summaryrefslogtreecommitdiff
path: root/ATRI/exceptions.py
blob: ab9e5d344cf3271d962a99fd7468bf1107eecf5a (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
import time
import json
import string
from pathlib import Path
from random import sample
from typing import Optional
from traceback import format_exc
from pydantic.main import BaseModel

from nonebot.adapters.cqhttp import MessageEvent
from nonebot.matcher import Matcher
from nonebot.message import run_postprocessor

from .log import logger


ERROR_FILE = Path('.') / 'ATRI' / 'data' / 'errors'
ERROR_FILE.parent.mkdir(exist_ok=True, parents=True)


class ExceptionInfo(BaseModel):
    error_id: str
    prompt: str
    time: str
    error_content: str


def store_error(error_id: str, prompt, error_content: str) -> None:
    data = ExceptionInfo(
        error_id=error_id,
        prompt=prompt,
        time=time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()),
        error_content=error_content
    )

    path = ERROR_FILE / f"{error_id}.json"
    path.parent.mkdir(exist_ok=True, parents=True)
    with open(path, 'w', encoding='utf-8') as target:
        target.write(
            json.dumps(
                data.dict(), indent=4
            )
        )


def read_error(error_id: str) -> dict:
    path = ERROR_FILE / f"{error_id}.json"
    try:
        with open(path, 'r', encoding='utf-8') as target:
            data = target.read()
        return json.loads(data)
    except FileNotFoundError:
        raise FileNotFoundError


class BaseBotException(BaseException):
    prompt: Optional[str] = 'ignore'

    def __init__(self, prompt: Optional[str]) -> None:
        super().__init__(self)
        self.prompt = prompt or self.__class__.prompt \
            or self.__class__.__name__
        self.error_content = format_exc()
        self.error_id = ''.join(
            sample(string.ascii_letters + string.digits, 16)
        )


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


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


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


class LoadingError(BaseBotException):
    prompt = "加载错误"


class RequestTimeOut(BaseBotException):
    prompt = "网页/接口请求超时"


@run_postprocessor  # type: ignore
async def _(matcher: Matcher, exception: Optional[Exception],
            event: MessageEvent, state: dict) -> None:
    """检测Bot运行中的报错,并进行提醒"""
    print(114514)
    if not exception:
        return
    
    try:
        raise exception
    except BaseBotException as Error:
        prompt = Error.prompt or Error.__class__.__name__
        error_id = Error.error_id
        # error_content = format_exc()
    except Exception as Error:
        prompt = "Unknown ERROR" + Error.__class__.__name__
        error_id = ''.join(
            sample(string.ascii_letters + string.digits, 16)
        )
        store_error(error_id, prompt, format_exc())

    logger.debug(f"A bug has been cumming, trace ID: {error_id}")
    msg = (
        "[WARNING] 这是一个错误...\n"
        f"Track ID: {error_id}\n"
        f"Reason: {prompt}\n"
        "ごんめなさい... ;w;"
    )
    
    await matcher.finish(msg)