Merge branch 'ClozyA_dev' into dev

This commit is contained in:
DLmaster361
2025-05-24 18:34:03 +08:00
11 changed files with 836 additions and 329 deletions

View File

@@ -234,10 +234,6 @@ class GlobalConfig(LQConfig):
self.notify_CompanyWebHookBotUrl = ConfigItem(
"Notify", "CompanyWebHookBotUrl", ""
)
self.notify_IfPushDeer = ConfigItem(
"Notify", "IfPushDeer", False, BoolValidator()
)
self.notify_IfPushDeerKey = ConfigItem("Notify", "PushDeerKey", "")
self.update_IfAutoUpdate = ConfigItem(
"Update", "IfAutoUpdate", False, BoolValidator()
@@ -439,6 +435,31 @@ class MaaUserConfig(LQConfig):
"Data", "CustomInfrastPlanIndex", "0"
)
# 新增用户单独通知字段
self.Notify_Enabled = ConfigItem("Notify", "Enabled", False, BoolValidator())
self.Notify_IfSendStatistic = ConfigItem(
"Notify", "IfSendStatistic", False, BoolValidator()
)
self.Notify_IfSendSixStar = ConfigItem(
"Notify", "IfSendSixStar", False, BoolValidator()
)
self.Notify_IfSendMail = ConfigItem(
"Notify", "IfSendMail", False, BoolValidator()
)
self.Notify_ToAddress = ConfigItem("Notify", "ToAddress", "")
self.Notify_IfServerChan = ConfigItem(
"Notify", "IfServerChan", False, BoolValidator()
)
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, BoolValidator()
)
self.Notify_CompanyWebHookBotUrl = ConfigItem(
"Notify", "CompanyWebHookBotUrl", ""
)
class MaaPlanConfig(LQConfig):
"""MAA计划表配置"""
@@ -563,7 +584,7 @@ class MaaPlanConfig(LQConfig):
class AppConfig(GlobalConfig):
VERSION = "4.3.8.3"
VERSION = "4.3.8.4"
gameid_refreshed = Signal()
PASSWORD_refreshed = Signal()

View File

@@ -608,13 +608,14 @@ class MaaManager(QObject):
Config.app_path
/ f"history/{curdate}/{user_data["Info"]["Name"]}/{start_time.strftime("%H-%M-%S")}.json",
)
if Config.get(Config.notify_IfSendSixStar) and if_six_star:
if if_six_star:
self.push_notification(
"公招六星",
f"喜报:用户 {user[0]} 公招出六星啦!",
{"user_name": user_data["Info"]["Name"]},
{
"user_name": user_data["Info"]["Name"],
},
user_data,
)
# 执行MAA解压更新动作
@@ -640,26 +641,23 @@ class MaaManager(QObject):
logger.info(f"{self.name} | 更新动作结束")
if Config.get(Config.notify_IfSendStatistic):
statistics = Config.merge_maa_logs("指定项", user_logs_list)
statistics["user_info"] = user[0]
statistics["start_time"] = user_start_time.strftime(
"%Y-%m-%d %H:%M:%S"
)
statistics["end_time"] = datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
statistics["maa_result"] = (
"代理任务全部完成"
if (run_book["Annihilation"] and run_book["Routine"])
else "代理任务未全部完成"
)
self.push_notification(
"统计信息",
f"{current_date} | 用户 {user[0]} 的自动代理统计报告",
statistics,
)
# 发送统计信息
statistics = Config.merge_maa_logs("指定项", user_logs_list)
statistics["user_index"] = user[2]
statistics["user_info"] = user[0]
statistics["start_time"] = user_start_time.strftime("%Y-%m-%d %H:%M:%S")
statistics["end_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
statistics["maa_result"] = (
"代理任务全部完成"
if (run_book["Annihilation"] and run_book["Routine"])
else "代理任务未全部完成"
)
self.push_notification(
"统计信息",
f"{current_date} | 用户 {user[0]} 的自动代理统计报告",
statistics,
user_data,
)
if run_book["Annihilation"] and run_book["Routine"]:
# 成功完成代理的用户修改相关参数
@@ -852,6 +850,17 @@ class MaaManager(QObject):
self.data[_]["Config"]["Info"]["Name"] for _ in wait_index
],
}
# 生成结果文本
result_text = (
f"任务开始时间:{result["start_time"]},结束时间:{result["end_time"]}\n"
f"已完成数:{result["completed_count"]},未完成数:{result["uncompleted_count"]}\n\n"
)
if len(result["failed_user"]) > 0:
result_text += f"{self.mode[2:4]}未成功的用户:\n{"\n".join(result["failed_user"])}\n"
if len(result["waiting_user"]) > 0:
result_text += f"\n未开始{self.mode[2:4]}的用户:\n{"\n".join(result["waiting_user"])}\n"
# 推送代理结果通知
Notify.push_plyer(
title.replace("报告", "已完成!"),
@@ -859,15 +868,7 @@ class MaaManager(QObject):
f"已完成用户数:{len(over_index)},未完成用户数:{len(error_index) + len(wait_index)}",
10,
)
if Config.get(Config.notify_SendTaskResultTime) == "任何时刻" or (
Config.get(Config.notify_SendTaskResultTime) == "仅失败时"
and len(error_index) + len(wait_index) != 0
):
result_text = self.push_notification("代理结果", title, result)
else:
result_text = self.push_notification(
"代理结果", title, result, if_get_text_only=True
)
self.push_notification("代理结果", title, result)
self.agree_bilibili(False)
self.log_monitor.deleteLater()
@@ -1741,16 +1742,21 @@ class MaaManager(QObject):
mode: str,
title: str,
message: Union[str, dict],
if_get_text_only: bool = False,
) -> str:
user_data: Dict[str, Dict[str, Union[str, int, bool]]] = None,
) -> None:
"""通过所有渠道推送通知"""
env = Environment(
loader=FileSystemLoader(str(Config.app_path / "resources/html"))
)
if mode == "代理结果":
if mode == "代理结果" and (
Config.get(Config.notify_SendTaskResultTime) == "任何时刻"
or (
Config.get(Config.notify_SendTaskResultTime) == "仅失败时"
and message["uncompleted_count"] != 0
)
):
# 生成文本通知内容
message_text = (
f"任务开始时间:{message["start_time"]},结束时间:{message["end_time"]}\n"
@@ -1762,9 +1768,6 @@ class MaaManager(QObject):
if len(message["waiting_user"]) > 0:
message_text += f"\n未开始{self.mode[2:4]}的用户:\n{"\n".join(message["waiting_user"])}\n"
if if_get_text_only:
return message_text
# 生成HTML通知内容
message["failed_user"] = "".join(message["failed_user"])
message["waiting_user"] = "".join(message["waiting_user"])
@@ -1772,11 +1775,31 @@ 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 敬上")
# ServerChan的换行是两个换行符。故而将\n替换为\n\n
serverchan_message = message_text.replace("\n", "\n\n")
return message_text
# 发送全局通知
if Config.get(Config.notify_IfSendMail):
Notify.send_mail(
"网页", title, message_html, Config.get(Config.notify_ToAddress)
)
if Config.get(Config.notify_IfServerChan):
Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
Config.get(Config.notify_ServerChanKey),
Config.get(Config.notify_ServerChanTag),
Config.get(Config.notify_ServerChanChannel),
)
if Config.get(Config.notify_IfCompanyWebHookBot):
Notify.CompanyWebHookBotPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
Config.get(Config.notify_CompanyWebHookBotUrl),
)
elif mode == "统计信息":
@@ -1805,18 +1828,155 @@ 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 敬上")
# 发送全局通知
if Config.get(Config.notify_IfSendStatistic):
if Config.get(Config.notify_IfSendMail):
Notify.send_mail(
"网页", title, message_html, Config.get(Config.notify_ToAddress)
)
if Config.get(Config.notify_IfServerChan):
Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
Config.get(Config.notify_ServerChanKey),
Config.get(Config.notify_ServerChanTag),
Config.get(Config.notify_ServerChanChannel),
)
if Config.get(Config.notify_IfCompanyWebHookBot):
Notify.CompanyWebHookBotPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
Config.get(Config.notify_CompanyWebHookBotUrl),
)
# 发送用户单独通知
if (
user_data["Notify"]["Enabled"]
and user_data["Notify"]["IfSendStatistic"]
):
# 发送邮件通知
if user_data["Notify"]["IfSendMail"]:
if user_data["Notify"]["ToAddress"]:
Notify.send_mail(
"网页",
title,
message_html,
user_data["Notify"]["ToAddress"],
)
else:
logger.error(
f"{self.name} | 用户邮箱地址为空,无法发送用户单独的邮件通知"
)
# 发送ServerChan通知
if user_data["Notify"]["IfServerChan"]:
if user_data["Notify"]["ServerChanKey"]:
Notify.ServerChanPush(
title,
f"{serverchan_message}\n\nAUTO_MAA 敬上",
user_data["Notify"]["ServerChanKey"],
user_data["Notify"]["ServerChanTag"],
user_data["Notify"]["ServerChanChannel"],
)
else:
logger.error(
f"{self.name} |用户ServerChan密钥为空无法发送用户单独的ServerChan通知"
)
# 推送CompanyWebHookBot通知
if user_data["Notify"]["IfCompanyWebHookBot"]:
if user_data["Notify"]["CompanyWebHookBotUrl"]:
Notify.CompanyWebHookBotPush(
title,
f"{message_text}\n\nAUTO_MAA 敬上",
user_data["Notify"]["CompanyWebHookBotUrl"],
)
else:
logger.error(
f"{self.name} |用户CompanyWebHookBot密钥为空无法发送用户单独的CompanyWebHookBot通知"
)
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 敬上")
# 发送全局通知
if Config.get(Config.notify_IfSendSixStar):
if Config.get(Config.notify_IfSendMail):
Notify.send_mail(
"网页", title, message_html, Config.get(Config.notify_ToAddress)
)
if Config.get(Config.notify_IfServerChan):
Notify.ServerChanPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
Config.get(Config.notify_ServerChanKey),
Config.get(Config.notify_ServerChanTag),
Config.get(Config.notify_ServerChanChannel),
)
if Config.get(Config.notify_IfCompanyWebHookBot):
Notify.CompanyWebHookBotPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
Config.get(Config.notify_CompanyWebHookBotUrl),
)
# 发送用户单独通知
if user_data["Notify"]["Enabled"] and user_data["Notify"]["IfSendSixStar"]:
# 发送邮件通知
if user_data["Notify"]["IfSendMail"]:
if user_data["Notify"]["ToAddress"]:
Notify.send_mail(
"网页",
title,
message_html,
user_data["Notify"]["ToAddress"],
)
else:
logger.error(
f"{self.name} | 用户邮箱地址为空,无法发送用户单独的邮件通知"
)
# 发送ServerChan通知
if user_data["Notify"]["IfServerChan"]:
if user_data["Notify"]["ServerChanKey"]:
Notify.ServerChanPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
user_data["Notify"]["ServerChanKey"],
user_data["Notify"]["ServerChanTag"],
user_data["Notify"]["ServerChanChannel"],
)
else:
logger.error(
f"{self.name} |用户ServerChan密钥为空无法发送用户单独的ServerChan通知"
)
# 推送CompanyWebHookBot通知
if user_data["Notify"]["IfCompanyWebHookBot"]:
if user_data["Notify"]["CompanyWebHookBotUrl"]:
Notify.CompanyWebHookBotPush(
title,
"好羡慕~\n\nAUTO_MAA 敬上",
user_data["Notify"]["CompanyWebHookBotUrl"],
)
else:
logger.error(
f"{self.name} |用户CompanyWebHookBot密钥为空无法发送用户单独的CompanyWebHookBot通知"
)
return None

View File

@@ -25,18 +25,20 @@ v4.3
作者DLmaster_361
"""
from PySide6.QtWidgets import QWidget
from PySide6.QtCore import Signal
import requests
import time
from loguru import logger
from plyer import notification
import re
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import time
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.utils import formataddr
import requests
from PySide6.QtCore import Signal
from PySide6.QtWidgets import QWidget
from loguru import logger
from plyer import notification
from app.core import Config
from app.services.security import Crypto
@@ -65,218 +67,210 @@ class Notification(QWidget):
return True
def send_mail(self, mode, title, content) -> None:
def send_mail(self, mode, title, content, to_address) -> None:
"""推送邮件通知"""
if Config.get(Config.notify_IfSendMail):
if (
Config.get(Config.notify_SMTPServerAddress) == ""
or Config.get(Config.notify_AuthorizationCode) == ""
or not bool(
re.match(
r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
Config.get(Config.notify_FromAddress),
)
)
or not bool(
re.match(
r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
Config.get(Config.notify_ToAddress),
)
)
):
logger.error(
"请正确设置邮件通知的SMTP服务器地址、授权码、发件人地址和收件人地址"
)
self.push_info_bar.emit(
"error",
"邮件通知推送异常",
"请正确设置邮件通知的SMTP服务器地址、授权码、发件人地址和收件人地址",
-1,
)
return None
try:
# 定义邮件正文
if mode == "文本":
message = MIMEText(content, "plain", "utf-8")
elif mode == "网页":
message = MIMEMultipart("alternative")
message["From"] = formataddr(
(
Header("AUTO_MAA通知服务", "utf-8").encode(),
Config.get(Config.notify_FromAddress),
)
) # 发件人显示的名字
message["To"] = formataddr(
(
Header("AUTO_MAA用户", "utf-8").encode(),
Config.get(Config.notify_ToAddress),
)
) # 收件人显示的名字
message["Subject"] = Header(title, "utf-8")
if mode == "网页":
message.attach(MIMEText(content, "html", "utf-8"))
smtpObj = smtplib.SMTP_SSL(
Config.get(Config.notify_SMTPServerAddress),
465,
)
smtpObj.login(
if (
Config.get(Config.notify_SMTPServerAddress) == ""
or Config.get(Config.notify_AuthorizationCode) == ""
or not bool(
re.match(
r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
Config.get(Config.notify_FromAddress),
Crypto.win_decryptor(Config.get(Config.notify_AuthorizationCode)),
)
smtpObj.sendmail(
)
or not bool(
re.match(
r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
to_address,
)
)
):
logger.error(
"请正确设置邮件通知的SMTP服务器地址、授权码、发件人地址和收件人地址"
)
self.push_info_bar.emit(
"error",
"邮件通知推送异常",
"请正确设置邮件通知的SMTP服务器地址、授权码、发件人地址和收件人地址",
-1,
)
return None
try:
# 定义邮件正文
if mode == "文本":
message = MIMEText(content, "plain", "utf-8")
elif mode == "网页":
message = MIMEMultipart("alternative")
message["From"] = formataddr(
(
Header("AUTO_MAA通知服务", "utf-8").encode(),
Config.get(Config.notify_FromAddress),
Config.get(Config.notify_ToAddress),
message.as_string(),
)
smtpObj.quit()
logger.success("邮件发送成功")
except Exception as e:
logger.error(f"发送邮件时出错:\n{e}")
self.push_info_bar.emit("error", "发送邮件时出错", f"{e}", -1)
def ServerChanPush(self, title, content):
"""使用Server酱推送通知支持 tag 和 channel避免使用SDK"""
if Config.get(Config.notify_IfServerChan):
send_key = Config.get(Config.notify_ServerChanKey)
if not send_key:
logger.error("请正确设置Server酱的SendKey")
self.push_info_bar.emit(
"error", "Server酱通知推送异常", "请正确设置Server酱的SendKey", -1
) # 发件人显示的名字
message["To"] = formataddr(
(
Header("AUTO_MAA用户", "utf-8").encode(),
to_address,
)
return None
) # 收件人显示的名字
message["Subject"] = Header(title, "utf-8")
try:
# 构造 URL
if send_key.startswith("sctp"):
match = re.match(r"^sctp(\d+)t", send_key)
if match:
url = f"https://{match.group(1)}.push.ft07.com/send/{send_key}.send"
else:
raise ValueError("SendKey 格式错误sctp")
if mode == "网页":
message.attach(MIMEText(content, "html", "utf-8"))
smtpObj = smtplib.SMTP_SSL(
Config.get(Config.notify_SMTPServerAddress),
465,
)
smtpObj.login(
Config.get(Config.notify_FromAddress),
Crypto.win_decryptor(Config.get(Config.notify_AuthorizationCode)),
)
smtpObj.sendmail(
Config.get(Config.notify_FromAddress),
to_address,
message.as_string(),
)
smtpObj.quit()
logger.success("邮件发送成功")
return None
except Exception as e:
logger.error(f"发送邮件时出错:\n{e}")
self.push_info_bar.emit("error", "发送邮件时出错", f"{e}", -1)
return None
return None
def ServerChanPush(self, title, content, send_key, tag, channel):
"""使用Server酱推送通知"""
if not send_key:
logger.error("请正确设置Server酱的SendKey")
self.push_info_bar.emit(
"error", "Server酱通知推送异常", "请正确设置Server酱的SendKey", -1
)
return None
try:
# 构造 URL
if send_key.startswith("sctp"):
match = re.match(r"^sctp(\d+)t", send_key)
if match:
url = f"https://{match.group(1)}.push.ft07.com/send/{send_key}.send"
else:
url = f"https://sctapi.ftqq.com/{send_key}.send"
# 构建 tags 和 channel
def is_valid(s):
return s == "" or (
s == "|".join(s.split("|"))
and (s.count("|") == 0 or all(s.split("|")))
)
tags = "|".join(
_.strip()
for _ in Config.get(Config.notify_ServerChanTag).split("|")
)
channels = "|".join(
_.strip()
for _ in Config.get(Config.notify_ServerChanChannel).split("|")
)
options = {}
if is_valid(tags):
options["tags"] = tags
else:
logger.warning("Server酱 Tag 配置不正确,将被忽略")
self.push_info_bar.emit(
"warning",
"Server酱通知推送异常",
"请正确设置 ServerChan 的 Tag",
-1,
)
if is_valid(channels):
options["channel"] = channels
else:
logger.warning("Server酱 Channel 配置不正确,将被忽略")
self.push_info_bar.emit(
"warning",
"Server酱通知推送异常",
"请正确设置 ServerChan 的 Channel",
-1,
)
# 请求发送
params = {"title": title, "desp": content, **options}
headers = {"Content-Type": "application/json;charset=utf-8"}
response = requests.post(url, json=params, headers=headers, timeout=10)
result = response.json()
if result.get("code") == 0:
logger.info("Server酱推送通知成功")
return True
else:
error_code = result.get("code", "-1")
logger.error(f"Server酱通知推送失败响应码{error_code}")
self.push_info_bar.emit(
"error", "Server酱通知推送失败", f"响应码:{error_code}", -1
)
return f"Server酱通知推送失败{error_code}"
except Exception as e:
logger.exception("Server酱通知推送异常")
self.push_info_bar.emit(
"error", "Server酱通知推送异常", f"请检查相关设置,如还有问题可联系开发者", -1
)
return f"Server酱通知推送异常{str(e)}"
def CompanyWebHookBotPush(self, title, content):
"""使用企业微信群机器人推送通知"""
if Config.get(Config.notify_IfCompanyWebHookBot):
if Config.get(Config.notify_CompanyWebHookBotUrl) == "":
logger.error("请正确设置企业微信群机器人的WebHook地址")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送异常",
"请正确设置企业微信群机器人的WebHook地址",
-1,
)
return None
content = f"{title}\n{content}"
data = {"msgtype": "text", "text": {"content": content}}
# 从远程服务器获取最新主题图像
for _ in range(3):
try:
response = requests.post(
url=Config.get(Config.notify_CompanyWebHookBotUrl),
json=data,
timeout=10,
)
info = response.json()
break
except Exception as e:
err = e
time.sleep(0.1)
raise ValueError("SendKey 格式错误sctp")
else:
logger.error(f"推送企业微信群机器人时出错:{err}")
url = f"https://sctapi.ftqq.com/{send_key}.send"
# 构建 tags 和 channel
def is_valid(s):
return s == "" or (
s == "|".join(s.split("|"))
and (s.count("|") == 0 or all(s.split("|")))
)
tags = "|".join(_.strip() for _ in tag.split("|"))
channels = "|".join(_.strip() for _ in channel.split("|"))
options = {}
if is_valid(tags):
options["tags"] = tags
else:
logger.warning("Server酱 Tag 配置不正确,将被忽略")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错:{info["errmsg"]}',
"warning",
"Server酱通知推送异常",
"请正确设置 ServerChan 的 Tag",
-1,
)
return None
if info["errcode"] == 0:
logger.info("企业微信群机器人推送通知成功")
if is_valid(channels):
options["channel"] = channels
else:
logger.warning("Server酱 Channel 配置不正确,将被忽略")
self.push_info_bar.emit(
"warning",
"Server酱通知推送异常",
"请正确设置 ServerChan 的 Channel",
-1,
)
# 请求发送
params = {"title": title, "desp": content, **options}
headers = {"Content-Type": "application/json;charset=utf-8"}
response = requests.post(url, json=params, headers=headers, timeout=10)
result = response.json()
if result.get("code") == 0:
logger.info("Server酱推送通知成功")
return True
else:
logger.error(f"企业微信群机器人推送通知失败:{info}")
error_code = result.get("code", "-1")
logger.error(f"Server酱通知推送失败响应码{error_code}")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错:{info["errmsg"]}',
-1,
"error", "Server酱通知推送失败", f"响应码:{error_code}", -1
)
return f'使用企业微信群机器人推送通知时出错:{info["errmsg"]}'
return f"Server酱通知推送失败{error_code}"
except Exception as e:
logger.exception("Server酱通知推送异常")
self.push_info_bar.emit(
"error",
"Server酱通知推送异常",
"请检查相关设置和网络连接。如全部配置正确,请稍后再试。",
-1,
)
return f"Server酱通知推送异常{str(e)}"
def CompanyWebHookBotPush(self, title, content, webhook_url):
"""使用企业微信群机器人推送通知"""
if webhook_url == "":
logger.error("请正确设置企业微信群机器人的WebHook地址")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送异常",
"请正确设置企业微信群机器人的WebHook地址",
-1,
)
return None
content = f"{title}\n{content}"
data = {"msgtype": "text", "text": {"content": content}}
for _ in range(3):
try:
response = requests.post(
url=webhook_url,
json=data,
timeout=10,
)
info = response.json()
break
except Exception as e:
err = e
time.sleep(0.1)
else:
logger.error(f"推送企业微信群机器人时出错:{err}")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错:{info["errmsg"]}',
-1,
)
return None
if info["errcode"] == 0:
logger.info("企业微信群机器人推送通知成功")
return True
else:
logger.error(f"企业微信群机器人推送通知失败:{info}")
self.push_info_bar.emit(
"error",
"企业微信群机器人通知推送失败",
f'使用企业微信群机器人推送通知时出错:{info["errmsg"]}',
-1,
)
return f'使用企业微信群机器人推送通知时出错:{info["errmsg"]}'
def send_test_notification(self):
"""发送测试通知到所有已启用的通知渠道"""
@@ -294,6 +288,7 @@ class Notification(QWidget):
"文本",
"AUTO_MAA测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容,说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作!",
Config.get(Config.notify_ToAddress),
)
# 发送Server酱通知
@@ -301,6 +296,9 @@ class Notification(QWidget):
self.ServerChanPush(
"AUTO_MAA测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容,说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作!",
Config.get(Config.notify_ServerChanKey),
Config.get(Config.notify_ServerChanTag),
Config.get(Config.notify_ServerChanChannel),
)
# 发送企业微信机器人通知
@@ -308,6 +306,7 @@ class Notification(QWidget):
self.CompanyWebHookBotPush(
"AUTO_MAA测试通知",
"这是 AUTO_MAA 外部通知测试信息。如果你看到了这段内容,说明 AUTO_MAA 的通知功能已经正确配置且可以正常工作!",
Config.get(Config.notify_CompanyWebHookBotUrl),
)
return True

View File

@@ -25,17 +25,24 @@ v4.3
作者DLmaster_361
"""
import os
import re
from datetime import datetime
from functools import partial
from typing import Optional, Union, List, Dict
from urllib.parse import urlparse
import markdown
from PySide6.QtCore import Qt, QTime, QTimer, QEvent, QSize
from PySide6.QtGui import QIcon, QPixmap, QPainter, QPainterPath
from PySide6.QtWidgets import (
QApplication,
QWidget,
QWidget,
QLabel,
QHBoxLayout,
QVBoxLayout,
QSizePolicy,
)
from PySide6.QtCore import Qt, QTime, QTimer, QEvent, QSize
from PySide6.QtGui import QIcon, QPixmap, QPainter, QPainterPath
from qfluentwidgets import (
LineEdit,
PasswordLineEdit,
@@ -74,15 +81,9 @@ from qfluentwidgets import (
ScrollArea,
Pivot,
PivotItem,
FlyoutViewBase,
)
from qfluentwidgets.common.overload import singledispatchmethod
import os
import re
import markdown
from datetime import datetime
from urllib.parse import urlparse
from functools import partial
from typing import Optional, Union, List, Dict
from app.core import Config
from app.services import Crypto
@@ -271,6 +272,39 @@ class NoticeMessageBox(MessageBoxBase):
self.Layout.addStretch(1)
class SettingFlyoutView(FlyoutViewBase):
"""设置卡二级菜单弹出组件"""
def __init__(
self,
parent,
title: str,
setting_cards: List[Union[SettingCard, HeaderCardWidget]],
):
super().__init__(parent)
self.title = SubtitleLabel(title)
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setSpacing(0)
content_layout.setContentsMargins(0, 0, 11, 0)
for setting_card in setting_cards:
content_layout.addWidget(setting_card)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
scrollArea.setWidget(content_widget)
self.viewLayout = QVBoxLayout(self)
self.viewLayout.setSpacing(12)
self.viewLayout.setContentsMargins(20, 16, 9, 16)
self.viewLayout.addWidget(self.title)
self.viewLayout.addWidget(scrollArea)
class SwitchSettingCard(SettingCard):
"""Setting card with switch button"""
@@ -933,6 +967,93 @@ class TimeEditSettingCard(SettingCard):
self.TimeEdit.setTime(QTime.fromString(value, "HH:mm"))
class UserNoticeSettingCard(PushAndSwitchButtonSettingCard):
"""Setting card with User's Notice"""
def __init__(
self,
icon: Union[str, QIcon, FluentIconBase],
title: str,
content: Union[str, None],
text: str,
qconfig: QConfig,
configItem: ConfigItem,
configItems: Dict[str, ConfigItem],
parent=None,
):
super().__init__(icon, title, content, text, qconfig, configItem, parent)
self.qconfig = qconfig
self.configItems = configItems
self.Lable = SubtitleLabel(self)
if configItems:
for config_item in configItems.values():
config_item.valueChanged.connect(self.setValues)
self.setValues()
self.hBoxLayout.addWidget(self.Lable, 0, Qt.AlignRight)
self.hBoxLayout.addSpacing(16)
def setValues(self):
def short_str(s: str) -> str:
if s.startswith(("SC", "sc")):
# SendKey首4 + 末4
return f"{s[:4]}***{s[-4:]}" if len(s) > 8 else s
elif s.startswith(("http://", "https://")):
# Webhook URL域名 + 路径尾3
parsed_url = urlparse(s)
domain = parsed_url.netloc
path_tail = (
parsed_url.path[-3:]
if len(parsed_url.path) > 3
else parsed_url.path
)
return f"{domain}***{path_tail}"
elif "@" in s:
# 邮箱:@前3/6 + 域名
username, domain = s.split("@", 1)
displayed_name = f"{username[:3]}***" if len(username) > 6 else username
return f"{displayed_name}@{domain}"
else:
# 普通字符串末尾3字符
return f"***{s[-3:]}" if len(s) > 3 else s
content_list = []
if self.configItems:
if not (
self.qconfig.get(self.configItems["IfSendStatistic"])
or self.qconfig.get(self.configItems["IfSendSixStar"])
):
content_list.append("未启用任何通知项")
if self.qconfig.get(self.configItems["IfSendStatistic"]):
content_list.append("统计信息已启用")
if self.qconfig.get(self.configItems["IfSendSixStar"]):
content_list.append("六星喜报已启用")
if self.qconfig.get(self.configItems["IfSendMail"]):
content_list.append(
f"邮箱通知:{short_str(self.qconfig.get(self.configItems["ToAddress"]))}"
)
if self.qconfig.get(self.configItems["IfServerChan"]):
content_list.append(
f"Server酱通知{short_str(self.qconfig.get(self.configItems["ServerChanKey"]))}"
)
if self.qconfig.get(self.configItems["IfCompanyWebHookBot"]):
content_list.append(
f"企业微信通知:{short_str(self.qconfig.get(self.configItems["CompanyWebHookBotUrl"]))}"
)
self.setContent(" | ".join(content_list))
class StatusSwitchSetting(SwitchButton):
def __init__(

View File

@@ -333,28 +333,24 @@ class DispatchCenter(QWidget):
self.setObjectName(name)
layout = QVBoxLayout()
self.top_bar = self.DispatchTopBar(self, name)
self.info = self.DispatchInfoCard(self)
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setContentsMargins(0, 0, 0, 0)
content_layout.addWidget(self.top_bar)
content_layout.addWidget(self.info)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
self.top_bar = self.DispatchTopBar(self, name)
self.info = self.DispatchInfoCard(self)
content_layout.addWidget(self.top_bar)
content_layout.addWidget(self.info)
scrollArea.setWidget(content_widget)
layout = QVBoxLayout(self)
layout.addWidget(scrollArea)
self.setLayout(layout)
class DispatchTopBar(CardWidget):
def __init__(self, parent=None, name: str = None):

View File

@@ -61,21 +61,22 @@ class History(QWidget):
super().__init__(parent)
self.setObjectName("历史记录")
self.history_top_bar = self.HistoryTopBar(self)
self.history_top_bar.search_history.connect(self.reload_history)
content_widget = QWidget()
self.content_layout = QVBoxLayout(content_widget)
self.history_top_bar = self.HistoryTopBar(self)
self.history_top_bar.search_history.connect(self.reload_history)
self.content_layout.setContentsMargins(0, 0, 11, 0)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
scrollArea.setWidget(content_widget)
layout = QVBoxLayout()
layout = QVBoxLayout(self)
layout.addWidget(self.history_top_bar)
layout.addWidget(scrollArea)
self.setLayout(layout)
self.history_card_list = []

View File

@@ -62,14 +62,6 @@ class Home(QWidget):
self.banner = Banner()
self.banner_text = TextBrowser()
widget = QWidget()
Layout = QVBoxLayout(widget)
Layout.addWidget(self.banner)
Layout.addWidget(self.banner_text)
Layout.setStretch(0, 2)
Layout.setStretch(1, 3)
v_layout = QVBoxLayout(self.banner)
v_layout.setContentsMargins(0, 0, 0, 15)
v_layout.setSpacing(5)
@@ -146,14 +138,22 @@ class Home(QWidget):
# 将底部水平布局添加到垂直布局
v_layout.addLayout(h2_layout)
layout = QVBoxLayout()
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setContentsMargins(0, 0, 0, 0)
content_layout.addWidget(self.banner)
content_layout.addWidget(self.banner_text)
content_layout.setStretch(0, 2)
content_layout.setStretch(1, 3)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
scrollArea.setWidget(widget)
scrollArea.setWidget(content_widget)
layout = QVBoxLayout(self)
layout.addWidget(scrollArea)
self.setLayout(layout)
self.set_banner()

View File

@@ -47,6 +47,8 @@ from qfluentwidgets import (
PushSettingCard,
TableWidget,
PrimaryToolButton,
Flyout,
FlyoutAnimationType,
)
from PySide6.QtCore import Signal
from datetime import datetime
@@ -64,6 +66,7 @@ from .Widget import (
LineEditSettingCard,
SpinBoxSettingCard,
ComboBoxMessageBox,
SettingFlyoutView,
EditableComboBoxSettingCard,
PasswordLineEditSettingCard,
UserLableSettingCard,
@@ -72,6 +75,7 @@ from .Widget import (
PushAndSwitchButtonSettingCard,
PushAndComboBoxSettingCard,
StatusSwitchSetting,
UserNoticeSettingCard,
PivotArea,
)
@@ -564,29 +568,25 @@ class MemberManager(QWidget):
self.setObjectName(f"脚本_{uid}")
self.config = Config.member_dict[f"脚本_{uid}"]["Config"]
layout = QVBoxLayout()
self.app_setting = self.AppSettingCard(f"脚本_{uid}", self.config, self)
self.user_setting = self.UserManager(f"脚本_{uid}", self)
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setContentsMargins(0, 0, 11, 0)
content_layout.addWidget(self.app_setting)
content_layout.addWidget(self.user_setting)
content_layout.addStretch(1)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
self.app_setting = self.AppSettingCard(f"脚本_{uid}", self.config, self)
self.user_setting = self.UserManager(f"脚本_{uid}", self)
content_layout.addWidget(self.app_setting)
content_layout.addWidget(self.user_setting)
content_layout.addStretch(1)
scrollArea.setWidget(content_widget)
layout = QVBoxLayout(self)
layout.addWidget(scrollArea)
self.setLayout(layout)
class AppSettingCard(HeaderCardWidget):
def __init__(self, name: str, config: MaaConfig, parent=None):
@@ -1529,6 +1529,49 @@ class MemberManager(QWidget):
parent=self,
)
# 新增单独通知卡片
self.card_NotifySet = UserNoticeSettingCard(
icon=FluentIcon.MAIL,
title="用户单独通知设置",
content="未启用任何通知项",
text="设置",
qconfig=self.config,
configItem=self.config.Notify_Enabled,
configItems={
"IfSendStatistic": self.config.Notify_IfSendStatistic,
"IfSendSixStar": self.config.Notify_IfSendSixStar,
"IfSendMail": self.config.Notify_IfSendMail,
"ToAddress": self.config.Notify_ToAddress,
"IfServerChan": self.config.Notify_IfServerChan,
"ServerChanKey": self.config.Notify_ServerChanKey,
"IfCompanyWebHookBot": self.config.Notify_IfCompanyWebHookBot,
"CompanyWebHookBotUrl": self.config.Notify_CompanyWebHookBotUrl,
},
parent=self,
)
self.card_NotifyContent = self.NotifyContentSettingCard(
self.config, self
)
self.card_EMail = self.EMailSettingCard(self.config, self)
self.card_ServerChan = self.ServerChanSettingCard(
self.config, self
)
self.card_CompanyWebhookBot = (
self.CompanyWechatPushSettingCard(self.config, self)
)
self.card_NotifySet_list = [
self.card_NotifyContent,
self.card_EMail,
self.card_ServerChan,
self.card_CompanyWebhookBot,
]
self.NotifySetCard = SettingFlyoutView(
self, "用户通知设置", self.card_NotifySet_list
)
self.NotifySetCard.setVisible(False)
h1_layout = QHBoxLayout()
h1_layout.addWidget(self.card_Name)
h1_layout.addWidget(self.card_Id)
@@ -1566,6 +1609,7 @@ class MemberManager(QWidget):
Layout.addLayout(h6_layout)
Layout.addLayout(h7_layout)
Layout.addLayout(h8_layout)
Layout.addWidget(self.card_NotifySet)
self.viewLayout.addLayout(Layout)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
@@ -1585,6 +1629,7 @@ class MemberManager(QWidget):
self.card_InfrastMode.clicked.connect(
self.set_infrastructure
)
self.card_NotifySet.clicked.connect(self.set_notify)
Config.gameid_refreshed.connect(self.refresh_gameid)
Config.PASSWORD_refreshed.connect(self.refresh_password)
@@ -1709,3 +1754,166 @@ class MemberManager(QWidget):
}
},
)
def set_notify(self) -> None:
"""设置用户通知相关配置"""
self.NotifySetCard.setVisible(True)
Flyout.make(
self.NotifySetCard,
self.card_NotifySet,
self,
aniType=FlyoutAnimationType.PULL_UP,
isDeleteOnClose=False,
)
class NotifyContentSettingCard(HeaderCardWidget):
def __init__(self, config: MaaUserConfig, parent=None):
super().__init__(parent)
self.setTitle("用户通知内容选项")
self.config = config
self.card_IfSendStatistic = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送统计信息",
content="推送自动代理统计信息的通知",
qconfig=self.config,
configItem=self.config.Notify_IfSendStatistic,
parent=self,
)
self.card_IfSendSixStar = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送公招高资喜报",
content="公招出现六星词条时推送喜报",
qconfig=self.config,
configItem=self.config.Notify_IfSendSixStar,
parent=self,
)
Layout = QVBoxLayout()
Layout.addWidget(self.card_IfSendStatistic)
Layout.addWidget(self.card_IfSendSixStar)
self.viewLayout.addLayout(Layout)
self.viewLayout.setSpacing(3)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
class EMailSettingCard(HeaderCardWidget):
def __init__(self, config: MaaUserConfig, parent=None):
super().__init__(parent)
self.setTitle("用户邮箱通知")
self.config = config
self.card_IfSendMail = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送用户邮件通知",
content="是否启用用户邮件通知功能",
qconfig=self.config,
configItem=self.config.Notify_IfSendMail,
parent=self,
)
self.card_ToAddress = LineEditSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="用户收信邮箱地址",
content="接收用户通知的邮箱地址",
text="请输入用户收信邮箱地址",
qconfig=self.config,
configItem=self.config.Notify_ToAddress,
parent=self,
)
Layout = QVBoxLayout()
Layout.addWidget(self.card_IfSendMail)
Layout.addWidget(self.card_ToAddress)
self.viewLayout.addLayout(Layout)
self.viewLayout.setSpacing(3)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
class ServerChanSettingCard(HeaderCardWidget):
def __init__(self, config: MaaUserConfig, parent=None):
super().__init__(parent)
self.setTitle("用户ServerChan通知")
self.config = config
self.card_IfServerChan = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送用户Server酱通知",
content="是否启用用户Server酱通知功能",
qconfig=self.config,
configItem=self.config.Notify_IfServerChan,
parent=self,
)
self.card_ServerChanKey = LineEditSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="用户SendKey",
content="SC3与SCT均须填写",
text="请输入用户SendKey",
qconfig=self.config,
configItem=self.config.Notify_ServerChanKey,
parent=self,
)
self.card_ServerChanChannel = LineEditSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="用户ServerChanChannel代码",
content="留空则默认,多个请使用“|”隔开",
text="请输入Channel代码仅SCT生效",
qconfig=self.config,
configItem=self.config.Notify_ServerChanChannel,
parent=self,
)
self.card_ServerChanTag = LineEditSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="用户Tag内容",
content="留空则默认,多个请使用“|”隔开",
text="请输入加入推送的Tag仅SC3生效",
qconfig=self.config,
configItem=self.config.Notify_ServerChanTag,
parent=self,
)
Layout = QVBoxLayout()
Layout.addWidget(self.card_IfServerChan)
Layout.addWidget(self.card_ServerChanKey)
Layout.addWidget(self.card_ServerChanChannel)
Layout.addWidget(self.card_ServerChanTag)
self.viewLayout.addLayout(Layout)
self.viewLayout.setSpacing(3)
self.viewLayout.setContentsMargins(3, 0, 3, 3)
class CompanyWechatPushSettingCard(HeaderCardWidget):
def __init__(self, config: MaaUserConfig, parent=None):
super().__init__(parent)
self.setTitle("用户企业微信推送")
self.config = config
self.card_IfCompanyWebHookBot = SwitchSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="推送用户企业微信机器人通知",
content="是否启用用户企微机器人通知功能",
qconfig=self.config,
configItem=self.config.Notify_IfCompanyWebHookBot,
parent=self,
)
self.card_CompanyWebHookBotUrl = LineEditSettingCard(
icon=FluentIcon.PAGE_RIGHT,
title="WebhookUrl",
content="用户企微群机器人Webhook地址",
text="请输入用户Webhook的Url",
qconfig=self.config,
configItem=self.config.Notify_CompanyWebHookBotUrl,
parent=self,
)
Layout = QVBoxLayout()
Layout.addWidget(self.card_IfCompanyWebHookBot)
Layout.addWidget(self.card_CompanyWebHookBotUrl)
self.viewLayout.addLayout(Layout)
self.viewLayout.setSpacing(3)
self.viewLayout.setContentsMargins(3, 0, 3, 3)

View File

@@ -379,16 +379,6 @@ class QueueManager(QWidget):
self.setObjectName(f"调度队列_{uid}")
self.config = Config.queue_dict[f"调度队列_{uid}"]["Config"]
layout = QVBoxLayout()
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
self.queue_set = self.QueueSetSettingCard(self.config, self)
self.time = self.TimeSettingCard(self.config, self)
self.task = self.TaskSettingCard(self.config, self)
@@ -398,18 +388,24 @@ class QueueManager(QWidget):
parent=self,
)
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setContentsMargins(0, 0, 11, 0)
content_layout.addWidget(self.queue_set)
content_layout.addWidget(self.time)
content_layout.addWidget(self.task)
content_layout.addWidget(self.history)
content_layout.addStretch(1)
scrollArea = ScrollArea()
scrollArea.setWidgetResizable(True)
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
scrollArea.setWidget(content_widget)
layout = QVBoxLayout(self)
layout.addWidget(scrollArea)
self.setLayout(layout)
class QueueSetSettingCard(HeaderCardWidget):
def __init__(self, config: QueueConfig, parent=None):

View File

@@ -70,9 +70,6 @@ class Setting(QWidget):
super().__init__(parent)
self.setObjectName("设置")
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
self.function = FunctionSettingCard(self)
self.start = StartSettingCard(self)
self.ui = UiSettingCard(self)
@@ -93,6 +90,9 @@ class Setting(QWidget):
)
self.other.card_Notice.clicked.connect(lambda: self.show_notice(if_show=True))
content_widget = QWidget()
content_layout = QVBoxLayout(content_widget)
content_layout.setContentsMargins(0, 0, 11, 0)
content_layout.addWidget(self.function)
content_layout.addWidget(self.start)
content_layout.addWidget(self.ui)
@@ -106,9 +106,9 @@ class Setting(QWidget):
scrollArea.setContentsMargins(0, 0, 0, 0)
scrollArea.setStyleSheet("background: transparent; border: none;")
scrollArea.setWidget(content_widget)
layout = QVBoxLayout()
layout = QVBoxLayout(self)
layout.addWidget(scrollArea)
self.setLayout(layout)
def agree_bilibili(self) -> None:
"""授权bilibili游戏隐私政策"""

View File

@@ -1,6 +1,11 @@
{
"main_version": "4.3.8.3",
"main_version": "4.3.8.4",
"version_info": {
"4.3.8.4": {
"新增功能": [
"支持为每一个用户执行独立通知"
]
},
"4.3.8.3": {
"新增功能": [
"用户仪表盘支持直接控制用户状态"