feat(push-webhook): 支持传参动态原始数据

README.md

@@ -55,7 +55,7 @@ docker run -d -v [配置文件的绝对路径]/config.yml:/mnt/config.yml nfew/a
 | Telegram机器人   | telegram_bot     |    ✅    | 🪜需要自备网络环境<br/>👉https://core.telegram.org/bots                                                             |
 | Bark          | bark             |    ❌    | 🍎适合苹果系用户，十分轻量，但没法推送图片<br/>👉https://apps.apple.com/cn/app/id1403753865                                     |
 | Gotify        | gotify           |    ❌    | 🖥️适合自建服务器<br/>👉https://gotify.net                                                                         |
-| Webhook       | webhook          |    ❌    | ⚡️通用的方式                                                                                                     |
+| Webhook       | webhook          | ✅(POST) | ⚡️通用的方式，请求格式详见附录                                                                                            |
 | 电子邮件          | email            |    ✅    | 📧通用的方式                                                                                                     |
 
 ## 开发说明
@@ -79,6 +79,43 @@ docker run -d -v [配置文件的绝对路径]/config.yml:/mnt/config.yml nfew/a
 
 `nohup python3 -u main.py >& aio-dynamic-push.log &`
 
+## 附录
+
+### Webhook 支持的请求格式
+
+#### GET 请求
+
+```http request
+GET https://xxx.api.com?title={{title}}&content={{content}}
+```
+
+#### POST 请求
+
+```http request
+POST https://xxx.api.com
+Content-Type: application/json
+
+{
+  "query_task_config": {
+    "name": "任务名称",
+    "enable": true,
+    "type": "bilibili/weibo/xhs/douyin",
+    "intervals_second": 600,
+    "begin_time": "00:00",
+    "end_time": "23:59",
+    "target_push_name_list": [
+      "推送通道名称"
+    ],
+    "enable_living_check": false,
+    "enable_dynamic_check": true
+  },
+  "dynamic_raw_data": {
+    "key1": "value1",
+    "key2": "value2"
+  }
+}
+```
+
 ## 声明:
 
 - 本仓库发布的`aio-dynamic-push`项目中涉及的任何脚本，仅用于测试和学习研究，禁止用于商业用途

push_channel/__init__.py

@@ -12,10 +12,10 @@
 from .wecom_apps import WeComApps
 from .wecom_bot import WeComBot
 
-push_channel_dict = {}
+push_channel_dict: dict[str, PushChannel] = {}
 
 
-def get_push_channel(config):
+def get_push_channel(config) -> PushChannel:
     channel_type = config.get("type", None)
     if channel_type == "serverChan_turbo":
         return ServerChanTurbo(config)

push_channel/_push_channel.py

@@ -4,11 +4,12 @@ def __init__(self, config):
         self.enable = config.get("enable", False)
         self.type = config.get("type", "")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         """
         :param title: 标题
         :param content: 内容
         :param jump_url: 跳转url
         :param pic_url: 图片url
+        :param extend_data: 扩展数据
         """
         raise NotImplementedError("Subclasses must implement the push method")

push_channel/bark.py

@@ -10,7 +10,7 @@ def __init__(self, config):
         if self.key == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         push_url = f"https://api.day.app/{self.key}/{title}/{content}?url={jump_url}"
         response = util.requests_post(push_url, self.name)
         push_result = "成功" if util.check_response_is_ok(response) else "失败"

push_channel/demo.py

@@ -10,7 +10,7 @@ def __init__(self, config):
         if self.param == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         # 在这里实现推送逻辑，记得要在 push_channel/__init__.py 中注册推送通道
         push_result = "成功"
         log.info(f"【推送_{self.name}】{push_result}")

push_channel/dingtalk_bot.py

@@ -12,7 +12,7 @@ def __init__(self, config):
         if self.access_token == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         push_url = "https://oapi.dingtalk.com/robot/send"
         headers = {
             "Content-Type": "application/json"

push_channel/email.py

@@ -19,7 +19,7 @@ def __init__(self, config):
         if self.smtp_host == "" or self.smtp_port == "" or self.sender_email == "" or self.sender_password == "" or self.receiver_email == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         message = MIMEMultipart()
         message["Subject"] = title
         message["From"] = self.sender_email

push_channel/feishu_apps.py

@@ -21,7 +21,7 @@ def __init__(self, config):
         if self.app_id == "" or self.app_secret == "" or self.receive_id_type == "" or self.receive_id == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         tenant_access_token = self._get_tenant_access_token()
         if tenant_access_token is None:
             return None

push_channel/feishu_bot.py

@@ -12,7 +12,7 @@ def __init__(self, config):
         if self.webhook_key == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         push_url = f"https://open.feishu.cn/open-apis/bot/v2/hook/{self.webhook_key}"
         headers = {
             "Content-Type": "application/json"

push_channel/gotify.py

@@ -12,7 +12,7 @@ def __init__(self, config):
         if self.web_server_url == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         headers = {
             "Content-Type": "application/json",
         }

push_channel/server_chan_turbo.py

@@ -10,7 +10,7 @@ def __init__(self, config):
         if self.send_key == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         push_url = f"https://sctapi.ftqq.com/{self.send_key}.send"
         response = util.requests_post(push_url, self.name, params={"title": title, "desp": f"`{content}`[点我直达]({jump_url})"})
         push_result = "成功" if util.check_response_is_ok(response) else "失败"

push_channel/telegram_bot.py

@@ -13,7 +13,7 @@ def __init__(self, config):
         if self.api_token == "" or self.chat_id == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         push_url = f"https://api.telegram.org/bot{self.api_token}/sendMessage"
         headers = {
             "Content-Type": "application/json"

push_channel/webhook.py

@@ -11,15 +11,15 @@ def __init__(self, config):
         if self.webhook_url == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         if not self.webhook_url:
             log.warning(f"【推送_{self.name}】推送地址为空，跳过推送")
             return
         push_url = self.webhook_url.replace("{{title}}", title).replace("{{content}}", content)
         if self.request_method == "GET":
             response = util.requests_get(push_url, self.name)
         elif self.request_method == "POST":
-            response = util.requests_post(push_url, self.name)
+            response = util.requests_post(push_url, self.name, json=extend_data)
         else:
             log.error(f"【推送_{self.name}】不支持的请求方法：{self.request_method}")
             return

push_channel/wecom_apps.py

@@ -14,7 +14,7 @@ def __init__(self, config):
         if self.corp_id == "" or self.agent_id == "" or self.corp_secret == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         access_token = self._get_wechat_access_token()
         push_url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
         params = {

push_channel/wecom_bot.py

@@ -12,7 +12,7 @@ def __init__(self, config):
         if self.key == "":
             log.error(f"【推送_{self.name}】配置不完整，推送功能将无法正常使用")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         push_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
         headers = {
             "Content-Type": "application/json"

query_task/__init__.py

@@ -6,7 +6,7 @@
 from .query_xhs import QueryXhs
 
 
-def get_query_task(config):
+def get_query_task(config) -> QueryTask:
     task_type = config.get("type", None)
     if task_type == "bilibili":
         return QueryBilibili(config)

query_task/_query_task.py

@@ -49,13 +49,29 @@ def handle_for_result_null(self, null_id="-1", dict_key=None, module_name="未
                 # 当第一个元素不是 null_id 时，代表用户已经发布过动态了，此时可能是请求被拦截；否则代表用户持续没有发布动态
                 log.error(f"【{module_name}-查询动态状态-{self.name}】【{user_name}】动态列表为空")
 
-    def push(self, title, content, jump_url=None, pic_url=None):
+    def push(self, title, content, jump_url=None, pic_url=None, extend_data=None):
         for item in self.target_push_name_list:
             target_push_channel = push_channel.push_channel_dict.get(item, None)
             if target_push_channel is None:
                 log.error(f"【{self.type}】推送通道【{item}】不存在")
             else:
                 try:
-                    target_push_channel.push(title, content, jump_url, pic_url)
+                    if extend_data is None:
+                        extend_data = {}
+                    extend_data = {
+                        **extend_data,
+                        'query_task_config': {
+                            'name': self.name,
+                            'enable': self.enable,
+                            'type': self.type,
+                            'intervals_second': self.intervals_second,
+                            'begin_time': self.begin_time,
+                            'end_time': self.end_time,
+                            'target_push_name_list': self.target_push_name_list,
+                            'enable_dynamic_check': self.enable_dynamic_check,
+                            'enable_living_check': self.enable_living_check,
+                        },
+                    }
+                    target_push_channel.push(title, content, jump_url, pic_url, extend_data)
                 except Exception as e:
                     log.error(f"【{self.type}】推送通道【{item}】出错：{e}", exc_info=True)

query_task/query_bilibili.py

@@ -180,7 +180,7 @@ def query_dynamic_v2(self, uid=None, is_retry_by_buvid3=False):
                         content = module_dynamic["major"]["opus"]["title"]
                         pic_url = module_dynamic["major"]["opus"]["pics"][0]["url"]
                     log.info(f"【B站-查询动态状态-{self.name}】【{uname}】动态有更新，准备推送：{content[:30]}")
-                    self.push_for_bili_dynamic(uname, dynamic_id, content, pic_url, dynamic_type, dynamic_time, title_msg)
+                    self.push_for_bili_dynamic(uname, dynamic_id, content, pic_url, dynamic_type, dynamic_time, title_msg, dynamic_raw_data=item)
 
     @DeprecationWarning
     def query_dynamic(self, uid=None):
@@ -265,7 +265,7 @@ def query_dynamic(self, uid=None):
                         content = card["title"]
                         pic_url = card["image_urls"][0]
                     log.info(f"【B站-查询动态状态-{self.name}】【{uname}】动态有更新，准备推送：{content[:30]}")
-                    self.push_for_bili_dynamic(uname, dynamic_id, content, pic_url, dynamic_type, dynamic_time, title_msg)
+                    self.push_for_bili_dynamic(uname, dynamic_id, content, pic_url, dynamic_type, dynamic_time, title_msg, dynamic_raw_data=item)
 
     def query_live_status_batch(self, uid_list=None):
         if uid_list is None:
@@ -327,7 +327,7 @@ def get_headers(uid):
             "sec-fetch-site": "same-site",
         }
 
-    def push_for_bili_dynamic(self, uname=None, dynamic_id=None, content=None, pic_url=None, dynamic_type=None, dynamic_time=None, title_msg='发动态了'):
+    def push_for_bili_dynamic(self, uname=None, dynamic_id=None, content=None, pic_url=None, dynamic_type=None, dynamic_time=None, title_msg='发动态了', dynamic_raw_data=None):
         """
         B站动态提醒推送
         :param uname: up主名字
@@ -337,6 +337,7 @@ def push_for_bili_dynamic(self, uname=None, dynamic_id=None, content=None, pic_u
         :param dynamic_type: 动态类型
         :param dynamic_time: 动态发送时间
         :param title_msg: 推送标题
+        :param dynamic_raw_data: 动态原始数据
         """
         if uname is None or dynamic_id is None or content is None:
             log.error(f"【B站-动态提醒推送-{self.name}】缺少参数，uname:[{uname}]，dynamic_id:[{dynamic_id}]，content:[{content[:30]}]")
@@ -345,7 +346,7 @@ def push_for_bili_dynamic(self, uname=None, dynamic_id=None, content=None, pic_u
         title = f"【B站】【{uname}】{title_msg}"
         content = f"{content[:100] + (content[100:] and '...')}[{dynamic_time}]"
         dynamic_url = f"https://www.bilibili.com/opus/{dynamic_id}"
-        super().push(title, content, dynamic_url, pic_url)
+        super().push(title, content, dynamic_url, pic_url, extend_data={'dynamic_raw_data': dynamic_raw_data})
 
     def push_for_bili_live(self, uname=None, room_id=None, room_title=None, room_cover_url=None):
         """

query_task/query_demo.py

@@ -30,7 +30,7 @@ def query_dynamic(self, uid=None):
         # 在这里实现检测逻辑，记得要在 query_task/__init__.py 中注册任务类型
         self.push_for_xxx(username="用户名", dynamic_id="动态id", content="动态内容", pic_url="图片地址", jump_url="跳转地址", dynamic_time="动态发送时间")
 
-    def push_for_xxx(self, username=None, dynamic_id=None, content=None, pic_url=None, jump_url=None, dynamic_time=None):
+    def push_for_xxx(self, username=None, dynamic_id=None, content=None, pic_url=None, jump_url=None, dynamic_time=None, dynamic_raw_data=None):
         """
         XXX动态提醒推送
         :param username: 用户名
@@ -39,10 +39,11 @@ def push_for_xxx(self, username=None, dynamic_id=None, content=None, pic_url=Non
         :param pic_url: 图片地址
         :param jump_url: 跳转地址
         :param dynamic_time: 动态发送时间
+        :param dynamic_raw_data: 动态原始数据
         """
         if username is None or dynamic_id is None or content is None:
             log.error(f"【XXX-动态提醒推送-{self.name}】缺少参数，username:[{username}]，dynamic_id:[{dynamic_id}]，content:[{content[:30]}]")
             return
         title = f"【XXX】【{username}】发动态了"
         content = f"{content[:100] + (content[100:] and '...')}[{dynamic_time}]"
-        super().push(title, content, jump_url, pic_url)
+        super().push(title, content, jump_url, pic_url, extend_data={'dynamic_raw_data': dynamic_raw_data})

query_task/query_douyin.py

@@ -83,7 +83,7 @@ def query_dynamic(self, nickname=None, sec_uid=None):
                         pic_url = aweme["video"]["cover"]["url_list"][0]
                         video_url = f"https://www.douyin.com/video/{aweme_id}"
                         log.info(f"【抖音-查询动态状态-{self.name}】【{nickname}】动态有更新，准备推送：{content[:30]}")
-                        self.push_for_douyin_dynamic(nickname, aweme_id, content, pic_url, video_url)
+                        self.push_for_douyin_dynamic(nickname, aweme_id, content, pic_url, video_url, dynamic_raw_data=aweme)
                     except AttributeError:
                         log.error(f"【抖音-查询动态状态-{self.name}】dict取值错误，nickname：{nickname}")
                         return
@@ -178,21 +178,22 @@ def get_headers_for_live():
             "upgrade-insecure-requests": "1",
         }
 
-    def push_for_douyin_dynamic(self, nickname=None, aweme_id=None, content=None, pic_url=None, video_url=None):
+    def push_for_douyin_dynamic(self, nickname=None, aweme_id=None, content=None, pic_url=None, video_url=None, dynamic_raw_data=None):
         """
         抖音动态提醒推送
         :param nickname: 作者名
         :param aweme_id: 动态id
         :param content: 动态内容
         :param pic_url: 封面图片
         :param video_url: 视频地址
+        :param dynamic_raw_data: 动态原始数据
         """
         if nickname is None or aweme_id is None or content is None:
             log.error(f"【抖音-动态提醒推送-{self.name}】缺少参数，nickname:[{nickname}]，aweme_id:[{aweme_id}]，content:[{content[:30]}]")
             return
         title = f"【抖音】【{nickname}】发视频了"
         content = content[:100] + (content[100:] and "...")
-        super().push(title, content, video_url, pic_url)
+        super().push(title, content, video_url, pic_url, extend_data={'dynamic_raw_data': dynamic_raw_data})
 
     def push_for_douyin_live(self, nickname=None, room_stream_url=None, room_title=None, room_cover_url=None):
         """