diff options
Diffstat (limited to 'ATRI/service')
-rw-r--r-- | ATRI/service/__init__.py | 95 | ||||
-rw-r--r-- | ATRI/service/banlist.py | 36 | ||||
-rw-r--r-- | ATRI/service/dormant.py | 19 | ||||
-rw-r--r-- | ATRI/service/httppost.py | 41 | ||||
-rw-r--r-- | ATRI/service/limit.py | 78 | ||||
-rw-r--r-- | ATRI/service/plugin.py | 60 | ||||
-rw-r--r-- | ATRI/service/send.py | 13 | ||||
-rw-r--r-- | ATRI/service/switch.py | 79 | ||||
-rw-r--r-- | ATRI/service/temp.py | 6 |
9 files changed, 209 insertions, 218 deletions
diff --git a/ATRI/service/__init__.py b/ATRI/service/__init__.py index 5dddb97..6b18ba1 100644 --- a/ATRI/service/__init__.py +++ b/ATRI/service/__init__.py @@ -1,10 +1,95 @@ 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_PATH = Path('.') / 'ATRI' / 'data' / 'service' -ERROR_PATH = Path('.') / 'ATRI' / 'data' / 'error' -os.makedirs(SERVICE_PATH, exist_ok=True) -os.makedirs(ERROR_PATH, exist_ok=True) -state = 0 +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 index 06cc381..e1ce6b7 100644 --- a/ATRI/service/banlist.py +++ b/ATRI/service/banlist.py @@ -2,32 +2,32 @@ import json import aiofiles from typing import Optional -from ATRI.exceptions import InvalidWriteText +from ATRI.exceptions import WriteError +from . import SERVICE_DIR -from . import SERVICE_PATH - -class BanList: - filename = 'banlist.service.json' - path = SERVICE_PATH / filename +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_banlist(cls) -> dict: + 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_list(cls, user: Optional[str]) -> None: + async def add_to_list(cls, user: Optional[str]) -> None: + cls.data[user] = user try: - cls.data[user] = user async with aiofiles.open( cls.path, 'w', encoding='utf-8') as target: await target.write( @@ -35,13 +35,13 @@ class BanList: cls.data, indent=4 ) ) - except InvalidWriteText: - raise InvalidWriteText('Writing file failed!') - + except WriteError: + raise WriteError("Writing file failed!") + @classmethod - async def del_list(cls, user: Optional[str]) -> None: + async def del_from_list(cls, user: Optional[str]) -> None: + del cls.data[user] try: - del cls.data[user] async with aiofiles.open( cls.path, 'w', encoding='utf-8') as target: await target.write( @@ -49,5 +49,5 @@ class BanList: cls.data, indent=4 ) ) - except InvalidWriteText: - raise InvalidWriteText('List writing file failed!') + except WriteError: + raise WriteError("Writing file failed!") diff --git a/ATRI/service/dormant.py b/ATRI/service/dormant.py deleted file mode 100644 index 7482d53..0000000 --- a/ATRI/service/dormant.py +++ /dev/null @@ -1,19 +0,0 @@ -from ATRI.exceptions import InvalidSetting - -from . import state - -class Dormant: - @staticmethod - def is_sleep() -> bool: - return True if state != 1 else False - - @staticmethod - def cont_wake(_type: bool) -> None: - global state - try: - if _type: - state = 0 - else: - state = 1 - except InvalidSetting: - raise InvalidSetting('Failed to modify variable!') diff --git a/ATRI/service/httppost.py b/ATRI/service/httppost.py index 9469115..75aff8f 100644 --- a/ATRI/service/httppost.py +++ b/ATRI/service/httppost.py @@ -2,18 +2,24 @@ 获取更多帮助 >> https://github.com/howmanybots/onebot/blob/master/v11/specs/api/public.md ''' import json -from typing import Optional, Union, Dict, Any +from typing import ( + Optional, + Union, + Dict, + Any +) from ATRI.log import logger from ATRI.config import config -from ATRI.request import Request +from ATRI.utils.request import post_bytes URL = ( - f"http://{config['http_post']['host']}:" - f"{config['http_post']['port']}/" + f"http://{config['HttpPost']['host']}:" + f"{config['HttpPost']['port']}/" ) + class HttpPost: @classmethod @@ -27,17 +33,17 @@ class HttpPost: "message": message, "auto_escape": f"{auto_escape}" } - result = json.loads(await Request.post_bytes(url, params)) + 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] = ..., @@ -46,42 +52,42 @@ class HttpPost: 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, @@ -89,20 +95,20 @@ class HttpPost: 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, @@ -208,4 +214,3 @@ class HttpPost: @classmethod def clean_cache(cls): ... -
\ No newline at end of file diff --git a/ATRI/service/limit.py b/ATRI/service/limit.py new file mode 100644 index 0000000..d5a8a04 --- /dev/null +++ b/ATRI/service/limit.py @@ -0,0 +1,78 @@ +import json +import aiofiles + +from ATRI.exceptions import WriteError +from . import SERVICE_DIR + + +class Limit: + + file_name = "limit.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 _filling_service(cls, service, group: int = None): + # Determine whether the service exists in global variables. + if service not in cls.data["global"]: + cls.data["global"][service] = True + # Similary, for group. + if service not in cls.data[group]: + cls.data[group][service] = True + + + @classmethod + def get_service(cls) -> dict: + return cls.data + + @classmethod + async def auth_service( + cls, service: str, group: int = None) -> bool: + cls.data.setdefault("global", {}) + cls.data.setdefault(group, {}) + cls._filling_service(service, group) + + try: + async with aiofiles.open( + cls.path, 'w', encoding='utf-8') as target: + await target.write( + json.dumps( + cls.data + ) + ) + except WriteError: + raise WriteError("Writing file failed!") + + if cls.data["global"][service]: + return True if cls.data[group][service] else False + else: + return False + + @classmethod + async def control_service( + cls, + service: str, + status: bool, + group: int = None + ) -> None: + cls._filling_service(service, group) + + if group: + cls.data[group][service] = status + else: + cls.data["global"][service] = status + + try: + async with aiofiles.open( + cls.path, 'w', encoding='utf-8') as target: + await target.write( + json.dumps( + cls.data + ) + ) + except WriteError: + raise WriteError("Writing file failed!") diff --git a/ATRI/service/plugin.py b/ATRI/service/plugin.py deleted file mode 100644 index e14ef34..0000000 --- a/ATRI/service/plugin.py +++ /dev/null @@ -1,60 +0,0 @@ -import json -from typing import Optional -from pydantic import BaseModel - -from ATRI.log import logger -from ATRI.exceptions import InvalidWriteText - -from . import SERVICE_PATH - - -class Plugin: - class PluginInfo(BaseModel): - name: str - _type: str - docs: Optional[str] = None - command: list - - @classmethod - def register(cls, plugin_name: str, _type: str, - doc: Optional[str] = None, - command: Optional[list] = None) -> None: - filename = f'{plugin_name}.plugins.json' - path = SERVICE_PATH / 'plugins' / filename - path.parent.mkdir(exist_ok=True, parents=True) - try: - data = json.loads(path.read_bytes()) - except: - data = {} - - data = cls.PluginInfo( - name=plugin_name, - _type=_type, - docs=doc, - command=command - ) - try: - with open(path, 'w', encoding='utf-8') as target: - target.write( - json.dumps( - data.dict(), indent=4 - ) - ) - except InvalidWriteText: - raise InvalidWriteText('Writing file failed!') - else: - pass - docs_judge = "N" if not doc else "Y" - - a = ' ' - log_center = '' - log_head = f"Success register plugin: [{plugin_name}]." - log_suffix = f"Docs [{docs_judge}]. Type [{_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) diff --git a/ATRI/service/send.py b/ATRI/service/send.py deleted file mode 100644 index 9b9c3f9..0000000 --- a/ATRI/service/send.py +++ /dev/null @@ -1,13 +0,0 @@ -from nonebot.adapters.cqhttp import Bot - - -class Send: - @staticmethod - async def send_to_superuser(message: str) -> None: - from ATRI.config import RUNTIME_CONFIG - async def _send_to_superuser(bot: Bot) -> None: - for sup in RUNTIME_CONFIG['superusers']: - await bot.send_private_msg( - user_id=sup, - message=message - ) diff --git a/ATRI/service/switch.py b/ATRI/service/switch.py deleted file mode 100644 index b97a0ec..0000000 --- a/ATRI/service/switch.py +++ /dev/null @@ -1,79 +0,0 @@ -import json -import aiofiles -from typing import Optional - -from ATRI.exceptions import InvalidWriteText - -from . import SERVICE_PATH - - -class Switch: - filename = 'switch.service.json' - path = SERVICE_PATH / filename - path.parent.mkdir(exist_ok=True, parents=True) - try: - data = json.loads(path.read_bytes()) - except: - data = {} - - @classmethod - def get_service(cls) -> dict: - return cls.data - - @classmethod - async def auth_service( - cls, service: str, group: Optional[int] = None) -> bool: - cls.data.setdefault('global', {}) - cls.data.setdefault(group, {}) - if service not in cls.data['global']: - cls.data['global'][service] = True - if service not in cls.data[group]: - cls.data[group][service] = True - try: - async with aiofiles.open( - cls.path, 'w', encoding='utf-8') as target: - await target.write( - json.dumps( - cls.data, indent=4 - ) - ) - except InvalidWriteText: - raise InvalidWriteText('Writing file failed!') - else: - pass - - if cls.data['global'][service]: - return True if cls.data[group][service] else False - else: - return False - - @classmethod - async def control_service(cls, service: str, _type: bool, - group: Optional[str]) -> None: - if service not in cls.data: - cls.data['global'][service] = True - cls.data[group][service] = True - try: - async with aiofiles.open( - cls.path, 'w', encoding='utf-8') as target: - await target.write( - json.dumps( - cls.data, indent=4 - ) - ) - except InvalidWriteText: - raise InvalidWriteText('Writing file failed!') - if group: - cls.data[group][service] = _type - else: - cls.data['global'][service] = _type - try: - async with aiofiles.open( - cls.path, 'w', encoding='utf-8') as target: - await target.write( - json.dumps( - cls.data, indent=4 - ) - ) - except InvalidWriteText: - raise InvalidWriteText('Writing file failed!') diff --git a/ATRI/service/temp.py b/ATRI/service/temp.py deleted file mode 100644 index 2b73fdc..0000000 --- a/ATRI/service/temp.py +++ /dev/null @@ -1,6 +0,0 @@ -import os -from pathlib import Path - - -TEMP_PATH = Path('.') / 'ATRI' / 'data' / 'temp' / 'img' -os.makedirs(TEMP_PATH, exist_ok=True) |