diff options
Diffstat (limited to 'ATRI/service')
-rw-r--r-- | ATRI/service/__init__.py | 10 | ||||
-rw-r--r-- | ATRI/service/banlist.py | 53 | ||||
-rw-r--r-- | ATRI/service/dormant.py | 19 | ||||
-rw-r--r-- | ATRI/service/httppost.py | 161 | ||||
-rw-r--r-- | ATRI/service/plugin.py | 60 | ||||
-rw-r--r-- | ATRI/service/send.py | 13 | ||||
-rw-r--r-- | ATRI/service/switch.py | 79 |
7 files changed, 395 insertions, 0 deletions
diff --git a/ATRI/service/__init__.py b/ATRI/service/__init__.py new file mode 100644 index 0000000..5dddb97 --- /dev/null +++ b/ATRI/service/__init__.py @@ -0,0 +1,10 @@ +import os +from pathlib import Path + + +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 diff --git a/ATRI/service/banlist.py b/ATRI/service/banlist.py new file mode 100644 index 0000000..06cc381 --- /dev/null +++ b/ATRI/service/banlist.py @@ -0,0 +1,53 @@ +import json +import aiofiles +from typing import Optional + +from ATRI.exceptions import InvalidWriteText + +from . import SERVICE_PATH + + +class BanList: + filename = 'banlist.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_banlist(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: + try: + cls.data[user] = user + 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!') + + @classmethod + async def del_list(cls, user: Optional[str]) -> None: + try: + del cls.data[user] + 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('List writing file failed!') diff --git a/ATRI/service/dormant.py b/ATRI/service/dormant.py new file mode 100644 index 0000000..7482d53 --- /dev/null +++ b/ATRI/service/dormant.py @@ -0,0 +1,19 @@ +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 new file mode 100644 index 0000000..d9b7360 --- /dev/null +++ b/ATRI/service/httppost.py @@ -0,0 +1,161 @@ +from ATRI.config import RUNTIME_CONFIG + + +URL = ( + f"http://{RUNTIME_CONFIG['http_post']['host']}" + f"{RUNTIME_CONFIG['http_post']['port']}" +) + +class HttpPost: + @classmethod + def send_private_msg(cls): + ... + + @classmethod + def send_group_msg(cls): + ... + + @classmethod + def send_msg(cls): + ... + + @classmethod + def delete_msg(cls): + ... + + @classmethod + def get_msg(cls): + ... + + @classmethod + def get_forward_msg(cls): + ... + + @classmethod + def send_like(cls): + ... + + @classmethod + def set_group_kick(cls): + ... + + @classmethod + def set_group_ban(cls): + ... + + @classmethod + def set_group_anonymous_ban(cls): + ... + + @classmethod + def set_group_whole_ban(cls): + ... + + @classmethod + def set_group_admin(cls): + ... + + @classmethod + def set_group_anonymous(cls): + ... + + @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): + ... +
\ No newline at end of file diff --git a/ATRI/service/plugin.py b/ATRI/service/plugin.py new file mode 100644 index 0000000..e14ef34 --- /dev/null +++ b/ATRI/service/plugin.py @@ -0,0 +1,60 @@ +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 new file mode 100644 index 0000000..9b9c3f9 --- /dev/null +++ b/ATRI/service/send.py @@ -0,0 +1,13 @@ +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 new file mode 100644 index 0000000..b97a0ec --- /dev/null +++ b/ATRI/service/switch.py @@ -0,0 +1,79 @@ +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!') |