增加 ntfy 通知 (#2537)

Co-authored-by: qiaoyun680 <qiaoyun680>

sample/notify.py

@@ -117,6 +117,10 @@ push_config = {
     'WEBHOOK_HEADERS': '',              # 自定义通知 请求头
     'WEBHOOK_METHOD': '',               # 自定义通知 请求方法
     'WEBHOOK_CONTENT_TYPE': ''          # 自定义通知 content-type
+
+    'NTFY_URL': '',                     # ntfy地址,如https://ntfy.sh
+    'NTFY_TOPIC': '',                   # ntfy的消息应用topic
+    'NTFY_PRIORITY':'3',                # 推送消息优先级,默认为3
 }
 # fmt: on
 
@@ -777,6 +781,32 @@ def chronocat(title: str, content: str) -> None:
                     print(f"QQ群消息:{ids}推送失败！")
 
 
+def ntfy(title: str, content: str) -> None:
+    """
+    通过 Ntfy 推送消息
+    """
+    if not push_config.get("NTFY_TOPIC"):
+        print("ntfy 服务的 NTFY_TOPIC 未设置!!\n取消推送")
+        return
+    print("ntfy 服务启动")
+    priority = '3'
+    if not push_config.get("NTFY_PRIORITY"):
+        print("ntfy 服务的NTFY_PRIORITY 未设置!!默认设置为3")
+    else:
+        priority = push_config.get("NTFY_PRIORITY")
+    data = (title + "\n" +content).encode(encoding='utf-8')
+    headers = {
+        "Title": "qinglong",
+        "Priority": priority
+    }
+    url = push_config.get("NTFY_URL") + "/" + push_config.get("NTFY_TOPIC")
+    response = requests.post(url, data=data, headers=headers)
+
+    if response["code"] == 200:
+        print("Ntfy 推送成功！")
+    else:
+        print("Ntfy 推送失败！错误信息：", response)
+
 def parse_headers(headers):
     if not headers:
         return {}
@@ -937,7 +967,8 @@ def add_notify_function():
         notify_function.append(chronocat)
     if push_config.get("WEBHOOK_URL") and push_config.get("WEBHOOK_METHOD"):
         notify_function.append(custom_notify)
-
+    if push_config.get("NTFY_TOPIC"):
+        notify_function.append(ntfy)
     if not notify_function:
         print(f"无推送渠道，请检查通知变量是否正确")
     return notify_function