Task routing

ayta/__init__.py

@@ -26,6 +26,15 @@ def create_app(test_config=None):
     config['CELERY']['beat_schedule'] = {}
     config['CELERY']['beat_schedule']['Renew WebSub endpoints'] = {'task': 'ayta.tasks.websub_renew_expiring', 'schedule': 4000}
     config['CELERY']['beat_schedule']['Process WebSub data'] = {'task': 'ayta.tasks.websub_process_data', 'schedule': 100}
+    config['CELERY']['beat_schedule']['Queue up new videos in static channel playlists'] = {'task': 'ayta.tasks.playlist_to_queue', 'schedule': 50000}
+    
+    # Celery task routing
+    # Tasks not defined in this configuration will be routed to the default queue "celery"
+    
+    config['CELERY']['task_routes'] = {
+        'ayta.tasks.video_download': {'queue': 'download'},
+        'ayta.tasks.video_queue': {'queue': 'download'}
+    }
     
     app = Flask(__name__)
     app.config.from_mapping(config)

ayta/tasks.py

@@ -167,12 +167,21 @@ def websub_renew_expiring(hours=6):
 @shared_task()
 def playlist_to_queue():
     from .nosql import get_nosql
+    import random
+    from datetime import datetime, timedelta
     
-    channels = get_nosql().list_all_channels(active=True)
+    pivot = datetime.utcnow() - timedelta(days=3)  # calculates 3 days before now
+    channels = list(get_nosql().list_all_channels(active=True))
+    random.shuffle(channels)  # randomize channelId order because otherwise the queue will follow the channel order as well
     
     for channel in channels:
         info = get_nosql().get_channel_info(channel)
         
+        # if last_run not set or last_run is older than the pivot (indicating it has not been updated)
+        if not info.get('last_run') or info.get('last_run') < pivot:
+            # skip channel
+            continue
+        
         for item in info['playlist']['entries']:
             videoId = item['id']
             get_nosql().queue_insertQueue(videoId, 'Playlist mirroring')