128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
from DrissionPage import ChromiumOptions, Chromium
|
|
import sys
|
|
import os
|
|
import logging
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
# PROXY_HOST = "h464.kdltpspro.com"
|
|
# PROXY_PORT = "15818"
|
|
PROXY_HOST = "h464.kdltpspro.com"
|
|
PROXY_PORT = "15818"
|
|
PROXY_URL = f"http://{PROXY_HOST}:{PROXY_PORT}"
|
|
|
|
class BrowserManager:
|
|
def __init__(self):
|
|
self.browser = None
|
|
|
|
def get_proxy(self, use_api=False):
|
|
"""
|
|
获取代理配置
|
|
Args:
|
|
use_api: 是否使用API获取代理
|
|
Returns:
|
|
str: 代理URL
|
|
"""
|
|
# 从日志可以看出,use_api=False,所以直接返回了默认代理配置
|
|
# 没有调用API获取新的代理,而是使用了预设的PROXY_URL
|
|
# PROXY_URL 定义为 http://h464.kdltpspro.com:15818
|
|
|
|
logging.info(f"获取代理配置, use_api={use_api}")
|
|
|
|
if use_api:
|
|
try:
|
|
# 从API获取代理
|
|
import requests
|
|
logging.info("正在从API获取代理...")
|
|
response = requests.get("http://127.0.0.1:46880/admin/api.proxyinfo/getproxyarmy")
|
|
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")
|
|
logging.info(f"解析到代理信息 - host:{proxy_host}, port:{proxy_port}")
|
|
|
|
if proxy_host and proxy_port:
|
|
proxy_url = f"http://{proxy_host}:{proxy_port}"
|
|
logging.info(f"使用API代理: {proxy_url}")
|
|
return proxy_url
|
|
|
|
logging.warning("从API获取代理失败,使用默认代理配置")
|
|
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)}")
|
|
default_proxy = os.getenv("BROWSER_PROXY", PROXY_URL)
|
|
logging.info(f"使用默认代理: {default_proxy}")
|
|
return default_proxy
|
|
|
|
# 不使用API时返回默认配置
|
|
default_proxy = os.getenv("BROWSER_PROXY", PROXY_URL)
|
|
logging.info(f"不使用API,直接返回默认代理: {default_proxy}")
|
|
return default_proxy
|
|
def init_browser(self, user_agent=None):
|
|
"""初始化浏览器"""
|
|
co = self._get_browser_options(user_agent)
|
|
self.browser = Chromium(co)
|
|
return self.browser
|
|
|
|
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)
|
|
logging.info(f"代理-----: {proxy}")
|
|
if proxy:
|
|
co.set_proxy(proxy)
|
|
|
|
co.auto_port()
|
|
if user_agent:
|
|
co.set_user_agent(user_agent)
|
|
|
|
# 设置为有头模式以便观察验证过程
|
|
co.headless(False) # 使用有头模式进行测试
|
|
|
|
# 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
|