feat(push): 使用Apprise推送

.github/workflows/deploy.yml

@@ -28,12 +28,13 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install certifi==2024.8.30 charset-normalizer==3.4.0 idna==3.10 requests==2.32.3 urllib3==2.2.3
+        pip install -r ./requirements
 
     - name: Run deployment script
       env:
         WXREAD_CURL_BASH: ${{ secrets.WXREAD_CURL_BASH }}
         PUSH_METHOD: ${{ secrets.PUSH_METHOD }}
+        APPRISE_URLS: ${{ secrets.APPRISE_URLS }}
         PUSHPLUS_TOKEN: ${{ secrets.PUSHPLUS_TOKEN }}
         WXPUSHER_SPT: ${{ secrets.WXPUSHER_SPT }}
         TELEGRAM_BOT_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }}

Dockerfile

@@ -12,15 +12,14 @@ RUN apt-get update && apt-get install -y cron && rm -rf /var/lib/apt/lists/*
 ENV PATH="/usr/local/bin:${PATH}"
 
 # 复制项目文件
-COPY main.py push.py config.py ./
+COPY main.py push.py config.py requirements.txt ./
+COPY apprise_plugins ./apprise_plugins
 
 # 创建日志目录并设置权限
 RUN mkdir -p /app/logs && chmod 777 /app/logs
 
 # 安装 Python 依赖
-RUN pip install --no-cache-dir \
-    requests>=2.32.3 \
-    urllib3>=2.2.3
+RUN pip install --no-cache-dir -r ./requirements.txt
 
 # 创建 cron 任务（每天凌晨1点执行）
 RUN echo "0 1 * * * cd /app && /usr/local/bin/python3 main.py >> /app/logs/\$(date +\%Y-\%m-\%d).log 2>&1" > /etc/cron.d/wxread-cron

README.md

@@ -49,6 +49,7 @@
 | `WXREAD_CURL_BASH`         | `read` 接口 `curl_bash`数据 | **必填**，必须提供有效指令                                   | secrets   |
 | `READ_NUM`                 | 阅读次数（每次 30 秒）              | **可选**，阅读时长，默认 60 分钟                           | variables |
 | `PUSH_METHOD`              | `pushplus`/`wxpusher`/`telegram`    | **可选**，推送方式，3选1，默认不推送                                       |    secrets     |
+| `APPRISE_URLS`             | Apprise支持的各类通知URL，包括Discord、Slack、Gotify、电子邮件等等，具体见[列表说明](https://github.com/caronc/apprise?tab=readme-ov-file#supported-notifications) <br> 此外本项目还额外支持`pushplus://{token}`和`wxpusherspt://{SPT}`两种形式 | **可选**，可以同时指定多个URL，不同URL之间以`,`隔开 <br> 当`PUSH_METHOD`不为空时此项无效 | secrets |
 | `PUSHPLUS_TOKEN`           | PushPlus 的 token                   | 当 `PUSH_METHOD=pushplus` 时必填，[获取地址](https://www.pushplus.plus/uc.html) | secrets   |
 | `WXPUSHER_SPT`             | WxPusher 的token                    | 当 `PUSH_METHOD=wxpusher` 时必填，[获取地址](https://wxpusher.zjiecode.com/docs/#/?id=获取spt) | secrets   |
 | `TELEGRAM_BOT_TOKEN`  <br>`TELEGRAM_CHAT_ID`   <br>`http_proxy`/`https_proxy`（可选）| 群组id以及机器人token                 | 当 `PUSH_METHOD=telegram` 时必填，[配置文档](https://www.nodeseek.com/post-22475-1) | secrets   |

apprise_plugins/pushplus.py

@@ -0,0 +1,25 @@
+import logging
+
+import requests
+from apprise.decorators import notify
+
+logger = logging.getLogger(__name__)
+
+
+@notify(on='pushplus', name='pushplus')
+def notify_pushplus(body: str, title: str, notify_type: str, meta: dict, *args, **kwargs):
+    """PushPlus消息推送"""
+    pushplus_url = "https://www.pushplus.plus/send"
+    token = meta.get('host', '')
+    response = requests.post(
+        pushplus_url,
+        json={
+            "token": token,
+            "title": title,
+            "content": body
+        },
+        headers={'Content-Type': 'application/json'},
+        timeout=10
+    )
+    response.raise_for_status()
+    logger.info("✅ PushPlus响应: %s", response.text)

apprise_plugins/wxpusherspt.py

@@ -0,0 +1,16 @@
+import logging
+
+import requests
+from apprise.decorators import notify
+
+logger = logging.getLogger(__name__)
+
+
+@notify(on='wxpusherspt', name='WxPusher SPT')
+def notify_wxpusher_spt(body: str, title: str, notify_type: str, meta: dict, *args, **kwargs):
+    """WxPusher消息推送（极简方式）"""
+    spt = meta.get('host', '')
+    wxpusher_simple_url = f"https://wxpusher.zjiecode.com/api/send/message/{spt}/{body}"
+    response = requests.get(wxpusher_simple_url, timeout=10)
+    response.raise_for_status()
+    logger.info("✅ WxPusher响应: %s", response.text)

config.py

@@ -11,6 +11,8 @@
 READ_NUM = int(os.getenv('READ_NUM') or 120)
 # 需要推送时可选，可选pushplus、wxpusher、telegram
 PUSH_METHOD = "" or os.getenv('PUSH_METHOD')
+# 通用推送配置
+APPRISE_URLS = [u.strip() for u in os.getenv("APPRISE_URLS", '').split(',')]
 # pushplus推送时需填
 PUSHPLUS_TOKEN = "" or os.getenv("PUSHPLUS_TOKEN")
 # telegram推送时需填

main.py

@@ -78,12 +78,10 @@ def get_wr_skey():
         else:
             ERROR_CODE = "❌ 无法获取新密钥或者WXREAD_CURL_BASH配置有误，终止运行。"
             logging.error(ERROR_CODE)
-            push(ERROR_CODE, PUSH_METHOD)
+            push(ERROR_CODE)
             raise Exception(ERROR_CODE)
     data.pop('s')
 
 logging.info("🎉 阅读脚本已完成！")
 
-if PUSH_METHOD not in (None, ''):
-    logging.info("⏱️ 开始推送...")
-    push(f"🎉 微信读书自动阅读完成！\n⏱️ 阅读时长：{(index - 1) * 0.5}分钟。", PUSH_METHOD)
+push(f"🎉 微信读书自动阅读完成！\n⏱️ 阅读时长：{(index - 1) * 0.5}分钟。")

push.py

@@ -1,108 +1,62 @@
-# push.py 支持 PushPlus 、wxpusher、Telegram 的消息推送模块
+# push.py 支持 Apprise、PushPlus 、wxpusher spt 的消息推送模块
+import logging
 import os
 import random
 import time
-import json
-import requests
-import logging
-from config import PUSHPLUS_TOKEN, TELEGRAM_CHAT_ID, TELEGRAM_BOT_TOKEN, WXPUSHER_SPT
-
-logger = logging.getLogger(__name__)
 
+from apprise import Apprise, AppriseAsset
 
-class PushNotification:
-    def __init__(self):
-        self.pushplus_url = "https://www.pushplus.plus/send"
-        self.telegram_url = "https://api.telegram.org/bot{}/sendMessage"
-        self.headers = {'Content-Type': 'application/json'}
-        # 从环境变量获取代理设置
-        self.proxies = {
-            'http': os.getenv('http_proxy'),
-            'https': os.getenv('https_proxy')
-        }
-        self.wxpusher_simple_url = "https://wxpusher.zjiecode.com/api/send/message/{}/{}"
+from config import (APPRISE_URLS, PUSH_METHOD, PUSHPLUS_TOKEN,
+                    TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, WXPUSHER_SPT)
 
-    def push_pushplus(self, content, token):
-        """PushPlus消息推送"""
-        attempts = 5
-        for attempt in range(attempts):
-            try:
-                response = requests.post(
-                    self.pushplus_url,
-                    data=json.dumps({
-                        "token": token,
-                        "title": "微信阅读推送...",
-                        "content": content
-                    }).encode('utf-8'),
-                    headers=self.headers,
-                    timeout=10
-                )
-                response.raise_for_status()
-                logger.info("✅ PushPlus响应: %s", response.text)
-                break  # 成功推送，跳出循环
-            except requests.exceptions.RequestException as e:
-                logger.error("❌ PushPlus推送失败: %s", e)
-                if attempt < attempts - 1:  # 如果不是最后一次尝试
-                    sleep_time = random.randint(180, 360)  # 随机3到6分钟
-                    logger.info("将在 %d 秒后重试...", sleep_time)
-                    time.sleep(sleep_time)
+logger = logging.getLogger(__name__)
 
-    def push_telegram(self, content, bot_token, chat_id):
-        """Telegram消息推送，失败时自动尝试直连"""
-        url = self.telegram_url.format(bot_token)
-        payload = {"chat_id": chat_id, "text": content}
 
-        try:
-            # 先尝试代理
-            response = requests.post(url, json=payload, proxies=self.proxies, timeout=30)
-            logger.info("✅ Telegram响应: %s", response.text)
-            response.raise_for_status()
+def push_one(content: str, url: str, apprise_obj: Apprise):
+    attempts = 5
+    result = False
+    for attempt in range(attempts):
+        result = apprise_obj.notify(body=content, title='微信阅读推送...')
+        if result:
+            logger.info(f"✅ 推送成功：{url}")
             return True
-        except Exception as e:
-            logger.error("❌ Telegram代理发送失败: %s", e)
-            try:
-                # 代理失败后直连
-                response = requests.post(url, json=payload, timeout=30)
-                response.raise_for_status()
-                return True
-            except Exception as e:
-                logger.error("❌ Telegram发送失败: %s", e)
-                return False
-
-    def push_wxpusher(self, content, spt):
-        """WxPusher消息推送（极简方式）"""
-        attempts = 5
-        url = self.wxpusher_simple_url.format(spt, content)
-
-        for attempt in range(attempts):
-            try:
-                response = requests.get(url, timeout=10)
-                response.raise_for_status()
-                logger.info("✅ WxPusher响应: %s", response.text)
-                break
-            except requests.exceptions.RequestException as e:
-                logger.error("❌ WxPusher推送失败: %s", e)
-                if attempt < attempts - 1:
-                    sleep_time = random.randint(180, 360)
-                    logger.info("将在 %d 秒后重试...", sleep_time)
-                    time.sleep(sleep_time)
+        else:
+            logger.error(f"❌ 推送失败：{url}")
+            if attempt < attempts - 1:
+                sleep_time = random.randint(180, 360)
+                logger.info(f"将在 {sleep_time} 秒后重试...")
+                time.sleep(sleep_time)
+    return False
 
 
 """外部调用"""
 
 
-def push(content, method):
-    """统一推送接口，支持 PushPlus、Telegram 和 WxPusher"""
-    notifier = PushNotification()
-
-    if method == "pushplus":
-        token = PUSHPLUS_TOKEN
-        return notifier.push_pushplus(content, token)
-    elif method == "telegram":
-        bot_token = TELEGRAM_BOT_TOKEN
-        chat_id = TELEGRAM_CHAT_ID
-        return notifier.push_telegram(content, bot_token, chat_id)
-    elif method == "wxpusher":
-        return notifier.push_wxpusher(content, WXPUSHER_SPT)
-    else:
-        raise ValueError("❌ 无效的通知渠道，请选择 'pushplus'、'telegram' 或 'wxpusher'")
+def push(content: str):
+    """统一推送接口，通过Apprise实现，额外支持 PushPlus 和 WxPusher 极简方式"""
+    urls = APPRISE_URLS
+    apprise_obj = Apprise(asset=AppriseAsset(
+        plugin_paths=os.path.join(__file__, '../apprise_plugins')
+    ))
+    if PUSH_METHOD == 'pushplus':
+        urls = [f'pushplus://{PUSHPLUS_TOKEN}']
+    elif PUSH_METHOD == 'wxpusher':
+        urls = [f'wxpusherspt://{WXPUSHER_SPT}']
+    elif PUSH_METHOD == 'telegram':
+        urls = [f'tgram://{TELEGRAM_BOT_TOKEN}/{TELEGRAM_CHAT_ID}']
+    if urls:
+        logging.info("⏱️ 开始推送...")
+    for url in urls:
+        # 由于Apprise不支持返回各个url成功与否的结果，故单独处理各个url
+        if not apprise_obj.add(url):
+            logger.error(f"❌ 通知渠道无效：{url}")
+            continue
+        if not push_one(content, url, apprise_obj):
+            old_no_proxy = os.getenv('no_proxy')
+            os.environ['no_proxy'] = '*'
+            push_one(content, url, apprise_obj)
+            if old_no_proxy:
+                os.environ['no_proxy'] = old_no_proxy
+            else:
+                os.environ.pop('no_proxy', None)
+        apprise_obj.clear()

requirements.txt

@@ -0,0 +1,3 @@
+requests>=2.32.3
+urllib3>=2.2.3
+apprise