From 2158a3227e1158078434dac2a0b538700bd3b433 Mon Sep 17 00:00:00 2001
From: Zebartin <16185081+Zebartin@users.noreply.github.com>
Date: Tue, 18 Mar 2025 21:55:45 +0800
Subject: [PATCH] =?UTF-8?q?feat(push):=20=E4=BD=BF=E7=94=A8Apprise?=
=?UTF-8?q?=E6=8E=A8=E9=80=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.github/workflows/deploy.yml | 3 +-
Dockerfile | 7 +-
README.md | 1 +
apprise_plugins/pushplus.py | 25 ++++++
apprise_plugins/wxpusherspt.py | 16 ++++
config.py | 2 +
main.py | 6 +-
push.py | 142 +++++++++++----------------------
requirements.txt | 3 +
9 files changed, 102 insertions(+), 103 deletions(-)
create mode 100644 apprise_plugins/pushplus.py
create mode 100644 apprise_plugins/wxpusherspt.py
create mode 100644 requirements.txt
diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml
index 8922c1db..ea89370c 100644
--- a/.github/workflows/deploy.yml
+++ b/.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 }}
diff --git a/Dockerfile b/Dockerfile
index 4e605031..167aa71a 100644
--- a/Dockerfile
+++ b/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
diff --git a/README.md b/README.md
index 603130fb..f5aa0efd 100644
--- a/README.md
+++ b/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)
此外本项目还额外支持`pushplus://{token}`和`wxpusherspt://{SPT}`两种形式 | **可选**，可以同时指定多个URL，不同URL之间以`,`隔开
当`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`
`TELEGRAM_CHAT_ID`
`http_proxy`/`https_proxy`（可选）| 群组id以及机器人token | 当 `PUSH_METHOD=telegram` 时必填，[配置文档](https://www.nodeseek.com/post-22475-1) | secrets |
diff --git a/apprise_plugins/pushplus.py b/apprise_plugins/pushplus.py
new file mode 100644
index 00000000..d8c5c1ae
--- /dev/null
+++ b/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)
diff --git a/apprise_plugins/wxpusherspt.py b/apprise_plugins/wxpusherspt.py
new file mode 100644
index 00000000..1f859d6d
--- /dev/null
+++ b/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)
diff --git a/config.py b/config.py
index ba702b21..e77577d3 100644
--- a/config.py
+++ b/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推送时需填
diff --git a/main.py b/main.py
index a89466cc..bf4a67bc 100644
--- a/main.py
+++ b/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}分钟。")
diff --git a/push.py b/push.py
index 02a967f9..f4008769 100644
--- a/push.py
+++ b/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()
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 00000000..236a3e80
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,3 @@
+requests>=2.32.3
+urllib3>=2.2.3
+apprise
\ No newline at end of file