跳至主要內容

浏览器自动化

HSXISAWD2026/3/20大约 4 分钟集成扩展HsxWorkFlow浏览器自动化

浏览器自动化

HsxWorkFlow 集成 DrissionPage 实现浏览器自动化操作，支持 Chromium 的页面操作、数据抓取等功能。

概览

DriverClient

DriverClient 是 DrissionPage 的客户端封装类。

创建客户端

from hsxworkflow.DP.DPClient import DriverClient

driver = DriverClient()

初始化浏览器

# 创建独立环境
driver.init_driver(new_env=True)

# 复用现有环境
driver.init_driver(new_env=False)

获取浏览器实例

# 获取最新标签页
browser = driver.get_browser(latest_tab=True)

# 创建新标签页
browser = driver.get_browser(new_tab=True)

# 获取浏览器对象
browser = driver.get_browser(latest_tab=False)

关闭页面

driver.close_page(
    tabs_range=None,        # 关闭范围 (start, end)
    remainder_one=True,     # 保留一个标签页
    clear_=True,            # 清理缓存
    close_interval=0,       # 关闭间隔（秒）
    cookies=True,           # 清除 cookies
    cache=True,             # 清除缓存
    session_storage=True,   # 清除 session storage
    local_storage=True,     # 清除 local storage
)

完整示例

from hsxworkflow import WorkRegisterHandler, StepActionHandler
from hsxworkflow.DP.DPClient import DriverClient

wf = WorkRegisterHandler(
    step_key="web_automation",
    title="浏览器自动化",
    desc="网页数据抓取"
)

@wf.register_class()
class WebAutomation(StepActionHandler):

    def __init__(self):
        super().__init__()
        self.driver = DriverClient()
        self.browser = None

    @wf.step(sort=1)
    def init_browser(self):
        """初始化浏览器"""
        self.socket_log("初始化浏览器...", base=True)

        # 初始化浏览器
        self.driver.init_driver(new_env=True)
        self.browser = self.driver.get_browser()

        return self.set_success(message="浏览器已初始化")

    @wf.step(sort=2)
    def navigate_to_page(self):
        """导航到页面"""
        self.socket_log("导航到页面...", base=True)

        # 导航到指定页面
        self.browser.get("https://example.com")

        # 等待页面加载
        self.browser.wait.load_start()

        return self.set_success(message="页面已加载")

    @wf.step(sort=3)
    def scrape_data(self):
        """抓取数据"""
        self.socket_log("抓取数据...", base=True)

        # 获取页面标题
        title = self.browser.title
        self.socket_log(f"页面标题: {title}", base=True)

        # 获取页面 URL
        url = self.browser.url
        self.socket_log(f"页面 URL: {url}", base=True)

        # 提取文本内容
        content = self.browser.html
        self.socket_log(f"内容长度: {len(content)}", base=True)

        return self.set_success(
            message="数据抓取完成",
            data={
                "title": title,
                "url": url,
                "content_length": len(content)
            }
        )

    @wf.step(sort=4)
    def cleanup(self):
        """清理浏览器"""
        self.socket_log("清理浏览器...", base=True)

        # 关闭所有页面
        self.driver.close_page(
            remainder_one=False,
            clear_=True,
            cookies=True,
            cache=True
        )

        return self.set_success(message="浏览器已清理")

页面操作

导航

# 导航到 URL
browser.get("https://example.com")

# 刷新页面
browser.refresh()

# 后退
browser.back()

# 前进
browser.forward()

元素定位

# 通过选择器定位
element = browser.ele('css:div.content')

# 通过文本定位
element = browser.ele('text:提交')

# 通过 XPath 定位
element = browser.ele('xpath://div[@class="content"]')

元素操作

# 点击
element.click()

# 输入文本
element.input("Hello World")

# 获取文本
text = element.text

# 获取属性
href = element.attr('href')

# 等待元素出现
browser.wait.ele_displayed('css:div.content', timeout=10)

表单操作

# 填写表单
browser.ele('css:#username').input("admin")
browser.ele('css:#password').input("password")
browser.ele('css:button[type="submit"]').click()

# 选择下拉框
browser.ele('css:#country').select("China")

# 勾选复选框
browser.ele('css:#agree').click()

数据抓取

文本提取

# 获取页面文本
text = browser.html

# 获取元素文本
element = browser.ele('css:div.content')
text = element.text

# 获取属性值
href = element.attr('href')
src = element.attr('src')

图片下载

# 获取图片元素
img = browser.ele('css:img.logo')

# 下载图片
img.download('logo.png')

列表数据

# 获取列表
items = browser.eles('css:li.item')

# 遍历列表
for item in items:
    text = item.text
    print(text)

资源管理

# 获取所有 cookies
cookies = browser.cookies()

# 获取特定 cookie
cookie = browser.get_cookie('session_id')

# 设置 cookie
browser.set_cookie('session_id', 'abc123')

# 删除 cookie
browser.del_cookie('session_id')

# 清除所有 cookies
browser.clear_cookies()

缓存管理

# 清除缓存
browser.clear_cache()

# 清除 session storage
browser.clear_session_storage()

# 清除 local storage
browser.clear_local_storage()

标签页管理

# 获取所有标签页
tabs = browser.get_tabs()

# 切换标签页
browser.to_tab(tab_id)

# 关闭当前标签页
browser.close_tab()

# 关闭所有标签页
browser.close_all_tabs()

最佳实践

1. 等待策略

# ✅ 使用显式等待
browser.wait.ele_displayed('css:div.content', timeout=10)

# ❌ 使用固定延迟
import time
time.sleep(5)

2. 异常处理

@wf.step(sort=1)
def scrape_with_retry(self):
    try:
        browser = self.driver.get_browser()
        browser.get("https://example.com")
        data = browser.ele('css:div.content').text
        return self.set_success(data=data)
    except Exception as e:
        return self.set_error(message=f"抓取失败: {str(e)}")

3. 资源清理

@wf.step(sort=3)
def cleanup(self):
    # 确保清理资源
    try:
        self.driver.close_page(remainder_one=False, clear_=True)
    except Exception as e:
        self.socket_log(f"清理失败: {e}", level="w")
    return self.set_success()

下一步

现在你已经了解了浏览器自动化的使用方法，可以继续学习：

Default footer
Copyright © 2026 HSXISAWD