我现在的代码是这样的:


# Note: driver variable allready initialzed (Chrome)

import trio # async library that selenium uses
import requests
from PIL import Image
from io import BytesIO

def show_image(url):
    try:
        response = requests.get(url)
        img = Image.open(BytesIO(response.content))
        img.show()
    except Exception as e:
        print(e)

def intercept(event, devtools):
    return devtools.fetch.fail_request(request_id=event.request_id,error_reason=devtools.network.ErrorReason.CONNECTION_REFUSED)

async def at_event(listener, connection, intercept):
    session, devtools = connection.session, connection.devtools
    async for event in listener:

        print({"frame_id": event.frame_id, "url": event.request.url})
        #show_image(event.request.url)
        await trio.sleep(0.0001)
        try:
            await session.execute(intercept(event=event, devtools=devtools))
            # await session.execute(devtools.fetch.fulfill_request(request_id = event.request_id, response_code=200))
            # await session.execute(devtools.fetch.continue_request(request_id=event.request_id))
        except Exception as e:
            print(e)

async def async_interceptor(pattern,main_func, intercept):
    async with driver.bidi_connection() as connection:
        session, devtools = connection.session, connection.devtools

        pattern = devtools.fetch.RequestPattern.from_json(pattern)
        await session.execute(devtools.fetch.enable(patterns =[pattern]))

        listener = session.listen(devtools.fetch.RequestPaused)
        async with trio.open_nursery() as nursery:
            nursery.start_soon(at_event,listener, connection, intercept)
            nursery.start_soon(main_func)

async def interceptor():
    await async_interceptor({"resourceType":"Image"},main, intercept)
async def main():
    driver.get("https://www.youtube.com/")

trio.run(interceptor)

我想要改变的是:

  • 我希望拦截器像线程中的100一样,而不是在异步函数中
  • 自动取款机,监听器并不总是被触发,也许是因为异步代码中的某些东西?

在我的示例代码中,我希望截取所有属于图像的请求并使其失败

推荐答案

我自己得到了答案:

import threading
import trio
from PIL import Image
from io import BytesIO
import requests

def show_image(url):
    try:
        response = requests.get(url)
        img = Image.open(BytesIO(response.content))
        img.show()
    except Exception as e:
        print(e)

def listener_helper(listener, at_event):
    async def async_listener_helper(listener, at_event):
        async with driver.bidi_connection() as connection:
            session, devtools = connection.session, connection.devtools

            my_listener = await listener(connection=connection)

            async for event in my_listener:
                print({"frame_id": event.frame_id, "url": event.request.url})
                try:
                    await session.execute(at_event(event=event, connection=connection))
                except Exception as e:
                    print(e)

    trio.run(async_listener_helper, listener,at_event)

def threaded_listener(listener, at_event):
    thread = threading.Thread(target=listener_helper, kwargs={"listener":listener, "at_event":at_event})
    thread.start()
    return thread

async def request_listener(connection):
    session, devtools = connection.session, connection.devtools
    pattern = devtools.fetch.RequestPattern.from_json({"resourceType":"Image"})
    await session.execute(devtools.fetch.enable(patterns=[pattern]))

    return session.listen(devtools.fetch.RequestPaused)

def at_event(event, connection):
    session, devtools = connection.session, connection.devtools
    # show_image(event.request.url)
    return devtools.fetch.fail_request(request_id=event.request_id,error_reason=devtools.network.ErrorReason.CONNECTION_REFUSED)

thread = threaded_listener(request_listener,at_event)

driver.get("https://www.youtube.com/")

Python-3.x相关问答推荐

一种基于绝对排序值的极框索引和列定位的Python方法

是否有必要使用Threads()中的args显式地将共享变量传递给Python中的线程函数或直接访问它?

如何绘制交叉验证的AUROC并找到最佳阈值?

msg-seviri l1.5本机文件

Django 模型类方法使用错误的 `self`

转换Pandas 数据框 - 添加行

切片的Python复杂性与元组的星号相结合

当参数名称与 typing.Protocol 中定义的名称不同时发生 mypy 错误

导入在不同目录中定义的函数

列表中的重复数字与列表理解

判断是否存在大文件而不下载它

将字符串表示与使用整数值的枚举相关联?

无法在 macOS 上的 Anaconda3 python3.6 上安装 OpenCV3

Python 3x 的最佳机器学习包?

为什么我在 Python 中收到错误消息无法导入名称 NoneType?

为什么 virtualenv 会有效地禁用 Python 3 制表符补全?

我们如何获得 __repr__() 的默认行为?

如何在 jupyter notebook 5 中逐行分析 python 3.5 代码

如何从Pandas 中的字符串中提取前8个字符

如何在 Python 3.4 中使用 pip 3?