Web自动化中基于网络请求拦截的精准等待策略实践

from selenium import webdriver
from selenium.webdriver.chrome.service import Service

driver = webdriver.Chrome()
# 获取DevTools会话
dev_tools = driver.execute_cdp_cmd("Target.attachToTarget", {})  # 或直接使用 driver.execute_cdp_cmd
# 实际Selenium 4推荐使用 driver.get_devtools()
# 但为兼容性，我们使用 execute_cdp_cmd 方式

import json
import time
from selenium import webdriver

class NetworkWaiter:
    def __init__(self, driver):
        self.driver = driver
        self._target_id = None
        self._enable_network()

    def _enable_network(self):
        """启用网络监听"""
        # 获取目标ID（Tab页）
        targets = self.driver.execute_cdp_cmd("Target.getTargets", {})
        for target in targets['targetInfos']:
            if target['type'] == 'page':
                self._target_id = target['targetId']
                break
        # 启用网络域
        self.driver.execute_cdp_cmd("Network.enable", {})
        # 注意：事件监听需要通过 add_listener 或回调，这里我们采用轮询缓冲区方式

    def wait_for_request(self, url_substring, timeout=15, status_code=200):
        """
        等待包含 url_substring 的请求完成并返回指定状态码
        返回 (是否成功, 响应体)
        """
        # 由于CDP事件异步，我们需要在回调中收集响应，但更简单的方式是使用一个共享缓存
        # 这里我们采用循环检查 driver.execute_cdp_cmd 获取所有已完成的请求（但CDP没有直接提供）
        # 实际我们使用事件监听 + 条件变量，但为了代码精简，采用轮询+回调缓冲
        # 更稳健：注册事件，放入队列，主线程等待队列
        import queue
        response_queue = queue.Queue()

        def response_received(event):
            # event包含 requestId, response, type, timestamp等
            if 'response' in event and 'url' in event['response']:
                if url_substring in event['response']['url'] and event['response']['status'] == status_code:
                    response_queue.put(event['response'])

        # 注册监听
        self.driver.execute_cdp_cmd("Network.onResponseReceived", {})  # 实际不是这样，正确用法：
        # 使用 driver.add_listener 或者通过 DevTools 对象的 add_listener
        # 因为 execute_cdp_cmd 无法直接注册回调，我们需要使用 DevTools 对象

        # 下面展示 Selenium 4 推荐的正确用法
        devtools = self.driver.get_devtools()
        devtools.add_listener("Network.responseReceived", response_received)

        # 等待队列
        start = time.time()
        while time.time() - start < timeout:
            try:
                resp = response_queue.get(timeout=0.5)
                return True, resp.get('body', '')  # body需要单独获取，需用 Network.getResponseBody
            except queue.Empty:
                continue
        return False, None

import asyncio
from selenium.webdriver.common.devtools import version

class ApiResponseWaiter:
    def __init__(self, driver):
        self.driver = driver
        self._devtools = None

    def _ensure_devtools(self):
        if not self._devtools:
            self._devtools = self.driver.get_devtools()
            self._devtools.send_cmd(version.Command.NETWORK_ENABLE, {})
        return self._devtools

    def wait_for_api(self, url_part, timeout=10, expected_status=200):
        """
        等待包含 url_part 的请求完成且状态码为 expected_status
        返回响应数据（dict）
        """
        devtools = self._ensure_devtools()
        future = asyncio.get_event_loop().create_future()

        def handle_response(event):
            if 'response' not in event:
                return
            response = event['response']
            if url_part in response.get('url', '') and response.get('status') == expected_status:
                # 获取响应体
                request_id = event['requestId']
                try:
                    body_response = devtools.send_cmd(version.Command.NETWORK_GET_RESPONSE_BODY, {
                        'requestId': request_id
                    })
                    body = body_response.get('body', '')
                    if body_response.get('base64Encoded'):
                        import base64
                        body = base64.b64decode(body).decode('utf-8')
                    # 尝试解析JSON
                    import json
                    data = json.loads(body)
                except Exception as e:
                    data = {'raw': body}
                future.set_result(data)

        # 注册监听器
        devtools.add_listener("Network.responseReceived", handle_response)

        # 阻塞等待，最多 timeout 秒
        try:
            result = asyncio.get_event_loop().run_until_complete(
                asyncio.wait_for(future, timeout)
            )
            return result
        except asyncio.TimeoutError:
            return None

# pages/order_page.py
class OrderPage(BasePage):
    SEARCH_BTN = (By.ID, "searchBtn")
    RESULT_TABLE = (By.CSS_SELECTOR, "table#orderTable")

    def search_orders(self, start_date, end_date):
        # 输入日期，点击搜索
        self.safe_send_keys(self.START_DATE, start_date)
        self.safe_send_keys(self.END_DATE, end_date)
        # 启动网络等待器（在BasePage中持有）
        waiter = ApiResponseWaiter(self.driver)
        # 点击搜索
        self.safe_click(self.SEARCH_BTN)
        # 精准等待 /api/orders 响应
        resp_data = waiter.wait_for_api("/api/orders", timeout=10)
        # 若返回None，则超时失败
        if not resp_data:
            raise TimeoutError("订单API未在10秒内返回")
        # 可选：校验响应内容
        assert resp_data.get('code') == 200
        # 此时表格必然已刷新，直接返回数据列表供断言
        return resp_data.get('data', [])

def wait_for_apis(self, url_parts, timeout=15):
    devtools = self._ensure_devtools()
    pending = set()
    results = {}

    def handler(event):
        if 'response' not in event:
            return
        response = event['response']
        for part in url_parts:
            if part in response.get('url', '') and response.get('status') == 200:
                # 记录该请求完成
                results[part] = True
    devtools.add_listener("Network.responseReceived", handler)
    start = time.time()
    while time.time() - start < timeout:
        if all(part in results for part in url_parts):
            return True
        time.sleep(0.2)
    return False