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