diff --git a/package.json b/package.json index 7b69783..c149936 100644 --- a/package.json +++ b/package.json @@ -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。", diff --git a/plugins/dynamicwechat/__init__.py b/plugins/dynamicwechat/__init__.py index efd006a..95f0544 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.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}服务", diff --git a/plugins/dynamicwechat/helper.py b/plugins/dynamicwechat/helper.py index 94b4d9e..8b12d65 100644 --- a/plugins/dynamicwechat/helper.py +++ b/plugins/dynamicwechat/helper.py @@ -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 地址