summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorLint Action <[email protected]>2021-08-13 09:53:08 +0000
committerLint Action <[email protected]>2021-08-13 09:53:08 +0000
commitae903cfe86a35fb0f276c51f666cd3b3d1404feb (patch)
tree66ca215c720d91f17c382740e03f9b11c8eee8dc
parent24644daeb153e781d162c638a79b3727df242802 (diff)
downloadATRI-ae903cfe86a35fb0f276c51f666cd3b3d1404feb.tar.gz
ATRI-ae903cfe86a35fb0f276c51f666cd3b3d1404feb.tar.bz2
ATRI-ae903cfe86a35fb0f276c51f666cd3b3d1404feb.zip
:rotating_light: 自动进行代码格式化
-rw-r--r--ATRI/plugins/manage/__init__.py108
-rw-r--r--ATRI/plugins/manage/data_source.py57
2 files changed, 101 insertions, 64 deletions
diff --git a/ATRI/plugins/manage/__init__.py b/ATRI/plugins/manage/__init__.py
index c1dfb3b..0be7b83 100644
--- a/ATRI/plugins/manage/__init__.py
+++ b/ATRI/plugins/manage/__init__.py
@@ -10,137 +10,152 @@ from .data_source import Manage
block_user = Manage().on_command("封禁用户", "对目标用户进行封禁", permission=SUPERUSER)
+
@block_user.handle()
async def _ready_block_user(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["block_user"] = msg
+
@block_user.got("block_user", "哪位?GKD!")
async def _deal_block_user(bot: Bot, event: MessageEvent, state: T_State):
user_id = state["block_user"]
quit_list = ["算了", "罢了"]
if user_id in quit_list:
await block_user.finish("...看来有人逃过一劫呢")
-
+
is_ok = Manage().block_user(user_id)
if not is_ok:
await block_user.finish("kuso!封禁失败了...")
-
+
await block_user.finish(f"用户 {user_id} 危!")
unblock_user = Manage().on_command("解封用户", "对目标用户进行解封", permission=SUPERUSER)
+
@unblock_user.handle()
async def _ready_unblock_user(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["unblock_user"] = msg
+
@unblock_user.got("unblock_user", "哪位?GKD!")
async def _deal_unblock_user(bot: Bot, event: MessageEvent, state: T_State):
user_id = state["unblock_user"]
quit_list = ["算了", "罢了"]
if user_id in quit_list:
await unblock_user.finish("...有人又得继续在小黑屋呆一阵子了")
-
+
is_ok = Manage().unblock_user(user_id)
if not is_ok:
await unblock_user.finish("kuso!解封失败了...")
-
+
await unblock_user.finish(f"好欸!{user_id} 重获新生!")
block_group = Manage().on_command("封禁群", "对目标群进行封禁", permission=SUPERUSER)
+
@block_group.handle()
async def _ready_block_group(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["block_group"] = msg
+
@block_group.got("block_group", "哪个群?GKD!")
async def _deal_block_group(bot: Bot, event: MessageEvent, state: T_State):
group_id = state["block_group"]
quit_list = ["算了", "罢了"]
if group_id in quit_list:
await block_group.finish("...看来有一群逃过一劫呢")
-
+
is_ok = Manage().block_group(group_id)
if not is_ok:
await block_group.finish("kuso!封禁失败了...")
-
+
await block_group.finish(f"群 {group_id} 危!")
unblock_group = Manage().on_command("解封群", "对目标群进行解封", permission=SUPERUSER)
+
@unblock_group.handle()
async def _ready_unblock_group(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["unblock_group"] = msg
+
@unblock_group.got("unblock_group", "哪个群?GKD!")
async def _deal_unblock_group(bot: Bot, event: MessageEvent, state: T_State):
group_id = state["unblock_group"]
quit_list = ["算了", "罢了"]
if group_id in quit_list:
await unblock_group.finish("...有一群又得继续在小黑屋呆一阵子了")
-
+
is_ok = Manage().unblock_group(group_id)
if not is_ok:
await unblock_group.finish("kuso!解封失败了...")
-
+
await unblock_group.finish(f"好欸!群 {group_id} 重获新生!")
global_block_service = Manage().on_command("全局禁用", "全局禁用某服务", permission=SUPERUSER)
+
@global_block_service.handle()
async def _ready_block_service(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["global_block_service"] = msg
+
@global_block_service.got("global_block_service", "阿...是哪个服务呢")
async def _deal_global_block_service(bot: Bot, event: MessageEvent, state: T_State):
block_service = state["global_block_service"]
quit_list = ["算了", "罢了"]
if block_service in quit_list:
await global_block_service.finish("好吧...")
-
+
is_ok = Manage().control_global_service(block_service, False)
if not is_ok:
await global_block_service.finish("kuso!禁用失败了...")
-
+
await global_block_service.finish(f"服务 {block_service} 已被禁用")
global_unblock_service = Manage().on_command("全局启用", "全局启用某服务", permission=SUPERUSER)
+
@global_unblock_service.handle()
async def _ready_unblock_service(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["global_unblock_service"] = msg
+
@global_unblock_service.got("global_unblock_service", "阿...是哪个服务呢")
async def _deal_global_unblock_service(bot: Bot, event: MessageEvent, state: T_State):
unblock_service = state["global_unblock_service"]
quit_list = ["算了", "罢了"]
if unblock_service in quit_list:
await global_unblock_service.finish("好吧...")
-
+
is_ok = Manage().control_global_service(unblock_service, True)
if not is_ok:
await global_unblock_service.finish("kuso!启用服务失败了...")
-
+
await global_unblock_service.finish(f"服务 {unblock_service} 已启用")
-user_block_service = Manage().on_regex(r"对用户(.*?)禁用(.*)", "针对某一用户禁用服务", permission=SUPERUSER)
+user_block_service = Manage().on_regex(
+ r"对用户(.*?)禁用(.*)", "针对某一用户禁用服务", permission=SUPERUSER
+)
+
@user_block_service.handle()
async def _user_block_service(bot: Bot, event: MessageEvent):
@@ -149,15 +164,17 @@ async def _user_block_service(bot: Bot, event: MessageEvent):
reg = re.findall(pattern, msg)
aim_user = reg[0]
aim_service = reg[1]
-
+
is_ok = Manage().control_user_service(aim_service, aim_user, False)
if not is_ok:
await user_block_service.finish("禁用失败...请检查服务名是否正确")
await user_block_service.finish(f"完成~已禁止用户 {aim_user} 使用 {aim_service}")
-
-user_unblock_service = Manage().on_regex(r"对用户(.*?)启用(.*)", "针对某一用户启用服务", permission=SUPERUSER)
+user_unblock_service = Manage().on_regex(
+ r"对用户(.*?)启用(.*)", "针对某一用户启用服务", permission=SUPERUSER
+)
+
@user_unblock_service.handle()
async def _user_unblock_service(bot: Bot, event: MessageEvent):
@@ -166,21 +183,27 @@ async def _user_unblock_service(bot: Bot, event: MessageEvent):
reg = re.findall(pattern, msg)
aim_user = reg[0]
aim_service = reg[1]
-
+
is_ok = Manage().control_user_service(aim_service, aim_user, True)
if not is_ok:
await user_unblock_service.finish("启用失败...请检查服务名是否正确,或者此人并不存在于名单中")
await user_unblock_service.finish(f"完成~已允许用户 {aim_user} 使用 {aim_service}")
-group_block_service = Manage().on_command("禁用", "针对所在群禁用某服务", permission=SUPERUSER | GROUP_OWNER | GROUP_ADMIN)
+group_block_service = Manage().on_command(
+ "禁用", "针对所在群禁用某服务", permission=SUPERUSER | GROUP_OWNER | GROUP_ADMIN
+)
+
@group_block_service.handle()
-async def _ready_group_block_service(bot: Bot, event: GroupMessageEvent, state: T_State):
+async def _ready_group_block_service(
+ bot: Bot, event: GroupMessageEvent, state: T_State
+):
msg = str(event.message).strip()
if msg:
state["group_block_service"] = msg
+
@group_block_service.got("group_block_service", "阿...是哪个服务呢")
async def _deal_group_block_service(bot: Bot, event: GroupMessageEvent, state: T_State):
aim_service = state["group_block_service"]
@@ -188,29 +211,37 @@ async def _deal_group_block_service(bot: Bot, event: GroupMessageEvent, state: T
quit_list = ["算了", "罢了"]
if aim_service in quit_list:
await group_block_service.finish("好吧...")
-
+
is_ok = Manage().control_group_service(aim_service, group_id, False)
if not is_ok:
await group_block_service.finish("禁用失败...请检查服务名是否输入正确")
await group_block_service.finish(f"完成!~已禁止本群使用服务:{aim_service}")
-group_unblock_service = Manage().on_command("启用", "针对所在群启用某服务", permission=SUPERUSER | GROUP_OWNER | GROUP_ADMIN)
+group_unblock_service = Manage().on_command(
+ "启用", "针对所在群启用某服务", permission=SUPERUSER | GROUP_OWNER | GROUP_ADMIN
+)
+
@group_unblock_service.handle()
-async def _ready_group_unblock_service(bot: Bot, event: GroupMessageEvent, state: T_State):
+async def _ready_group_unblock_service(
+ bot: Bot, event: GroupMessageEvent, state: T_State
+):
msg = str(event.message).strip()
if msg:
state["group_unblock_service"] = msg
+
@group_unblock_service.got("group_unblock_service", "阿...是哪个服务呢")
-async def _deal_group_unblock_service(bot: Bot, event: GroupMessageEvent, state: T_State):
+async def _deal_group_unblock_service(
+ bot: Bot, event: GroupMessageEvent, state: T_State
+):
aim_service = state["group_unblock_service"]
group_id = str(event.group_id)
quit_list = ["算了", "罢了"]
if aim_service in quit_list:
await group_unblock_service.finish("好吧...")
-
+
is_ok = Manage().control_group_service(aim_service, group_id, True)
if not is_ok:
await group_unblock_service.finish("启用失败...请检查服务名是否输入正确,或群不存在于名单中")
@@ -219,6 +250,7 @@ async def _deal_group_unblock_service(bot: Bot, event: GroupMessageEvent, state:
get_friend_add_list = Manage().on_command("获取好友申请", "获取好友申请列表", permission=SUPERUSER)
+
@get_friend_add_list.handle()
async def _get_friend_add_list(bot: Bot, event: MessageEvent):
data = Manage().load_friend_apply_list()
@@ -233,23 +265,25 @@ async def _get_friend_add_list(bot: Bot, event: MessageEvent):
msg0 = "申请人ID | 申请信息 | 申请码\n" + "\n".join(map(str, temp_list))
msg1 = msg0 + "\nTip: 使用 同意/拒绝好友 [申请码] 以决定"
await get_friend_add_list.finish(msg1)
-
+
approve_friend_add = Manage().on_command("同意好友", "同意好友申请", permission=SUPERUSER)
+
@approve_friend_add.handle()
async def _ready_approve_friend_add(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["approve_friend_add"]
+
@approve_friend_add.got("approve_friend_add", "申请码GKD!")
async def _deal_approve_friend_add(bot: Bot, event: MessageEvent, state: T_State):
apply_code = state["approve_friend_add"]
quit_list = ["算了", "罢了"]
if apply_code in quit_list:
await approve_friend_add.finish("好吧...")
-
+
try:
await bot.set_friend_add_request(flag=apply_code, approve=True)
except BaseException:
@@ -262,19 +296,21 @@ async def _deal_approve_friend_add(bot: Bot, event: MessageEvent, state: T_State
refuse_friend_add = Manage().on_command("拒绝好友", "拒绝好友申请", permission=SUPERUSER)
+
@refuse_friend_add.handle()
async def _ready_refuse_friend_add(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["refuse_friend_add"]
+
@refuse_friend_add.got("refuse_friend_add", "申请码GKD!")
async def _deal_refuse_friend_add(bot: Bot, event: MessageEvent, state: T_State):
apply_code = state["refuse_friend_add"]
quit_list = ["算了", "罢了"]
if apply_code in quit_list:
await refuse_friend_add.finish("好吧...")
-
+
try:
await bot.set_friend_add_request(flag=apply_code, approve=False)
except BaseException:
@@ -287,6 +323,7 @@ async def _deal_refuse_friend_add(bot: Bot, event: MessageEvent, state: T_State)
get_group_invite_list = Manage().on_command("获取邀请列表", "获取群邀请列表", permission=SUPERUSER)
+
@get_group_invite_list.handle()
async def _get_group_invite_list(bot: Bot, event: MessageEvent):
data = Manage().load_invite_apply_list()
@@ -305,21 +342,25 @@ async def _get_group_invite_list(bot: Bot, event: MessageEvent):
approve_group_invite = Manage().on_command("同意邀请", "同意群聊邀请", permission=SUPERUSER)
+
@approve_group_invite.handle()
async def _ready_approve_group_invite(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["approve_group_invite"]
+
@approve_group_invite.got("approve_group_invite", "申请码GKD!")
async def _deal_approve_group_invite(bot: Bot, event: MessageEvent, state: T_State):
apply_code = state["approve_group_invite"]
quit_list = ["算了", "罢了"]
if apply_code in quit_list:
await approve_group_invite.finish("好吧...")
-
+
try:
- await bot.set_group_add_request(flag=apply_code, sub_type="invite", approve=True)
+ await bot.set_group_add_request(
+ flag=apply_code, sub_type="invite", approve=True
+ )
except BaseException:
await approve_group_invite.finish("同意失败...尝试下手动?")
data = Manage().load_invite_apply_list()
@@ -330,21 +371,25 @@ async def _deal_approve_group_invite(bot: Bot, event: MessageEvent, state: T_Sta
refuse_group_invite = Manage().on_command("拒绝邀请", "拒绝群聊邀请", permission=SUPERUSER)
+
@refuse_group_invite.handle()
async def _ready_refuse_group_invite(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.message).strip()
if msg:
state["refuse_group_invite"]
+
@refuse_group_invite.got("refuse_group_invite", "申请码GKD!")
async def _deal_refuse_group_invite(bot: Bot, event: MessageEvent, state: T_State):
apply_code = state["refuse_group_invite"]
quit_list = ["算了", "罢了"]
if apply_code in quit_list:
await refuse_group_invite.finish("好吧...")
-
+
try:
- await bot.set_group_add_request(flag=apply_code, sub_type="invite", approve=False)
+ await bot.set_group_add_request(
+ flag=apply_code, sub_type="invite", approve=False
+ )
except BaseException:
await refuse_group_invite.finish("拒绝失败...尝试下手动?")
data = Manage().load_invite_apply_list()
@@ -355,6 +400,7 @@ async def _deal_refuse_group_invite(bot: Bot, event: MessageEvent, state: T_Stat
track_error = Manage().on_command("追踪", "获取报错信息,传入追踪码", aliases={"/track"})
+
@track_error.handle()
async def _track_error(bot: Bot, event: MessageEvent):
track_id = str(event.message).strip()
diff --git a/ATRI/plugins/manage/data_source.py b/ATRI/plugins/manage/data_source.py
index 74d81d4..1897d03 100644
--- a/ATRI/plugins/manage/data_source.py
+++ b/ATRI/plugins/manage/data_source.py
@@ -28,7 +28,6 @@ __doc__ = """
class Manage(Service):
-
def __init__(self):
Service.__init__(self, "管理", __doc__, True)
@@ -53,8 +52,7 @@ class Manage(Service):
except BaseException:
data = dict()
return data
-
-
+
@staticmethod
def _save_block_user_list(data: dict) -> None:
file_name = "block_user.json"
@@ -62,10 +60,10 @@ class Manage(Service):
if not path.is_file():
with open(path, "w", encoding="utf-8") as w:
w.write(json.dumps({}))
-
+
with open(path, "w", encoding="utf-8") as w:
w.write(json.dumps(data, indent=4))
-
+
@staticmethod
def _load_block_group_list() -> dict:
"""
@@ -82,13 +80,13 @@ class Manage(Service):
with open(path, "w", encoding="utf-8") as w:
w.write(json.dumps({}))
return dict()
-
+
try:
data = json.loads(path.read_bytes())
except BaseException:
data = dict()
return data
-
+
@staticmethod
def _save_block_group_list(data: dict) -> None:
file_name = "block_group.json"
@@ -96,55 +94,51 @@ class Manage(Service):
if not path.is_file():
with open(path, "w", encoding="utf-8") as w:
w.write(json.dumps({}))
-
+
with open(path, "w", encoding="utf-8") as w:
w.write(json.dumps(data, indent=4))
-
+
@classmethod
def block_user(cls, user_id: str) -> bool:
data = cls._load_block_user_list()
now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
- data[user_id] = {
- "time": now_time
- }
+ data[user_id] = {"time": now_time}
try:
cls._save_block_user_list(data)
return True
except BaseException:
return False
-
+
@classmethod
def unblock_user(cls, user_id: str) -> bool:
data: dict = cls._load_block_user_list()
if user_id not in data:
return False
-
+
try:
data.pop(user_id)
cls._save_block_user_list(data)
return True
except BaseException:
return False
-
+
@classmethod
def block_group(cls, group_id: str) -> bool:
data = cls._load_block_group_list()
now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
- data[group_id] = {
- "time": now_time
- }
+ data[group_id] = {"time": now_time}
try:
cls._save_block_group_list(data)
return True
except BaseException:
return False
-
+
@classmethod
def unblock_group(cls, group_id: str) -> bool:
data: dict = cls._load_block_group_list()
if group_id not in data:
return False
-
+
try:
data.pop(group_id)
cls._save_block_group_list(data)
@@ -164,7 +158,7 @@ class Manage(Service):
data["enabled"] = is_enabled
ServiceTools().save_service(data, service)
return True
-
+
@staticmethod
def control_user_service(service: str, user_id: str, is_enabled: bool) -> bool:
"""
@@ -175,7 +169,7 @@ class Manage(Service):
except BaseException:
return False
temp_list: list = data.get("disable_user", list())
-
+
if is_enabled:
try:
temp_list.remove(user_id)
@@ -186,7 +180,7 @@ class Manage(Service):
data["disable_user"] = temp_list
ServiceTools().save_service(data, service)
return True
-
+
@staticmethod
def control_group_service(service: str, group_id: str, is_enabled: bool) -> bool:
"""
@@ -198,7 +192,7 @@ class Manage(Service):
except BaseException:
return False
temp_list: list = data.get("disable_group", list())
-
+
if is_enabled:
try:
temp_list.remove(group_id)
@@ -224,7 +218,7 @@ class Manage(Service):
except BaseException:
data = dict()
return data
-
+
@staticmethod
def save_friend_apply_list(data: dict) -> None:
file_name = "friend_add.json"
@@ -235,7 +229,7 @@ class Manage(Service):
with open(path, "w", encoding="utf-8") as w:
w.write(json.dumps(data, indent=4))
-
+
@staticmethod
def load_invite_apply_list() -> dict:
file_name = "group_invite.json"
@@ -250,7 +244,7 @@ class Manage(Service):
except BaseException:
data = dict()
return data
-
+
@staticmethod
def save_invite_apply_list(data: dict) -> None:
file_name = "group_invite.json"
@@ -268,16 +262,13 @@ class Manage(Service):
data = load_error(track_id)
except ReadFileError:
return "请检查ID是否正确..."
-
+
prompt = data.get("prompt", "ignore")
time = data.get("time", "ignore")
content = data.get("content", "ignore")
-
+
msg0 = TRACK_BACK_FORMAT.format(
- track_id=track_id,
- prompt=prompt,
- time=time,
- content=content
+ track_id=track_id, prompt=prompt, time=time, content=content
)
repo = f"详细请移步此处~\n{await UbuntuPaste(content=msg0).paste()}"
return repo