From 13caed72076c5bfd268f1a8b132305e4980779c3 Mon Sep 17 00:00:00 2001 From: AoXuan Date: Sat, 27 Sep 2025 15:36:20 +0800 Subject: [PATCH] =?UTF-8?q?Revert=20"refactor:=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E8=87=AA=E5=AE=9A=E4=B9=89Webhook=E6=8E=A8=E9=80=81=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E5=8F=8A=E6=A8=A1=E6=9D=BF=E7=AE=A1=E7=90=86"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 57a3505cab773166e8876415787d8c6a0f0c2488. --- app/api/setting.py | 41 ---- app/models/schema.py | 24 -- app/services/notification.py | 429 ----------------------------------- 3 files changed, 494 deletions(-) diff --git a/app/api/setting.py b/app/api/setting.py index 80c660e..33a5210 100644 --- a/app/api/setting.py +++ b/app/api/setting.py @@ -28,7 +28,6 @@ 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=["全局设置"]) @@ -96,43 +95,3 @@ async def test_notify() -> OutBase: code=500, status="error", message=f"{type(e).__name__}: {str(e)}" ) return OutBase() - - -@router.post( - "/webhook/templates", summary="获取Webhook模板", response_model=InfoOut, status_code=200 -) -async def get_webhook_templates() -> InfoOut: - """获取所有可用的Webhook模板""" - - try: - templates = Notify.get_webhook_templates() - return InfoOut(data=templates) - except Exception as e: - return InfoOut( - code=500, - status="error", - message=f"{type(e).__name__}: {str(e)}", - data={} - ) - - -@router.post( - "/webhook/test", summary="测试单个Webhook", response_model=OutBase, status_code=200 -) -async def test_webhook(webhook_config: CustomWebhook = Body(...)) -> OutBase: - """测试单个Webhook配置""" - - try: - webhook_dict = webhook_config.model_dump() - await Notify.CustomWebhookPush( - "AUTO-MAS Webhook测试", - "这是一条测试消息，如果您收到此消息，说明Webhook配置正确！", - webhook_dict - ) - return OutBase(message="Webhook测试成功") - except Exception as e: - return OutBase( - code=500, - status="error", - message=f"Webhook测试失败: {type(e).__name__}: {str(e)}" - ) diff --git a/app/models/schema.py b/app/models/schema.py index 8af9012..9be5bb8 100644 --- a/app/models/schema.py +++ b/app/models/schema.py @@ -110,16 +110,6 @@ 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="自定义请求头") - body_template: Optional[str] = Field(default=None, description="自定义消息体模板") - - class GlobalConfig_Notify(BaseModel): SendTaskResultTime: Optional[Literal["不推送", "任何时刻", "仅失败时"]] = Field( default=None, description="任务结果推送时机" @@ -146,9 +136,6 @@ class GlobalConfig_Notify(BaseModel): CompanyWebHookBotUrl: Optional[str] = Field( default=None, description="企微Webhook Bot URL" ) - CustomWebhooks: Optional[List[CustomWebhook]] = Field( - default=None, description="自定义Webhook列表" - ) class GlobalConfig_Update(BaseModel): @@ -322,9 +309,6 @@ class UserConfig_Notify(BaseModel): CompanyWebHookBotUrl: Optional[str] = Field( default=None, description="企微Webhook Bot URL" ) - CustomWebhooks: Optional[List[CustomWebhook]] = Field( - default=None, description="自定义Webhook列表" - ) class MaaUserConfig(BaseModel): @@ -808,11 +792,3 @@ class UpdateCheckOut(OutBase): if_need_update: bool = Field(..., description="是否需要更新前端") latest_version: str = Field(..., description="最新前端版本号") update_info: Dict[str, List[str]] = Field(..., description="版本更新信息字典") - - -class WebhookTemplatesOut(OutBase): - data: Dict[str, Dict] = Field(..., description="Webhook模板数据") - - -class WebhookTestIn(BaseModel): - webhook: CustomWebhook = Field(..., description="要测试的Webhook配置") diff --git a/app/services/notification.py b/app/services/notification.py index bf680ba..146eb0d 100644 --- a/app/services/notification.py +++ b/app/services/notification.py @@ -22,13 +22,11 @@ import re import smtplib import requests -import json from email.header import Header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import formataddr from pathlib import Path -from typing import Dict, List, Optional from plyer import notification @@ -42,162 +40,6 @@ class Notification: def __init__(self): super().__init__() - self.webhook_templates = self._init_webhook_templates() - - def _init_webhook_templates(self) -> Dict[str, Dict]: - """初始化 Webhook 模板""" - return { - "企业微信": { - "name": "企业微信群机器人", - "description": "企业微信群机器人 Webhook 推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "msgtype": "text", - "text": {"content": "{title}\n{content}"} - }, - "image_template": { - "msgtype": "image", - "image": {"base64": "{image_base64}", "md5": "{image_md5}"} - }, - "url_example": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY" - }, - "钉钉": { - "name": "钉钉群机器人", - "description": "钉钉群机器人 Webhook 推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "msgtype": "text", - "text": {"content": "{title}\n{content}"} - }, - "url_example": "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN" - }, - "飞书": { - "name": "飞书群机器人", - "description": "飞书群机器人 Webhook 推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "msg_type": "text", - "content": {"text": "{title}\n{content}"} - }, - "url_example": "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_HOOK_ID" - }, - "Bark": { - "name": "Bark 推送", - "description": "Bark iOS 推送服务", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "title": "{title}", - "body": "{content}", - "sound": "default", - "group": "AUTO-MAS" - }, - "url_example": "https://api.day.app/YOUR_KEY/" - }, - "Bark_GET": { - "name": "Bark 推送 (GET方式)", - "description": "Bark iOS 推送服务 - GET 请求方式", - "headers": {}, - "body_template": {}, - "method": "GET", - "url_template": "https://api.day.app/YOUR_KEY/{title}/{content}?sound=default&group=AUTO-MAS", - "url_example": "https://api.day.app/YOUR_KEY/" - }, - "Server酱": { - "name": "Server酱推送", - "description": "Server酱微信推送服务", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "title": "{title}", - "desp": "{content}" - }, - "url_example": "https://sctapi.ftqq.com/YOUR_SEND_KEY.send" - }, - "PushPlus": { - "name": "PushPlus推送", - "description": "PushPlus 微信推送服务", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "token": "YOUR_TOKEN", - "title": "{title}", - "content": "{content}", - "template": "html" - }, - "url_example": "http://www.pushplus.plus/send" - }, - "QQ机器人": { - "name": "QQ机器人", - "description": "QQ 机器人推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "message": "{title}\n{content}" - }, - "url_example": "http://your-qq-bot-server/send" - }, - "Telegram": { - "name": "Telegram Bot", - "description": "Telegram 机器人推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "chat_id": "YOUR_CHAT_ID", - "text": "{title}\n{content}", - "parse_mode": "HTML" - }, - "url_example": "https://api.telegram.org/botYOUR_BOT_TOKEN/sendMessage" - }, - "Discord": { - "name": "Discord Webhook", - "description": "Discord 频道 Webhook 推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "content": "**{title}**\n{content}", - "username": "AUTO-MAS" - }, - "url_example": "https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_WEBHOOK_TOKEN" - }, - "Slack": { - "name": "Slack Webhook", - "description": "Slack 频道 Webhook 推送", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "text": "*{title}*\n{content}", - "username": "AUTO-MAS" - }, - "url_example": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK" - }, - "Gotify": { - "name": "Gotify 推送", - "description": "Gotify 自托管推送服务", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "title": "{title}", - "message": "{content}", - "priority": 5 - }, - "url_example": "https://your-gotify-server/message?token=YOUR_TOKEN" - }, - "Ntfy": { - "name": "Ntfy 推送", - "description": "Ntfy 推送服务", - "headers": {"Content-Type": "text/plain"}, - "body_template": "{title}\n{content}", - "url_example": "https://ntfy.sh/YOUR_TOPIC" - }, - "自定义": { - "name": "自定义格式", - "description": "完全自定义的 Webhook 格式", - "headers": {"Content-Type": "application/json"}, - "body_template": { - "title": "{title}", - "content": "{content}", - "timestamp": "{timestamp}" - }, - "url_example": "https://your-custom-webhook-url" - } - } - - def get_webhook_templates(self) -> Dict[str, Dict]: - """获取所有可用的 Webhook 模板""" - return self.webhook_templates async def push_plyer(self, title, message, ticker, t) -> bool: """ @@ -392,194 +234,6 @@ class Notification: else: raise Exception(f"企业微信群机器人推送图片失败: {response.text}") - async def CustomWebhookPush(self, title: str, content: str, webhook_config: Dict) -> 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_type = webhook_config.get("template", "自定义") - template = self.webhook_templates.get(template_type, self.webhook_templates["自定义"]) - - # 获取请求方法 - method = template.get("method", "POST").upper() - - # 设置请求头 - headers = webhook_config.get("headers", template.get("headers", {})) - - # 添加时间戳 - import time - timestamp = str(int(time.time())) - - try: - if method == "GET" and template.get("url_template"): - # 使用 URL 模板的 GET 请求（如 Bark GET 方式） - url_template = template["url_template"] - # URL 编码标题和内容 - import urllib.parse - encoded_title = urllib.parse.quote(title) - encoded_content = urllib.parse.quote(content) - - # 替换 URL 模板中的变量 - final_url = webhook_config["url"] - if not final_url.endswith('/'): - final_url += '/' - final_url += f"{encoded_title}/{encoded_content}" - - # 添加查询参数 - if "?" in url_template: - query_part = url_template.split("?", 1)[1] - query_part = query_part.replace("{title}", encoded_title).replace("{content}", encoded_content) - final_url += f"?{query_part}" - - response = requests.get( - url=final_url, - headers=headers, - timeout=10, - proxies=Config.get_proxies() - ) - else: - # POST 请求 - # 使用自定义模板或默认模板 - if webhook_config.get("body_template"): - try: - body_template = json.loads(webhook_config["body_template"]) - except json.JSONDecodeError: - body_template = template.get("body_template", {}) - else: - body_template = template.get("body_template", {}) - - # 处理不同的数据类型 - if isinstance(body_template, dict): - # JSON 格式 - body_str = json.dumps(body_template) - body_str = body_str.replace("{title}", title).replace("{content}", content).replace("{timestamp}", timestamp) - data = json.loads(body_str) - - response = requests.post( - url=webhook_config["url"], - json=data, - headers=headers, - timeout=10, - proxies=Config.get_proxies() - ) - else: - # 纯文本格式（如 Ntfy） - body_str = str(body_template) - body_str = body_str.replace("{title}", title).replace("{content}", content).replace("{timestamp}", timestamp) - - response = requests.post( - url=webhook_config["url"], - data=body_str, - headers=headers, - timeout=10, - proxies=Config.get_proxies() - ) - - # 检查响应 - if response.status_code in [200, 201, 204]: - # 尝试解析JSON响应 - try: - result = response.json() - # 企业微信/钉钉等返回格式检查 - if "errcode" in result: - if result["errcode"] == 0: - logger.success(f"自定义Webhook推送成功: {webhook_config.get('name', 'Unknown')} - {title}") - else: - raise Exception(f"Webhook推送失败: {result}") - elif "code" in result and result["code"] != 200: - raise Exception(f"Webhook推送失败: {result}") - else: - logger.success(f"自定义Webhook推送成功: {webhook_config.get('name', 'Unknown')} - {title}") - except json.JSONDecodeError: - # 非JSON响应，但状态码成功认为推送成功 - logger.success(f"自定义Webhook推送成功: {webhook_config.get('name', 'Unknown')} - {title}") - else: - raise Exception(f"HTTP {response.status_code}: {response.text}") - - except Exception as e: - logger.error(f"自定义Webhook推送失败 {webhook_config.get('name', 'Unknown')}: {str(e)}") - raise - - async def CustomWebhookPushImage(self, image_path: Path, webhook_config: Dict) -> None: - """ - 自定义 Webhook 推送图片通知 - - :param image_path: 图片文件路径 - :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_type = webhook_config.get("template", "自定义") - template = self.webhook_templates.get(template_type, self.webhook_templates["自定义"]) - - # 只有支持图片的模板才处理图片推送 - if "image_template" not in template: - logger.warning(f"Webhook模板 {template_type} 不支持图片推送") - return - - # 压缩图片 - ImageUtils.compress_image_if_needed(image_path) - - # 检查图片是否存在 - if not image_path.exists(): - raise FileNotFoundError(f"文件未找到: {image_path}") - - # 获取图片base64和md5 - image_base64 = ImageUtils.get_base64_from_file(str(image_path)) - image_md5 = ImageUtils.calculate_md5_from_file(str(image_path)) - - # 替换模板中的变量 - body_template = template["image_template"] - body_str = json.dumps(body_template) - body_str = body_str.replace("{image_base64}", image_base64).replace("{image_md5}", image_md5) - data = json.loads(body_str) - - # 设置请求头 - headers = webhook_config.get("headers", template["headers"]) - - try: - response = requests.post( - url=webhook_config["url"], - json=data, - headers=headers, - timeout=10, - proxies=Config.get_proxies() - ) - - if response.status_code == 200: - try: - result = response.json() - if "errcode" in result: - if result["errcode"] == 0: - logger.success(f"自定义Webhook图片推送成功: {webhook_config.get('name', 'Unknown')} - {image_path.name}") - else: - raise Exception(f"Webhook图片推送失败: {result}") - else: - logger.success(f"自定义Webhook图片推送成功: {webhook_config.get('name', 'Unknown')} - {image_path.name}") - except json.JSONDecodeError: - logger.success(f"自定义Webhook图片推送成功: {webhook_config.get('name', 'Unknown')} - {image_path.name}") - else: - raise Exception(f"HTTP {response.status_code}: {response.text}") - - except Exception as e: - logger.error(f"自定义Webhook图片推送失败 {webhook_config.get('name', 'Unknown')}: {str(e)}") - raise - async def send_test_notification(self) -> None: """发送测试通知到所有已启用的通知渠道""" @@ -622,90 +276,7 @@ class Notification: Config.get("Notify", "CompanyWebHookBotUrl"), ) - # 发送自定义Webhook通知 - custom_webhooks = Config.get("Notify", "CustomWebhooks", []) - for webhook in custom_webhooks: - if webhook.get("enabled", True): - try: - await self.CustomWebhookPush( - "AUTO-MAS测试通知", - "这是 AUTO-MAS 外部通知测试信息。如果你看到了这段内容, 说明 AUTO-MAS 的通知功能已经正确配置且可以正常工作！", - webhook - ) - # 如果支持图片推送，也测试图片 - if webhook.get("template") in ["企业微信"]: - await self.CustomWebhookPushImage( - Path.cwd() / "res/images/notification/test_notify.png", - webhook - ) - except Exception as e: - logger.error(f"自定义Webhook测试失败 {webhook.get('name', 'Unknown')}: {str(e)}") - logger.success("测试通知发送完成") - async def send_notification_to_all_channels(self, title: str, content: str, user_config: Optional[Dict] = None, image_path: Optional[Path] = None) -> None: - """ - 发送通知到所有已启用的通知渠道 - - :param title: 通知标题 - :param content: 通知内容 - :param user_config: 用户特定配置（可选） - :param image_path: 图片路径（可选） - """ - - # 使用用户配置或全局配置 - notify_config = user_config.get("Notify", {}) if user_config else {} - - # 发送系统通知 - if Config.get("Notify", "IfPushPlyer"): - try: - await self.push_plyer(title, content, title, 5) - except Exception as e: - logger.error(f"系统通知发送失败: {str(e)}") - - # 发送邮件通知 - if notify_config.get("IfSendMail") or (not user_config and Config.get("Notify", "IfSendMail")): - try: - to_address = notify_config.get("ToAddress") or Config.get("Notify", "ToAddress") - if to_address: - await self.send_mail("文本", title, content, to_address) - except Exception as e: - logger.error(f"邮件通知发送失败: {str(e)}") - - # 发送Server酱通知 - if notify_config.get("IfServerChan") or (not user_config and Config.get("Notify", "IfServerChan")): - try: - server_chan_key = notify_config.get("ServerChanKey") or Config.get("Notify", "ServerChanKey") - if server_chan_key: - await self.ServerChanPush(title, content, server_chan_key) - except Exception as e: - logger.error(f"Server酱通知发送失败: {str(e)}") - - # 发送企业微信Webhook通知 - if notify_config.get("IfCompanyWebHookBot") or (not user_config and Config.get("Notify", "IfCompanyWebHookBot")): - try: - webhook_url = notify_config.get("CompanyWebHookBotUrl") or Config.get("Notify", "CompanyWebHookBotUrl") - if webhook_url: - await self.WebHookPush(title, content, webhook_url) - if image_path and image_path.exists(): - await self.CompanyWebHookBotPushImage(image_path, webhook_url) - except Exception as e: - logger.error(f"企业微信Webhook通知发送失败: {str(e)}") - - # 发送自定义Webhook通知 - custom_webhooks = notify_config.get("CustomWebhooks", []) - if not custom_webhooks and not user_config: - custom_webhooks = Config.get("Notify", "CustomWebhooks", []) - - for webhook in custom_webhooks: - if webhook.get("enabled", True): - try: - await self.CustomWebhookPush(title, content, webhook) - # 如果支持图片推送且有图片 - if image_path and image_path.exists() and webhook.get("template") in ["企业微信"]: - await self.CustomWebhookPushImage(image_path, webhook) - except Exception as e: - logger.error(f"自定义Webhook通知发送失败 {webhook.get('name', 'Unknown')}: {str(e)}") - Notify = Notification()