automas_admin/AUTO-MAS-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加通知测试接口

Browse Source
This commit is contained in:
DLmaster361
2025-09-10 20:33:05 +08:00
parent 6138fc47b1
commit d207c65df7
4 changed files with 61 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
app/api/setting.py
View File

@@ -26,7 +26,7 @@ import shutil
from fastapi import APIRouter, Body
from app.core import Config
from app.services import System
from app.services import System, Notify
from app.models.schema import *
router = APIRouter(prefix="/api/setting", tags=["全局设置"])
@@ -80,3 +80,18 @@ async def update_script(script: SettingUpdateIn = Body(...)) -> OutBase:
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
return OutBase()
@router.post(
"/test_notify", summary="测试通知", response_model=OutBase, status_code=200
)
async def test_notify() -> OutBase:
"""测试通知"""
try:
await Notify.send_test_notification()
except Exception as e:
return OutBase(
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
return OutBase()

24
app/services/notification.py
View File

@@ -41,7 +41,7 @@ class Notification:
def __init__(self):
super().__init__()
def push_plyer(self, title, message, ticker, t) -> bool:
async def push_plyer(self, title, message, ticker, t) -> bool:
"""
推送系统通知
@@ -71,7 +71,7 @@ class Notification:
return True
def send_mail(self, mode, title, content, to_address) -> None:
async def send_mail(self, mode, title, content, to_address) -> None:
"""
推送邮件通知
@@ -134,7 +134,7 @@ class Notification:
smtpObj.quit()
logger.success(f"邮件发送成功: {title}")
def ServerChanPush(self, title, content, send_key) -> None:
async def ServerChanPush(self, title, content, send_key) -> None:
"""
使用Server酱推送通知
@@ -170,7 +170,7 @@ class Notification:
else:
raise Exception(f"ServerChan 推送通知失败: {response.text}")
def WebHookPush(self, title, content, webhook_url) -> None:
async def WebHookPush(self, title, content, webhook_url) -> None:
"""
WebHook 推送通知
@@ -195,7 +195,9 @@ class Notification:
else:
raise Exception(f"WebHook 推送通知失败: {response.text}")
def CompanyWebHookBotPushImage(self, image_path: Path, webhook_url: str) -> None:
async def CompanyWebHookBotPushImage(
self, image_path: Path, webhook_url: str
) -> None:
"""
使用企业微信群机器人推送图片通知
@@ -232,13 +234,13 @@ class Notification:
else:
raise Exception(f"企业微信群机器人推送图片失败: {response.text}")
def send_test_notification(self) -> None:
async def send_test_notification(self) -> None:
"""发送测试通知到所有已启用的通知渠道"""
logger.info("发送测试通知到所有已启用的通知渠道")
# 发送系统通知
self.push_plyer(
await self.push_plyer(
"测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容, 说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作！",
"测试通知",
@@ -247,7 +249,7 @@ class Notification:
# 发送邮件通知
if Config.get("Notify", "IfSendMail"):
self.send_mail(
await self.send_mail(
"文本",
"AUTO_MAA测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容, 说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作！",
@@ -256,7 +258,7 @@ class Notification:
# 发送Server酱通知
if Config.get("Notify", "IfServerChan"):
self.ServerChanPush(
await self.ServerChanPush(
"AUTO_MAA测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容, 说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作！",
Config.get("Notify", "ServerChanKey"),
@@ -264,12 +266,12 @@ class Notification:
# 发送WebHook通知
if Config.get("Notify", "IfCompanyWebHookBot"):
self.WebHookPush(
await self.WebHookPush(
"AUTO_MAA测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容, 说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作！",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
Notify.CompanyWebHookBotPushImage(
await self.CompanyWebHookBotPushImage(
Path.cwd() / "res/images/notification/test_notify.png",
Config.get("Notify", "CompanyWebHookBotUrl"),
)

40
app/task/MAA.py
View File

@@ -594,7 +594,7 @@ class MaaManager:
self.if_open_emulator = True
# 推送异常通知
Notify.push_plyer(
await Notify.push_plyer(
"用户自动代理出现异常！",
f"用户 {user['name']} 的{MOOD_BOOK[mode]}部分出现一次异常",
f"{user['name']}的{MOOD_BOOK[mode]}出现异常",
@@ -914,7 +914,7 @@ class MaaManager:
logger.success(
f"用户 {self.user_list[self.index]['user_id']} 的自动代理任务已完成"
)
Notify.push_plyer(
await Notify.push_plyer(
"成功完成一个自动代理任务！",
f"已完成用户 {self.user_list[self.index]['name']} 的自动代理任务",
f"已完成 {self.user_list[self.index]['name']} 的自动代理任务",
@@ -1055,7 +1055,7 @@ class MaaManager:
result_text += f"\n未开始{self.mode}的用户: \n{"\n".join(result["waiting_user"])}\n"
# 推送代理结果通知
Notify.push_plyer(
await Notify.push_plyer(
title.replace("报告", "已完成！"),
f"已完成用户数: {len(over_user)}, 未完成用户数: {len(error_user) + len(wait_user)}",
f"已完成用户数: {len(over_user)}, 未完成用户数: {len(error_user) + len(wait_user)}",
@@ -1901,19 +1901,19 @@ class MaaManager:
# 发送全局通知
if Config.get("Notify", "IfSendMail"):
Notify.send_mail(
await Notify.send_mail(
"网页", title, message_html, Config.get("Notify", "ToAddress")
)
if Config.get("Notify", "IfServerChan"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
@@ -1955,19 +1955,19 @@ class MaaManager:
if Config.get("Notify", "IfSendStatistic"):
if Config.get("Notify", "IfSendMail"):
Notify.send_mail(
await Notify.send_mail(
"网页", title, message_html, Config.get("Notify", "ToAddress")
)
if Config.get("Notify", "IfServerChan"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
@@ -1981,7 +1981,7 @@ class MaaManager:
# 发送邮件通知
if self.cur_user_data.get("Notify", "IfSendMail"):
if self.cur_user_data.get("Notify", "ToAddress"):
Notify.send_mail(
await Notify.send_mail(
"网页",
title,
message_html,
@@ -1993,7 +1993,7 @@ class MaaManager:
# 发送ServerChan通知
if self.cur_user_data.get("Notify", "IfServerChan"):
if self.cur_user_data.get("Notify", "ServerChanKey"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
self.cur_user_data.get("Notify", "ServerChanKey"),
@@ -2006,7 +2006,7 @@ class MaaManager:
# 推送CompanyWebHookBot通知
if self.cur_user_data.get("Notify", "IfCompanyWebHookBot"):
if self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
@@ -2027,24 +2027,24 @@ class MaaManager:
if Config.get("Notify", "IfSendSixStar"):
if Config.get("Notify", "IfSendMail"):
Notify.send_mail(
await Notify.send_mail(
"网页", title, message_html, Config.get("Notify", "ToAddress")
)
if Config.get("Notify", "IfServerChan"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
Notify.CompanyWebHookBotPushImage(
await Notify.CompanyWebHookBotPushImage(
Path.cwd() / "res/images/notification/six_star.png",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
@@ -2057,7 +2057,7 @@ class MaaManager:
# 发送邮件通知
if self.cur_user_data.get("Notify", "IfSendMail"):
if self.cur_user_data.get("Notify", "ToAddress"):
Notify.send_mail(
await Notify.send_mail(
"网页",
title,
message_html,
@@ -2070,7 +2070,7 @@ class MaaManager:
if self.cur_user_data.get("Notify", "IfServerChan"):
if self.cur_user_data.get("Notify", "ServerChanKey"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
self.cur_user_data.get("Notify", "ServerChanKey"),
@@ -2083,12 +2083,12 @@ class MaaManager:
# 推送CompanyWebHookBot通知
if self.cur_user_data.get("Notify", "IfCompanyWebHookBot"):
if self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
)
Notify.CompanyWebHookBotPushImage(
await Notify.CompanyWebHookBotPushImage(
Path.cwd() / "res/images/notification/six_star.png",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
)

24
app/task/general.py
View File

@@ -454,7 +454,7 @@ class GeneralManager:
await System.kill_process(self.game_path)
# 推送异常通知
Notify.push_plyer(
await Notify.push_plyer(
"用户自动代理出现异常！",
f"用户 {user['name']} 的自动代理出现一次异常",
f"{user['name']} 的自动代理出现异常",
@@ -576,7 +576,7 @@ class GeneralManager:
logger.success(
f"用户 {self.user_list[self.index]['user_id']} 的自动代理任务已完成"
)
Notify.push_plyer(
await Notify.push_plyer(
"成功完成一个自动代理任务！",
f"已完成用户 {self.user_list[self.index]['name']} 的自动代理任务",
f"已完成 {self.user_list[self.index]['name']} 的自动代理任务",
@@ -703,7 +703,7 @@ class GeneralManager:
result_text += f"\n未开始{self.mode}的用户: \n{"\n".join(result['waiting_user'])}\n"
# 推送代理结果通知
Notify.push_plyer(
await Notify.push_plyer(
title.replace("报告", "已完成！"),
f"已完成配置数: {len(over_user)}, 未完成配置数: {len(error_user) + len(wait_user)}",
f"已完成配置数: {len(over_user)}, 未完成配置数: {len(error_user) + len(wait_user)}",
@@ -979,19 +979,19 @@ class GeneralManager:
# 发送全局通知
if Config.get("Notify", "IfSendMail"):
Notify.send_mail(
await Notify.send_mail(
"网页", title, message_html, Config.get("Notify", "ToAddress")
)
if Config.get("Notify", "IfServerChan"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
@@ -1016,19 +1016,19 @@ class GeneralManager:
if Config.get("Notify", "IfSendStatistic"):
if Config.get("Notify", "IfSendMail"):
Notify.send_mail(
await Notify.send_mail(
"网页", title, message_html, Config.get("Notify", "ToAddress")
)
if Config.get("Notify", "IfServerChan"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
@@ -1042,7 +1042,7 @@ class GeneralManager:
# 发送邮件通知
if self.cur_user_data.get("Notify", "IfSendMail"):
if self.cur_user_data.get("Notify", "ToAddress"):
Notify.send_mail(
await Notify.send_mail(
"网页",
title,
message_html,
@@ -1054,7 +1054,7 @@ class GeneralManager:
# 发送ServerChan通知
if self.cur_user_data.get("Notify", "IfServerChan"):
if self.cur_user_data.get("Notify", "ServerChanKey"):
Notify.ServerChanPush(
await Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
self.cur_user_data.get("Notify", "ServerChanKey"),
@@ -1067,7 +1067,7 @@ class GeneralManager:
# 推送CompanyWebHookBot通知
if self.cur_user_data.get("Notify", "IfCompanyWebHookBot"):
if self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"):
Notify.WebHookPush(
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),