Scraping from Websocket

[Ehsan-U]

Wouldn't it be useful to have a crawler type specifically for scraping data from WebSocket endpoints?

[janbuchar]

Hello @Ehsan-U, that's an interesting idea! Do you have a particular use case for this? My intuition would be that websocket messages are usually structured data already, so the benefits of the crawler would be somewhat limited, but perhaps I'm missing something important.

[Ehsan-U]

Hi @janbuchar,
I recently came across a few projects where real-time data is accessible via WebSocket connections. To my surprise, none of the current scraping frameworks provide native support for handling WebSockets. The only option seems to be the websockets package, which requires a lot of boilerplate just to access the data.

Here's an example of a project where data is being scraped from Discord channel using a WebSocket connection:

class DiscordClient:
    ws_uri: str = "wss://gateway.discord.gg/?v=9&encoding=json"
    last_sq: int
    targets: Dict[str, Dict] = {}


    @asynccontextmanager
    async def connect(self):
        ws, heartbeat_task = None, None
        try:
            ws = await websockets.connect(self.ws_uri)
            hello_msg = await ws.recv()
            hello_data = json.loads(hello_msg)

            if hello_data['op'] == 10:
                heartbeat_interval = hello_data['d']['heartbeat_interval'] / 1000
                heartbeat_task = asyncio.create_task(self.send_heartbeat(ws, heartbeat_interval))

                await ws.send(json.dumps(IDENTIFY_PAYLOAD))

                msg = await ws.recv()
                data = json.loads(msg)
                self.last_sq = data['s']
                op = data['op']
                if op == 0:
                    yield ws
        except (Exception, asyncio.CancelledError) as e:
            log.error(event="connecting to ws", exception=str(e))
            raise e
        finally:
            if heartbeat_task and not heartbeat_task.done():
                heartbeat_task.cancel()
                try:
                    await heartbeat_task
                except asyncio.CancelledError:
                    pass
            if ws:
                await ws.close()


    async def send_heartbeat(self, ws: ClientConnection, interval: int):
        while True:
            await asyncio.sleep(interval)
            payload = {"op": 1, "d": self.last_sq}
            await ws.send(json.dumps(payload))
            log.debug(event="heartbeat SYN", data={"sq": self.last_sq})


    async def on_message(self, message: str):
        data = json.loads(message)
        op = data['op']
        if op == 0:
            t = data['t']
            if t == "MESSAGE_CREATE":
                channel_id = data['d']['channel_id']
                print(msg['d'])
                msg = data['d']['content']
                for server, channels in self.targets.items():
                    if channel_id in channels:
                        log.info(event="new message", data={"msg": msg,"channel": channel_id,"server": server})
                        break
        elif op == 1:
            log.debug(event="heartbeat ACK", data={"sq": data['s']})


    async def listen(self):
        async with self.connect() as ws:
            log.info(event="listening")
            while True:
                msg = await ws.recv()
                await self.on_message(msg)


    async def run(self):
        while True:
            try:
                await self.listen()
            except (Exception, asyncio.CancelledError) as e:
                log.error(event="listen", exception=str(e))
                await asyncio.sleep(5)


d = DiscordClient()
asyncio.run(d.run())

[janbuchar]

Thanks for the explanation. Honestly, I don't see how a scraping framework could help here - you are opening an asynchronous communication channel that uses messages with a well-defined structure. That is in sharp contrast to what crawlee works with - resources accessible via URLs, free-form data (HTML), links to other resources, a strict request-response communication pattern.