from DrissionPage import ChromiumOptions, Chromium import sys import os import logging from dotenv import load_dotenv import tempfile import zipfile import string load_dotenv() PROXY_HOST = "h464.kdltpspro.com" PROXY_PORT = "15818" PROXY_USERNAME = "h464" # 代理用户名 PROXY_PASSWORD = "kdltpspro" # 代理密码 PROXY_URL = f"http://{PROXY_HOST}:{PROXY_PORT}" class BrowserManager: def __init__(self): self.browser = None self.current_proxy_info = None def create_proxyauth_extension(self, proxy_host, proxy_port, proxy_username, proxy_password, scheme='http', plugin_folder=None): """ 创建Chrome代理认证插件 """ if plugin_folder is None: # 创建插件在脚本所在目录 current_directory = os.path.dirname(os.path.abspath(__file__)) plugin_folder = os.path.join(current_directory, 'kdl_Chromium_Proxy') # 确保文件夹存在 if not os.path.exists(plugin_folder): os.makedirs(plugin_folder) logging.info(f"创建代理插件,路径: {plugin_folder}") manifest_json = """ { "version": "1.0.0", "manifest_version": 2, "name": "kdl_Chromium_Proxy", "permissions": [ "proxy", "tabs", "unlimitedStorage", "storage", "", "webRequest", "webRequestBlocking", "browsingData" ], "background": { "scripts": ["background.js"] }, "minimum_chrome_version":"22.0.0" } """ background_js = string.Template(""" var config = { mode: "fixed_servers", rules: { singleProxy: { scheme: "${scheme}", host: "${host}", port: parseInt(${port}) }, bypassList: [] } }; chrome.proxy.settings.set({value: config, scope: "regular"}, function() {}); function callbackFn(details) { return { authCredentials: { username: "${username}", password: "${password}" } }; } chrome.webRequest.onAuthRequired.addListener( callbackFn, {urls: [""]}, ['blocking'] ); """).substitute( host=proxy_host, port=proxy_port, username=proxy_username, password=proxy_password, scheme=scheme, ) with open(os.path.join(plugin_folder, "manifest.json"), "w") as manifest_file: manifest_file.write(manifest_json) with open(os.path.join(plugin_folder, "background.js"), "w") as background_file: background_file.write(background_js) return plugin_folder def get_proxy(self, use_api=False, mode=False, proxy_choice=None, custom_api=None): """ 获取代理配置 Args: use_api: 是否使用API获取代理 mode: 如果开启 返回格式要改变成 host port username password proxy_choice: 代理选择 (1=本地代理, 2=全局代理) custom_api: 自定义代理API地址 Returns: str 或 tuple: 代理URL或(host, port, username, password)元组 """ logging.info(f"获取代理配置, use_api={use_api}, mode={mode}, proxy_choice={proxy_choice}") if use_api: try: # 从API获取代理 import requests logging.info("正在从API获取代理...") # 根据选择使用不同的API if custom_api: api_url = custom_api logging.info(f"使用自定义API: {api_url}") elif proxy_choice == 1: api_url = "https://cursorapi.nosqli.com/admin/api.proxyinfo/getproxyarmybendi" logging.info("使用本地代理API") elif proxy_choice == 2 or proxy_choice is None: # 如果未指定或选择全局代理 api_url = "https://cursorapi.nosqli.com/admin/api.GlobalProxyip/get_proxy" logging.info("使用全局代理API") else: logging.warning(f"未知的代理选择: {proxy_choice},使用cn代理") api_url = "https://cursorapi.nosqli.com/admin/api.proxyinfo/getproxyarmybendi" response = requests.get(api_url) logging.info(f"API响应状态码: {response.status_code}") if response.status_code == 200: proxy_data = response.json() logging.info(f"API返回数据: {proxy_data}") if proxy_data.get("code") == 0: proxy_info = proxy_data.get("data", {}) proxy_host = proxy_info.get("host") proxy_port = proxy_info.get("port") proxy_username = proxy_info.get("username", "") proxy_password = proxy_info.get("password", "") proxy_data = proxy_info.get("proxy_data", "") logging.info(f"解析到代理信息 - host:{proxy_host}, port:{proxy_port}") if proxy_host and proxy_port: if mode: logging.info(f"使用API代理(mode=True): 返回元组格式") return proxy_host, proxy_port, proxy_username, proxy_password, proxy_data else: proxy_url = f"http://{proxy_host}:{proxy_port}" logging.info(f"使用API代理: {proxy_url}") return proxy_url logging.warning("从API获取代理失败,使用默认代理配置") # 使用默认代理 if mode: # 使用环境变量或常量中的默认值 proxy_host = os.getenv("PROXY_HOST", PROXY_HOST) proxy_port = os.getenv("PROXY_PORT", PROXY_PORT) proxy_username = os.getenv("PROXY_USERNAME", PROXY_USERNAME ) proxy_password = os.getenv("PROXY_PASSWORD", PROXY_PASSWORD) proxy_data = os.getenv("PROXY_DATA", "") logging.info(f"使用默认代理信息(mode=True): host={proxy_host}, port={proxy_port}") return proxy_host, proxy_port, proxy_username, proxy_password, proxy_data else: default_proxy = os.getenv("BROWSER_PROXY", PROXY_URL) logging.info(f"使用默认代理: {default_proxy}") return default_proxy except Exception as e: logging.error(f"获取代理出错: {str(e)}") # 发生异常时使用默认代理 if mode: proxy_host = os.getenv("PROXY_HOST", PROXY_HOST) proxy_port = os.getenv("PROXY_PORT", PROXY_PORT) proxy_username = os.getenv("PROXY_USERNAME", PROXY_USERNAME) proxy_password = os.getenv("PROXY_PASSWORD", PROXY_PASSWORD) proxy_data = os.getenv("PROXY_DATA", "") logging.info(f"异常情况下使用默认代理信息(mode=True)") return proxy_host, proxy_port, proxy_username, proxy_password, proxy_data else: default_proxy = os.getenv("BROWSER_PROXY", PROXY_URL) logging.info(f"使用默认代理: {default_proxy}") return default_proxy # 不使用API时返回默认配置 if mode: proxy_host = os.getenv("PROXY_HOST", PROXY_HOST) proxy_port = os.getenv("PROXY_PORT", PROXY_PORT) proxy_username = os.getenv("PROXY_USERNAME", PROXY_USERNAME) proxy_password = os.getenv("PROXY_PASSWORD", PROXY_PASSWORD) proxy_data = os.getenv("PROXY_DATA", "") logging.info(f"不使用API,返回默认代理信息(mode=True)") return proxy_host, proxy_port, proxy_username, proxy_password, proxy_data else: default_proxy = os.getenv("BROWSER_PROXY", PROXY_URL) logging.info(f"不使用API,直接返回默认代理: {default_proxy}") return default_proxy def get_plugin_folder(self): """获取代理插件文件夹路径""" # 使用当前目录下的kdl_Chromium_Proxy文件夹 current_directory = os.path.dirname(os.path.abspath(__file__)) plugin_folder = os.path.join(current_directory, 'kdl_Chromium_Proxy') # 确保文件夹存在 if not os.path.exists(plugin_folder): os.makedirs(plugin_folder) logging.info(f"代理插件文件夹路径: {plugin_folder}") return plugin_folder def init_browser(self, user_agent=None, proxy_choice=None, custom_api=None): """ 初始化浏览器实例 Args: user_agent: 自定义User-Agent proxy_choice: 代理选择 (1=本地代理, 2=全局代理) custom_api: 自定义代理API地址 Returns: tuple: (browser实例, 代理信息元组) """ try: # 获取代理配置 proxy_info = self.get_proxy(use_api=True, mode=True, proxy_choice=proxy_choice, custom_api=custom_api) proxy_host, proxy_port, proxy_username, proxy_password, proxy_data = proxy_info # 创建代理认证插件 plugin_folder = self.create_proxyauth_extension( proxy_host=proxy_host, proxy_port=proxy_port, proxy_username=proxy_username, proxy_password=proxy_password ) # 设置浏览器选项 co = ChromiumOptions() co.set_argument("--hide-crash-restore-bubble") co.set_pref("credentials_enable_service", False) # 添加代理插件 co.add_extension(plugin_folder) # 设置用户代理 if user_agent: co.set_user_agent(user_agent) # 设置无头模式 co.headless(True) # Mac系统特殊处理 if sys.platform == "darwin": co.set_argument("--no-sandbox") co.set_argument("--disable-gpu") # 创建浏览器实例 browser = Chromium(co) self.browser = browser self.current_proxy_info = proxy_info logging.info("浏览器实例创建成功") return browser, proxy_info except Exception as e: logging.error(f"初始化浏览器失败: {str(e)}") raise def _get_browser_options(self, user_agent=None): """获取浏览器配置""" co = ChromiumOptions() try: extension_path = self._get_extension_path() co.add_extension(extension_path) except FileNotFoundError as e: logging.warning(f"警告: {e}") co.set_pref("credentials_enable_service", False) co.set_argument("--hide-crash-restore-bubble") proxy = self.get_proxy(True,True) logging.info(f"代理-----: {proxy}") # 在类的__init__方法中声明self.current_proxy_info = None if isinstance(proxy, tuple) and len(proxy) == 4: self.current_proxy_info = proxy else: logging.error("代理信息格式错误") self.current_proxy_info = None proxy_host, proxy_port, proxy_username, proxy_password = proxy try: # 使用类方法 # proxyauth_plugin_folder = BrowserManager.create_proxyauth_extension( proxyauth_plugin_folder = self.create_proxyauth_extension( proxy_host=proxy_host, proxy_port=proxy_port, proxy_username=proxy_username, proxy_password=proxy_password ) # 使用函数返回的插件文件夹路径而不是硬编码路径 logging.info(f"代理插件文件夹路径: {proxyauth_plugin_folder}") co.add_extension(proxyauth_plugin_folder) except Exception as e: logging.error(f"创建代理扩展失败: {str(e)}") # 如果创建扩展失败,尝试直接设置代理 proxy_url = f"http://{proxy_username}:{proxy_password}@{proxy_host}:{proxy_port}" co.set_proxy(proxy_url) logging.info(f"直接设置代理URL: {proxy_url}") co.auto_port() if user_agent: co.set_user_agent(user_agent) # 设置为有头模式以便观察验证过程 co.headless(True) # 使用有头模式进行测试 # Mac 系统特殊处理 if sys.platform == "darwin": co.set_argument("--no-sandbox") co.set_argument("--disable-gpu") return co def _get_extension_path(self): """获取插件路径""" root_dir = os.getcwd() extension_path = os.path.join(root_dir, "turnstilePatch") if hasattr(sys, "_MEIPASS"): extension_path = os.path.join(sys._MEIPASS, "turnstilePatch") if not os.path.exists(extension_path): raise FileNotFoundError(f"插件不存在: {extension_path}") return extension_path def quit(self): """关闭浏览器""" if self.browser: try: self.browser.quit() except: pass