d0zingcat/MoviePilot-Plugins
1
0
Fork 0
mirror of https://github.com/d0zingcat/MoviePilot-Plugins.git synced 2026-05-14 07:26:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #620 from RamenRa/master

Browse Source
This commit is contained in:
jxxghp
2024-12-28 17:58:05 +08:00
committed by GitHub
parent 6693698d63 d38dab82d7
commit ea6d9b8baa
3 changed files with 73 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
package.json
View File

@@ -875,12 +875,13 @@
"name": "动态企微可信IP",
"description": "修改企微应用可信IP，支持Srever酱等第三方通知。验证码以？结尾发送到企业微信应用",
"labels": "消息通知",
"version": "1.7.0",
"version": "1.7.1",
"icon": "Wecom_A.png",
"author": "RamenRa",
"level": 2,
"v2": true,
"history": {
"v1.7.1": "允许使用'||wan2'选项及无法使用'立即检测一次'",
"v1.7.0": "使用第三方通知时可IP变动后通知，拟支持多网络出口检查。",
"v1.6.0": "忽略因网络波动导致获取ip错误。自定义的类合并为helper.py。后续核心功能没问题将不再更新",
"v1.5.2": "可以从指定url获取ip,修复不使用cc时cookie失效过快，v1可配置第三方为备用通知，server酱可以将文本发送到server3,二维码给服务号",
@@ -890,6 +891,7 @@
"v1.4.0": "修复强制更改IP时配置面板延时过长的问题。庆祝v2进入正式版，显示了一个没用的参数"
}
},
"SyncCookieCloud": {
"name": "同步CookieCloud",
"description": "同步MoviePilot站点Cookie到本地CookieCloud。",

76
plugins/dynamicwechat/__init__.py
View File

@@ -20,7 +20,7 @@ from app.log import logger
from app.plugins import _PluginBase
from app.schemas.types import EventType
from app.plugins.dynamicwechat.helper import PyCookieCloud, MySender
from app.plugins.dynamicwechat.helper import PyCookieCloud, MySender, IpLocationParser
class DynamicWeChat(_PluginBase):
@@ -31,7 +31,7 @@ class DynamicWeChat(_PluginBase):
# 插件图标
plugin_icon = "Wecom_A.png"
# 插件版本
plugin_version = "1.7.0"
plugin_version = "1.7.1"
# 插件作者
plugin_author = "RamenRa"
# 作者主页
@@ -152,7 +152,6 @@ class DynamicWeChat(_PluginBase):
if "||wan2" in self._input_id_list: # 多wan口
self.wan2 = IpLocationParser(self._settings_file_path)
self._current_ip_address = self.wan2.read_ips("ips") # 从文件中读取
logger.info(f"当前记录的IP：{self._current_ip_address}，如为空或不一致，预计检测1-2轮内会修正")
else:
self.wan2 = None
_, self._current_ip_address = self.get_ip_from_url() # 直接从网页获取
@@ -163,14 +162,24 @@ class DynamicWeChat(_PluginBase):
self._scheduler = BackgroundScheduler(timezone=settings.TZ)
# 运行一次定时服务
if self._onlyonce: # 多网口ip检测禁用立即检测
if not self.wan2:
if self.wan2:
if not self._forced_update or not self._local_scan:
logger.info("多网络出口检查需要时间较长，预计25秒内完成")
self._scheduler.add_job(func=self.write_wan2_ip, trigger='date',
run_date=datetime.now(tz=pytz.timezone(settings.TZ)) + timedelta(
seconds=3),
name="多网络出口获取IP") # 添加任务
self._scheduler.add_job(func=self.check, trigger='date',
run_date=datetime.now(tz=pytz.timezone(settings.TZ)) + timedelta(
seconds=20),
name="多网络出口检查IP") # 添加任务
else:
if not self._forced_update or not self._local_scan:
# logger.info("立即检测公网IP")
self._scheduler.add_job(func=self.check, trigger='date',
run_date=datetime.now(tz=pytz.timezone(settings.TZ)) + timedelta(seconds=3),
name="检测公网IP") # 添加任务
else:
logger.info("启用多网口检测时禁用‘立即检测一次’功能")
# logger.info("启用多网口检测时禁用‘立即检测一次’功能")
# 关闭一次性开关
self._onlyonce = False
@@ -222,7 +231,7 @@ class DynamicWeChat(_PluginBase):
image=None, force_send=False
)
if error:
logger.info(f"cookie失效通知发送失败,原因：{result}")
logger.info(f"cookie失效通知发送失败,原因：{error}")
@eventmanager.register(EventType.PluginAction)
def forced_change(self, event: Event = None):
@@ -301,6 +310,41 @@ class DynamicWeChat(_PluginBase):
except Exception as e:
logger.error(f"本地扫码任务: 本地扫码失败: {e}")
@eventmanager.register(EventType.PluginAction)
def write_wan2_ip(self):
if not self._enabled:
logger.error("插件未开启")
return
if event:
event_data = event.event_data
if not event_data or event_data.get("action") != "dynamicwechat":
return
urls = ["https://ip.skk.moe/multi", "https://ip.m27.tech", "https://ip.orz.tools"]
random.shuffle(urls)
self.wan2_url = None
# 创建一个 Playwright 实例
with sync_playwright() as p:
browser = None # 定义浏览器变量
for url in urls:
try:
# 启动浏览器
if url == "https://ip.skk.moe/multi":
browser = p.chromium.launch(headless=False, args=['--lang=zh-CN'])
else:
browser = p.chromium.launch(headless=True, args=['--lang=zh-CN'])
page = browser.new_page()
china_ips = self.wan2.get_ipv4(page, url)
if china_ips:
self.wan2.overwrite_ips("url_ip", china_ips) # 将获取到的IP写入文件 覆盖写入
self.wan2_url = url
break
except Exception as e:
logger.warning(f"{url} 多出口IP获取失败, Error: {e}")
finally:
if browser:
browser.close()
browser = None # 重置浏览器变量
@eventmanager.register(EventType.PluginAction)
def check(self, event: Event = None):
"""
@@ -425,14 +469,13 @@ class DynamicWeChat(_PluginBase):
if response.status_code == 200:
ip_address = re.search(self._ip_pattern, response.text)
if ip_address:
# self.wan1.overwrite_ips("url_ip", ip_address.group())
return url, ip_address.group() # 返回匹配的 IP 地址
except Exception as e:
if "104" not in str(e) and 'Read timed out' not in str(e): # 忽略网络波动,都失败会返回None, "获取IP失败"
logger.warning(f"{url} 获取IP失败, Error: {e}")
return None, "获取IP失败"
else:
urls = ["https://ip.skk.moe/multi", "https://ip.orz.tools"]
urls = ["https://ip.skk.moe/multi", "https://ip.m27.tech", "https://ip.orz.tools"]
random.shuffle(urls)
# 创建一个 Playwright 实例
with sync_playwright() as p:
@@ -456,6 +499,7 @@ class DynamicWeChat(_PluginBase):
if browser:
browser.close()
browser = None # 重置浏览器变量
self.wan2_url = None
return None, "获取IP失败"
def find_qrc(self, page):
@@ -697,7 +741,7 @@ class DynamicWeChat(_PluginBase):
captcha_panel = page.wait_for_selector('.receive_captcha_panel', timeout=5000) # 检查验证码面板
if captcha_panel: # 出现了短信验证界面
if task == 'local_scanning':
time.sleep(6)
time.sleep(3)
else:
logger.info("等待30秒,请将短信验证码请以'？'结束,发送到<企业微信应用> 如： 110301？")
time.sleep(30) # 多等30秒
@@ -728,7 +772,8 @@ class DynamicWeChat(_PluginBase):
def click_app_management_buttons(self, page):
self._cookie_valid = True
self._my_send.reset_limit() # 解除限制 可以发送cookie失效提醒
if self._my_send:
self._my_send.reset_limit() # 解除限制 可以发送cookie失效提醒
bash_url = "https://work.weixin.qq.com/wework_admin/frame#apps/modApiApp/"
# 按钮的选择器和名称
buttons = [
@@ -1277,9 +1322,10 @@ class DynamicWeChat(_PluginBase):
}]
"""
if self._enabled and self._cron:
logger.info(f"服务启动")
if self.wan2:
logger.info("多网口检测第一次获取IP可能会失败")
if not self.wan2:
logger.info(f"服务启动")
else:
logger.info(f"当前记录的IP：{self._current_ip_address}，首次使用可能为空或检测IP失败")
return [{
"id": self.__class__.__name__,
"name": f"{self.plugin_name}服务",

18
plugins/dynamicwechat/helper.py
View File

@@ -316,7 +316,7 @@ class MySender:
class IpLocationParser:
def __init__(self, settings_file_path, max_ips=4):
def __init__(self, settings_file_path, max_ips=3):
self._settings_file_path = settings_file_path
self._max_ips = max_ips # 最大历史IP数量
self._ips = self.read_ips("ips") # 初始化时读取已存储的 IP 地址
@@ -327,7 +327,7 @@ class IpLocationParser:
parser_methods = {
"https://ip.orz.tools": IpLocationParser._parse_ip_orz_tools,
"https://ip.skk.moe/multi": IpLocationParser._parse_ip_skk_moe,
"http://revproxy.ustc.edu.cn:8000": IpLocationParser._parse_ip_ustc,
"https://ip.m27.tech": IpLocationParser._parse_ip_m27,
}
parser_method = parser_methods.get(url)
if parser_method is None:
@@ -405,25 +405,25 @@ class IpLocationParser:
return IpLocationParser._remove_duplicates(ipv4_addresses, locations)
@staticmethod
def _parse_ip_ustc(page):
def _parse_ip_m27(page):
"""解析 https://ip.m27.tech 页面中的 IP 和归属地"""
rows = page.query_selector_all(
'body > div:nth-child(4) > center > table > tbody > tr > td:nth-child(2)'
'body > div > div.panel.panel-success > div.panel-body > table > tbody > tr'
)
# print(f"ip_ustc共找到 {len(rows)} 行数据")
# print(f"共找到 {len(rows)} 行数据")
ipv4_addresses, locations = [], []
for row in rows:
row_text = row.inner_text().strip()
# 提取 IP 地址
ip_match = re.match(r'(\d+\.\d+\.\d+\.\d+)', row_text)
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', row_text)
if ip_match:
ip = ip_match.group(1).strip()
if not IpLocationParser._is_valid_ipv4(ip):
continue
else:
continue
#
# 提取归属地
location_match = re.search(r'(China|中国).*', row_text)
location = location_match.group(0).strip() if location_match else "未知"
@@ -439,7 +439,7 @@ class IpLocationParser:
# 导航到目标页面
page.goto(url)
# 等待一段时间，让所有动态渲染的内容加载完成
page.wait_for_timeout(10000) # 等待 10 秒钟
page.wait_for_timeout(8000) # 等待 8 秒钟
# 调用解析器解析数据
ipv4_addresses, locations = IpLocationParser._parse(page, url)
# 筛选出属于中国的 IP 地址