commit f7c819fe2761794fc2609ae2f2c88362f1ed9989 · edavis.dev/bsky-tools

+2 -2

feed_manager.py

···

       22
       22
        
               if feed is not None:

     

       23
       23
        
                   return feed.serve_feed_debug(limit, offset, langs)

     

       24
       24
        
       

     

       25
       25
       -
           def run_tasks_minute(self):

     

       25
       25
       +
           def commit_changes(self):

     

       26
       26
        
               for feed in self.feeds.values():

     

       27
       27
       -
                   feed.run_tasks_minute()

     

       27
       27
       +
                   feed.commit_changes()

     

       28
       28
        
       

     

       29
       29
        
       feed_manager = FeedManager()

     

       30
       30
        
       # feed_manager.register(PopularFeed)

+4 -7

feedgen.py

···

       1
       1
        
       #!/usr/bin/env python3

     

       2
       2
        
       

     

       3
       3
        
       import asyncio

     

       4
       4
       -
       from datetime import datetime, timezone

     

       5
       4
        
       from io import BytesIO

     

       6
       5
        
       import logging

     

       7
       6
        
       

     
···

       56
       55
        
       

     

       57
       56
        
       async def main():

     

       58
       57
        
           firehose_manager = FirehoseManager()

     

       58
       58
       +
           event_count = 0

     

       59
       59
        
       

     

       60
       60
       -
           current_minute = None

     

       61
       60
        
           async for commit in firehose_events(firehose_manager):

     

       62
       61
        
               feed_manager.process_commit(commit)

     

       63
       63
       -
       

     

       64
       64
       -
               now = datetime.now(timezone.utc)

     

       65
       65
       -
               if now.minute != current_minute:

     

       66
       66
       -
                   current_minute = now.minute

     

       67
       67
       -
                   feed_manager.run_tasks_minute()

     

       62
       62
       +
               event_count += 1

     

       63
       63
       +
               if event_count % 500 == 0:

     

       64
       64
       +
                   feed_manager.commit_changes()

     

       68
       65
        
                   firehose_manager.set_sequence_number(commit['seq'])

     

       69
       66
        
       

     

       70
       67
        
       if __name__ == '__main__':

+2 -2

feeds/__init__.py

···

       7
       7
        
           def serve_feed(self, limit, offset, langs):

     

       8
       8
        
               raise NotImplementedError

     

       9
       9
        
       

     

       10
       10
       -
           def run_tasks_minute(self):

     

       11
       11
       -
               pass

     

       10
       10
       +
           def commit_changes(self):

     

       11
       11
       +
               raise NotImplementedError

     

       12
       12
        
       

     

       13
       13
        
           def parse_timestamp(self, timestamp):

     

       14
       14
        
               # https://atproto.com/specs/lexicon#datetime