♻️⚡️ 重构 Service，优化部分代码

4 files changed, 0 insertions, 417 deletions

diff --git a/ATRI/service/__init__.py b/ATRI/service/__init__.py
deleted file mode 100644
index 6b18ba1..0000000
--- a/ATRI/service/__init__.py
+++ /dev/null
@@ -1,95 +0,0 @@
-import os
-import json
-from pathlib import Path
-from pydantic import BaseModel
-from typing import Optional
-
-from ATRI.log import logger
-from ATRI.exceptions import WriteError
-
-
-SERVICE_DIR = Path('.') / 'ATRI' / 'data' / 'service'
-PLUGIN_INFO_DIR = Path('.') / 'ATRI' / 'data' / 'service' / 'plugins'
-os.makedirs(SERVICE_DIR, exist_ok=True)
-os.makedirs(PLUGIN_INFO_DIR, exist_ok=True)
-
-sleep = False
-
-class Service:
-    class ServiceInfo(BaseModel):
-        name: str
-        _type: str
-        docs: Optional[str] = None
-        command: Optional[list] = None
-    
-    @classmethod
-    def register(
-        cls,
-        service_name: str,
-        service_type: str,
-        docs: Optional[str] = None,
-        command: Optional[list] = None
-    ) -> None:
-        """
-        启动时保存各个服务信息，便于后续网页控制台调用
-        增强管理
-        """
-        file_name = f"{service_name}.function.json"
-        path = SERVICE_DIR / file_name
-        path.parent.mkdir(exist_ok=True, parents=True)
-        try:
-            data = json.loads(path.read_bytes())
-        except:
-            data = {}
-        
-        data = cls.ServiceInfo(
-            name=service_name,
-            _type=service_type,
-            docs=docs,
-            command=command
-        )
-        try:
-            with open(path, 'w', encoding='utf-8') as target:
-                target.write(
-                    json.dumps(
-                        data.dict(), indent=4
-                    )
-                )
-        except WriteError:
-            raise WriteError("Writing file failed!")
-        else:
-            pass
-        
-        # TODO: shitcode-style
-        docs_judge = "N" if not docs else "Y"
-        a = " "
-        log_center = ""
-        log_head = f"Success register service: [{service_name}]."
-        log_suffix = f"Docs [{docs_judge}]. Type [{service_type}]"
-        log_head_lenght = len(log_head)
-        log_suffix_lenght = len(log_suffix)
-        log_center_lenght = 120 - (
-            log_head_lenght + log_suffix_lenght
-        )
-        for _ in range(log_center_lenght): log_center = log_center + a
-        log_print = log_head + log_center + log_suffix
-        logger.info(log_print)
-    
-    @staticmethod
-    def is_dormant() -> bool:
-        """
-        为bot提供休眠，期间不会响应除了 superusers 以外的用户的信息
-        触发在于 Rule
-        """
-        return False if sleep else True
-    
-    @staticmethod
-    def control_dormant(_type: bool) -> None:
-        """
-        更改休眠状态
-        """
-        global sleep
-        if _type:
-            sleep = True
-        else:
-            sleep = False
diff --git a/ATRI/service/banlist.py b/ATRI/service/banlist.py
deleted file mode 100644
index e1ce6b7..0000000
--- a/ATRI/service/banlist.py
+++ /dev/null
@@ -1,53 +0,0 @@
-import json
-import aiofiles
-from typing import Optional
-
-from ATRI.exceptions import WriteError
-from . import SERVICE_DIR
-
-
-class BanSystem:
-    
-    file_name = "banlist.service.json"
-    path = SERVICE_DIR / file_name
-    path.parent.mkdir(exist_ok=True, parents=True)
-    try:
-        data = json.loads(path.read_bytes())
-    except:
-        data = {}
-    
-    @classmethod
-    def get_list(cls) -> dict:
-        return cls.data
-    
-    @classmethod
-    def is_in_list(cls, user: Optional[str]) -> bool:
-        return False if user in cls.data else True
-
-    @classmethod
-    async def add_to_list(cls, user: Optional[str]) -> None:
-        cls.data[user] = user
-        try:
-            async with aiofiles.open(
-                cls.path, 'w', encoding='utf-8') as target:
-                await target.write(
-                    json.dumps(
-                        cls.data, indent=4
-                    )
-                )
-        except WriteError:
-            raise WriteError("Writing file failed!")
-    
-    @classmethod
-    async def del_from_list(cls, user: Optional[str]) -> None:
-        del cls.data[user]
-        try:
-            async with aiofiles.open(
-                cls.path, 'w', encoding='utf-8') as target:
-                await target.write(
-                    json.dumps(
-                        cls.data, indent=4
-                    )
-                )
-        except WriteError:
-            raise WriteError("Writing file failed!")
diff --git a/ATRI/service/httppost.py b/ATRI/service/httppost.py
deleted file mode 100644
index 75aff8f..0000000
--- a/ATRI/service/httppost.py
+++ /dev/null
@@ -1,216 +0,0 @@
-'''
-获取更多帮助 >> https://github.com/howmanybots/onebot/blob/master/v11/specs/api/public.md
-'''
-import json
-from typing import (
-    Optional,
-    Union,
-    Dict,
-    Any
-)
-
-from ATRI.log import logger
-from ATRI.config import config
-from ATRI.utils.request import post_bytes
-
-
-URL = (
-    f"http://{config['HttpPost']['host']}:"
-    f"{config['HttpPost']['port']}/"
-)
-
-
-class HttpPost:
-
-    @classmethod
-    async def send_private_msg(cls,
-                               user_id: int,
-                               message: str,
-                               auto_escape: bool = False): # -> Dict[str, Any]
-        url = URL + "send_private_msg?"
-        params = {
-            "user_id": user_id,
-            "message": message,
-            "auto_escape": f"{auto_escape}"
-        }
-        result = json.loads(await post_bytes(url, params))
-        logger.debug(result)
-        return result
-
-    @classmethod
-    def send_group_msg(cls,
-                       group_id: int,
-                       message: Union[str],
-                       auto_escape: Optional[bool] = ...) -> Dict[str, Any]:
-        ...
-
-    @classmethod
-    def send_msg(cls,
-                 message_type: Optional[str] = ...,
-                 user_id: Optional[int] = ...,
-                 group_id: Optional[int] = ...,
-                 message = Union[str],
-                 auto_escape: bool = ...) -> Dict[str, Any]:
-        ...
-
-    @classmethod
-    def delete_msg(cls,
-                   message_id: int):
-        ...
-
-    @classmethod
-    def get_msg(cls,
-                message_id: int) -> Dict[str, Any]:
-        ...
-
-    @classmethod
-    def get_forward_msg(cls,
-                        id: int):
-        ...
-
-    @classmethod
-    def send_like(cls,
-                  user_id: int,
-                  times: int = ...):
-        ...
-
-    @classmethod
-    def set_group_kick(cls,
-                       group_id: int,
-                       user_id: int,
-                       reject_add_request: bool = ...):
-        ...
-
-    @classmethod
-    def set_group_ban(cls,
-                      group_id: int,
-                      user_id: int,
-                      duration: int = ...):
-        ...
-
-    @classmethod
-    def set_group_anonymous_ban(cls,
-                                group_id: int,
-                                anonymous: Optional[Dict[str, Any]] = ...,
-                                flag: Optional[str] = ...,
-                                duration: int = ...):
-        ...
-
-    @classmethod
-    def set_group_whole_ban(cls,
-                            group_id: int,
-                            enable: bool = ...):
-        ...
-
-    @classmethod
-    def set_group_admin(cls,
-                        group_id: int,
-                        user_id: int,
-                        enable: bool = ...):
-        ...
-
-    @classmethod
-    def set_group_anonymous(cls,
-                            group_id: int,
-                            enable: bool = ...):
-        ...
-    
-    @classmethod
-    def set_group_card(cls):
-        ...
-
-    @classmethod
-    def set_group_name(cls):
-        ...
-    
-    @classmethod
-    def set_group_leave(cls):
-        ...
-    
-    @classmethod
-    def set_group_special_title(cls):
-        ...
-    
-    @classmethod
-    def set_friend_add_request(cls):
-        ...
-    
-    @classmethod
-    def set_group_add_request(cls):
-        ...
-
-    @classmethod
-    def get_login_info(cls):
-        ...
-    
-    @classmethod
-    def get_stranger_info(cls):
-        ...
-    
-    @classmethod
-    def get_friend_list(cls):
-        ...
-    
-    @classmethod
-    def get_group_info(cls):
-        ...
-    
-    @classmethod
-    def get_group_list(cls):
-        ...
-    
-    @classmethod
-    def get_group_member_info(cls):
-        ...
-    
-    @classmethod
-    def get_group_member_list(cls):
-        ...
-
-    @classmethod
-    def get_group_honor_info(cls):
-        ...
-    
-    @classmethod
-    def get_cookies(cls):
-        ...
-    
-    @classmethod
-    def get_csrf_token(cls):
-        ...
-    
-    @classmethod
-    def get_credentials(cls):
-        ...
-    
-    @classmethod
-    def get_record(cls):
-        ...
-    
-    @classmethod
-    def get_image(cls):
-        ...
-    
-    @classmethod
-    def can_send_image(cls):
-        ...
-    
-    @classmethod
-    def can_send_record(cls):
-        ...
-    
-    @classmethod
-    def get_status(cls):
-        ...
-    
-    @classmethod
-    def get_version_info(cls):
-        ...
-    
-    @classmethod
-    def set_restart(cls):
-        ...
-    
-    @classmethod
-    def clean_cache(cls):
-        ...
diff --git a/ATRI/service/limit.py b/ATRI/service/limit.py
deleted file mode 100644
index 2ea3f95..0000000
--- a/ATRI/service/limit.py
+++ /dev/null
@@ -1,53 +0,0 @@
-import json
-from pathlib import Path
-from typing import Optional
-
-from ATRI.utils.file import write_file
-from . import SERVICE_DIR
-
-
-class Limit:
-    @staticmethod
-    def _get_file(group: Optional[int] = None) -> Path:
-        file_name = f"{group}.service.json"
-        LIMIT_DIR = SERVICE_DIR / "limit"
-        path = LIMIT_DIR / file_name
-        
-        if not LIMIT_DIR.exists():
-            LIMIT_DIR.mkdir()
-        return path
-    
-    @classmethod
-    def _read_file(cls, group: Optional[int] = None) -> dict:
-        try:
-            data = json.loads(cls._get_file(group).read_bytes())
-        except:
-            data = {}
-        return data
-
-    @classmethod
-    async def auth_service(cls, service: str, group: Optional[int] = None) -> bool:
-        data = cls._read_file(group)
-        if service not in data:
-            data[service] = True
-            await write_file(cls._get_file(group), json.dumps(data))
-        
-        if data[service]:
-            return True
-        else:
-            return False
-    
-    @classmethod
-    async def control_service(
-        cls,
-        service: str,
-        status: bool,
-        group: Optional[int] = None
-    ) -> None:
-        data = cls._read_file(group)
-        if service not in data:
-            data[service] = True
-            await write_file(cls._get_file(group), json.dumps(data))
-        
-        data[service] = status
-        await write_file(cls._get_file(group), json.dumps(data))