From 1127d88f3b7cc4cb18cc6f2a7a983d15800c7e21 Mon Sep 17 00:00:00 2001 From: ramen <1205925392@qq.com> Date: Fri, 15 Nov 2024 00:33:16 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8Dv2=E5=BE=AE=E4=BF=A1?= =?UTF-8?q?=E9=80=9A=E7=9F=A5=EF=BC=8C=E5=8F=AF=E4=BB=A5=E6=8C=87=E5=AE=9A?= =?UTF-8?q?=E5=BE=AE=E4=BF=A1=E9=80=9A=E7=9F=A5ID?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- package.json | 5 +- plugins/dynamicwechat/__init__.py | 45 +++++++++------- plugins/dynamicwechat/notify_helper.py | 75 +++++++++++++++----------- 3 files changed, 75 insertions(+), 50 deletions(-) diff --git a/package.json b/package.json index f73f643..764f2bb 100644 --- a/package.json +++ b/package.json @@ -861,14 +861,15 @@ }, "DynamicWeChat": { "name": "动态企微可信IP", - "description": "修改企微应用可信IP，详细说明查看'作者主页'，支持第三方通知。验证码以？结尾发送到企业微信应用", + "description": "修改企微应用可信IP，支持Srever酱等第三方通知。验证码以？结尾发送到企业微信应用", "labels": "消息通知", - "version": "1.5.0", + "version": "1.5.1", "icon": "Wecom_A.png", "author": "RamenRa", "level": 2, "v2": true, "history": { + "v1.5.1": "修复v2微信通知，可以指定微信通知ID", "v1.5.0": "支持企微应用通知和第Serve酱等第三方推送。按要求修改插件名称", "v1.4.1": "完善面板说明", "v1.4.0": "修复强制更改IP时配置面板延时过长的问题。庆祝v2进入正式版，显示了一个没用的参数", diff --git a/plugins/dynamicwechat/__init__.py b/plugins/dynamicwechat/__init__.py index 0b1942a..aa16cfc 100644 --- a/plugins/dynamicwechat/__init__.py +++ b/plugins/dynamicwechat/__init__.py @@ -20,7 +20,7 @@ from app.log import logger from app.plugins import _PluginBase from app.plugins.dynamicwechat.update_help import PyCookieCloud from app.plugins.dynamicwechat.notify_helper import MySender -from app.schemas.types import EventType, NotificationType +from app.schemas.types import EventType class DynamicWeChat(_PluginBase): @@ -31,7 +31,7 @@ class DynamicWeChat(_PluginBase): # 插件图标 plugin_icon = "Wecom_A.png" # 插件版本 - plugin_version = "1.5.0" + plugin_version = "1.5.1" # 插件作者 plugin_author = "RamenRa" # 作者主页 @@ -42,6 +42,8 @@ class DynamicWeChat(_PluginBase): plugin_order = 47 # 可使用的用户级别 auth_level = 2 + # 检测间隔时间,默认10分钟 + _refresh_cron = '*/20 * * * *' # ------------------------------------------私有属性------------------------------------------ _enabled = False # 开关 @@ -70,8 +72,7 @@ class DynamicWeChat(_PluginBase): _current_ip_address = '0.0.0.0' # 企业微信登录 _wechatUrl = 'https://work.weixin.qq.com/wework_admin/loginpage_wx?from=myhome' - # 检测间隔时间,默认10分钟 - _refresh_cron = '*/20 * * * *' + # 输入的企业应用id _input_id_list = '' # 二维码 @@ -99,6 +100,10 @@ class DynamicWeChat(_PluginBase): # 定时器 _scheduler: Optional[BackgroundScheduler] = None + if hasattr(settings, 'VERSION_FLAG'): + version = settings.VERSION_FLAG # V2 + else: + version = "v1" def init_plugin(self, config: dict = None): # 清空配置 self._notification_token = '' @@ -123,7 +128,10 @@ class DynamicWeChat(_PluginBase): self._use_cookiecloud = config.get("use_cookiecloud") self._cookie_header = config.get("cookie_header") self._ip_changed = config.get("ip_changed") - self._my_send = MySender(self._notification_token) + if self.version != "v1": + self._my_send = MySender(self._notification_token, func=self.post_message) + else: + self._my_send = MySender(self._notification_token) if not self._my_send.init_success: # 没有输入通知方式，不通知 self._my_send = None # 停止现有任务 @@ -258,12 +266,14 @@ class DynamicWeChat(_PluginBase): if not event_data or event_data.get("action") != "dynamicwechat": return - logger.info("开始检测公网IP") - if self.CheckIP(): - self.ChangeIP() - self.__update_config() - - logger.info("----------------------本次任务结束----------------------") + if self._cookie_valid: + logger.info("开始检测公网IP") + if self.CheckIP(): + self.ChangeIP() + self.__update_config() + logger.info("----------------------本次任务结束----------------------") + else: + logger.warning("cookie已失效请及时更新，不检测公网IP") def CheckIP(self): retry_urls = random.sample(self._ip_urls, len(self._ip_urls)) @@ -381,7 +391,7 @@ class DynamicWeChat(_PluginBase): img_src, refuse_time = self.find_qrc(page) if img_src: if self._my_send: - result = self._my_send.send("企业微信登录二维码", content=None, image=img_src, force_send=False) + result = self._my_send.send(title="企业微信登录二维码", image=img_src) if result: logger.info(f"二维码发送失败，原因：{result}") browser.close() @@ -815,7 +825,7 @@ class DynamicWeChat(_PluginBase): 'props': { 'type': 'info', 'variant': 'tonal', - 'text': '建议启用内建或自定义CookieCloud。支持微信、Server酱、PushPlus、AnPush，具体请查看作者主页' + 'text': '建议启用内建或自定义CookieCloud。支持微信、Server酱等第三方通知，具体请查看作者主页' } } ] @@ -835,7 +845,7 @@ class DynamicWeChat(_PluginBase): 'component': 'VAlert', 'props': { 'type': 'info', - 'text': 'cookie失效时通知用户，用户使用/push_qr让插件推送二维码。使用第三方通知时填写对应Token/API' + 'text': 'Cookie失效时通知用户，用户使用/push_qr让插件推送二维码。使用第三方通知时填写对应Token/API' } } ] @@ -1004,8 +1014,7 @@ class DynamicWeChat(_PluginBase): image_src, refuse_time = self.find_qrc(page) if image_src: if self._my_send: - # logger.info(f"远程推送任务: {image_src}") - result = self._my_send.send("企业微信登录二维码", content=None, image=image_src, force_send=False) + result = self._my_send.send("企业微信登录二维码", image=image_src) if result: logger.info(f"远程推送任务: 二维码发送失败，原因：{result}") browser.close() @@ -1019,7 +1028,7 @@ class DynamicWeChat(_PluginBase): # logger.info("远程推送任务: 没有可用的CookieCloud服务器，只修改可信IP") self.click_app_management_buttons(page) else: - logger.warning("远程推送任务: 任何通知方式") + logger.warning("远程推送任务: 没有找到可用的通知方式") else: logger.warning("远程推送任务: 未找到二维码") browser.close() @@ -1032,7 +1041,7 @@ class DynamicWeChat(_PluginBase): { "cmd": "/push_qr", "event": EventType.PluginAction, - "desc": "立即推送登录二维码到", + "desc": "立即推送登录二维码", "category": "", "data": { "action": "push_qrcode" diff --git a/plugins/dynamicwechat/notify_helper.py b/plugins/dynamicwechat/notify_helper.py index 92c2e35..6c70a5d 100644 --- a/plugins/dynamicwechat/notify_helper.py +++ b/plugins/dynamicwechat/notify_helper.py @@ -1,48 +1,44 @@ import re import requests -from app.modules.wechat.wechat import WeChat +from app.modules.wechat import WeChat +from app.schemas.types import NotificationType class MySender: - def __init__(self, token=None): - if not token: # 如果 token 为空 - self.token = None - self.channel = None - self.init_success = False # 标识初始化失败 - else: - self.token = token - self.channel = self.send_channel() # 初始化时确定发送渠道 - self.first_text_sent = False # 记录是否已经发送过纯文本消息 - self.init_success = True # 标识初始化成功 + def __init__(self, token=None, func=None): + self.token = token + self.channel = self.send_channel() if token else None # 初始化时确定发送渠道 + self.first_text_sent = False # 记录是否已发送过纯文本消息 + self.init_success = bool(token) # 标识初始化成功 + self.post_message_func = func # V2微信模式的 post_message 方法 def send_channel(self): - if self.token: - if self.token == "WeChat": - return "WeChat" + if "WeChat" in self.token: + return "WeChat" - letters_only = ''.join(re.findall(r'[A-Za-z]', self.token)) - # 判断其他推送渠道 - if self.token.startswith("SCT"): - return "ServerChan" - elif letters_only.isupper(): - return "AnPush" - else: - return "PushPlus" - return None + letters_only = ''.join(re.findall(r'[A-Za-z]', self.token)) + if self.token.lower().startswith("sct".lower()): + return "ServerChan" + elif letters_only.isupper(): + return "AnPush" + else: + return "PushPlus" # 标题，内容，图片，是否强制发送 - def send(self, title, content, image=None, force_send=False, diy_channel=None): + def send(self, title, content=None, image=None, force_send=False, diy_channel=None): if not self.init_success: return # 如果初始化失败，直接返回 - # 判断发送的内容类型 - contains_image = bool(image) # 是否包含图片 - if not contains_image and not force_send: + if not image and not force_send: if self.first_text_sent: return else: self.first_text_sent = True + # # 如果是 V2 微信通知直接处理 + if self.channel == "WeChat" and self.post_message_func: + return self.send_v2_wechat(title, content, image) + try: if not diy_channel: channel = self.channel @@ -50,7 +46,7 @@ class MySender: channel = diy_channel if channel == "WeChat": - return MySender.send_wechat(title, content, image) + return MySender.send_wechat(title, content, image, self.token) elif channel == "ServerChan": return self.send_serverchan(title, content, image) elif channel == "AnPush": @@ -63,10 +59,14 @@ class MySender: return f"Error occurred: {str(e)}" @staticmethod - def send_wechat(title, content, image): + def send_wechat(title, content, image, token): wechat = WeChat() + if ',' in token: + channel, actual_userid = token.split(',', 1) + else: + actual_userid = None if image: - send_status = wechat.send_msg(title='企业微信登录二维码', image=image, link=image) + send_status = wechat.send_msg(title='企业微信登录二维码', image=image, link=image, userid=actual_userid) else: send_status = wechat.send_msg(title=title, text=content) @@ -128,6 +128,21 @@ class MySender: return f"PushPlus send failed: {result.get('msg')}" return None + def send_v2_wechat(self, title, content, image): + """V2 微信通知发送""" + if not self.token or ',' not in self.token: + return '没有指定V2微信用户ID' + channel, actual_userid = self.token.split(',', 1) + self.post_message_func( + mtype=NotificationType.Plugin, + title=title, + text=content, + image=image, + link=image, + userid=actual_userid + ) + return None + def reset_limit(self): """解除限制，允许再次发送纯文本消息""" self.first_text_sent = False