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

feat(core): 初步完成新增用户单独通知功能

Browse Source
This commit is contained in:
aoxuan
2025-05-18 01:57:48 +08:00
parent c0f887ff9d
commit 0d904b229e
4 changed files with 381 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
app/core/config.py
View File

@@ -439,6 +439,20 @@ class MaaUserConfig(LQConfig):
"Data", "CustomInfrastPlanIndex", "0"
)
# 新增用户单独通知字段
self.Notify_Enable = ConfigItem("Notify_Enable", False, False)
self.Notify_IfSMTP = ConfigItem("Notify_IfSMTP", False, False)
self.Notify_SMTPServerAddress = ConfigItem("Notify_SMTPServerAddress", "", "")
self.Notify_AuthorizationCode = ConfigItem("Notify_AuthorizationCode", "", "")
self.Notify_FromAddress = ConfigItem("Notify_FromAddress", "", "")
self.Notify_ToAddress = ConfigItem("Notify_ToAddress", "", "")
self.Notify_IfServerChan = ConfigItem("Notify_IfServerChan", False, False)
self.Notify_ServerChanKey = ConfigItem("Notify_ServerChanKey", "", "")
self.Notify_ServerChanChannel = ConfigItem("Notify_ServerChanChannel", "", "")
self.Notify_ServerChanTag = ConfigItem("Notify_ServerChanTag", "", "")
self.Notify_IfCompanyWebHookBot = ConfigItem("Notify_IfCompanyWebHookBot", False, False)
self.Notify_CompanyWebHookBotUrl = ConfigItem("Notify_CompanyWebHookBotUrl", "", "")
class MaaPlanConfig(LQConfig):
"""MAA计划表配置"""

41
app/models/MAA.py
View File

@@ -40,6 +40,7 @@ from typing import Union, List, Dict
from app.core import Config, MaaConfig, MaaUserConfig
from app.services import Notify, System
from app.services.notification import Notification, UserNotification
class MaaManager(QObject):
@@ -1746,7 +1747,6 @@ class MaaManager(QObject):
)
if mode == "代理结果":
# 生成文本通知内容
message_text = (
f"任务开始时间：{message["start_time"]}，结束时间：{message["end_time"]}\n"
@@ -1768,14 +1768,26 @@ class MaaManager(QObject):
template = env.get_template("MAA_result.html")
message_html = template.render(message)
# 发送全局通知
Notify.send_mail("网页", title, message_html)
Notify.ServerChanPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
# 发送用户单独通知
for user_name in message["failed_user"].split("、") + message["waiting_user"].split("、"):
if not user_name: # 跳过空字符串
continue
user_config = Config.member_dict.get(user_name)
if user_config and user_config.get(user_config.Notify_Enable):
user_notify = UserNotification(user_config)
user_notify.send_notification(
f"{self.mode[2:4]}任务未完成通知",
f"您的{self.mode[2:4]}任务未完成,请检查相关设置。\n\n{message_text}"
)
return message_text
elif mode == "统计信息":
# 生成文本通知内容
formatted = []
for stage, items in message["drop_statistics"].items():
@@ -1801,18 +1813,41 @@ class MaaManager(QObject):
template = env.get_template("MAA_statistics.html")
message_html = template.render(message)
# 发送全局通知
Notify.send_mail("网页", title, message_html)
# ServerChan的换行是两个换行符。故而将\n替换为\n\n
serverchan_message = message_text.replace("\n", "\n\n")
Notify.ServerChanPush(title, f"{serverchan_message}\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, f"{message_text}\n\nAUTO_MAA 敬上")
elif mode == "公招六星":
# 发送用户单独通知
user_name = message.get("user_info")
if user_name:
user_config = Config.member_dict.get(user_name)
if user_config and user_config.get(user_config.Notify_Enable):
user_notify = UserNotification(user_config)
user_notify.send_notification(
f"{self.mode[2:4]}任务统计报告",
f"您的{self.mode[2:4]}任务统计报告如下:\n\n{message_text}"
)
elif mode == "公招六星":
# 生成HTML通知内容
template = env.get_template("MAA_six_star.html")
message_html = template.render(message)
# 发送全局通知
Notify.send_mail("网页", title, message_html)
Notify.ServerChanPush(title, "好羡慕~\n\nAUTO_MAA 敬上")
Notify.CompanyWebHookBotPush(title, "好羡慕~\n\nAUTO_MAA 敬上")
# 发送用户单独通知
user_name = message.get("user_name")
if user_name:
user_config = Config.member_dict.get(user_name)
if user_config and user_config.get(user_config.Notify_Enable):
user_notify = UserNotification(user_config)
user_notify.send_notification(
"公招六星通知",
"恭喜您在公招中获得了六星干员!"
)

122
app/services/notification.py
View File

@@ -313,4 +313,126 @@ class Notification(QWidget):
return True
class UserNotification:
"""用户单独通知服务"""
def __init__(self, user_config):
self.config = user_config
def send_notification(self, title: str, content: str) -> bool:
"""发送用户通知
Args:
title: 通知标题
content: 通知内容
Returns:
bool: 是否发送成功
"""
if not self.config.get(self.config.Notify_Enable):
return False
success = False
# 发送邮件通知
if self._check_smtp_config():
try:
self._send_email(title, content)
success = True
except Exception as e:
logger.error(f"发送邮件通知失败: {str(e)}")
# 发送ServerChan通知
if self.config.get(self.config.Notify_IfServerChan) and self._check_serverchan_config():
try:
self._send_serverchan(title, content)
success = True
except Exception as e:
logger.error(f"发送ServerChan通知失败: {str(e)}")
# 发送企业微信机器人通知
if self.config.get(self.config.Notify_IfCompanyWebHookBot) and self._check_webhook_config():
try:
self._send_webhook(title, content)
success = True
except Exception as e:
logger.error(f"发送企业微信机器人通知失败: {str(e)}")
return success
def _check_smtp_config(self) -> bool:
"""检查SMTP配置是否完整"""
return all([
self.config.get(self.config.Notify_IfSMTP),
self.config.get(self.config.Notify_SMTPServerAddress),
self.config.get(self.config.Notify_AuthorizationCode),
self.config.get(self.config.Notify_FromAddress),
self.config.get(self.config.Notify_ToAddress)
])
def _check_serverchan_config(self) -> bool:
"""检查ServerChan配置是否完整"""
return bool(self.config.get(self.config.Notify_ServerChanKey))
def _check_webhook_config(self) -> bool:
"""检查企业微信机器人配置是否完整"""
return bool(self.config.get(self.config.Notify_CompanyWebHookBotUrl))
def _send_email(self, title: str, content: str):
"""发送邮件通知"""
import smtplib
from email.mime.text import MIMEText
from email.header import Header
msg = MIMEText(content, 'plain', 'utf-8')
msg['Subject'] = Header(title, 'utf-8')
msg['From'] = self.config.get(self.config.Notify_FromAddress)
msg['To'] = self.config.get(self.config.Notify_ToAddress)
server = smtplib.SMTP_SSL(self.config.get(self.config.Notify_SMTPServerAddress))
server.login(
self.config.get(self.config.Notify_FromAddress),
self.config.get(self.config.Notify_AuthorizationCode)
)
server.send_message(msg)
server.quit()
def _send_serverchan(self, title: str, content: str):
"""发送ServerChan通知"""
import requests
key = self.config.get(self.config.Notify_ServerChanKey)
channel = self.config.get(self.config.Notify_ServerChanChannel)
tag = self.config.get(self.config.Notify_ServerChanTag)
url = f"https://sctapi.ftqq.com/{key}.send"
data = {
"title": title,
"desp": content,
"channel": channel if channel else 9, # 默认使用企业微信通道
"tag": tag if tag else ""
}
response = requests.post(url, data=data)
if response.status_code != 200:
raise Exception(f"ServerChan API返回错误: {response.text}")
def _send_webhook(self, title: str, content: str):
"""发送企业微信机器人通知"""
import requests
import json
url = self.config.get(self.config.Notify_CompanyWebHookBotUrl)
data = {
"msgtype": "markdown",
"markdown": {
"content": f"### {title}\n{content}"
}
}
response = requests.post(url, json=data)
if response.status_code != 200:
raise Exception(f"企业微信机器人API返回错误: {response.text}")
Notify = Notification()

207
app/ui/member_manager.py
View File

@@ -1529,6 +1529,140 @@ class MemberManager(QWidget):
parent=self,
)
# 新增单独通知卡片
self.card_NotifyEnable = SwitchSettingCard(
icon=FluentIcon.INFO,
title="启用单独通知",
content="启用后，任务结束将向该用户单独推送通知",
qconfig=self.config,
configItem=self.config.Notify_Enable,
parent=self
)
self.card_NotifyIfSMTP = SwitchSettingCard(
icon=FluentIcon.INFO,
title="启用SMTP通知",
content="是否启用单独通知的SMTP邮件推送",
qconfig=self.config,
configItem=self.config.Notify_IfSMTP,
parent=self
)
self.card_NotifySMTP = LineEditSettingCard(
icon=FluentIcon.INFO,
title="SMTP服务器地址",
content="单独通知的SMTP服务器地址",
text="请输入SMTP服务器地址",
qconfig=self.config,
configItem=self.config.Notify_SMTPServerAddress,
parent=self
)
self.card_NotifyAuthCode = PasswordLineEditSettingCard(
icon=FluentIcon.INFO,
title="授权码",
content="单独通知的邮箱授权码",
text="请输入授权码",
algorithm="AUTO",
qconfig=self.config,
configItem=self.config.Notify_AuthorizationCode,
parent=self
)
self.card_NotifyFrom = LineEditSettingCard(
icon=FluentIcon.INFO,
title="发件人邮箱",
content="单独通知的发件人邮箱",
text="请输入发件人邮箱",
qconfig=self.config,
configItem=self.config.Notify_FromAddress,
parent=self
)
self.card_NotifyTo = LineEditSettingCard(
icon=FluentIcon.INFO,
title="收件人邮箱",
content="单独通知的收件人邮箱",
text="请输入收件人邮箱",
qconfig=self.config,
configItem=self.config.Notify_ToAddress,
parent=self
)
self.card_NotifyIfServerChan = SwitchSettingCard(
icon=FluentIcon.INFO,
title="启用ServerChan",
content="是否启用单独通知的ServerChan推送",
qconfig=self.config,
configItem=self.config.Notify_IfServerChan,
parent=self
)
self.card_NotifyServerChanKey = LineEditSettingCard(
icon=FluentIcon.INFO,
title="ServerChanKey",
content="单独通知的ServerChan SendKey",
text="请输入ServerChanKey",
qconfig=self.config,
configItem=self.config.Notify_ServerChanKey,
parent=self
)
self.card_NotifyServerChanChannel = LineEditSettingCard(
icon=FluentIcon.INFO,
title="ServerChanChannel",
content="单独通知的ServerChan Channel",
text="请输入ServerChanChannel",
qconfig=self.config,
configItem=self.config.Notify_ServerChanChannel,
parent=self
)
self.card_NotifyServerChanTag = LineEditSettingCard(
icon=FluentIcon.INFO,
title="ServerChanTag",
content="单独通知的ServerChan Tag",
text="请输入ServerChanTag",
qconfig=self.config,
configItem=self.config.Notify_ServerChanTag,
parent=self
)
self.card_NotifyIfCompanyWebHookBot = SwitchSettingCard(
icon=FluentIcon.INFO,
title="启用企业微信机器人",
content="是否启用单独通知的企业微信机器人推送",
qconfig=self.config,
configItem=self.config.Notify_IfCompanyWebHookBot,
parent=self
)
self.card_NotifyCompanyWebHookBotUrl = LineEditSettingCard(
icon=FluentIcon.INFO,
title="企业微信机器人WebhookUrl",
content="单独通知的企业微信机器人Webhook地址",
text="请输入WebhookUrl",
qconfig=self.config,
configItem=self.config.Notify_CompanyWebHookBotUrl,
parent=self
)
# 设置通知卡片默认隐藏
self.card_NotifyIfSMTP.setVisible(False)
self.card_NotifySMTP.setVisible(False)
self.card_NotifyAuthCode.setVisible(False)
self.card_NotifyFrom.setVisible(False)
self.card_NotifyTo.setVisible(False)
self.card_NotifyIfServerChan.setVisible(False)
self.card_NotifyServerChanKey.setVisible(False)
self.card_NotifyServerChanChannel.setVisible(False)
self.card_NotifyServerChanTag.setVisible(False)
self.card_NotifyIfCompanyWebHookBot.setVisible(False)
self.card_NotifyCompanyWebHookBotUrl.setVisible(False)
# 连接通知启用开关的信号
self.card_NotifyEnable.checkedChanged.connect(self.toggle_notify_settings)
self.card_NotifyIfSMTP.checkedChanged.connect(self.toggle_smtp_settings)
# 根据配置状态初始化显示
if self.config.get(self.config.Notify_Enable):
self.toggle_notify_settings(True)
if self.config.get(self.config.Notify_IfSMTP):
self.toggle_smtp_settings(True)
if self.config.get(self.config.Notify_IfServerChan):
self.toggle_serverchan_settings(True)
if self.config.get(self.config.Notify_IfCompanyWebHookBot):
self.toggle_webhook_settings(True)
h1_layout = QHBoxLayout()
h1_layout.addWidget(self.card_Name)
h1_layout.addWidget(self.card_Id)
@@ -1566,6 +1700,27 @@ class MemberManager(QWidget):
Layout.addLayout(h6_layout)
Layout.addLayout(h7_layout)
Layout.addLayout(h8_layout)
# 创建通知设置容器
notify_container = QWidget()
notify_layout = QVBoxLayout(notify_container)
notify_layout.setContentsMargins(0, 0, 0, 0)
notify_layout.setSpacing(0)
notify_layout.addWidget(self.card_NotifyEnable)
notify_layout.addWidget(self.card_NotifyIfSMTP)
notify_layout.addWidget(self.card_NotifySMTP)
notify_layout.addWidget(self.card_NotifyAuthCode)
notify_layout.addWidget(self.card_NotifyFrom)
notify_layout.addWidget(self.card_NotifyTo)
notify_layout.addWidget(self.card_NotifyIfServerChan)
notify_layout.addWidget(self.card_NotifyServerChanKey)
notify_layout.addWidget(self.card_NotifyServerChanChannel)
notify_layout.addWidget(self.card_NotifyServerChanTag)
notify_layout.addWidget(self.card_NotifyIfCompanyWebHookBot)
notify_layout.addWidget(self.card_NotifyCompanyWebHookBotUrl)
Layout.addWidget(notify_container)
self.viewLayout.addLayout(Layout)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
@@ -1588,6 +1743,14 @@ class MemberManager(QWidget):
Config.gameid_refreshed.connect(self.refresh_gameid)
Config.PASSWORD_refreshed.connect(self.refresh_password)
# 连接ServerChan和企业微信机器人的开关信号
self.card_NotifyIfServerChan.checkedChanged.connect(
lambda checked: self.toggle_serverchan_settings(checked)
)
self.card_NotifyIfCompanyWebHookBot.checkedChanged.connect(
lambda checked: self.toggle_webhook_settings(checked)
)
self.switch_mode()
self.switch_infrastructure()
@@ -1709,3 +1872,47 @@ class MemberManager(QWidget):
}
},
)
def toggle_notify_settings(self, checked: bool):
"""切换通知设置卡片的显示状态"""
self.card_NotifyIfSMTP.setVisible(checked)
self.card_NotifyIfServerChan.setVisible(checked)
self.card_NotifyIfCompanyWebHookBot.setVisible(checked)
# 根据SMTP开关状态控制相关设置
if checked and self.config.get(self.config.Notify_IfSMTP):
self.toggle_smtp_settings(True)
else:
self.toggle_smtp_settings(False)
# 根据ServerChan开关状态控制相关设置
if checked and self.config.get(self.config.Notify_IfServerChan):
self.toggle_serverchan_settings(True)
else:
self.toggle_serverchan_settings(False)
# 根据企业微信机器人开关状态控制相关设置
if checked and self.config.get(self.config.Notify_IfCompanyWebHookBot):
self.toggle_webhook_settings(True)
else:
self.toggle_webhook_settings(False)
def toggle_smtp_settings(self, checked: bool):
"""切换SMTP相关设置的显示状态"""
if self.config.get(self.config.Notify_Enable):
self.card_NotifySMTP.setVisible(checked)
self.card_NotifyAuthCode.setVisible(checked)
self.card_NotifyFrom.setVisible(checked)
self.card_NotifyTo.setVisible(checked)
def toggle_serverchan_settings(self, checked: bool):
"""切换ServerChan相关设置的显示状态"""
if self.config.get(self.config.Notify_Enable):
self.card_NotifyServerChanKey.setVisible(checked)
self.card_NotifyServerChanChannel.setVisible(checked)
self.card_NotifyServerChanTag.setVisible(checked)
def toggle_webhook_settings(self, checked: bool):
"""切换企业微信机器人相关设置的显示状态"""
if self.config.get(self.config.Notify_Enable):
self.card_NotifyCompanyWebHookBotUrl.setVisible(checked)