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