this repo has no description
firehose.py: reworked
+46
+46
firehose.py
+46
firehose.py
···
1
+
#!/usr/bin/env python3
2
+
3
+
import asyncio
4
+
import dag_cbor
5
+
import redis
6
+
import sys
7
+
import websockets
8
+
from datetime import datetime, timezone
9
+
from io import BytesIO
10
+
11
+
async def main():
12
+
redis_cnx = redis.Redis()
13
+
relay_url = 'wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos'
14
+
firehose_seq = redis_cnx.get('bsky-tools:firehose:subscribe-repos:seq')
15
+
if firehose_seq:
16
+
relay_url += f'?cursor={firehose_seq.decode()}'
17
+
18
+
sys.stdout.write(f'opening websocket to {relay_url}\n')
19
+
sys.stdout.flush()
20
+
21
+
async with websockets.connect(relay_url, ping_timeout=None) as firehose:
22
+
current_minute = None
23
+
while True:
24
+
message = BytesIO(await firehose.recv())
25
+
header = dag_cbor.decode(message, allow_concat=True)
26
+
if header['op'] != 1 or header['t'] == '#info':
27
+
continue
28
+
29
+
redis_cnx.publish('bsky-tools:firehose:stream', message.getvalue())
30
+
31
+
# checkpoint the seq
32
+
now = datetime.now(timezone.utc)
33
+
if now.time().minute != current_minute:
34
+
current_minute = now.time().minute
35
+
36
+
payload = dag_cbor.decode(message)
37
+
payload_seq = payload['seq']
38
+
payload_time = datetime.strptime(payload['time'], '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)
39
+
payload_lag = now - payload_time
40
+
41
+
redis_cnx.set('bsky-tools:firehose:subscribe-repos:seq', payload_seq)
42
+
sys.stdout.write(f'seq: {payload_seq}, lag: {payload_lag.total_seconds()}\n')
43
+
sys.stdout.flush()
44
+
45
+
if __name__ == '__main__':
46
+
asyncio.run(main())