mirror of
https://github.com/jxxghp/MoviePilot-Plugins.git
synced 2026-03-27 10:05:57 +00:00
Merge pull request #548 from RamenRa/master
This commit is contained in:
jxxghp
GitHub
14
package.json
14
package.json
@@ -857,23 +857,21 @@
|
||||
"v2": true
|
||||
},
|
||||
"DynamicWeChat": {
|
||||
"name": "假的企业微信可信IP",
|
||||
"description": "优先使用cookie，可本地扫码刷新Cookie，验证码以？结尾发送到企业微信应用",
|
||||
"name": "动态企微可信IP",
|
||||
"description": "修改企微应用可信IP，详细说明查看'作者主页'，支持第三方通知。验证码以？结尾发送到企业微信应用",
|
||||
"labels": "消息通知",
|
||||
"version": "1.4.1",
|
||||
"version": "1.5.0",
|
||||
"icon": "Wecom_A.png",
|
||||
"author": "RamenRa",
|
||||
"level": 2,
|
||||
"v2": true,
|
||||
"history": {
|
||||
"v1.4.1": "完善面板说明，为了和原仓库做出明显区别在插件名称添加了特殊字符",
|
||||
"v1.5.0": "支持企微应用通知和第Serve酱等第三方推送。按要求修改插件名称",
|
||||
"v1.4.1": "完善面板说明",
|
||||
"v1.4.0": "修复强制更改IP时配置面板延时过长的问题。庆祝v2进入正式版，显示了一个没用的参数",
|
||||
"v1.3.1": "修正一些逻辑判断，修改ip成功会通知一次",
|
||||
"v1.3.0": "兼容v2，操作cookie前检查一次CookieCloud",
|
||||
"v1.2.0": "远程命令/push_qr，立即推送一次二维码到pushplus。添加<本地扫码刷新cookie>",
|
||||
"v1.1.5": "将chromium运行设置为headless模式",
|
||||
"v1.1.4": "放弃self.post_message()的消息推送，还原成send_pushplus_message()",
|
||||
"v1.1.3": "关闭cookie输入框，延长cookie任务成功时不输出日志，使用设定中的CookieCloud设置"
|
||||
"v1.2.0": "远程命令/push_qr，立即推送一次二维码到pushplus。添加<本地扫码刷新cookie>"
|
||||
}
|
||||
},
|
||||
"SyncCookieCloud": {
|
||||
|
||||
@@ -19,18 +19,19 @@ from app.helper.cookiecloud import CookieCloudHelper
|
||||
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
|
||||
|
||||
|
||||
class DynamicWeChat(_PluginBase):
|
||||
# 插件名称
|
||||
plugin_name = "假的企业微信可信IP"
|
||||
plugin_name = "动态企微可信IP"
|
||||
# 插件描述
|
||||
plugin_desc = "优先使用cookie，可本地扫码刷新Cookie，验证码以？结尾发送到企业微信应用"
|
||||
plugin_desc = "修改企微应用可信IP，详细说明查看'作者主页'，支持第三方通知。验证码以？结尾发送到企业微信应用"
|
||||
# 插件图标
|
||||
plugin_icon = "Wecom_A.png"
|
||||
# 插件版本
|
||||
plugin_version = "1.4.1"
|
||||
plugin_version = "1.5.0"
|
||||
# 插件作者
|
||||
plugin_author = "RamenRa"
|
||||
# 作者主页
|
||||
@@ -56,6 +57,10 @@ class DynamicWeChat(_PluginBase):
|
||||
_local_scan = False
|
||||
# 类初始化时添加标记变量
|
||||
_is_special_upload = False
|
||||
# 聚合通知
|
||||
_my_send = None
|
||||
# 通知方式token/api
|
||||
_notification_token = ''
|
||||
|
||||
# 匹配ip地址的正则
|
||||
_ip_pattern = r'\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b'
|
||||
@@ -69,17 +74,16 @@ class DynamicWeChat(_PluginBase):
|
||||
_refresh_cron = '*/20 * * * *'
|
||||
# 输入的企业应用id
|
||||
_input_id_list = ''
|
||||
# helloimg的token
|
||||
_helloimg_s_token = ""
|
||||
# pushplus的token
|
||||
_pushplus_token = ""
|
||||
# 二维码
|
||||
_qr_code_image = None
|
||||
# 用户消息
|
||||
text = ""
|
||||
# 手机验证码
|
||||
_verification_code = ''
|
||||
# 过期时间
|
||||
_future_timestamp = 0
|
||||
# 配置文件路径
|
||||
_settings_file_path = None
|
||||
|
||||
# cookie有效检测
|
||||
_cookie_valid = True
|
||||
@@ -97,30 +101,31 @@ class DynamicWeChat(_PluginBase):
|
||||
|
||||
def init_plugin(self, config: dict = None):
|
||||
# 清空配置
|
||||
self._helloimg_s_token = ''
|
||||
self._pushplus_token = ''
|
||||
self._notification_token = ''
|
||||
self._ip_changed = True
|
||||
self._forced_update = False
|
||||
self._use_cookiecloud = True
|
||||
self._local_scan = False
|
||||
self._input_id_list = ''
|
||||
self._cookie_header = ""
|
||||
self._current_ip_address = self.get_ip_from_url(self._ip_urls[0])
|
||||
self._cookie_lifetime = PyCookieCloud.load_cookie_lifetime()
|
||||
self._current_ip_address = self.get_ip_from_url(random.choice(self._ip_urls))
|
||||
self._settings_file_path = self.get_data_path() / "settings.json"
|
||||
# self._cookie_lifetime = PyCookieCloud.load_cookie_lifetime()
|
||||
if config:
|
||||
self._enabled = config.get("enabled")
|
||||
self._notification_token = config.get("notification_token")
|
||||
self._cron = config.get("cron")
|
||||
self._onlyonce = config.get("onlyonce")
|
||||
self._input_id_list = config.get("input_id_list")
|
||||
self._current_ip_address = config.get("current_ip_address")
|
||||
self._pushplus_token = config.get("pushplus_token")
|
||||
self._helloimg_s_token = config.get("helloimg_s_token")
|
||||
self._forced_update = config.get("forced_update")
|
||||
self._local_scan = config.get("local_scan")
|
||||
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 not self._my_send.init_success: # 没有输入通知方式，不通知
|
||||
self._my_send = None
|
||||
# 停止现有任务
|
||||
self.stop_service()
|
||||
if (self._enabled or self._onlyonce) and self._input_id_list:
|
||||
@@ -174,8 +179,25 @@ class DynamicWeChat(_PluginBase):
|
||||
event_data = event.event_data
|
||||
if not event_data or event_data.get("action") != "dynamicwechat":
|
||||
return
|
||||
self.ChangeIP()
|
||||
self.__update_config()
|
||||
# 先尝试cookie登陆
|
||||
try:
|
||||
with sync_playwright() as p:
|
||||
browser = p.chromium.launch(headless=True, args=['--lang=zh-CN'])
|
||||
context = browser.new_context()
|
||||
cookie = self.get_cookie()
|
||||
if cookie:
|
||||
context.add_cookies(cookie)
|
||||
page = context.new_page()
|
||||
page.goto(self._wechatUrl)
|
||||
time.sleep(3)
|
||||
if self.check_login_status(page, task='forced_change'):
|
||||
self.click_app_management_buttons(page)
|
||||
else:
|
||||
logger.error("cookie失效,强制修改IP失败：请使用'本地扫码修改IP'")
|
||||
browser.close()
|
||||
except Exception as err:
|
||||
logger.error(f"强制修改IP失败：{err}")
|
||||
|
||||
logger.info("----------------------本次任务结束----------------------")
|
||||
|
||||
@eventmanager.register(EventType.PluginAction)
|
||||
@@ -198,7 +220,6 @@ class DynamicWeChat(_PluginBase):
|
||||
page = context.new_page()
|
||||
page.goto(self._wechatUrl)
|
||||
time.sleep(3) # 页面加载等待时间
|
||||
|
||||
if self.find_qrc(page):
|
||||
current_time = datetime.now()
|
||||
future_time = current_time + timedelta(seconds=110)
|
||||
@@ -242,12 +263,7 @@ class DynamicWeChat(_PluginBase):
|
||||
self.ChangeIP()
|
||||
self.__update_config()
|
||||
|
||||
# logger.info("检测公网IP完毕")
|
||||
logger.info("----------------------本次任务结束----------------------")
|
||||
# if event:
|
||||
# self.post_message(channel=event.event_data.get("channel"),
|
||||
# title="检测公网IP完毕",
|
||||
# userid=event.event_data.get("user"))
|
||||
|
||||
def CheckIP(self):
|
||||
retry_urls = random.sample(self._ip_urls, len(self._ip_urls))
|
||||
@@ -338,22 +354,16 @@ class DynamicWeChat(_PluginBase):
|
||||
|
||||
qr_code_data = requests.get(qr_code_url).content
|
||||
self._qr_code_image = io.BytesIO(qr_code_data)
|
||||
return True
|
||||
refuse_time = (datetime.now() + timedelta(seconds=115)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
return qr_code_url, refuse_time
|
||||
else:
|
||||
logger.warning("未找到二维码")
|
||||
return False
|
||||
return None, None
|
||||
except Exception as e:
|
||||
logger.debug(str(e))
|
||||
return False
|
||||
return None, None
|
||||
|
||||
|
||||
def send_pushplus_message(self, title, content):
|
||||
pushplus_url = f"http://www.pushplus.plus/send/{self._pushplus_token}"
|
||||
pushplus_data = {
|
||||
"title": title,
|
||||
"content": content,
|
||||
"template": "html"
|
||||
}
|
||||
response = requests.post(pushplus_url, json=pushplus_data)
|
||||
|
||||
def ChangeIP(self):
|
||||
logger.info("开始请求企业微信管理更改可信IP")
|
||||
@@ -362,25 +372,20 @@ class DynamicWeChat(_PluginBase):
|
||||
# 启动 Chromium 浏览器并设置语言为中文
|
||||
browser = p.chromium.launch(headless=True, args=['--lang=zh-CN'])
|
||||
context = browser.new_context()
|
||||
# ----------cookie addd-----------------
|
||||
cookie = self.get_cookie()
|
||||
if cookie:
|
||||
context.add_cookies(cookie)
|
||||
# ----------cookie END-----------------
|
||||
page = context.new_page()
|
||||
page.goto(self._wechatUrl)
|
||||
time.sleep(3)
|
||||
if self.find_qrc(page):
|
||||
if self._pushplus_token and self._helloimg_s_token:
|
||||
img_src, refuse_time = self.upload_image(self._qr_code_image)
|
||||
self.send_pushplus_message(refuse_time, f"企业微信登录二维码<br/><img src='{img_src}' />")
|
||||
# if img_src:
|
||||
# self.post_message(
|
||||
# mtype=NotificationType.Plugin,
|
||||
# title="企业微信登录二维码",
|
||||
# text=refuse_time,
|
||||
# image=img_src
|
||||
# )
|
||||
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)
|
||||
if result:
|
||||
logger.info(f"二维码发送失败，原因：{result}")
|
||||
browser.close()
|
||||
return
|
||||
logger.info("二维码已经发送，等待用户 90 秒内扫码登录")
|
||||
# logger.info("如收到短信验证码请以？结束，发送到<企业微信应用> 如： 110301？")
|
||||
time.sleep(90) # 等待用户扫码
|
||||
@@ -391,17 +396,15 @@ class DynamicWeChat(_PluginBase):
|
||||
else:
|
||||
self._ip_changed = False
|
||||
else:
|
||||
logger.info("cookie失效，请重新上传或者配置pushplus_token和helloimg_s_token。")
|
||||
self._ip_changed = False
|
||||
logger.info("cookie已失效")
|
||||
else: # 如果直接进入企业微信
|
||||
logger.info("尝试cookie登录")
|
||||
# ----------cookie addd-----------------
|
||||
login_status = self.check_login_status(page, "")
|
||||
if login_status:
|
||||
self.click_app_management_buttons(page)
|
||||
else:
|
||||
# ----------cookie END-----------------
|
||||
self._ip_changed = False
|
||||
return
|
||||
browser.close()
|
||||
|
||||
except Exception as e:
|
||||
@@ -411,7 +414,7 @@ class DynamicWeChat(_PluginBase):
|
||||
|
||||
def _update_cookie(self, page, context):
|
||||
self._future_timestamp = 0 # 标记二维码失效
|
||||
PyCookieCloud.save_cookie_lifetime(0) # 重置cookie存活时间
|
||||
PyCookieCloud.save_cookie_lifetime(self._settings_file_path, 0) # 重置cookie存活时间
|
||||
if self._use_cookiecloud:
|
||||
if not self._cc_server: # 连接失败返回 False
|
||||
self.try_connect_cc() # 再尝试一次连接
|
||||
@@ -487,26 +490,34 @@ class DynamicWeChat(_PluginBase):
|
||||
return cookies
|
||||
|
||||
def refresh_cookie(self): # 保活
|
||||
try:
|
||||
with sync_playwright() as p:
|
||||
browser = p.chromium.launch(headless=True, args=['--lang=zh-CN'])
|
||||
context = browser.new_context()
|
||||
cookie = self.get_cookie()
|
||||
if cookie:
|
||||
context.add_cookies(cookie)
|
||||
page = context.new_page()
|
||||
page.goto(self._wechatUrl)
|
||||
time.sleep(3)
|
||||
if not self.check_login_status(page, task='refresh_cookie'):
|
||||
self._cookie_valid = False
|
||||
logger.warning("cookie已失效，下次IP变动推送二维码")
|
||||
else:
|
||||
self._cookie_valid = True
|
||||
PyCookieCloud.increase_cookie_lifetime(1200)
|
||||
self._cookie_lifetime = PyCookieCloud.load_cookie_lifetime()
|
||||
browser.close()
|
||||
except Exception as e:
|
||||
logger.error(f"cookie校验失败:{e}")
|
||||
if self._use_cookiecloud:
|
||||
try:
|
||||
with sync_playwright() as p:
|
||||
browser = p.chromium.launch(headless=True, args=['--lang=zh-CN'])
|
||||
context = browser.new_context()
|
||||
cookie = self.get_cookie()
|
||||
if cookie:
|
||||
context.add_cookies(cookie)
|
||||
page = context.new_page()
|
||||
page.goto(self._wechatUrl)
|
||||
time.sleep(3)
|
||||
if not self.check_login_status(page, task='refresh_cookie'):
|
||||
self._cookie_valid = False
|
||||
if self._my_send:
|
||||
result = self._my_send.send(title="cookie已失效，请及时更新",
|
||||
content="请在企业微信应用发送/push_qr，让插件推送二维码。如果是使用微信通知请确保公网IP还没有变动",
|
||||
image=None, force_send=False) # 标题，内容，图片，是否强制发送
|
||||
if result:
|
||||
logger.info(f"cookie失效通知发送失败，原因：{result}")
|
||||
else:
|
||||
self._cookie_valid = True
|
||||
if self._my_send:
|
||||
self._my_send.reset_limit()
|
||||
PyCookieCloud.increase_cookie_lifetime(self._settings_file_path, 1200)
|
||||
self._cookie_lifetime = PyCookieCloud.load_cookie_lifetime(self._settings_file_path)
|
||||
browser.close()
|
||||
except Exception as e:
|
||||
logger.error(f"cookie校验失败:{e}")
|
||||
|
||||
#
|
||||
def check_login_status(self, page, task):
|
||||
@@ -600,81 +611,15 @@ class DynamicWeChat(_PluginBase):
|
||||
logger.info(f"应用: {app_id} 输入IP：" + self._current_ip_address)
|
||||
ip_parts = self._current_ip_address.split('.')
|
||||
masked_ip = f"{ip_parts[0]}.{len(ip_parts[1]) * '*'}.{len(ip_parts[2]) * '*'}.{ip_parts[3]}"
|
||||
self.post_message(
|
||||
mtype=NotificationType.Plugin,
|
||||
title="更新可信IP成功",
|
||||
text='应用: ' + app_id + ' 输入IP：' + masked_ip,
|
||||
# image=img_src
|
||||
)
|
||||
if self._my_send:
|
||||
result = self._my_send.send(title="更新可信IP成功",
|
||||
content='应用: ' + app_id + ' 输入IP：' + masked_ip,
|
||||
force_send=True, diy_channel="WeChat")
|
||||
return
|
||||
else:
|
||||
logger.error("未找到应用id，修改IP失败")
|
||||
return
|
||||
|
||||
def upload_image(self, file_obj, permission=1, strategy_id=1, album_id=1):
|
||||
"""
|
||||
上传图片到 helloimg 图床，支持传入文件路径或 BytesIO 对象。
|
||||
|
||||
:param file_obj: 文件对象，可以是路径 (str) 或 BytesIO 对象
|
||||
:param permission: 上传图片的权限设置，默认 1
|
||||
:param strategy_id: 上传策略 ID，默认 1
|
||||
:param album_id: 相册 ID，默认 1
|
||||
:return: 上传成功返回图片链接，失败返回 None
|
||||
"""
|
||||
helloimg_token = "Bearer " + self._helloimg_s_token
|
||||
helloimg_url = "https://www.helloimg.com/api/v1/upload"
|
||||
headers = {
|
||||
"Authorization": helloimg_token,
|
||||
"Accept": "application/json",
|
||||
}
|
||||
|
||||
# 构造上传的文件，支持传入 BytesIO 或文件路径
|
||||
if isinstance(file_obj, io.BytesIO):
|
||||
# 如果是 BytesIO 对象，直接使用
|
||||
files = {
|
||||
"file": ('qr_code.png', file_obj, 'image/png')
|
||||
}
|
||||
else:
|
||||
# 如果是文件路径，打开文件进行读取
|
||||
files = {
|
||||
"file": open(file_obj, "rb")
|
||||
}
|
||||
|
||||
expired_at = (datetime.now() + timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
helloimg_data = {
|
||||
"token": "你的临时上传 Token", # 确保这里的 token 是有效的
|
||||
"permission": permission,
|
||||
"strategy_id": strategy_id,
|
||||
"album_id": album_id,
|
||||
"expired_at": expired_at
|
||||
}
|
||||
refuse_time = (datetime.now() + timedelta(seconds=110)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
# 发送上传请求
|
||||
response = requests.post(helloimg_url, headers=headers, files=files, data=helloimg_data)
|
||||
|
||||
# 检查响应内容是否符合预期
|
||||
response_data = None
|
||||
try:
|
||||
response_data = response.json()
|
||||
if not response_data['status']:
|
||||
if response_data['message'] == "Unauthenticated.":
|
||||
logger.error("Token失效，无法上传图片。请检查你的上传Token。")
|
||||
logger.info(f"使用的Token: {helloimg_token}")
|
||||
# self._ip_changed = False
|
||||
return
|
||||
else:
|
||||
logger.error(f"上传到图床失败: {response_data['message']}")
|
||||
self._ip_changed = False
|
||||
return
|
||||
|
||||
img_src = response_data['data']['links']['html']
|
||||
return img_src.split('"')[1], refuse_time # 提取 img src
|
||||
except KeyError as e:
|
||||
logger.error(f"上传图片时解析响应失败: {e}, 响应内容: {response_data}")
|
||||
self._ip_changed = False
|
||||
return
|
||||
|
||||
def __update_config(self):
|
||||
"""
|
||||
更新配置
|
||||
@@ -683,13 +628,11 @@ class DynamicWeChat(_PluginBase):
|
||||
"enabled": self._enabled,
|
||||
"onlyonce": self._onlyonce,
|
||||
"cron": self._cron,
|
||||
# "wechatUrl": self._wechatUrl,
|
||||
"notification_token": self._notification_token,
|
||||
"current_ip_address": self._current_ip_address,
|
||||
"ip_changed": self._ip_changed,
|
||||
"forced_update": self._forced_update,
|
||||
"local_scan": self._local_scan,
|
||||
"helloimg_s_token": self._helloimg_s_token,
|
||||
"pushplus_token": self._pushplus_token,
|
||||
"input_id_list": self._input_id_list,
|
||||
"cookie_header": self._cookie_header,
|
||||
"use_cookiecloud": self._use_cookiecloud,
|
||||
@@ -759,7 +702,6 @@ class DynamicWeChat(_PluginBase):
|
||||
}
|
||||
]
|
||||
},
|
||||
# 添加 "使用CookieCloud获取cookie" 开关按钮
|
||||
{
|
||||
'component': 'VRow',
|
||||
'content': [
|
||||
@@ -816,6 +758,24 @@ class DynamicWeChat(_PluginBase):
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
'component': 'VCol',
|
||||
'props': {
|
||||
'cols': 12,
|
||||
'md': 6
|
||||
},
|
||||
'content': [
|
||||
{
|
||||
'component': 'VTextarea',
|
||||
'props': {
|
||||
'model': 'notification_token',
|
||||
'label': '[可选] 通知方式',
|
||||
'rows': 1,
|
||||
'placeholder': '支持微信、Server酱、PushPlus、AnPush等Token或API'
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -841,47 +801,6 @@ class DynamicWeChat(_PluginBase):
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
'component': 'VRow',
|
||||
'content': [
|
||||
{
|
||||
'component': 'VCol',
|
||||
'props': {
|
||||
'cols': 12,
|
||||
'md': 6
|
||||
},
|
||||
'content': [
|
||||
{
|
||||
'component': 'VTextarea',
|
||||
'props': {
|
||||
'model': 'pushplus_token',
|
||||
'label': '[可选]pushplus_token',
|
||||
'rows': 1,
|
||||
'placeholder': '[可选] 请输入 pushplus_token'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
'component': 'VCol',
|
||||
'props': {
|
||||
'cols': 12,
|
||||
'md': 6
|
||||
},
|
||||
'content': [
|
||||
{
|
||||
'component': 'VTextarea',
|
||||
'props': {
|
||||
'model': 'helloimg_s_token',
|
||||
'label': '[可选]helloimg_s_token',
|
||||
'rows': 1,
|
||||
'placeholder': '[可选] 请输入 helloimg_token'
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
'component': 'VRow',
|
||||
'content': [
|
||||
@@ -896,7 +815,7 @@ class DynamicWeChat(_PluginBase):
|
||||
'props': {
|
||||
'type': 'info',
|
||||
'variant': 'tonal',
|
||||
'text': '使用内建CookieCloud 或 自定义 或 填写两个token 至少三选一。具体请查看作者主页'
|
||||
'text': '建议启用内建或自定义CookieCloud。支持微信、Server酱、PushPlus、AnPush，具体请查看作者主页'
|
||||
}
|
||||
}
|
||||
]
|
||||
@@ -916,7 +835,7 @@ class DynamicWeChat(_PluginBase):
|
||||
'component': 'VAlert',
|
||||
'props': {
|
||||
'type': 'info',
|
||||
'text': '优先使用cookie，当IP变动 且 cookie失效 且 填写两个token才会调用API推送登录二维码。',
|
||||
'text': 'cookie失效时通知用户，用户使用/push_qr让插件推送二维码。使用第三方通知时填写对应Token/API'
|
||||
}
|
||||
}
|
||||
]
|
||||
@@ -931,11 +850,10 @@ class DynamicWeChat(_PluginBase):
|
||||
"onlyonce": False,
|
||||
"forceUpdate": False,
|
||||
"use_cookiecloud": True,
|
||||
"use_local_qr": False, # 默认关闭本地扫码
|
||||
"use_local_qr": False,
|
||||
"cookie_header": "",
|
||||
"pushplus_token": "",
|
||||
"helloimg_token": "",
|
||||
"input_id_list": "",
|
||||
"notification_token": "",
|
||||
"input_id_list": ""
|
||||
}
|
||||
|
||||
def get_page(self) -> List[dict]:
|
||||
@@ -1083,10 +1001,15 @@ class DynamicWeChat(_PluginBase):
|
||||
page = context.new_page()
|
||||
page.goto(self._wechatUrl)
|
||||
time.sleep(3)
|
||||
if self.find_qrc(page):
|
||||
if self._pushplus_token and self._helloimg_s_token:
|
||||
img_src, refuse_time = self.upload_image(self._qr_code_image)
|
||||
self.send_pushplus_message(refuse_time, f"企业微信登录二维码<br/><img src='{img_src}' />")
|
||||
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)
|
||||
if result:
|
||||
logger.info(f"远程推送任务: 二维码发送失败，原因：{result}")
|
||||
browser.close()
|
||||
return
|
||||
logger.info("远程推送任务: 二维码已经发送，等待用户 90 秒内扫码登录")
|
||||
# logger.info("远程推送任务: 如收到短信验证码请以？结束，发送到<企业微信应用> 如： 110301？")
|
||||
time.sleep(90)
|
||||
@@ -1096,9 +1019,10 @@ class DynamicWeChat(_PluginBase):
|
||||
# logger.info("远程推送任务: 没有可用的CookieCloud服务器，只修改可信IP")
|
||||
self.click_app_management_buttons(page)
|
||||
else:
|
||||
logger.warning("远程推送任务: 未配置pushplus_token和helloimg_s_token")
|
||||
logger.warning("远程推送任务: 任何通知方式")
|
||||
else:
|
||||
logger.warning("远程推送任务: 未找到二维码")
|
||||
browser.close()
|
||||
except Exception as e:
|
||||
logger.error(f"远程推送任务: 推送二维码失败: {e}")
|
||||
|
||||
@@ -1108,7 +1032,7 @@ class DynamicWeChat(_PluginBase):
|
||||
{
|
||||
"cmd": "/push_qr",
|
||||
"event": EventType.PluginAction,
|
||||
"desc": "立即推送登录二维码到pushplus",
|
||||
"desc": "立即推送登录二维码到",
|
||||
"category": "",
|
||||
"data": {
|
||||
"action": "push_qrcode"
|
||||
|
||||
133
plugins/dynamicwechat/notify_helper.py
Normal file
133
plugins/dynamicwechat/notify_helper.py
Normal file
@@ -0,0 +1,133 @@
|
||||
import re
|
||||
import requests
|
||||
from app.modules.wechat.wechat import WeChat
|
||||
|
||||
|
||||
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 send_channel(self):
|
||||
if self.token:
|
||||
if self.token == "WeChat":
|
||||
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
|
||||
|
||||
# 标题，内容，图片，是否强制发送
|
||||
def send(self, title, content, 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 self.first_text_sent:
|
||||
return
|
||||
else:
|
||||
self.first_text_sent = True
|
||||
|
||||
try:
|
||||
if not diy_channel:
|
||||
channel = self.channel
|
||||
else:
|
||||
channel = diy_channel
|
||||
|
||||
if channel == "WeChat":
|
||||
return MySender.send_wechat(title, content, image)
|
||||
elif channel == "ServerChan":
|
||||
return self.send_serverchan(title, content, image)
|
||||
elif channel == "AnPush":
|
||||
return self.send_anpush(title, content, image)
|
||||
elif channel == "PushPlus":
|
||||
return self.send_pushplus(title, content, image)
|
||||
else:
|
||||
return "Unknown channel"
|
||||
except Exception as e:
|
||||
return f"Error occurred: {str(e)}"
|
||||
|
||||
@staticmethod
|
||||
def send_wechat(title, content, image):
|
||||
wechat = WeChat()
|
||||
if image:
|
||||
send_status = wechat.send_msg(title='企业微信登录二维码', image=image, link=image)
|
||||
else:
|
||||
send_status = wechat.send_msg(title=title, text=content)
|
||||
|
||||
if send_status is None:
|
||||
return "微信通知发送错误"
|
||||
return None
|
||||
|
||||
def send_serverchan(self, title, content, image):
|
||||
if self.token.startswith('sctp'):
|
||||
match = re.match(r'sctp(\d+)t', self.token)
|
||||
if match:
|
||||
num = match.group(1)
|
||||
url = f'https://{num}.push.ft07.com/send/{self.token}.send'
|
||||
else:
|
||||
raise ValueError('Invalid sendkey format for sctp')
|
||||
else:
|
||||
url = f'https://sctapi.ftqq.com/{self.token}.send'
|
||||
|
||||
params = {'title': title, 'desp': f'' if image else content}
|
||||
headers = {'Content-Type': 'application/json;charset=utf-8'}
|
||||
response = requests.post(url, json=params, headers=headers)
|
||||
result = response.json()
|
||||
if result.get('code') != 0:
|
||||
return f"Server酱通知错误: {result.get('message')}"
|
||||
return None
|
||||
|
||||
def send_anpush(self, title, content, image):
|
||||
if ',' in self.token:
|
||||
channel, token = self.token.split(',', 1)
|
||||
else:
|
||||
return
|
||||
url = f"https://api.anpush.com/push/{token}"
|
||||
payload = {
|
||||
"title": title,
|
||||
"content": f"<img src=\"{image}\" width=\"100%\">" if image else content,
|
||||
"channel": channel
|
||||
}
|
||||
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
||||
response = requests.post(url, headers=headers, data=payload)
|
||||
result = response.json()
|
||||
# 判断返回的code和msgIds
|
||||
if result.get('code') != 200:
|
||||
return f"AnPush: {result.get('msg')}"
|
||||
elif not result.get('data') or not result['data'].get('msgIds'):
|
||||
return "AnPush 消息通道未找到"
|
||||
return None
|
||||
|
||||
def send_pushplus(self, title, content, image):
|
||||
pushplus_url = f"http://www.pushplus.plus/send/{self.token}"
|
||||
# PushPlus发送逻辑
|
||||
data = {
|
||||
"title": title,
|
||||
"content": f"企业微信登录二维码<br/><img src='{image}' />" if image else content,
|
||||
"template": "html"
|
||||
}
|
||||
response = requests.post(pushplus_url, json=data)
|
||||
result = response.json()
|
||||
if result.get('code') != 200:
|
||||
return f"PushPlus send failed: {result.get('msg')}"
|
||||
return None
|
||||
|
||||
def reset_limit(self):
|
||||
"""解除限制，允许再次发送纯文本消息"""
|
||||
self.first_text_sent = False
|
||||
@@ -7,9 +7,6 @@ from typing import Dict, Any
|
||||
from Crypto import Random
|
||||
from Crypto.Cipher import AES
|
||||
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
settings_file = os.path.join(script_dir, 'settings.json')
|
||||
|
||||
|
||||
def bytes_to_key(data: bytes, salt: bytes, output=48) -> bytes:
|
||||
# 兼容v2 将bytes_to_key和encrypt导入
|
||||
@@ -98,7 +95,7 @@ class PyCookieCloud:
|
||||
return md5.hexdigest()[:16]
|
||||
|
||||
@staticmethod
|
||||
def load_cookie_lifetime(): # 返回时间戳 单位秒
|
||||
def load_cookie_lifetime(settings_file: str = None): # 返回时间戳 单位秒
|
||||
if os.path.exists(settings_file):
|
||||
with open(settings_file, 'r') as file:
|
||||
settings = json.load(file)
|
||||
@@ -107,13 +104,18 @@ class PyCookieCloud:
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def save_cookie_lifetime(cookie_lifetime): # 传入时间戳 单位秒
|
||||
def save_cookie_lifetime(settings_file, cookie_lifetime): # 传入时间戳 单位秒
|
||||
with open(settings_file, 'w') as file:
|
||||
json.dump({'_cookie_lifetime': cookie_lifetime}, file)
|
||||
|
||||
@staticmethod
|
||||
def increase_cookie_lifetime(seconds: int):
|
||||
current_lifetime = PyCookieCloud.load_cookie_lifetime()
|
||||
def increase_cookie_lifetime(settings_file, seconds: int):
|
||||
if os.path.exists(settings_file):
|
||||
with open(settings_file, 'r') as file:
|
||||
settings = json.load(file)
|
||||
current_lifetime = settings.get('_cookie_lifetime', 0)
|
||||
else:
|
||||
current_lifetime = 0
|
||||
new_lifetime = current_lifetime + seconds
|
||||
# 保存新的 _cookie_lifetime
|
||||
PyCookieCloud.save_cookie_lifetime(new_lifetime)
|
||||
PyCookieCloud.save_cookie_lifetime(settings_file, new_lifetime)
|
||||
|
||||
Reference in New Issue
Block a user