[Python] How to intercept response body using nodriver (python)?

Discussão em 'Python' iniciado por Stack, Fevereiro 21, 2025.

  1. Stack

    Stack Membro Participativo

    I'm trying to intercept the response from the web server and extract the body. it uses the module nodriver to successfully load the page and capture the request event. However when it attempts to send the get_response_body generator to the tab (body = await self.tab.send(gen)) it gets stuck. What am i missing?

    import nodriver
    import asyncio
    import os

    url = "https://www.getfpv.com/graphql?query=%0A%7B%0A%20%20products(%0A%20%20%20%20filter%3A%20%7B%0A%20%20%20%20%20%20%0A%20%20%20%20%20%20sku%3A%20%7B%20in%3A%20%5B%2222817%22%5D%20%7D%2C%0A%20%20%20%20%20%20%0A%20%20%20%20%20%20%0A%20%20%20%20%20%20%7D%0A%20%20%20%20pageSize%3A%208%2C%0A%20%20%20%20sort%3A%20%7Bposition%3A%20ASC%7D%0A%20%20%20%20)%20%7B%0A%20%20%20%20items%20%7B%0A%20%20%20%20%20%20%20%20related_products%20%7B%0A%20%20%20%20%20%20%20%20%0A%20%20%20%20%20%20sku%0A%20%20%20%20%20%20id%0A%20%20%20%20%20%20name%0A%20%20%20%20%20%20small_image%20%7B%0A%20%20%20%20%20%20%20%20label%0A%20%20%20%20%20%20%20%20url%0A%20%20%20%20%20%20%7D%0A%20%20%20%20%20%20thumbnail%20%7B%0A%20%20%20%20%20%20%20%20url%0A%20%20%20%20%20%20%7D%0A%20%20%20%20%20%20url_key%0A%20%20%20%20%20%20url_suffix%0A%20%20%20%20%20%20visibility%0A%20%20%20%20%20%20status%0A%20%20%20%20%20%20price_range%20%7B%0A%20%20%20%20%20%20%20%20minimum_price%20%7B%0A%20%20%20%20%20%20%20%20%20%20regular_price%20%7B%0A%20%20%20%20%20%20%20%20%20%20%20%20value%0A%20%20%20%20%20%20%20%20%20%20%20%20currency%0A%20%20%20%20%20%20%20%20%20%20%7D%0A%20%20%20%20%20%20%20%20%20%20final_price%20%7B%0A%20%20%20%20%20%20%20%20%20%20%20%20value%0A%20%20%20%20%20%20%20%20%20%20%20%20currency%0A%20%20%20%20%20%20%20%20%20%20%7D%0A%20%20%20%20%20%20%20%20%7D%0A%20%20%20%20%20%20%7D%0A%20%20%20%20%7D%0A%20%20%20%20%7D%0A%20%20%7D%0A%7D"

    class Scraper:
    tab: nodriver.Tab

    def __init__(self):
    nodriver.loop().run_until_complete(self.main())

    async def handle_request(self, event):
    print(f"Event: {event}")
    request_id = event.request_id
    print(f"Request ID: {request_id}")
    gen = nodriver.cdp.network.get_response_body(request_id)
    print(f"Generator: {gen}")
    if event.request.url == url:
    print("Probably about to get stuck...")
    body = await self.tab.send(gen) # Gets stuck here
    print(f"Body: {body}")

    async def main(self):
    browser = await nodriver.start()
    self.tab = browser.main_tab
    await self.tab.send(nodriver.cdp.network.enable())
    self.tab.add_handler(nodriver.cdp.network.RequestWillBeSent, self.handle_request)
    await self.tab.get(url)
    await self.tab.wait(t=10)


    if __name__ == "__main__":
    Scraper()

    Continue reading...

Compartilhe esta Página