summaryrefslogtreecommitdiff
path: root/utils
diff options
context:
space:
mode:
Diffstat (limited to 'utils')
-rw-r--r--utils/utils_banList/__init__.py55
-rw-r--r--utils/utils_banList/banList_group.json1
-rw-r--r--utils/utils_banList/banList_user.json1
-rw-r--r--utils/utils_error/__init__.py64
-rw-r--r--utils/utils_history/__init__.py112
-rw-r--r--utils/utils_request/__init__.py57
-rw-r--r--utils/utils_switch/__init__.py151
-rw-r--r--utils/utils_switch/switch.json1
8 files changed, 442 insertions, 0 deletions
diff --git a/utils/utils_banList/__init__.py b/utils/utils_banList/__init__.py
new file mode 100644
index 0000000..66142cc
--- /dev/null
+++ b/utils/utils_banList/__init__.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import json
+from pathlib import Path
+from typing import Optional
+
+def banList(user: str, group: Optional[str] = None) -> bool:
+ """
+ :说明:
+
+ 判断某一 用户/群 是否处于封禁名单中。
+
+ :参数:
+
+ * ``user: str``: 用户QQ号
+ * ``group: Optional[str] = None``: 用户所在群号，若不传入则只检测用户
+
+ :返回:
+
+ 是：False | 否：True
+
+ :用法:
+
+ .. code-block:: python
+
+ banList(user=123456789, group=123456789)
+
+ """
+ file_user = Path('.') / 'utils' / 'utils_banList' / 'banList_user.json'
+ file_group = Path('.') / 'utils' / 'utils_banList' / 'banList_group.json'
+
+ try:
+ with open(file_user, 'r') as f:
+ data_user = json.load(f)
+ except:
+ data_user = {}
+
+ try:
+ with open(file_group, 'r') as f:
+ data_group = json.load(f)
+ except:
+ data_group = {}
+
+ if user not in data_user:
+ if group:
+ if group not in data_group:
+ return True
+ else:
+ return False
+ else:
+ return True
+ else:
+ print(3)
+ return False \ No newline at end of file
diff --git a/utils/utils_banList/banList_group.json b/utils/utils_banList/banList_group.json
new file mode 100644
index 0000000..9e26dfe
--- /dev/null
+++ b/utils/utils_banList/banList_group.json
@@ -0,0 +1 @@
+{} \ No newline at end of file
diff --git a/utils/utils_banList/banList_user.json b/utils/utils_banList/banList_user.json
new file mode 100644
index 0000000..9e26dfe
--- /dev/null
+++ b/utils/utils_banList/banList_user.json
@@ -0,0 +1 @@
+{} \ No newline at end of file
diff --git a/utils/utils_error/__init__.py b/utils/utils_error/__init__.py
new file mode 100644
index 0000000..e037b30
--- /dev/null
+++ b/utils/utils_error/__init__.py
@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import json
+import string
+from pathlib import Path
+from random import sample
+from typing import Optional
+from datetime import datetime
+from traceback import format_exc
+
+from nonebot.rule import keyword
+
+def errorRepo(repo_msg: Optional[str] = None) -> str:
+ """
+ :说明:
+
+ 返回错误堆栈。
+
+ :参数:
+
+ * ``repo_msg: Optional[str] = None``: 此错误发生时指定的错误信息，若不传入则返回 unknown
+
+ :返回:
+
+ 错误信息
+
+ :用法:
+
+ .. code-block:: python
+
+ try:
+ ...
+ except Exception:
+ print(errorRepo(repo_msg="message"))
+
+ """
+ file_error = Path('.') / 'ATRI' / 'data' / 'data_Error' / 'error.json'
+ try:
+ with open(file_error, 'r') as f:
+ data_error = json.load(f)
+ except:
+ data_error = {}
+
+ key_error = ''.join(sample(string.ascii_letters + string.digits, 16))
+ msg_error = f"{datetime.now()}\n"
+ msg_error = f"{format_exc()}"
+ data_error[f"{key_error}"] = f"{msg_error}"
+
+ with open(file_error, 'w') as f:
+ f.write(json.dumps(data_error))
+ f.close()
+
+ if repo_msg:
+ pass
+ else:
+ repo_msg = 'unknown'
+
+ msg0 = f'ERROR! Reason: [{repo_msg}]\n'
+ msg0 += f'trackID: {key_error}\n'
+ msg0 += "请使用[来杯红茶]功能以联系维护者\n"
+ msg0 += "并附上 trackID"
+
+ return msg0 \ No newline at end of file
diff --git a/utils/utils_history/__init__.py b/utils/utils_history/__init__.py
new file mode 100644
index 0000000..d6a6d33
--- /dev/null
+++ b/utils/utils_history/__init__.py
@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import json
+from pathlib import Path
+from typing import Optional
+
+def saveMessage(message_id: str, message: str, user: str, group: Optional[str] = None) -> None:
+ """
+ :说明:
+
+ 获取信息并进行存储。
+ :参数:
+
+ * ``message_id: str``: 消息id
+ * ``message: str``: 目标信息
+ * ``user: str``: 发出用户
+ * ``group: Optional[str] = None``: 发出群号，若不传入则归入私聊消息
+
+ :返回:
+
+ None
+
+ :用法:
+
+ .. code-block:: python
+
+ getMessage(message='test', user=123456789, group=123456789)
+
+ """
+ file_group = Path('.') / 'ATRI' / 'data' / 'data_Group' / f'{group}' / 'message.json'
+ file_private = Path('.') / 'ATRI' / 'data' / 'data_Log' / 'message_private.json'
+
+ try:
+ with open(file_group, 'r') as f:
+ data_group = json.load(f)
+ except:
+ data_group = {}
+
+ try:
+ with open(file_private, 'r') as f:
+ data_private = json.load(f)
+ except:
+ data_private = {}
+
+ if group:
+ data_group[f"{message_id}"] = {"message": f"{message}", "user_id": f"{user}", "group_id": f"{group}"}
+
+ try:
+ with open(file_group, 'w') as f:
+ f.write(json.dumps(data_group))
+ f.close()
+ except:
+ os.mkdir(Path('.') / 'ATRI' / 'data' / 'data_Group' / f'{group}')
+ with open(file_group, 'w') as f:
+ f.write(json.dumps(data_group))
+ f.close()
+ else:
+ data_private[f"{message_id}"] = {"message": f"{message}", "user_id": f"{user}"}
+
+ try:
+ with open(file_private, 'w') as f:
+ f.write(json.dumps(data_private))
+ f.close()
+ except:
+ os.mkdir(Path('.') / 'ATRI' / 'data' / 'data_Log')
+ with open(file_private, 'w') as f:
+ f.write(json.dumps(data_private))
+ f.close()
+
+def getMessage(message_id: str, group: Optional[str] = None) -> dict:
+ """
+ :说明:
+
+ 通过 message_id 获取到对应消息参数: message, user, group
+
+ :参数:
+
+ * ``message_id: str``: 目标消息id
+ * ``group: Optional[str] = None``: 对应群号，若不传入则获取私聊消息
+
+ :返回:
+
+ 消息内容，类型为: dict
+
+ :用法:
+
+ .. code-block:: python
+
+ loadMessage(message_id=123456789)
+
+ """
+ file_group = Path('.') / 'ATRI' / 'data' / 'data_Group' / f'{group}' / 'message.json'
+ file_private = Path('.') / 'ATRI' / 'data' / 'data_Log' / 'message_private.json'
+
+ if group:
+ try:
+ with open(file_group, 'r') as f:
+ data_group = json.load(f)
+ return data_group
+
+ except:
+ return {"status": "None"}
+ else:
+ try:
+ with open(file_private, 'r') as f:
+ data_private = json.load(f)
+ return data_private
+
+ except:
+ return {"status": "None"} \ No newline at end of file
diff --git a/utils/utils_request/__init__.py b/utils/utils_request/__init__.py
new file mode 100644
index 0000000..cc0992a
--- /dev/null
+++ b/utils/utils_request/__init__.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import requests
+from typing import Optional
+from aiohttp import ClientSession
+
+def request_get(url: str, params: Optional[dict] = None) -> bytes:
+ """
+ :说明:
+
+ 通过 GET 方式请求 url。
+
+ :参数:
+
+ * ``url: str``: 目标网址
+ * ``params: Optional[dict] = None``: 参数，若不传入则为空
+
+ :返回:
+
+ requests.content
+
+ :用法:
+
+ .. code-block:: python
+
+ request_get(url="www.demo.com", params=params)
+
+ """
+ return requests.get(url, params).content
+
+async def aio_get_bytes(url: str, headers: Optional[dict] = None):
+ """
+ :说明:
+
+ 通过 GET 以 异步 方式请求 url。
+
+ :参数:
+
+ * ``url: str``: 目标网址
+ * ``headers: Optional[dict] = None``: 参数，若不传入则为空
+
+ :返回:
+
+ bytes
+
+ :用法:
+
+ .. code-block:: python
+
+ aio_get_bytes(url="www.demo.com", headers=headers)
+
+ """
+ async with ClientSession() as asyncSession:
+ async with asyncSession.get(url, headers=headers) as resp:
+ result = await resp.read()
+ return result \ No newline at end of file
diff --git a/utils/utils_switch/__init__.py b/utils/utils_switch/__init__.py
new file mode 100644
index 0000000..5c10a85
--- /dev/null
+++ b/utils/utils_switch/__init__.py
@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import json
+from pathlib import Path
+from typing import Optional
+
+def checkSwitch(func_name: str, group: str) -> bool:
+ """
+ :说明:
+
+ 判断此功能针对 群 或 全体 是否开启。
+
+ :参数:
+
+ * ``func_name: str``: 功能名称
+ * ``group: str``: 功能触发所在群号
+
+ :返回:
+
+ 是：True | 否：False
+
+ :用法:
+
+ .. code-block:: python
+
+ switch(func_name=Func, group=123456789)
+
+ """
+ file_switch_all = Path('.') / 'utils' / 'utils_switch' / 'switch.json'
+ file_switch_alone = Path('.') / 'ATRI' / 'data' / 'data_Group' / f'{group}' / 'switch.json'
+
+ try:
+ with open(file_switch_all, 'r') as f:
+ data_switch_all = json.load(f)
+ except:
+ data_switch_all = {}
+ data_switch_all["anime-setu"] = "True"
+ data_switch_all["anime-pic-search"] = "True"
+ data_switch_all["anime-vid-search"] = "True"
+ data_switch_all["ai-face"] = "True"
+
+ with open(file_switch_all, 'w') as f:
+ f.write(json.dumps(data_switch_all))
+ f.close()
+
+ try:
+ with open(file_switch_alone, 'r') as f:
+ data_switch_alone = json.load(f)
+ except:
+ data_switch_alone = {}
+ try:
+ os.mkdir(Path('.') / 'ATRI' / 'data' / 'data_Group' / f'{group}')
+ except:
+ pass
+
+ data_switch_alone["anime-setu"] = "True"
+ data_switch_alone["anime-pic-search"] = "True"
+ data_switch_alone["anime-vid-search"] = "True"
+ data_switch_alone["ai-face"] = "True"
+
+ with open(file_switch_alone, 'w') as f:
+ f.write(json.dumps(data_switch_alone))
+ f.close()
+
+ if data_switch_all[func_name] == "True":
+ if data_switch_alone[func_name] == "True":
+ return True
+ else:
+ return False
+ else:
+ return False
+
+def controlSwitch(func_name: str, control: bool, group: Optional[str] = None) -> str:
+ """
+ :说明:
+
+ 目标功能针对 群 或 全体 开启或关闭。
+
+ :参数:
+
+ * ``func_name: str``: 功能名称
+ * ``control: bool``: 开启 / 关闭
+ * ``group: Optional[str] = None``: 对应群号，若不传入则为全局
+
+ :返回:
+
+ None
+
+ :用法:
+
+ .. code-block:: python
+
+ controlSwitch(func_name=Func, group=123456789)
+
+ """
+ file_switch_all = Path('.') / 'utils' / 'utils_switch' / 'switch.json'
+
+ if group:
+ file_switch_group = Path('.') / 'ATRI' / 'data' / 'data_Group' / f'{group}' / 'switch.json'
+ try:
+ with open(file_switch_group, 'r') as f:
+ data_switch_group = json.load(f)
+ except:
+ data_switch_group = {}
+
+ if data_switch_group[f"{func_name}"]:
+ pass
+ else:
+ return f"Can't find func({func_name})"
+
+ data_switch_group[f"{func_name}"] = f"{control}"
+
+ with open(file_switch_group, 'w') as f:
+ f.write(json.dumps(data_switch_group))
+ f.close()
+
+ else:
+ pass
+
+ try:
+ with open(file_switch_all, 'r') as f:
+ data_switch_all = json.load(f)
+ except:
+ data_switch_all = {}
+
+ if data_switch_all[f"{func_name}"]:
+ pass
+ else:
+ return f"Can't find func({func_name})"
+
+ data_switch_all[f"{func_name}"] = f"{control}"
+
+ with open(file_switch_all, 'w') as f:
+ f.write(json.dumps(data_switch_all))
+ f.close()
+
+ if control == True:
+ if group:
+ msg = f"({func_name}) has been opened for group ({group})!"
+ else:
+ msg = f"({func_name}) has been opened!"
+
+ else:
+ if group:
+ msg = f"({func_name}) has been closed for group ({group})!"
+ else:
+ msg = f"({func_name}) has been closed!"
+
+ return msg \ No newline at end of file
diff --git a/utils/utils_switch/switch.json b/utils/utils_switch/switch.json
new file mode 100644
index 0000000..9b6bb22
--- /dev/null
+++ b/utils/utils_switch/switch.json
@@ -0,0 +1 @@
+{"anime-setu": "True", "anime-pic-search": "True", "anime-vid-search": "True", "all-off-anime-setu": "True"} \ No newline at end of file
Powered by cgit, Theme cgit-dark created by Ted Yin, updated by Mole Shang.