crawlab_feader/FworkSpider/feapder/utils/webdriver/drissionpage_driver.py

# -*- coding: utf-8 -*-
"""
Created on 2025-05-14 
---------
@summary:
---------
@author: Dzr
"""

from time import perf_counter

from DrissionPage import Chromium, ChromiumOptions
from DrissionPage.common import Settings

from feapder.utils import tools
from feapder.utils.log import log
from feapder.utils.webdriver.webdirver import *


class SingletonMeta(type):
    """单例元类"""
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

    def clear_instance(cls):
        """清除元类中保存的实例引用"""
        if cls in cls._instances:
            del cls._instances[cls]


class Browser(metaclass=SingletonMeta):
    _browser: Chromium = None

    def __init__(
        self,
        load_images=True,
        user_agent=None,
        port=None,
        user_data_path=None,
        proxy=None,
        headless=False,
        singleton_tab=True,
        driver_type="Chromium",
        timeout=30,
        custom_argument=None,
        download_path=None,
        browser_path=None,
        **kwargs
    ):
        """
        webdriver 封装，仅支持Chromium
        Args:
            load_images: 是否加载图片
            port: 浏览器端口
            user_data_path: 用户数据目录
            scope: 自动端口范围，与port 同时只能生效一个
            user_agent: 字符串 或 无参函数，返回值为user_agent
            proxy: xxx.xxx.xxx.xxx:xxxx 或 无参函数，返回值为代理地址
            headless: 是否启用无头模式
            driver_type: Chromium
            singleton_tab: 标签页是否开启多例支持，True=单例 False=多例
            timeout: 请求超时时间
            custom_argument: 自定义参数，浏览器启动配置参数
            download_path: 文件下载保存路径；
            browser_path: 浏览器可执行文件路径；
            **kwargs:
        """

        # 如果实例已存在，则不再重新初始化
        if self._browser is not None:
            return

        self._singleton_tab = singleton_tab
        self._driver_type = driver_type
        self._headless = headless
        self._user_agent = user_agent or setting.DEFAULT_USERAGENT
        self._proxy = proxy
        self._timeout = timeout
        self._load_images = load_images
        self._download_path = download_path
        self._browser_path = browser_path
        self._custom_argument = custom_argument
        self._kwargs = kwargs

        Settings.set_language("zh_cn")  # DrissionPage 的报错信息及提示设置为中文
        Settings.set_singleton_tab_obj(self._singleton_tab)

        co = ChromiumOptions()
        if self._browser_path is not None:
            co.set_browser_path(self._browser_path)

        port = port or setting.DRISSIONPAGE.get("port")
        user_data_path = user_data_path or setting.DRISSIONPAGE.get("user_data_path")
        if port is not None:
            co.set_local_port(int(port))
            if user_data_path is not None:
                co.set_user_data_path(user_data_path)
        else:
            # 设置自动端口范围
            co.auto_port(scope=setting.DRISSIONPAGE.get("scope"))

        # 设置默认超时时间，用于元素等待、alert 等待、WebPage的 s 模式连接等等
        if self._timeout is not None:
            co.set_timeouts(base=self._timeout)

        # 设置是否以无界面模式启动浏览器
        co.headless(on_off=self._headless)

        # 设置初始窗口大小
        window_size = setting.DRISSIONPAGE.get("window_size")
        if window_size is not None:
            window_size = ",".join((str(n) for n in window_size))
            co.set_argument("--window-size", window_size)

        # 设置 useragent
        co.set_user_agent(self._user_agent)

        # 设置浏览器代理
        if self._proxy is not None:
            co.set_argument("--proxy-server", value=self._proxy)

        # 设置是否禁止加载图片
        co.no_imgs(on_off=not self._load_images)

        # 设置下载路径
        if self._download_path is not None:
            co.set_download_path(self._download_path)

        # 添加自定义的配置参数
        if self._custom_argument:
            for arg in self._custom_argument:
                co.set_argument(arg)

        self._browser = Chromium(addr_or_opts=co)

    @property
    def is_running(self):
        return self._browser.states.is_alive if self._browser is not None else False

    def new_tab(self):
        if self.is_running:
            return self._browser.new_tab()

    def tabs_count(self):
        if self.is_running:
            return self._browser.tabs_count
        else:
            return 0

    def get_browser(self):
        return self._browser

    def quit(self):
        if self._browser is not None:
            self._browser.quit(del_data=True)
            self._browser = None
            SingletonMeta.clear_instance(self.__class__)  # 释放资源

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_val:
            log.error(exc_val)

        self.quit()
        return True


class DrissionPageDriver(WebDriver):

    def __init__(self, **kwargs):
        super(DrissionPageDriver, self).__init__(**kwargs)
        # 创建全局浏览器实例(单例)
        self._browser = Browser(**kwargs)

        # 创建新标签页
        tab = self._browser.new_tab()

        # 设置自动确认弹窗
        tab.set.auto_handle_alert()

        # 设置网页加载策略
        tab.set.load_mode(setting.DRISSIONPAGE.get("load_mode", "normal"))

        # 设置浏览器标识
        ua = kwargs.get("user_agent")
        if ua is not None:
            tab.set.user_agent(ua)

        self.url = None
        self.tab = tab

    @property
    def browser(self):
        return self._browser.get_browser()

    def get_tab(self):
        """获取当前标签页，启用多例-可支持多个实例控制同一个标签页"""
        return self.browser.get_tab(id_or_num=self.tab.tab_id)

    def get_dom_hash(self):
        """获取当前DOM的哈希值"""
        return tools.get_md5(self.tab.html)

    def wait_for_dom_stable(self, duration=None):
        """
        计算指定时间内页面的DOM变化次数，可忽略特定元素

        :param duration: 监听时长（秒）
        :return: DOM变化次数
        """
        duration = self.tab.timeouts.base if duration is None else duration
        script = f'''
            return new Promise((resolve) => {{
                let mutationCount = 0;

                // 创建MutationObserver监听DOM变化
                const observer = new MutationObserver((mutations) => {{
                    if (mutations.length > 0) {{
                        mutationCount += mutations.length;
                    }}
                }});

                // 开始监听所有DOM变化
                observer.observe(document, {{
                    childList: true,
                    attributes: true,
                    subtree: true,
                    characterData: true
                }});

                // 指定时间后停止监听并返回变化次数
                setTimeout(() => {{
                    observer.disconnect();
                    resolve(mutationCount);
                }}, {duration * 1000});
            }});
        '''
        count = self.tab.run_js(script)
        return True if count == 0 else False

    def reload(self):
        # 尝试触发重绘(通过重绘修复图片加载完成后可能导致布局错乱问题)
        self.tab.run_js('document.body.style.display="none";document.body.style.display="block";')

    def wait_for_dom_load(self, timeout=None, raise_err=False):
        """等待页面加载"""

        assert "ERR_CONNECTION_CLOSED" not in self.tab.raw_data
        assert self.tab.url_available is True

        timeout = self.tab.timeouts.page_load if timeout is None else timeout
        end_time = perf_counter() + timeout
        while perf_counter() < end_time:
            init_hash = self.get_dom_hash()
            render_time = (end_time - perf_counter()) / 10
            timeout = render_time if render_time > 0 else 1
            self.reload()
            self.tab.wait(timeout)
            current_hash = self.get_dom_hash()
            if current_hash == init_hash:
                return True

        if raise_err is True:
            raise TimeoutError("等待页面加载超时")

        return False

    @property
    def domain(self):
        return tools.get_domain(self.url or self.tab.url)

    def quit(self):
        if self.tab:
            self.tab.close()

        if self._browser.is_running and self._browser.tabs_count() <= 1:
            self._browser.quit()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_val:
            log.error(exc_val)

        self.quit()
        return True