diff options
| -rw-r--r-- | ATRI/plugins/manege/data_source.py | 86 |
1 files changed, 52 insertions, 34 deletions
diff --git a/ATRI/plugins/manege/data_source.py b/ATRI/plugins/manege/data_source.py index e936ef5..0a5a93e 100644 --- a/ATRI/plugins/manege/data_source.py +++ b/ATRI/plugins/manege/data_source.py @@ -29,6 +29,7 @@ __doc__ = """ class Manege(Service): + def __init__(self): Service.__init__(self, "管理", __doc__, True) @@ -47,11 +48,14 @@ class Manege(Service): if not path.is_file(): with open(path, "w", encoding="utf-8") as w: w.write(json.dumps({})) + return dict() + try: + data = json.loads(path.read_bytes()) + except BaseException: data = dict() - - data = json.loads(path.read_bytes()) return data - + + @staticmethod def _save_block_user_list(data: dict) -> None: file_name = "block_user.json" @@ -59,11 +63,10 @@ class Manege(Service): if not path.is_file(): with open(path, "w", encoding="utf-8") as w: w.write(json.dumps({})) - data = dict() - + with open(path, "w", encoding="utf-8") as w: w.write(json.dumps(data, indent=4)) - + @staticmethod def _load_block_group_list() -> dict: """ @@ -79,11 +82,14 @@ class Manege(Service): if not path.is_file(): with open(path, "w", encoding="utf-8") as w: w.write(json.dumps({})) + return dict() + + try: + data = json.loads(path.read_bytes()) + except BaseException: data = dict() - - data = json.loads(path.read_bytes()) return data - + @staticmethod def _save_block_group_list(data: dict) -> None: file_name = "block_group.json" @@ -91,52 +97,55 @@ class Manege(Service): if not path.is_file(): with open(path, "w", encoding="utf-8") as w: w.write(json.dumps({})) - data = dict() - + with open(path, "w", encoding="utf-8") as w: w.write(json.dumps(data, indent=4)) - + @classmethod def block_user(cls, user_id: str) -> bool: data = cls._load_block_user_list() - now_time = datetime.now() - data[user_id] = {"time": now_time} + now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + data[user_id] = { + "time": now_time + } try: cls._save_block_user_list(data) return True except BaseException: return False - + @classmethod def unblock_user(cls, user_id: str) -> bool: data: dict = cls._load_block_user_list() if user_id not in data: return False - + try: data.pop(user_id) cls._save_block_user_list(data) return True except BaseException: return False - + @classmethod def block_group(cls, group_id: str) -> bool: data = cls._load_block_group_list() - now_time = datetime.now() - data[group_id] = {"time": now_time} + now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + data[group_id] = { + "time": now_time + } try: cls._save_block_group_list(data) return True except BaseException: return False - + @classmethod def unblock_group(cls, group_id: str) -> bool: data: dict = cls._load_block_group_list() if group_id not in data: return False - + try: data.pop(group_id) cls._save_block_group_list(data) @@ -156,7 +165,7 @@ class Manege(Service): data["enabled"] = is_enabled ServiceTools().save_service(data, service) return True - + @staticmethod def control_user_service(service: str, user_id: str, is_enabled: bool) -> bool: """ @@ -167,7 +176,7 @@ class Manege(Service): except BaseException: return False temp_list: list = data.get("disable_user", list()) - + if is_enabled: try: temp_list.remove(user_id) @@ -178,7 +187,7 @@ class Manege(Service): data["disable_user"] = temp_list ServiceTools().save_service(data, service) return True - + @staticmethod def control_group_service(service: str, group_id: str, is_enabled: bool) -> bool: """ @@ -190,7 +199,7 @@ class Manege(Service): except BaseException: return False temp_list: list = data.get("disable_group", list()) - + if is_enabled: try: temp_list.remove(group_id) @@ -211,9 +220,12 @@ class Manege(Service): w.write(json.dumps({})) return dict() - data = json.loads(path.read_bytes()) + try: + data = json.loads(path.read_bytes()) + except BaseException: + data = dict() return data - + @staticmethod def save_friend_apply_list(data: dict) -> None: file_name = "friend_add.json" @@ -224,7 +236,7 @@ class Manege(Service): with open(path, "w", encoding="utf-8") as w: w.write(json.dumps(data, indent=4)) - + @staticmethod def load_invite_apply_list() -> dict: file_name = "group_invite.json" @@ -234,9 +246,12 @@ class Manege(Service): w.write(json.dumps({})) return dict() - data = json.loads(path.read_bytes()) + try: + data = json.loads(path.read_bytes()) + except BaseException: + data = dict() return data - + @staticmethod def save_invite_apply_list(data: dict) -> None: file_name = "group_invite.json" @@ -254,19 +269,22 @@ class Manege(Service): data = load_error(track_id) except ReadFileError: return "请检查ID是否正确..." - + prompt = data.get("prompt", "ignore") time = data.get("time", "ignore") content = data.get("content", "ignore") - + msg0 = TRACK_BACK_FORMAT.format( - track_id=track_id, prompt=prompt, time=time, content=content + track_id=track_id, + prompt=prompt, + time=time, + content=content ) f_data = FormData() f_data.add_field("poster", "ATRI running log") f_data.add_field("syntax", "text") f_data.add_field("expiration", "day") f_data.add_field("content", msg0) - + repo = f"详细请移步此处~\n{await UbuntuPaste(f_data).paste()}" return repo |