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