···

       
1
       
1
       
-
       
#!/usr/bin/env python3

     

       
2
       
2
       
-
       


     

       
3
       
3
       
-
       
import asyncio

     

       
4
       
4
       
-
       
from datetime import datetime, timezone

     

       
5
       
5
       
-
       
import json

     

       
6
       
6
       
-
       
import os

     

       
7
       
7
       
-
       
import sqlite3

     

       
8
       
8
       
-
       
import sys

     

       
9
       
9
       
-
       


     

       
10
       
10
       
-
       
import redis

     

       
11
       
11
       
-
       
import websockets

     

       
12
       
12
       
-
       


     

       
13
       
13
       
-
       
app_bsky_allowlist = (

     

       
14
       
14
       
-
       
    'app.bsky.actor.profile',

     

       
15
       
15
       
-
       
    'app.bsky.feed.generator',

     

       
16
       
16
       
-
       
    'app.bsky.feed.like',

     

       
17
       
17
       
-
       
    'app.bsky.feed.post',

     

       
18
       
18
       
-
       
    'app.bsky.feed.postgate',

     

       
19
       
19
       
-
       
    'app.bsky.feed.repost',

     

       
20
       
20
       
-
       
    'app.bsky.feed.threadgate',

     

       
21
       
21
       
-
       
    'app.bsky.graph.block',

     

       
22
       
22
       
-
       
    'app.bsky.graph.follow',

     

       
23
       
23
       
-
       
    'app.bsky.graph.list',

     

       
24
       
24
       
-
       
    'app.bsky.graph.listblock',

     

       
25
       
25
       
-
       
    'app.bsky.graph.listitem',

     

       
26
       
26
       
-
       
    'app.bsky.graph.starterpack',

     

       
27
       
27
       
-
       
    'app.bsky.labeler.service',

     

       
28
       
28
       
-
       
    'chat.bsky.actor.declaration',

     

       
29
       
29
       
-
       
)

     

       
30
       
30
       
-
       


     

       
31
       
31
       
-
       
other_allowlist = (

     

       
32
       
32
       
-
       
    'social.psky',

     

       
33
       
33
       
-
       
    'blue.zio.atfile',

     

       
34
       
34
       
-
       
    'com.shinolabs.pinksea',

     

       
35
       
35
       
-
       
    'com.whtwnd',

     

       
36
       
36
       
-
       
    'events.smokesignal',

     

       
37
       
37
       
-
       
    'fyi.unravel',

     

       
38
       
38
       
-
       
    'xyz.statusphere',

     

       
39
       
39
       
-
       
)

     

       
40
       
40
       
-
       


     

       
41
       
41
       
-
       
async def bsky_activity():

     

       
42
       
42
       
-
       
    relay_url = 'wss://jetstream1.us-west.bsky.network/subscribe'

     

       
43
       
43
       
-
       


     

       
44
       
44
       
-
       
    sys.stdout.write(f'opening websocket connection to {relay_url}\n')

     

       
45
       
45
       
-
       
    sys.stdout.flush()

     

       
46
       
46
       
-
       


     

       
47
       
47
       
-
       
    async with websockets.connect(relay_url, ping_timeout=60) as firehose:

     

       
48
       
48
       
-
       
        while True:

     

       
49
       
49
       
-
       
            yield json.loads(await firehose.recv())

     

       
50
       
50
       
-
       


     

       
51
       
51
       
-
       
async def main():

     

       
52
       
52
       
-
       
    redis_cnx = redis.Redis()

     

       
53
       
53
       
-
       
    redis_pipe = redis_cnx.pipeline()

     

       
54
       
54
       
-
       


     

       
55
       
55
       
-
       
    sys.stdout.write('starting up\n')

     

       
56
       
56
       
-
       
    sys.stdout.flush()

     

       
57
       
57
       
-
       


     

       
58
       
58
       
-
       
    op_count = 0

     

       
59
       
59
       
-
       
    async for event in bsky_activity():

     

       
60
       
60
       
-
       
        if event['kind'] != 'commit':

     

       
61
       
61
       
-
       
            continue

     

       
62
       
62
       
-
       


     

       
63
       
63
       
-
       
        payload = event.get('commit')

     

       
64
       
64
       
-
       
        if payload is None:

     

       
65
       
65
       
-
       
            continue

     

       
66
       
66
       
-
       


     

       
67
       
67
       
-
       
        if payload['operation'] != 'create':

     

       
68
       
68
       
-
       
            continue

     

       
69
       
69
       
-
       


     

       
70
       
70
       
-
       
        collection = payload['collection']

     

       
71
       
71
       
-
       
        if not collection.startswith(app_bsky_allowlist + other_allowlist):

     

       
72
       
72
       
-
       
            continue

     

       
73
       
73
       
-
       


     

       
74
       
74
       
-
       
        for prefix in other_allowlist:

     

       
75
       
75
       
-
       
            if collection.startswith(prefix):

     

       
76
       
76
       
-
       
                redis_pipe.incr('dev.edavis.atproto.collection.' + prefix.replace('.', '_'))

     

       
77
       
77
       
-
       


     

       
78
       
78
       
-
       
        if collection == 'app.bsky.feed.post':

     

       
79
       
79
       
-
       
            embed = payload['record'].get('embed')

     

       
80
       
80
       
-
       
            if embed is not None and embed.get('$type', ''):

     

       
81
       
81
       
-
       
                embed_type = embed['$type']

     

       
82
       
82
       
-
       
                redis_pipe.incr(f'app.bsky.feed.post:embed:{embed_type}')

     

       
83
       
83
       
-
       


     

       
84
       
84
       
-
       
        redis_pipe \

     

       
85
       
85
       
-
       
            .incr(collection) \

     

       
86
       
86
       
-
       
            .incr('dev.edavis.muninsky.ops')

     

       
87
       
87
       
-
       


     

       
88
       
88
       
-
       
        op_count += 1

     

       
89
       
89
       
-
       
        if op_count % 500 == 0:

     

       
90
       
90
       
-
       
            event_time_ms = event['time_us'] / 1_000_000

     

       
91
       
91
       
-
       
            sys.stdout.write(f'timestamp: {event_time_ms}\n')

     

       
92
       
92
       
-
       
            redis_pipe.execute()

     

       
93
       
93
       
-
       
            sys.stdout.flush()

     

       
94
       
94
       
-
       


     

       
95
       
95
       
-
       
if __name__ == '__main__':

     

       
96
       
96
       
-
       
    asyncio.run(main())