···
-
class RapidFireFeed(BaseFeed):
-
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/rapidfire'
-
self.db_cnx = apsw.Connection('db/rapidfire.db')
self.db_cnx.pragma('journal_mode', 'WAL')
self.db_cnx.pragma('wal_autocheckpoint', '0')
-
create table if not exists posts (uri text, create_ts timestamp, lang text);
-
create index if not exists create_ts_idx on posts(create_ts);
-
self.logger = logging.getLogger('feeds.rapidfire')
def process_commit(self, commit):
···
-
grapheme.length(record['text']) <= MAX_TEXT_LENGTH,
-
record.get('reply') is None,
-
record.get('embed') is None,
-
record.get('facets') is None
-
post_uri = f'at://{repo}/{path}'
-
ts = self.safe_timestamp(record['createdAt']).timestamp()
-
self.transaction_begin(self.db_cnx)
-
langs = record.get('langs') or ['']
-
'insert into posts (uri, create_ts, lang) values (:uri, :ts, :lang)',
-
dict(uri=post_uri, ts=ts, lang=lang)
-
def delete_old_posts(self):
-
"delete from posts where create_ts < unixepoch('now', '-15 minutes')"
def commit_changes(self):
self.logger.debug('committing changes')
-
self.delete_old_posts()
self.transaction_commit(self.db_cnx)
self.wal_checkpoint(self.db_cnx, 'RESTART')
def serve_feed(self, limit, offset, langs):
-
cur = self.db_cnx.execute(
-
"select uri from posts order by create_ts desc limit :limit offset :offset",
-
dict(limit=limit, offset=offset)
-
return [uri for (uri,) in cur]
-
lang_values = list(langs.values())
-
lang_selects = ['select uri, create_ts from posts where lang = ?'] * len(lang_values)
-
lang_clause = ' union '.join(lang_selects)
-
cur = self.db_cnx.execute(
-
lang_clause + ' order by create_ts desc limit ? offset ?',
-
[*lang_values, limit, offset]
-
return [uri for (uri, create_ts) in cur]
def serve_feed_debug(self, limit, offset, langs):
-
select *, unixepoch('now') - create_ts as age_seconds
-
order by create_ts desc
limit :limit offset :offset
bindings = dict(limit=limit, offset=offset)
···
+
class BattleFeed(BaseFeed):
+
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/battle'
+
self.db_cnx = apsw.Connection('db/battle.db')
self.db_cnx.pragma('journal_mode', 'WAL')
self.db_cnx.pragma('wal_autocheckpoint', '0')
+
create table if not exists posts (
+
grapheme_length integer unique,
+
self.logger = logging.getLogger('feeds.battle')
def process_commit(self, commit):
···
+
post_uri = f'at://{repo}/{path}'
+
l = grapheme.length(record['text'])
+
ts = self.safe_timestamp(record['createdAt']).timestamp()
+
self.transaction_begin(self.db_cnx)
+
langs = record.get('langs') or ['']
+
self.db_cnx.execute("""
+
insert into posts(uri, grapheme_length, create_ts, lang)
+
values(:uri, :length, :ts, :lang)
+
on conflict(grapheme_length) do update set uri = :uri
+
""", dict(uri=post_uri, length=l, ts=ts, lang=lang))
def commit_changes(self):
self.logger.debug('committing changes')
self.transaction_commit(self.db_cnx)
self.wal_checkpoint(self.db_cnx, 'RESTART')
def serve_feed(self, limit, offset, langs):
+
# cur = self.db_cnx.execute(
+
# "select uri from posts order by grapheme_length asc limit :limit offset :offset",
+
# dict(limit=limit, offset=offset)
+
# return [uri for (uri,) in cur]
+
# lang_values = list(langs.values())
+
# lang_selects = ['select uri, grapheme_length from posts where lang = ?'] * len(lang_values)
+
# lang_clause = ' union '.join(lang_selects)
+
# cur = self.db_cnx.execute(
+
# lang_clause + ' order by grapheme_length asc limit ? offset ?',
+
# [*lang_values, limit, offset]
+
# return [uri for (uri, create_ts) in cur]
+
cur = self.db_cnx.execute("""
+
order by grapheme_length asc
+
limit :limit offset :offset
+
""", dict(limit=limit, offset=offset))
+
return [uri for (uri,) in cur]
def serve_feed_debug(self, limit, offset, langs):
+
order by grapheme_length asc
limit :limit offset :offset
bindings = dict(limit=limit, offset=offset)
edavis.dev