Revert "refactor: 添加自定义Webhook推送功能及模板管理"

This reverts commit 57a3505cab.

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)}"
-        )

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配置")

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": "<b>{title}</b>\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()