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 if commit['opType'] != 'c':
26 return
27
28 if commit['collection'] != 'app.bsky.feed.like':
29 return
30
31 record = commit.get('record')
32 if record is None:
33 return
34
35 ts = self.safe_timestamp(record.get('createdAt')).timestamp()
36 like_subject_uri = record['subject']['uri']
37
38 self.transaction_begin(self.db_cnx)
39
40 self.db_cnx.execute("""
41 insert into posts (uri, create_ts, update_ts, temperature)
42 values (:uri, :ts, :ts, 1)
43 on conflict (uri) do update set temperature = temperature + 1, update_ts = :ts
44 """, dict(uri=like_subject_uri, ts=ts))
45
46 def delete_old_posts(self):
47 self.db_cnx.execute("""
48 delete from posts
49 where
50 temperature * exp( -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 ) ) < 1.0
51 and create_ts < unixepoch('now', '-15 minutes')
52 """)
53 self.logger.debug('deleted {} old posts'.format(self.db_cnx.changes()))
54
55 def commit_changes(self):
56 self.delete_old_posts()
57 self.logger.debug('committing changes')
58 self.transaction_commit(self.db_cnx)
59 self.wal_checkpoint(self.db_cnx, 'RESTART')
60
61 def serve_feed(self, limit, offset, langs):
62 cur = self.db_cnx.execute("""
63 select uri from posts
64 order by temperature * exp( -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 ) )
65 desc limit :limit offset :offset
66 """, dict(limit=limit, offset=offset))
67 return [uri for (uri,) in cur]
68
69 def serve_feed_debug(self, limit, offset, langs):
70 query = """
71 select
72 uri, temperature,
73 unixepoch('now') - create_ts as age_seconds,
74 exp(
75 -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 )
76 ) as decay,
77 temperature * exp(
78 -1 * ( ( unixepoch('now') - create_ts ) / 1800.0 )
79 ) as score
80 from posts
81 order by score desc
82 limit :limit offset :offset
83 """
84 bindings = dict(limit=limit, offset=offset)
85 return apsw.ext.format_query_table(
86 self.db_cnx, query, bindings,
87 string_sanitize=2, text_width=9999, use_unicode=True
88 )