summaryrefslogtreecommitdiff
path: root/ATRI/service
diff options
context:
space:
mode:
Diffstat (limited to 'ATRI/service')
-rw-r--r--ATRI/service/__init__.py10
-rw-r--r--ATRI/service/banlist.py53
-rw-r--r--ATRI/service/dormant.py19
-rw-r--r--ATRI/service/httppost.py161
-rw-r--r--ATRI/service/plugin.py60
-rw-r--r--ATRI/service/send.py13
-rw-r--r--ATRI/service/switch.py79
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!')
Powered by cgit, Theme cgit-dark created by Ted Yin, updated by Mole Shang.