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

feat(notify): 实现自定义Webhook通知功能

Browse Source
This commit is contained in:
AoXuan
2025-09-27 16:22:04 +08:00
parent 13caed7207
commit d2066e9631
5 changed files with 386 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
app/api/setting.py
View File

@@ -28,6 +28,7 @@ from fastapi import APIRouter, Body
from app.core import Config
from app.services import System, Notify
from app.models.schema import *
import uuid
router = APIRouter(prefix="/api/setting", tags=["全局设置"])
@@ -95,3 +96,164 @@ async def test_notify() -> OutBase:
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
return OutBase()
@router.post(
"/webhook/create", summary="创建自定义Webhook", response_model=OutBase, status_code=200
)
async def create_webhook(webhook_data: dict = Body(...)) -> OutBase:
"""创建自定义Webhook"""
try:
# 生成唯一ID
webhook_id = str(uuid.uuid4())
# 创建webhook配置
webhook_config = {
"id": webhook_id,
"name": webhook_data.get("name", ""),
"url": webhook_data.get("url", ""),
"template": webhook_data.get("template", ""),
"enabled": webhook_data.get("enabled", True),
"headers": webhook_data.get("headers", {}),
"method": webhook_data.get("method", "POST")
}
# 获取当前配置
current_config = await Config.get_setting()
custom_webhooks = current_config.get("Notify", {}).get("CustomWebhooks", [])
# 添加新webhook
custom_webhooks.append(webhook_config)
# 更新配置
update_data = {
"Notify": {
"CustomWebhooks": custom_webhooks
}
}
await Config.update_setting(update_data)
return OutBase(message=f"Webhook '{webhook_config['name']}' 创建成功")
except Exception as e:
return OutBase(
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
@router.post(
"/webhook/update", summary="更新自定义Webhook", response_model=OutBase, status_code=200
)
async def update_webhook(webhook_data: dict = Body(...)) -> OutBase:
"""更新自定义Webhook"""
try:
webhook_id = webhook_data.get("id")
if not webhook_id:
return OutBase(code=400, status="error", message="缺少Webhook ID")
# 获取当前配置
current_config = await Config.get_setting()
custom_webhooks = current_config.get("Notify", {}).get("CustomWebhooks", [])
# 查找并更新webhook
updated = False
for i, webhook in enumerate(custom_webhooks):
if webhook.get("id") == webhook_id:
custom_webhooks[i].update({
"name": webhook_data.get("name", webhook.get("name", "")),
"url": webhook_data.get("url", webhook.get("url", "")),
"template": webhook_data.get("template", webhook.get("template", "")),
"enabled": webhook_data.get("enabled", webhook.get("enabled", True)),
"headers": webhook_data.get("headers", webhook.get("headers", {})),
"method": webhook_data.get("method", webhook.get("method", "POST"))
})
updated = True
break
if not updated:
return OutBase(code=404, status="error", message="Webhook不存在")
# 更新配置
update_data = {
"Notify": {
"CustomWebhooks": custom_webhooks
}
}
await Config.update_setting(update_data)
return OutBase(message="Webhook更新成功")
except Exception as e:
return OutBase(
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
@router.post(
"/webhook/delete", summary="删除自定义Webhook", response_model=OutBase, status_code=200
)
async def delete_webhook(webhook_data: dict = Body(...)) -> OutBase:
"""删除自定义Webhook"""
try:
webhook_id = webhook_data.get("id")
if not webhook_id:
return OutBase(code=400, status="error", message="缺少Webhook ID")
# 获取当前配置
current_config = await Config.get_setting()
custom_webhooks = current_config.get("Notify", {}).get("CustomWebhooks", [])
# 查找并删除webhook
original_length = len(custom_webhooks)
custom_webhooks = [w for w in custom_webhooks if w.get("id") != webhook_id]
if len(custom_webhooks) == original_length:
return OutBase(code=404, status="error", message="Webhook不存在")
# 更新配置
update_data = {
"Notify": {
"CustomWebhooks": custom_webhooks
}
}
await Config.update_setting(update_data)
return OutBase(message="Webhook删除成功")
except Exception as e:
return OutBase(
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
@router.post(
"/webhook/test", summary="测试自定义Webhook", response_model=OutBase, status_code=200
)
async def test_webhook(webhook_data: dict = Body(...)) -> OutBase:
"""测试自定义Webhook"""
try:
webhook_config = {
"name": webhook_data.get("name", "测试Webhook"),
"url": webhook_data.get("url", ""),
"template": webhook_data.get("template", ""),
"enabled": True,
"headers": webhook_data.get("headers", {}),
"method": webhook_data.get("method", "POST")
}
await Notify.CustomWebhookPush(
"AUTO-MAS Webhook测试",
"这是一条测试消息，如果您收到此消息，说明Webhook配置正确！",
webhook_config
)
return OutBase(message="Webhook测试成功")
except Exception as e:
return OutBase(
code=500, status="error", message=f"Webhook测试失败: {str(e)}"
)

24
app/models/schema.py
View File

@@ -110,6 +110,16 @@ class GlobalConfig_UI(BaseModel):
IfToTray: Optional[bool] = Field(default=None, description="是否最小化到托盘")
class CustomWebhook(BaseModel):
id: str = Field(..., description="Webhook唯一标识")
name: str = Field(..., description="Webhook名称")
url: str = Field(..., description="Webhook URL")
template: str = Field(..., description="消息模板")
enabled: bool = Field(default=True, description="是否启用")
headers: Optional[Dict[str, str]] = Field(default=None, description="自定义请求头")
method: Optional[Literal["POST", "GET"]] = Field(default="POST", description="请求方法")
class GlobalConfig_Notify(BaseModel):
SendTaskResultTime: Optional[Literal["不推送", "任何时刻", "仅失败时"]] = Field(
default=None, description="任务结果推送时机"
@@ -130,11 +140,8 @@ class GlobalConfig_Notify(BaseModel):
default=None, description="是否使用ServerChan推送"
)
ServerChanKey: Optional[str] = Field(default=None, description="ServerChan推送密钥")
IfCompanyWebHookBot: Optional[bool] = Field(
default=None, description="是否使用企微Webhook推送"
)
CompanyWebHookBotUrl: Optional[str] = Field(
default=None, description="企微Webhook Bot URL"
CustomWebhooks: Optional[List[CustomWebhook]] = Field(
default=None, description="自定义Webhook列表"
)
@@ -303,11 +310,8 @@ class UserConfig_Notify(BaseModel):
default=None, description="是否使用Server酱推送"
)
ServerChanKey: Optional[str] = Field(default=None, description="ServerChanKey")
IfCompanyWebHookBot: Optional[bool] = Field(
default=None, description="是否使用Webhook推送"
)
CompanyWebHookBotUrl: Optional[str] = Field(
default=None, description="企微Webhook Bot URL"
CustomWebhooks: Optional[List[CustomWebhook]] = Field(
default=None, description="用户自定义Webhook列表"
)

118
app/services/notification.py
View File

@@ -170,9 +170,101 @@ class Notification:
else:
raise Exception(f"ServerChan 推送通知失败: {response.text}")
async def CustomWebhookPush(self, title, content, webhook_config) -> None:
"""
自定义 Webhook 推送通知
:param title: 通知标题
:param content: 通知内容
:param webhook_config: Webhook配置对象
"""
if not webhook_config.get("url"):
raise ValueError("Webhook URL 不能为空")
if not webhook_config.get("enabled", True):
logger.info(f"Webhook {webhook_config.get('name', 'Unknown')} 已禁用，跳过推送")
return
# 解析模板
template = webhook_config.get("template", '{"title": "{title}", "content": "{content}"}')
# 替换模板变量
try:
import json
from datetime import datetime
# 准备模板变量
template_vars = {
"title": title,
"content": content,
"datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"date": datetime.now().strftime("%Y-%m-%d"),
"time": datetime.now().strftime("%H:%M:%S")
}
# 替换模板中的变量
formatted_template = template.format(**template_vars)
# 尝试解析为JSON
try:
data = json.loads(formatted_template)
except json.JSONDecodeError:
# 如果不是JSON格式，作为纯文本发送
data = formatted_template
except Exception as e:
logger.warning(f"模板解析失败，使用默认格式: {e}")
data = {"title": title, "content": content}
# 准备请求头
headers = {"Content-Type": "application/json"}
if webhook_config.get("headers"):
headers.update(webhook_config["headers"])
# 发送请求
method = webhook_config.get("method", "POST").upper()
try:
if method == "POST":
if isinstance(data, dict):
response = requests.post(
url=webhook_config["url"],
json=data,
headers=headers,
timeout=10,
proxies=Config.get_proxies()
)
else:
response = requests.post(
url=webhook_config["url"],
data=data,
headers=headers,
timeout=10,
proxies=Config.get_proxies()
)
else: # GET
params = data if isinstance(data, dict) else {"message": data}
response = requests.get(
url=webhook_config["url"],
params=params,
headers=headers,
timeout=10,
proxies=Config.get_proxies()
)
# 检查响应
if response.status_code == 200:
logger.success(f"自定义Webhook推送成功: {webhook_config.get('name', 'Unknown')} - {title}")
else:
raise Exception(f"HTTP {response.status_code}: {response.text}")
except Exception as e:
raise Exception(f"自定义Webhook推送失败 ({webhook_config.get('name', 'Unknown')}): {str(e)}")
async def WebHookPush(self, title, content, webhook_url) -> None:
"""
WebHook 推送通知
WebHook 推送通知 (兼容旧版企业微信格式)
:param title: 通知标题
:param content: 通知内容
@@ -264,17 +356,19 @@ class Notification:
Config.get("Notify", "ServerChanKey"),
)
# 发送WebHook通知
if Config.get("Notify", "IfCompanyWebHookBot"):
await self.WebHookPush(
"AUTO-MAS测试通知",
"这是 AUTO-MAS 外部通知测试信息。如果你看到了这段内容, 说明 AUTO-MAS 的通知功能已经正确配置且可以正常工作！",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
await self.CompanyWebHookBotPushImage(
Path.cwd() / "res/images/notification/test_notify.png",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送自定义Webhook通知
custom_webhooks = Config.get("Notify", "CustomWebhooks", [])
if custom_webhooks:
for webhook in custom_webhooks:
if webhook.get("enabled", True):
try:
await self.CustomWebhookPush(
"AUTO-MAS测试通知",
"这是 AUTO-MAS 外部通知测试信息。如果你看到了这段内容, 说明 AUTO-MAS 的通知功能已经正确配置且可以正常工作！",
webhook
)
except Exception as e:
logger.error(f"自定义Webhook测试失败 ({webhook.get('name', 'Unknown')}): {e}")
logger.success("测试通知发送完成")

105
app/task/MAA.py
View File

@@ -1906,12 +1906,19 @@ class MaaManager:
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送自定义Webhook通知
custom_webhooks = Config.get("Notify", "CustomWebhooks", [])
if custom_webhooks:
for webhook in custom_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
elif mode == "统计信息":
@@ -1962,12 +1969,19 @@ class MaaManager:
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送自定义Webhook通知
custom_webhooks = Config.get("Notify", "CustomWebhooks", [])
if custom_webhooks:
for webhook in custom_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
# 发送用户单独通知
if self.cur_user_data.get("Notify", "Enabled") and self.cur_user_data.get(
@@ -2000,13 +2014,19 @@ class MaaManager:
)
# 推送CompanyWebHookBot通知
if self.cur_user_data.get("Notify", "IfCompanyWebHookBot"):
if self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"):
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送用户自定义Webhook通知
user_webhooks = self.cur_user_data.get("Notify", {}).get("CustomWebhooks", [])
if user_webhooks:
for webhook in user_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"用户自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
else:
logger.error(
"用户CompanyWebHookBot密钥为空, 无法发送用户单独的CompanyWebHookBot通知"
@@ -2034,16 +2054,19 @@ class MaaManager:
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
await Notify.WebHookPush(
title,
"好羡慕~\n\nAUTO-MAS 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
await Notify.CompanyWebHookBotPushImage(
Path.cwd() / "res/images/notification/six_star.png",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送自定义Webhook通知（六星喜报）
custom_webhooks = Config.get("Notify", "CustomWebhooks", [])
if custom_webhooks:
for webhook in custom_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
"好羡慕~\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
# 发送用户单独通知
if self.cur_user_data.get("Notify", "Enabled") and self.cur_user_data.get(
@@ -2077,17 +2100,19 @@ class MaaManager:
)
# 推送CompanyWebHookBot通知
if self.cur_user_data.get("Notify", "IfCompanyWebHookBot"):
if self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"):
await Notify.WebHookPush(
title,
"好羡慕~\n\nAUTO-MAS 敬上",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
)
await Notify.CompanyWebHookBotPushImage(
Path.cwd() / "res/images/notification/six_star.png",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送用户自定义Webhook通知（六星喜报）
user_webhooks = self.cur_user_data.get("Notify", {}).get("CustomWebhooks", [])
if user_webhooks:
for webhook in user_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
"好羡慕~\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"用户自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
else:
logger.error(
"用户CompanyWebHookBot密钥为空, 无法发送用户单独的CompanyWebHookBot通知"

58
app/task/general.py
View File

@@ -982,12 +982,19 @@ class GeneralManager:
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送自定义Webhook通知
custom_webhooks = Config.get("Notify", "CustomWebhooks", [])
if custom_webhooks:
for webhook in custom_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
elif mode == "统计信息":
@@ -1019,12 +1026,19 @@ class GeneralManager:
Config.get("Notify", "ServerChanKey"),
)
if Config.get("Notify", "IfCompanyWebHookBot"):
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
Config.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送自定义Webhook通知
custom_webhooks = Config.get("Notify", "CustomWebhooks", [])
if custom_webhooks:
for webhook in custom_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
# 发送用户单独通知
if self.cur_user_data.get("Notify", "Enabled") and self.cur_user_data.get(
@@ -1057,13 +1071,19 @@ class GeneralManager:
)
# 推送CompanyWebHookBot通知
if self.cur_user_data.get("Notify", "IfCompanyWebHookBot"):
if self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"):
await Notify.WebHookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
self.cur_user_data.get("Notify", "CompanyWebHookBotUrl"),
)
# 发送用户自定义Webhook通知
user_webhooks = self.cur_user_data.get("Notify", {}).get("CustomWebhooks", [])
if user_webhooks:
for webhook in user_webhooks:
if webhook.get("enabled", True):
try:
await Notify.CustomWebhookPush(
title,
f"{message_text}\n\nAUTO-MAS 敬上",
webhook
)
except Exception as e:
logger.error(f"用户自定义Webhook推送失败 ({webhook.get('name', 'Unknown')}): {e}")
else:
logger.error(
"用户CompanyWebHookBot密钥为空, 无法发送用户单独的CompanyWebHookBot通知"