From d2066e9631834b1ecfe1a62f3e94e063b25f9b64 Mon Sep 17 00:00:00 2001 From: AoXuan Date: Sat, 27 Sep 2025 16:22:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(notify):=20=E5=AE=9E=E7=8E=B0=E8=87=AA?= =?UTF-8?q?=E5=AE=9A=E4=B9=89Webhook=E9=80=9A=E7=9F=A5=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/setting.py | 162 +++++++++++++++++++++++++++++++++++ app/models/schema.py | 24 +++--- app/services/notification.py | 118 ++++++++++++++++++++++--- app/task/MAA.py | 105 ++++++++++++++--------- app/task/general.py | 58 +++++++++---- 5 files changed, 386 insertions(+), 81 deletions(-) diff --git a/app/api/setting.py b/app/api/setting.py index 33a5210..465b969 100644 --- a/app/api/setting.py +++ b/app/api/setting.py @@ -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)}" + ) diff --git a/app/models/schema.py b/app/models/schema.py index 9be5bb8..1317d70 100644 --- a/app/models/schema.py +++ b/app/models/schema.py @@ -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列表" ) diff --git a/app/services/notification.py b/app/services/notification.py index 146eb0d..4b580fe 100644 --- a/app/services/notification.py +++ b/app/services/notification.py @@ -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("测试通知发送完成") diff --git a/app/task/MAA.py b/app/task/MAA.py index 341d5b9..6580f4a 100644 --- a/app/task/MAA.py +++ b/app/task/MAA.py @@ -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通知" diff --git a/app/task/general.py b/app/task/general.py index 5c656ea..4a6b794 100644 --- a/app/task/general.py +++ b/app/task/general.py @@ -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通知"