this repo has no description
edavis.dev
1import logging
2
3import apsw
4import apsw.ext
5
6from . import BaseFeed
7
8class PopularFeed(BaseFeed):
9 FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/popular'
10
11 def __init__(self):
12 self.db_cnx = apsw.Connection('db/popular.db')
13 self.db_cnx.pragma('journal_mode', 'WAL')
14 self.db_cnx.pragma('wal_autocheckpoint', '0')
15
16 with self.db_cnx:
17 self.db_cnx.execute("""
18 create table if not exists posts (uri text, create_ts timestamp, update_ts timestamp, temperature int);
19 create unique index if not exists uri_idx on posts(uri);
20 """)
21
22 self.logger = logging.getLogger('feeds.popular')
23
24 def process_commit(self, commit):
25 op = commit['op']
26 if op['action'] != 'create':
27 return
28
29 collection, _ = op['path'].split('/')
30 if collection != 'app.bsky.feed.like':
31 return
32
33 record = op.get('record')
34 if record is None:
35 return
36
37 ts = self.safe_timestamp(record.get('createdAt')).timestamp()
38 like_subject_uri = op['record']['subject']['uri']
39
40 self.transaction_begin(self.db_cnx)
41
42 self.db_cnx.execute("""
43 insert into posts (uri, create_ts, update_ts, temperature)
44 values (:uri, :ts, :ts, 1)
45 on conflict (uri) do update set temperature = temperature + 1, update_ts = :ts
46 """, dict(uri=like_subject_uri, ts=ts))
47
48 def delete_old_posts(self):
49 self.db_cnx.execute("""
50 delete from posts
51 where
52 temperature * exp( -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 ) ) < 1.0
53 and create_ts < unixepoch('now', '-15 minutes')
54 """)
55
56 def commit_changes(self):
57 self.logger.debug('committing changes')
58 self.delete_old_posts()
59 self.transaction_commit(self.db_cnx)
60 self.wal_checkpoint(self.db_cnx, 'RESTART')
61
62 def serve_feed(self, limit, offset, langs):
63 cur = self.db_cnx.execute("""
64 select uri from posts
65 order by temperature * exp( -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 ) )
66 desc limit :limit offset :offset
67 """, dict(limit=limit, offset=offset))
68 return [uri for (uri,) in cur]
69
70 def serve_feed_debug(self, limit, offset, langs):
71 query = """
72 select
73 uri, temperature,
74 unixepoch('now') - create_ts as age_seconds,
75 exp(
76 -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 )
77 ) as decay,
78 temperature * exp(
79 -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 )
80 ) as score
81 from posts
82 order by score desc
83 limit :limit offset :offset
84 """
85 bindings = dict(limit=limit, offset=offset)
86 return apsw.ext.format_query_table(
87 self.db_cnx, query, bindings,
88 string_sanitize=2, text_width=9999, use_unicode=True
89 )