this repo has no description
First pass of bsky-activity and commit_ops
+98
+87
bsky-activity.py
+87
bsky-activity.py
···
1
+
#!/usr/bin/env python3
2
+
3
+
import dag_cbor
4
+
import os
5
+
import redis
6
+
import sqlite3
7
+
import sys
8
+
from datetime import datetime, timezone
9
+
from firehose_utils import commit_ops
10
+
from io import BytesIO
11
+
12
+
app_bsky_allowlist = set([
13
+
'app.bsky.actor.profile',
14
+
'app.bsky.feed.generator',
15
+
'app.bsky.feed.like',
16
+
'app.bsky.feed.post',
17
+
'app.bsky.feed.repost',
18
+
'app.bsky.feed.threadgate',
19
+
'app.bsky.graph.block',
20
+
'app.bsky.graph.follow',
21
+
'app.bsky.graph.list',
22
+
'app.bsky.graph.listblock',
23
+
'app.bsky.graph.listitem',
24
+
'app.bsky.labeler.service',
25
+
])
26
+
27
+
def main():
28
+
redis_cnx = redis.Redis()
29
+
redis_pipe = redis_cnx.pipeline()
30
+
redis_sub = redis_cnx.pubsub(ignore_subscribe_messages=True)
31
+
32
+
db_fname = '/opt/muninsky/users.db'
33
+
db_fname = 'users.db'
34
+
35
+
db_cnx = sqlite3.connect(db_fname)
36
+
with db_cnx:
37
+
db_cnx.executescript("""
38
+
PRAGMA journal_mode = WAL;
39
+
PRAGMA synchronous = off;
40
+
CREATE TABLE IF NOT EXISTS users (did TEXT, ts TIMESTAMP);
41
+
CREATE UNIQUE INDEX IF NOT EXISTS did_idx on users(did);
42
+
CREATE INDEX IF NOT EXISTS ts_idx on users(ts);
43
+
""")
44
+
45
+
op_count = 0
46
+
redis_sub.subscribe('bsky-tools:firehose:stream')
47
+
for event in redis_sub.listen():
48
+
frame = BytesIO(event['data'])
49
+
header = dag_cbor.decode(frame, allow_concat=True)
50
+
if header['op'] != 1 or header['t'] != '#commit':
51
+
continue
52
+
53
+
payload = dag_cbor.decode(frame)
54
+
if payload['tooBig']:
55
+
# TODO(ejd): how handle these?
56
+
continue
57
+
58
+
for op in commit_ops(payload):
59
+
if op['action'] != 'create':
60
+
continue
61
+
62
+
collection, _ = op['path'].split('/')
63
+
if collection not in app_bsky_allowlist:
64
+
continue
65
+
66
+
repo_did = payload['repo']
67
+
ts = datetime.now(timezone.utc).timestamp()
68
+
db_cnx.execute(
69
+
'insert into users values (:did, :ts) on conflict (did) do update set ts = :ts',
70
+
{'did': repo_did, 'ts': ts}
71
+
)
72
+
73
+
redis_pipe \
74
+
.incr(collection) \
75
+
.incr('dev.edavis.muninsky.ops')
76
+
77
+
op_count += 1
78
+
if op_count % 500 == 0:
79
+
payload_seq = payload['seq']
80
+
sys.stdout.write(f'checkpoint: seq: {payload_seq}\n')
81
+
redis_pipe.set('dev.edavis.muninsky.seq', payload_seq)
82
+
redis_pipe.execute()
83
+
db_cnx.commit()
84
+
sys.stdout.flush()
85
+
86
+
if __name__ == '__main__':
87
+
main()
+11
firehose_utils.py
+11
firehose_utils.py
···
1
+
from atproto import CAR
2
+
3
+
def commit_ops(payload):
4
+
# TODO(ejd): figure out how to validate blocks
5
+
car_parsed = CAR.from_bytes(payload['blocks'])
6
+
for op in payload['ops']:
7
+
repo_op = op.copy()
8
+
if op['cid'] is not None:
9
+
repo_op['cid'] = op['cid'].encode('base32')
10
+
repo_op['record'] = car_parsed.blocks[repo_op['cid']]
11
+
yield repo_op