Small concurrency and logging fix

.gitea/workflows/frontend-release.yaml

@@ -1,4 +1,4 @@
-name: Generate release
+name: Generate docker image
 
 on:
   release:
@@ -22,13 +22,4 @@ jobs:
       uses: docker/build-push-action@v5
       with:
         push: true
-        tags: git.ventilaar.nl/ventilaar/ayta:latest
+        tags: git.ventilaar.nl/ventilaar/ayta:latest
-        
-    - name: Update worker server
-      uses: appleboy/ssh-action@v1.0.3
-      with:
-        host: 192.168.66.109
-        username: root
-        key: ${{ secrets.SERVER_KEY }}
-        port: 22
-        script: /root/update_worker.sh

.gitea/workflows/workers-tasks.yaml

@@ -0,0 +1,18 @@
+name: Update worker server
+
+on:
+  release:
+    types: [published]
+
+jobs:
+  build-and-publish:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Update worker server
+      uses: appleboy/ssh-action@v1.0.3
+      with:
+        host: 192.168.66.109
+        username: root
+        key: ${{ secrets.SERVER_KEY }}
+        port: 22
+        script: /root/update_worker.sh

Dockerfile

@@ -1,7 +1,7 @@
-FROM python:3-alpine
+FROM python:3.12-alpine
 WORKDIR /app
 COPY requirements.txt /app
 RUN pip install --no-cache-dir -r requirements.txt
 COPY . /app
 EXPOSE 8000
-CMD ["gunicorn", "--bind", "0.0.0.0:8000", "ayta:create_app()"]
+CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "1", "ayta:create_app()"]

README.md

@@ -6,7 +6,7 @@ current cronjob yt-dlp archive service.
 Partially inspired by [hobune](https://github.com/rebane2001/hobune). While that project is amazingby it's own, it's just not scaleable.
 
 ## The idea
-Having over 250k videos, scaling the current cronjob yt-dlp archive task is just really hard. Filetypes change, things get partially downloaded and such.
+Having over 350k videos, scaling the current cronjob yt-dlp archive task is just really hard. Filetypes change, things get partially downloaded and such.
 Partially yt-dlp is to blame because it's a package that needs to change all the time. But with this some changes are not accounted for.
 yt-dlp will still do the downloads. But a flask frontend will be developed to make all downloaded videos easily indexable.
 For it to be quick (unlike hobune) a database has to be implemented. This could get solved by a static site generator type of software, but that is not my choice.
@@ -54,7 +54,7 @@ Extra functionality for further development of features.
 Mainly focused on retiring the cronjob based scripts and moving it to celery based tasks
 - [ ] manage videos by ID's instead of per channel basis
 - [ ] download videos from queue
-- [ ] Manage websub callbacks
+- [x] Manage websub callbacks
 
 ### Stage 4
 Mongodb finally has it's limitations.

ayta/__init__.py

@@ -9,15 +9,24 @@ def create_app(test_config=None):
     config = {'MONGO_CONNECTION': os.environ.get('AYTA_MONGOCONNECTION', 'mongodb://root:example@192.168.66.140:27017'),
               'OIDC_PROVIDER': os.environ.get('AYTA_OIDC_PROVIDER', 'https://auth.ventilaar.nl'),
               'OIDC_ID': os.environ.get('AYTA_OIDC_ID', 'ayta'),
-              'CACHE_TYPE': os.environ.get('AYTA_CACHETYPE', 'SimpleCache'),
               'CACHE_DEFAULT_TIMEOUT': int(os.environ.get('AYTA_CACHETIMEOUT', 6)),
-              'SECRET_KEY': os.environ.get('AYTA_SECRETKEY', secrets.token_hex(32)),
               'DEBUG': bool(os.environ.get('AYTA_DEBUG', False)),
               'DOMAIN': os.environ.get('AYTA_DOMAIN', 'https://testing.mashallah.nl'),
-              'CELERY': dict(broker_url=str(os.environ.get('AYTA_CELERYBROKER', 'amqp://guest:guest@192.168.66.140:5672/')), 
+              'CELERY': {'broker_url': str(os.environ.get('AYTA_CELERYBROKER', 'amqp://guest:guest@192.168.66.140:5672/'))}
-                             task_ignore_result=True,)
              }
     
+    # Static Flask configuration options
+    
+    config['CELERY']['task_ignore_result'] = True
+    config['CACHE_TYPE'] = 'SimpleCache'
+    config['SECRET_KEY'] = secrets.token_bytes(32)
+    
+    # Celery Periodic tasks
+    
+    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}
+    
     app = Flask(__name__)
     app.config.from_mapping(config)
     
@@ -32,7 +41,8 @@ def create_app(test_config=None):
     app.jinja_env.filters['pretty_time'] = filters.pretty_time
     app.jinja_env.filters['current_time'] = filters.current_time
     app.jinja_env.filters['epoch_time'] = filters.epoch_time
-
+    app.jinja_env.filters['epoch_date'] = filters.epoch_date
+    
     from .blueprints import watch
     from .blueprints import index
     from .blueprints import admin

ayta/blueprints/admin.py

@@ -1,8 +1,8 @@
-from flask import Blueprint, render_template, request, redirect, url_for, flash
+from flask import Blueprint, render_template, request, redirect, url_for, flash, current_app
 from ..nosql import get_nosql
 from ..dlp import checkChannelId, getChannelInfo
 from ..decorators import login_required
-from ..tasks import subscribe_websub_callback, unsubscribe_websub_callback
+from ..tasks import test_sleep, websub_subscribe_callback, websub_unsubscribe_callback, video_download, video_queue
 from datetime import datetime
 from secrets import token_urlsafe
 
@@ -71,15 +71,15 @@ def channel(channelId):
         value = request.form.get('value', None)
         
         if task == 'subscribe-websub':
-            task = subscribe_websub_callback.delay(channelId)
+            task = websub_subscribe_callback.delay(channelId)
             flash(f"Started task {task.id}")
             return redirect(url_for('admin.channel', channelId=channelId))
         
         if task == 'update-value':
-            if key == 'active':
+            if key in ['active', 'websub']:
                 value = True if value else False
 
-            if key == 'added_date':
+            if key in ['added_date']:
                 value = datetime.strptime(value, '%Y-%m-%d')
               
             get_nosql().update_channel_key(channelId, key, value)
@@ -109,29 +109,41 @@ def run(runId):
 @bp.route('/websub', methods=['GET', 'POST'])
 @login_required
 def websub():
+    render = {}
+    
     if request.method == 'POST':
         task = request.form.get('task', None)
         value = request.form.get('value', None)
         
         if task == 'unsubscribe':
-            channelId = get_nosql().websub_getCallback(value).get('channel')
+            task = websub_unsubscribe_callback.delay(value)
-            
-            task = unsubscribe_websub_callback.delay(value, channelId)
             
             flash(f"Started task {task.id}")
             return redirect(url_for('admin.websub'))
-            
         elif task == 'clean-retired':
             get_nosql().websub_cleanRetired()
             return redirect(url_for('admin.websub'))
+        elif task == 'unsubscribe-callbacks':
+            for callbackId in get_nosql().websub_getCallbacks():
+                websub_unsubscribe_callback.delay(callbackId)
+            flash(f"Started unsubscribe tasks for all callbacks")
+            return redirect(url_for('admin.websub'))
+        elif task == 'subscribe-channels':
+            for channelId in get_nosql().list_all_channels(websub=True):
+                websub_subscribe_callback.delay(channelId)
+            flash(f'Started subscribe tasks for activated channels')
+            return redirect(url_for('admin.websub'))
+            
     
     callbackIds = get_nosql().websub_getCallbacks()
     callbacks = {}
     
+    render['stats'] = get_nosql().websub_statistics()
+    
     for callbackId in callbackIds:
         callbacks[callbackId] = get_nosql().websub_getCallback(callbackId)
     
-    return render_template('admin/websub.html', callbacks=callbacks)
+    return render_template('admin/websub.html', callbacks=callbacks, render=render)
 
 @bp.route('/reports', methods=['GET', 'POST'])
 @login_required
@@ -149,9 +161,9 @@ def reports():
    
     return render_template('admin/reports.html', reports=reports)
 
-@bp.route('/posters', methods=['GET', 'POST'])
+@bp.route('/queue', methods=['GET', 'POST'])
 @login_required
-def posters():
+def queue():
     if request.method == 'POST':
         task = request.form.get('task', None)
         value = request.form.get('value', None)
@@ -162,34 +174,51 @@ def posters():
                 flash('Description must be at least 8 characters long')
             
             if value and len(value) >= 12:
-                get_nosql().poster_newEndpoint(value, description)
+                get_nosql().queue_newEndpoint(value, description)
                 flash(f'Created endpoint ID: {value}')
             else:
                 value = token_urlsafe(16)
-                get_nosql().poster_newEndpoint(value, description)
+                get_nosql().queue_newEndpoint(value, description)
                 flash(f'Created endpoint ID: {value}')
+                
         elif task == 'retire':
-            get_nosql().poster_retireEndpoint(value)
+            get_nosql().queue_retireEndpoint(value)
             flash(f'Endpoint retired: {value}')
             
         elif task == 'clean-retired':
-            get_nosql().poster_cleanRetired()
+            get_nosql().queue_cleanRetired()
             flash(f'Cleaned retired endpoints')
             
         elif task == 'manual-queue':
-            get_nosql().poster_insertQueue('manual', value)
+            if not get_nosql().check_exists(value):
-            flash(f'Added to queue: {value}')
+                direct = request.form.get('direct', None)
+                if direct:
+                    task = video_download.delay(value)
+                    flash(f"Started task {task.id}")
+                else:
+                    get_nosql().queue_insertQueue(value, 'webui')
+                    flash(f'Added to queue: {value}')
+            else:
+                flash(f'This video ID already exists in the archive: {value}')
             
         elif task == 'delete-queue':
-            get_nosql().poster_deleteQueue(value)
+            get_nosql().queue_deleteQueue(value)
             flash(f'Deleted from queue: {value}')
+            
+        elif task == 'empty-queue':
+            get_nosql().queue_emptyQueue()
+            flash(f'Queue has been emptied')
+            
+        elif task == 'queue-run-once':
+            task =  video_queue.delay()
+            flash(f'Task has been started on the oldest queued item: {task.id}')
         
-        return redirect(url_for('admin.posters'))
+        return redirect(url_for('admin.queue'))
         
-    endpoints = get_nosql().poster_getEndpoints()
+    endpoints = get_nosql().queue_getEndpoints()
-    queue = get_nosql().poster_getQueue()
+    queue = get_nosql().queue_getQueue()
    
-    return render_template('admin/posters.html', endpoints=endpoints, queue=queue)
+    return render_template('admin/queue.html', endpoints=endpoints, queue=queue)
 
 @bp.route('/users', methods=['GET', 'POST'])
 @login_required
@@ -217,4 +246,17 @@ def users():
 
     users = get_nosql().list_all_users()
    
-    return render_template('admin/users.html', users=users)
+    return render_template('admin/users.html', users=users)
+    
+@bp.route('/workers', methods=['GET', 'POST'])
+#@login_required
+def workers():
+    if request.method == 'POST':
+        task = request.form.get('task', None)
+        if task == 'test-sleep':
+            test_sleep.delay()
+
+    celery = current_app.extensions.get('celery')
+    
+    tasks = celery.control.inspect().active()
+    return render_template('admin/workers.html', tasks=tasks)

ayta/blueprints/api.py

@@ -39,10 +39,10 @@ def websub(cap):
     
     return abort(404)
     
-@bp.route('/poster/<cap>', methods=['POST'])
+@bp.route('/queue/<cap>', methods=['POST'])
-def poster(cap):
+def queue(cap):
     # if endpoint does not exist
-    if not get_nosql().poster_isActive(cap):
+    if not get_nosql().queue_isActive(cap):
         return abort(404)
     
     videoId = request.form.get('v')
@@ -60,7 +60,7 @@ def poster(cap):
         return abort(409)
     
     # try to insert
-    if get_nosql().poster_insertQueue(cap, videoId):
+    if get_nosql().queue_insertQueue(videoId, cap):
         return '', 202
     else:
         return abort(409)

ayta/blueprints/watch.py

@@ -36,4 +36,9 @@ def base():
 
     render['info'] = get_nosql().get_video_info(vGet)
     render['params'] = request.args.get('v')
+    
+    if render['info'].get('_status') != 'available':
+        flash(render['info'].get('_status_description', 'Video unavailable because of technical errors. Come back later.'))
+        return redirect(url_for('index.base'))
+        
     return render_template('watch/index.html', render=render)

ayta/filters.py

@@ -16,9 +16,15 @@ def pretty_time(time):
         except:
             return time  # return given time
 
-def epoch_time(time):
+def epoch_date(epoch):
     try:
-        return datetime.fromtimestamp(time).strftime('%d %b %Y')
+        return datetime.fromtimestamp(epoch).strftime('%d %b %Y')
+    except:
+        return None
+        
+def epoch_time(epoch):
+    try:
+        return datetime.fromtimestamp(epoch).strftime('%d %b %Y %H:%M:%S')
     except:
         return None
 

ayta/nosql.py

@@ -36,13 +36,13 @@ class Mango:
         self.db = self.client['ayta']
         self.channels = self.db['channels']
         self.info_json = self.db['info_json']
-        self.posters_queue = self.db['posters_queue']
+        self.download_queue = self.db['download_queue']
         self.run_log = self.db['run_log']
         self.channel_log = self.db['channel_log']
         self.websub_callbacks = self.db['websub_callbacks']
         self.websub_data = self.db['websub_data']
         self.reports = self.db['reports']
-        self.posters_endpoints = self.db['posters_endpoints']
+        self.queue_endpoints = self.db['queue_endpoints']
         self.users = self.db['users']
             
         self.ensure_indexes()
@@ -98,7 +98,7 @@ class Mango:
         
         stats['videos'] = self.info_json.count_documents({})
         stats['channels'] = self.channels.count_documents({})
-        stats['queue'] = self.posters_queue.count_documents({})
+        stats['queue'] = self.download_queue.count_documents({})
         
         return stats
             
@@ -142,12 +142,14 @@ class Mango:
     #            channel operations          #
     ##########################################
     
-    def list_all_channels(self, active=False):
+    def list_all_channels(self, active=False, websub=False):
         """ Returns a SET of YouTube channel ID's; Depending on given positional BOOL only active channels or everything"""
         search_terms = {}
         
         if active:
             search_terms['active'] = True
+        elif websub:
+            search_terms['websub'] = True
         
         channels = []
         for channel in self.channels.find(search_terms, {'id': 1}):
@@ -169,9 +171,6 @@ class Mango:
     def get_channel_info(self, channelId):
         return self.channels.find_one({'id': channelId})
         
-    def update_channel_state(self, channelId, state):
-        self.channels.update_one({'id': channelId}, {"$set": {"active": bool(state)}})
-        return True
         
     def update_channel_key(self, channelId, key, value):
         self.channels.update_one({'id': channelId}, {"$set": {key: value}})
@@ -196,16 +195,19 @@ class Mango:
     def get_orphaned_videos(self):
         """ Returns a SET of YouTube video ID's which have info_jsons in the collection but no permanent channel is defined. SLOW OPERATION """
         # Ok lemme explain. Perform inner join from channel collection on channel_id key. match only the fields which are empty. return video id
-        pipeline = [{'$lookup': {'from': 'channels', 'localField': 'channel_id', 'foreignField': 'id', 'as': 'channel'}}, {'$match': {'channel': {'$size': 0}}},{'$project': {'id': 1}}]
+        pipeline = [{'$match': {'_status': 'available'}},
-
+                    {'$lookup': {'from': 'channels', 'localField': 'channel_id', 'foreignField': 'id', 'as': 'channel'}},
+                    {'$match': {'channel': {'$size': 0}}},{'$project': {'id': 1}},
+                    {'$project': {'id': 1}}]
+        
         results = self.info_json.aggregate(pipeline)
         ids = [result['id'] for result in results]
-        
+
         return tuple(ids)
     
     def get_recent_videos(self, count=99):
         """ Returns a SET of YouTube video ID's which have been added last to the info_json collection """
-        result = self.info_json.find({}, {'_id': 0, 'id': 1}, sort=[('_id', pymongo.DESCENDING)]).limit(count)
+        result = self.info_json.find({'_status': 'available'}, {'_id': 0, 'id': 1}, sort=[('_id', pymongo.DESCENDING)]).limit(count)
 
         ids = []
         
@@ -294,22 +296,22 @@ class Mango:
         status = status.get('status')
         
         if status in ['requesting']:
-            self.websub_callbacks.update_one({'id': callbackId}, {'$set': {'status': 'active', 'activation_time': current_time(object=True), 'lease': lease}})
+            self.websub_callbacks.update_one({'id': callbackId}, {'$set': {'status': 'active', 'activation_time': current_time(object=True), 'lease': int(lease)}})
             return True
         
         return False
 
-    def websub_existsCallback(self, callbackId):
+    def websub_existsCallback(self, callbackId, channel=False):
-        status = self.websub_callbacks.find_one({'id': callbackId}, {'status': 1})
+        if channel:
+            query = {'channel': callbackId, 'status': {'$in': ['requesting', 'active', 'retiring']}}
+        else:
+            query = {'id': callbackId, 'status': {'$in': ['requesting', 'active', 'retiring']}}
+            
+        status = self.websub_callbacks.find_one(query, {'id': 1, 'status': 1})
 
-        if not status:
+        if status:
-            return False
+            return status.get('id')
-            
-        status = status.get('status')
         
-        if status in ['requesting', 'active', 'retiring']:
-            return True
-            
         return False
         
     def websub_retiringCallback(self, callbackId):
@@ -330,7 +332,7 @@ class Mango:
     def websub_getCallback(self, callbackId):
         return self.websub_callbacks.find_one({'id': callbackId})
         
-    def websub_getCallbacks(self, channelId=''):
+    def websub_getCallbacks(self, channelId=None):
         callbacks = []
         
         if channelId:
@@ -338,7 +340,6 @@ class Mango:
         else:
             filter = {}
         
-        
         for callback in self.websub_callbacks.find(filter, {'id': 1}):
             callbacks.append(callback['id'])
             
@@ -360,65 +361,93 @@ class Mango:
     def websub_deletePostProcessing(self, _id):
         self.websub_data.delete_one({'_id': _id})
         
-    def websub_cleanRetired(self, days=3):
+    def websub_cleanRetired(self, days=1):
         days = self.datetime.utcnow() - self.timedelta(days=days)
 
         self.websub_callbacks.delete_many({'status': 'retired', 'retired_time': {'$lt': days}})
         self.websub_callbacks.delete_many({'status': 'requesting', 'requesting_time': {'$lt': days}})
         
         return True
+        
+    def websub_statistics(self):
+        stats = {}
+        
+        stats['unprocessed_data'] = self.websub_data.count_documents({'state': 'unprocessed'})
+        stats['active_callbacks'] = self.websub_callbacks.count_documents({'status': 'active'})
+        
+        return stats
 
     ##########################################
-    #           POSTER FUNCTIONS             #
+    #            QUEUE FUNCTIONS             #
     ##########################################
     
-    def poster_newEndpoint(self, endpointId, description=''):
+    def queue_newEndpoint(self, endpointId, description=''):
-        self.posters_endpoints.insert_one({'id': endpointId, 'description': description, 'status': 'active', 'created_time': current_time(object=True)})
+        self.queue_endpoints.insert_one({'id': endpointId, 'description': description, 'status': 'active', 'created_time': current_time(object=True)})
         return endpointId
-        
+    
-    def poster_insertQueue(self, endpointId, videoId):
+    def queue_retireEndpoint(self, endpointId):
-        # if no document exists
+        return self.queue_endpoints.update_one({'id': endpointId}, {'$set': {'status': 'retired', 'retired_time': current_time(object=True)}})
-        if not self.posters_queue.count_documents({'id': videoId}) >= 1:
+    
-            self.posters_queue.insert_one({'id': videoId, 'endpoint': endpointId, 'created_time': current_time(object=True), 'status': 'queued'}).inserted_id
+    def queue_isActive(self, endpointId):
-            return True
+        status = self.queue_endpoints.find_one({'id': endpointId}, {'status': 1})
-            
-        # key already in queue
-        return False
-        
-    def poster_deleteQueue(self, videoId):
-        if self.posters_queue.delete_one({'id': videoId}):
-            return True
-        return False
-
-    def poster_retireEndpoint(self, endpointId):
-        return self.posters_endpoints.update_one({'id': endpointId}, {'$set': {'status': 'retired', 'retired_time': current_time(object=True)}})
-        
-    def poster_isActive(self, endpointId):
-        status = self.posters_endpoints.find_one({'id': endpointId}, {'status': 1})
 
         if not status:
             return False
-            
+        elif status.get('status') == 'active':
-        status = status.get('status')
-        
-        if status == 'active':
             return True
+        else:
+            return False
             
-        return False
+    def queue_getEndpoints(self):
+        return self.queue_endpoints.find({})
         
-    def poster_getEndpoints(self):
+    def queue_cleanRetired(self, days=3):
-        return self.posters_endpoints.find({})
-        
-    def poster_getQueue(self):
-        return self.posters_queue.find({})
-        
-    def poster_cleanRetired(self, days=3):
         days = self.datetime.utcnow() - self.timedelta(days=days)
 
-        self.posters_endpoints.delete_many({'status': 'retired', 'retired_time': {'$lt': days}})
+        self.queue_endpoints.delete_many({'status': 'retired', 'retired_time': {'$lt': days}})
         
         return True
+        
+    ##########################################
+        
+    def queue_insertQueue(self, videoId, endpointId=None):
+        # if already queued
+        if self.download_queue.count_documents({'id': videoId}) >= 1:
+            return False
+        
+        # if already in archive
+        if self.check_exists(videoId):
+            return False
+        
+        # add to queue
+        self.download_queue.insert_one({'id': videoId, 'endpoint': endpointId, 'created_time': current_time(object=True), 'status': 'queued'}).inserted_id
+        return True
+        
+    def queue_deleteQueue(self, videoId):
+        if self.download_queue.delete_one({'id': videoId}):
+            return True
+        return False
 
+    def queue_getQueue(self):
+        return self.download_queue.find({})
+        
+    def queue_emptyQueue(self):
+        return self.download_queue.delete_many({})
+        
+    def queue_setFailed(self, videoId):
+        return self.download_queue.update_one({'id': videoId}, {'$set': {'status': 'failed'}})
+        
+    def queue_getNext(self):
+        """ Returns a LIST of queue parameters. Function first checks if ID exists, if so deletes and then checks the next queued until queue is empty (None) or queued id does not exist yet."""
+        while True:
+            queueItem = self.download_queue.find_one({'status': 'queued'})
+            if not queueItem:
+                return None
+            elif self.check_exists(queueItem['id']):
+                self.queue_deleteQueue(queueItem['id'])
+            self.download_queue.update_one({'id': queueItem['id']}, {'$set': {'status': 'working'}})
+            return queueItem
+        
 ##########################################
 #           HELPER FUNCTIONS             #
 ##########################################
@@ -442,7 +471,4 @@ def clean_info_json(originalInfo, format='dict'):
         return json.dumps(originalInfo)
     else:
         print('The requested output format is not supported!')
-    
+    
-   
-if __name__ == '__main__':
-    mango = Mango('mongodb://root:example@192.168.66.140:27017')

ayta/oidc.py

@@ -1,4 +1,10 @@
 class OIDC():
+    """
+    This function class is nothing more than a nonce and state store for security in the authentication mechanism.
+    Additionally this class provides the function to generate redirect url's and check bearer tokens on their validity as well as caching jwt signing keys.
+    Fairly barebones and should be 100% secure. (famous last words)
+    This is made for form posted JWT's. While not the most secure it is the most easy way to implement. Moving on to a code based solution might be preferred in the future.
+    """
     def __init__(self, app=None):
         self.states = {}
         self.nonces = {}
@@ -15,27 +21,34 @@ class OIDC():
         self.client_id = config['OIDC_ID']
         self.provider = config['OIDC_PROVIDER']
         self.domain = config['DOMAIN']
+        self.window = 120  # the time window to allow states and nonces in seconds
         
+        # Authentication provider url must be HTTPS and end on a TLD
         if self.provider[:8] != 'https://' or self.provider[-1] == '/':
             print('Incorrect OIDC provider URI', flush=True)
             exit()
-            
+        
+        # Get the provider configuration endpoints
         configuration = requests.get(f'{self.provider}/.well-known/openid-configuration').json()
         
         jwks_uri = configuration.get('jwks_uri')
         self.authorize_uri = configuration.get('authorization_endpoint')
         
+        # Start the JWKS management client, it will load the keys and maintain them
         self.jwks_manager = jwt.PyJWKClient(jwks_uri)
 
-    #################################
+    #######################################################
 
     def state_maintenance(self):
         from datetime import datetime
         
-        pivot = datetime.now().timestamp() - 120
+        # Current time minus the acceptable window
+        pivot = datetime.now().timestamp() - self.window
         
+        # List with expired states
         expired_states = [state for state, timestamp in self.states.items() if timestamp <= pivot]
         
+        # Remove expired states from store
         for state in expired_states:
             del self.states[state]
 
@@ -43,30 +56,40 @@ class OIDC():
         import secrets
         from datetime import datetime
         
+        # Clean state store first
         self.state_maintenance()
         
+        # Generate token and paired timestamp
         state = secrets.token_urlsafe(8)
         timestamp = datetime.now().timestamp()
-
+        
+        # Add token to the state store
         self.states[state] = timestamp
-
+        
+        # Return the state
         return state
         
     def state_check(self, state):
+        # Clean state store first
         self.state_maintenance()
 
+        # If given state is actively stored
         if state in self.states:
+            # Delete state and return True
             del self.states[state]
             return True
-            
+        
+        # Given state is not stored
         return False
         
-    #################################
+    #######################################################
+    # Same code as above but a different store for nonces #
+    #######################################################
     
     def nonce_maintenance(self):
         from datetime import datetime
         
-        pivot = datetime.now().timestamp() - 120
+        pivot = datetime.now().timestamp() - self.window
         
         expired_nonces = [nonce for nonce, timestamp in self.nonces.items() if timestamp <= pivot]
         
@@ -95,7 +118,7 @@ class OIDC():
             
         return False
     
-    #################################
+    #######################################################
     
     def generate_redirect(self):
         return str(f'{self.authorize_uri}'
@@ -107,21 +130,33 @@ class OIDC():
         
     def check_bearer(self, token):
         import jwt
-
+        
+        # Test given JWT
         try:
+            # Get the signed public key from the token
             signing_key = self.jwks_manager.get_signing_key_from_jwt(token).key
+            
+            # Try to decode the token, this will also check the validity in these points:
+            # 1. Token is signed by expected keys
+            # 2. Token is issued by the expected provider
+            # 3. Expected parameters are really in the token
+            # 4. Token is really intended for us
+            # 5. Token is still valid (with 5 sec margin)
             decoded = jwt.decode(token, signing_key,
                                  algorithms=jwt.algorithms.get_default_algorithms(),
                                  issuer=self.provider,
                                  require=['aud', 'client_id', 'exp', 'iat', 'iss', 'rat', 'sub'],
                                  audience=self.client_id,
                                  leeway=5)
+                                 
+        # Any exception (invalid JWT, invalid formatting etc...) must return False
         except Exception as e:
             print(e, flush=True)
             return False
 
-        # double check if given token is really requested by us
+        # Double check if given token is really requested by us by matching the nonce in the signed key
         if not self.nonce_check(decoded.get('nonce', None)):
             return False
-
+        
+        # Return the unique user identifier
         return decoded.get('sub', False)

ayta/tasks.py

@@ -1,12 +1,62 @@
 from celery import shared_task
 from flask import current_app
 
+##########################################
+#              CELERY TASKS              #
+##########################################
+
 @shared_task()
-def subscribe_websub_callback(channelId):
+def test_sleep(time=60):
+    from time import sleep
+    sleep(time)
+    return True
+    
+@shared_task()
+def video_download(videoId):
+    """ 
+    I do not want to deal with the quirks of native yt-dlp in python, hence the subprocess.
+    """
+    import subprocess
+    
+    process = subprocess.run(['/usr/local/bin/yt-dlp', '--config-location', '/var/www/archive.ventilaar.net/goodstuff/config_video.conf', '--', f'https://www.youtube.com/watch?v={videoId}'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    
+    if process.returncode != 0:
+        return False
+    return True
+    
+@shared_task()
+def video_queue():
+    """ 
+    Gets the oldest video ID from the queue and runs video_download() on it.
+    """
+    from .nosql import get_nosql
+    
+    videoId = get_nosql().queue_getNext()
+    
+    if videoId:
+        videoId = videoId['id']
+    else:
+        return None
+    
+    if video_download(videoId):
+        get_nosql().queue_deleteQueue(videoId)
+        return True
+    else:
+        get_nosql().queue_setFailed(videoId)
+        return False
+
+@shared_task()
+def websub_subscribe_callback(channelId):
     import requests
     from .nosql import get_nosql
     
-    callbackId = get_nosql().websub_newCallback(channelId)
+    # check if a callback already exists for channel
+    answer = get_nosql().websub_existsCallback(channelId, channel=True)
+    
+    if not answer:
+        callbackId = get_nosql().websub_newCallback(channelId)
+    else:
+        callbackId = answer
     
     url = 'https://pubsubhubbub.appspot.com/subscribe'
     data = {
@@ -16,21 +66,30 @@ def subscribe_websub_callback(channelId):
         'hub.mode': 'subscribe',
         'hub.verify_token': '',
         'hub.secret': '',
-        'hub.lease_numbers': '86400',
+        'hub.lease_numbers': '432000',
     }
     
     get_nosql().websub_requestingCallback(callbackId)
     response = requests.post(url, data=data)
     if response.status_code == 202:
         return True
-        
+    
+    # maybe handle errors?
+    
     return False
     
 @shared_task()
-def unsubscribe_websub_callback(callbackId, channelId):
+def websub_unsubscribe_callback(callbackId):
     import requests
     from .nosql import get_nosql
     
+    answer = get_nosql().websub_existsCallback(callbackId)
+    
+    if not answer:
+        return False
+        
+    channelId = get_nosql().websub_getCallback(callbackId).get('channel')
+    
     url = 'https://pubsubhubbub.appspot.com/subscribe'
     data = {'hub.callback': f'{current_app.config["DOMAIN"]}/api/websub/{callbackId}',
             'hub.topic': f'https://www.youtube.com/xml/feeds/videos.xml?channel_id={channelId}',
@@ -44,4 +103,101 @@ def unsubscribe_websub_callback(callbackId, channelId):
     if response.status_code == 202:
         return True
         
-    return False
+    # maybe handle errors?
+        
+    return False
+    
+@shared_task()
+def websub_process_data():
+    from .nosql import get_nosql
+    
+    while True:
+        blob = get_nosql().websub_getFirstPostData()
+        if not blob:
+            break
+        
+        _id, data = blob
+        
+        parsed = do_parse_data(data)
+        if parsed:
+            state, channelId, videoId = parsed
+        
+            if state == 'added':
+                if not get_nosql().check_exists(videoId):  # if video not exists
+                   get_nosql().queue_insertQueue(videoId, 'WebSub')
+                   # note for future me
+                   # the websub notifications report ALL videos, including shorts and livestreams
+                   # so if you are going to work on individual video downloading make sure you filter them!
+
+            elif state == 'removed':
+                # we currently do not do anything with removed videos
+                # but the idea is to trigger a full channel mirror in case a creator started to mass delete videos
+                pass
+            
+        get_nosql().websub_deletePostProcessing(_id)
+
+@shared_task()
+def websub_renew_expiring(hours=6):
+    from .nosql import get_nosql
+    from datetime import datetime, timedelta
+    
+    count = 0
+
+    for callbackId in get_nosql().websub_getCallbacks():
+        data = get_nosql().websub_getCallback(callbackId)
+        
+        if data.get('status') not in ['active']:  # callback not active
+            continue
+        
+        pivot = datetime.utcnow() + timedelta(hours=hours)  # hours past now
+        expires = data.get('activation_time') + timedelta(seconds=data.get('lease'))  # callback expires at
+        
+        if pivot <= expires:  # expiration happens after n hours fron now
+            continue  # skip callback
+        
+        # expiration happens within n hours
+        websub_subscribe_callback.delay(data.get('channel'))
+        
+        # limit amount of subscribe requests to spread out the requests over time
+        # with an expiration pivot of 6h and a maximum validity of 5 days we can currently handle 3072 channels
+        count = count + 1
+        if count >= 256:
+            break
+
+##########################################
+#              TASK MODULES              #
+##########################################
+
+def do_parse_data(data):
+    import xml.etree.ElementTree as ET
+    
+    data = data.decode('utf-8')
+
+    try:
+        root = ET.fromstring(data)
+    except ET.ParseError:
+        print('Not XML')
+        return False
+
+    yt = any(child.tag.startswith('{http://www.youtube.com/xml/schemas/2015}') for child in root.iter())
+    at = any(child.tag.startswith('{http://purl.org/atompub/tombstones/1.0}') for child in root.iter())
+
+    if yt and not at:
+        # Video published
+        state = 'added'
+        ns = {'yt': 'http://www.youtube.com/xml/schemas/2015', '': 'http://www.w3.org/2005/Atom'}
+        entry = root.find('.//{http://www.w3.org/2005/Atom}entry')
+        videoId = entry.find('./yt:videoId', ns).text
+        channelId = entry.find('./yt:channelId', ns).text
+    elif not yt and at:
+        # Video hidden
+        state = 'removed'
+        ns = {'at': 'http://purl.org/atompub/tombstones/1.0', '': 'http://www.w3.org/2005/Atom'}
+        deleted_entry = root.find('.//{http://purl.org/atompub/tombstones/1.0}deleted-entry')
+        videoId = deleted_entry.attrib['ref'].split(':')[-1]
+        channelId = deleted_entry.find('./at:by/uri', ns).text.split('/')[-1]
+    else:
+        print('Unknown xml')
+        return False
+        
+    return (state, channelId, videoId)

ayta/templates/admin/channel.html

@@ -19,7 +19,7 @@
     {% for item in channelInfo %}
 	  <form method="POST">
 	    <div class="input-field">
-		  <span class="supporting-text">{{ item }}</span>
+		  <span class="supporting-text mb-2">{{ item }}</span>
 		  <input class="validate" type="text" value="{{ item }}" name="key" hidden>
 		</div>
 		  

ayta/templates/admin/index.html

@@ -11,79 +11,89 @@
 <div class="divider"></div>
 <div class="row">
   <div class="col s12">
-	<h5>Global channel options</h5>
+    <h5>Global channel options</h5>
   </div>
 </div>
 <div class="row">
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.system') }}">
+    <a href="{{ url_for('admin.system') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
           <span class="card-title">System</span>
-		  <p class="grey-text">Internal system settings</p>
+          <p class="grey-text">Internal system settings</p>
         </div>
       </div>
-	</a>
+    </a>
   </div>
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.channels') }}">
+    <a href="{{ url_for('admin.channels') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
           <span class="card-title">Channels</span>
-		  <p class="grey-text">Manage channels in the system</p>
+          <p class="grey-text">Manage channels in the system</p>
         </div>
       </div>
-	</a>
+    </a>
   </div>
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.runs') }}">
+    <a href="{{ url_for('admin.runs') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
           <span class="card-title">Archive runs</span>
-		  <p class="grey-text">Look at the cron run logs</p>
+          <p class="grey-text">Look at the cron run logs</p>
         </div>
       </div>
-	</a>
+    </a>
   </div>
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.websub') }}">
+    <a href="{{ url_for('admin.websub') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
           <span class="card-title">WebSub</span>
-		  <p class="grey-text">Edit WebSub YouTube links</p>
+          <p class="grey-text">Edit WebSub YouTube links</p>
         </div>
       </div>
-	</a>
+    </a>
   </div>
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.reports') }}">
+    <a href="{{ url_for('admin.reports') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
           <span class="card-title">Reports</span>
-		  <p class="grey-text">View user reports</p>
+          <p class="grey-text">View user reports</p>
         </div>
       </div>
-	</a>
+    </a>
   </div>
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.posters') }}">
+    <a href="{{ url_for('admin.queue') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
-          <span class="card-title">Posters</span>
+          <span class="card-title">Queue</span>
-		  <p class="grey-text">Extension posters</p>
+          <p class="grey-text">Video download queue and API access</p>
         </div>
       </div>
-	</a>
+    </a>
   </div>
   <div class="col s6 l4 m-4">
-	<a href="{{ url_for('admin.users') }}">
+    <a href="{{ url_for('admin.users') }}">
-	  <div class="card black-text">
+      <div class="card black-text">
         <div class="card-content">
           <span class="card-title">Users</span>
-		  <p class="grey-text">Authenticated users</p>
+          <p class="grey-text">Authenticated users</p>
         </div>
       </div>
-	</a>
+    </a>
+  </div>
+  <div class="col s6 l4 m-4">
+    <a href="{{ url_for('admin.workers') }}">
+      <div class="card black-text">
+        <div class="card-content">
+          <span class="card-title">Workers</span>
+          <p class="grey-text">Worker and task management</p>
+        </div>
+      </div>
+    </a>
   </div>
 </div>
 {% endblock %}

ayta/templates/admin/posters.html

@@ -1,150 +0,0 @@
-{% extends 'material_base.html' %}
-{% block title %}Posters administration page{% endblock %}
-{% block description %}Posters administration page of the AYTA system{% endblock %}
-
-{% block content %}
-<div class="row">
-  <div class="col s12 l11">
-    <h4>Posters administration page</h4>
-  </div>
-  <div class="col s12 l1 m-5">
-    <form method="POST">
-      <input title="Prunes all deleted endpoints, but keeps last 3 days" type="submit" value="clean-retired" name="task">
-    </form>
-  </div>
-</div>
-<div class="divider"></div>
-<div class="row">
-  <div class="col s12">
-	<h5>Poster options</h5>
-  </div>
-</div>
-<div class="row">
-  <div class="col s12 l4 m-4">
-	<div class="card">
-      <div class="card-content">
-        <span class="card-title">Create new endpoint</span>
-	    <form method="post">
-		  <div class="row">
-            <div class="col s12 input-field">
-              <input placeholder="Custom endpoint" name="value" type="text" class="validate" minlength="12">
-			  <span class="supporting-text">Leaving this empty will create a random secure string</span>
-	        </div>
-			<div class="col s12 input-field">
-              <input placeholder="Description" name="description" type="text" class="validate" minlength="8" maxlength="64" required>
-			  <span class="supporting-text">Description for the endpoint for better administration</span>
-	        </div>
-          <button class="btn mt-4" type="submit" name="task" value="add-endpoint">Create</button>
-		  </div>
-	    </form>
-      </div>
-    </div>
-  </div>
-  <div class="col s12 l4 m-4">
-	<div class="card">
-      <div class="card-content">
-        <span class="card-title">Queue manually</span>
-	    <form method="post">
-		  <div class="row">
-            <div class="col s12 input-field">
-              <input placeholder="Youtube video ID" name="value" type="text" class="validate" minlength="11" maxlength="11" required>
-			  <span class="supporting-text">Must be a valid Youtube video ID</span>
-	        </div>
-			<div class="col s12 mt-5 input-field">
-              <div class="switch">
-		        <label>Queue<input type="checkbox" value="direct" name="value" disabled><span class="lever"></span>Direct</label>
-				<span class="supporting-text">Queue up or start directly</span>
-		      </div>
-	        </div>
-          <button class="btn mt-4" type="submit" name="task" value="manual-queue">Queue</button>
-		  </div>
-	    </form>
-      </div>
-    </div>
-  </div>
-</div>
-<div class="divider"></div>
-<div class="row">
-  <div class="col s6 l9">
-	<h5>Registered endpoints</h5>
-  </div>
-  <div class="col s6 l3 m-4 input-field">
-	<input id="filter_query" type="text">
-    <label for="filter_query">Filter results</label>
-  </div>
-</div>
-<div class="row">
-  <div class="col s12">
-    <table class="striped highlight responsive-table">
-      <thead>
-        <tr>
-		  <th>Actions</th>
-		  <th>id</th>
-		  <th>description</th>
-		  <th>status</th>
-		  <th>created_time</th>
-		  <th>retired_time</th>
-        </tr>
-      </thead>
-      <tbody>
-	    {% for endpoint in endpoints %}
-        <tr class="filterable">
-		  <td>
-		    <form method="post">
-			  <input type="text" value="{{ endpoint.get('id') }}" name="value" hidden>
-		      <button class="btn-small waves-effect waves-light" type="submit" name="task" value="retire" title="Retire endpoint" {% if endpoint.get('status') != 'active' %}disabled{% endif %}>🗑️</button>
-			</form>
-		  </td>
-		  <td>{{ endpoint.get('id') }}</td>
-		  <td>{{ endpoint.get('description') }}</td>
-		  <td>{{ endpoint.get('status') }}</td>
-		  <td>{{ endpoint.get('created_time') }}</td>
-		  <td>{{ endpoint.get('retired_time') }}</td>
-        </tr>
-		{% endfor %}
-      </tbody>
-    </table>
-  </div>
-</div>
-<div class="divider"></div>
-<div class="row">
-  <div class="col s6 l9">
-	<h5>Queued ID's</h5>
-  </div>
-  <div class="col s6 l3 m-4 input-field">
-	<input id="filter_query" type="text">
-    <label for="filter_query">Filter results</label>
-  </div>
-</div>
-<div class="row">
-  <div class="col s12">
-    <table class="striped highlight responsive-table">
-      <thead>
-        <tr>
-		  <th>Actions</th>
-		  <th>id</th>
-		  <th>endpoint</th>
-		  <th>status</th>
-		  <th>created_time</th>
-        </tr>
-      </thead>
-      <tbody>
-	    {% for id in queue %}
-        <tr class="filterable">
-		  <td>
-		    <form method="post">
-			  <input type="text" value="{{ id.get('id') }}" name="value" hidden>
-		      <button class="btn-small waves-effect waves-light" type="submit" name="task" value="delete-queue" title="Delete from queue" {% if id.get('status') != 'queued' %}disabled{% endif %}>🗑️</button>
-			</form>
-		  </td>
-		  <td>{{ id.get('id') }}</td>
-		  <td>{{ id.get('endpoint') }}</td>
-		  <td>{{ id.get('status') }}</td>
-		  <td>{{ id.get('created_time') }}</td>
-        </tr>
-		{% endfor %}
-      </tbody>
-    </table>
-  </div>
-</div>
-{% endblock %}

ayta/templates/admin/queue.html

@@ -0,0 +1,172 @@
+{% extends 'material_base.html' %}
+{% block title %}Queue administration page{% endblock %}
+{% block description %}Queue administration page of the AYTA system{% endblock %}
+
+{% block content %}
+<div class="row">
+  <div class="col s12">
+    <h4>Queue administration page</h4>
+  </div>
+</div>
+<div class="divider"></div>
+<div class="row">
+  <div class="col s12">
+    <h5>Options</h5>
+  </div>
+</div>
+<div class="row">
+  <div class="col s12 l4 m-4">
+    <div class="card">
+      <div class="card-content">
+        <span class="card-title">Direct actions</span>
+        <form class="mt-4" method="post" onsubmit="return confirm('Are you sure?');">
+          <button class="btn mb-2 red" type="submit" name="task" value="empty-queue">Empty Queue</button>
+          <br>
+          <span class="supporting-text">Removes all queued ids</span>
+        </form>
+        <form class="mt-4" method="post" onsubmit="return confirm('Are you sure?');">
+          <button class="btn mb-2" type="submit" name="task" value="clean-retired">Clean retired</button>
+          <br>
+          <span class="supporting-text">Prunes all deactivated endpoints, but keeps last 3 days</span>
+        </form>
+        <form class="mt-4" method="post" onsubmit="return confirm('Are you sure?');">
+          <button class="btn mb-2 green" type="submit" name="task" value="queue-run-once">Download oldest queued</button>
+          <br>
+          <span class="supporting-text">Will download the oldest queued video ID</span>
+        </form>
+      </div>
+    </div>
+  </div>
+  <div class="col s12 l4 m-4">
+    <div class="card">
+      <div class="card-content">
+        <span class="card-title">Create new endpoint</span>
+        <form method="post">
+          <div class="row">
+            <div class="col s12 input-field">
+              <input placeholder="Custom endpoint" name="value" type="text" class="validate" minlength="12">
+              <span class="supporting-text">Leaving this empty will create a random secure string</span>
+            </div>
+            <div class="col s12 input-field">
+              <input placeholder="Description" name="description" type="text" class="validate" minlength="8" maxlength="64" required>
+              <span class="supporting-text">Description for the endpoint for better administration</span>
+            </div>
+          <button class="btn mt-4" type="submit" name="task" value="add-endpoint">Create</button>
+          </div>
+        </form>
+      </div>
+    </div>
+  </div>
+  <div class="col s12 l4 m-4">
+    <div class="card">
+      <div class="card-content">
+        <span class="card-title">Queue manually</span>
+        <form method="post">
+          <div class="row">
+            <div class="col s12 input-field">
+              <input placeholder="Youtube video ID" name="value" type="text" class="validate" minlength="11" maxlength="11" required>
+              <span class="supporting-text">Must be a valid Youtube video ID</span>
+            </div>
+            <div class="col s12 mt-5 input-field">
+              <div class="switch">
+                <label>Queue<input type="checkbox" value="direct" name="direct"><span class="lever"></span>Direct</label>
+                <span class="supporting-text">Queue up or start directly</span>
+              </div>
+            </div>
+          <button class="btn mt-4" type="submit" name="task" value="manual-queue">Queue</button>
+          </div>
+        </form>
+      </div>
+    </div>
+  </div>
+</div>
+<div class="divider"></div>
+<div class="row">
+  <div class="col s6 l9">
+    <h5>Registered endpoints</h5>
+  </div>
+  <div class="col s6 l3 m-4 input-field">
+    <input id="filter_query" type="text">
+    <label for="filter_query">Filter results</label>
+  </div>
+</div>
+<div class="row">
+  <div class="col s12">
+    <table class="striped highlight responsive-table">
+      <thead>
+        <tr>
+          <th>Actions</th>
+          <th>id</th>
+          <th>description</th>
+          <th>status</th>
+          <th>created_time</th>
+          <th>retired_time</th>
+        </tr>
+      </thead>
+      <tbody>
+        {% for endpoint in endpoints %}
+        <tr class="filterable">
+          <td>
+            <form method="post">
+              <input type="text" value="{{ endpoint.get('id') }}" name="value" hidden>
+              <button class="btn-small waves-effect waves-light" type="submit" name="task" value="retire" title="Retire endpoint" {% if endpoint.get('status') != 'active' %}disabled{% endif %}>🗑️</button>
+            </form>
+          </td>
+          <td>{{ endpoint.get('id') }}</td>
+          <td>{{ endpoint.get('description') }}</td>
+          <td>{{ endpoint.get('status') }}</td>
+          <td>{{ endpoint.get('created_time') }}</td>
+          <td>{{ endpoint.get('retired_time') }}</td>
+        </tr>
+        {% endfor %}
+      </tbody>
+    </table>
+  </div>
+</div>
+<div class="divider"></div>
+<div class="row">
+  <div class="col s6 l9">
+    <h5>Queued ID's</h5>
+  </div>
+  <div class="col s6 l3 m-4 input-field">
+    <input id="filter_query" type="text">
+    <label for="filter_query">Filter results</label>
+  </div>
+</div>
+<div class="row">
+  <div class="col s12">
+    <table class="striped highlight responsive-table">
+      <thead>
+        <tr>
+          <th>Actions</th>
+          <th>id</th>
+          <th>endpoint</th>
+          <th>status</th>
+          <th>created_time</th>
+        </tr>
+      </thead>
+      <tbody>
+        {% for id in queue %}
+        <tr class="filterable">
+          <td>
+            <form method="post">
+              <input type="text" value="{{ id.get('id') }}" name="value" hidden>
+              <button class="btn-small waves-effect waves-light" type="submit" name="task" value="delete-queue" title="Delete from queue" {% if id.get('status') != 'queued' %}disabled{% endif %}>🗑️</button>
+            </form>
+            <form method="post">
+              <input type="text" value="{{ id.get('id') }}" name="value" hidden>
+              <button class="btn-small waves-effect waves-light" type="submit" name="task" value="run-download" title="Run download task" disabled>⏩</button>
+			  <!-- This function fill not work until the download queue and video download process is rewritten -->
+            </form>
+          </td>
+          <td>{{ id.get('id') }}</td>
+          <td>{{ id.get('endpoint') }}</td>
+          <td>{{ id.get('status') }}</td>
+          <td>{{ id.get('created_time') }}</td>
+        </tr>
+        {% endfor %}
+      </tbody>
+    </table>
+  </div>
+</div>
+{% endblock %}

ayta/templates/admin/websub.html

@@ -4,14 +4,9 @@
 
 {% block content %}
 <div class="row">
-  <div class="col s12 l11">
+  <div class="col s12">
     <h4>WebSub administration page</h4>
   </div>
-  <div class="col s12 l1 m-5">
-    <form method="POST">
-	  <input title="Prunes all retired callbacks, but keeps last 3 days" type="submit" value="clean-retired" name="task">
-    </form>
-  </div>
 </div>
 <div class="divider"></div>
 <div class="row">
@@ -19,6 +14,43 @@
 	<h5>WebSub options</h5>
   </div>
 </div>
+<div class="row">
+  <div class="col s12 l4 m-4">
+	<div class="card">
+      <div class="card-content">
+        <span class="card-title">Direct actions</span>
+	    <form method="post" onsubmit="return confirm('Are you sure?');">
+          <button class="btn mb-2 green" type="submit" name="task" value="subscribe-channels">Subscribe channels</button>
+		  <br>
+		  <span class="supporting-text">Send WebSub subscription request for all activated channels. (This will renew existing ones as well)</span>
+	    </form>
+		<form class="mt-4" method="post" onsubmit="return confirm('Are you sure?');">
+          <button class="btn mb-2 red" type="submit" name="task" value="unsubscribe-callbacks">Unsubscribe channels</button>
+		  <br>
+		  <span class="supporting-text">Send WebSub unsubscription request for all activated endpoints. (This will only unsubscribe, not disable)</span>
+	    </form>
+		<form class="mt-4" method="post" onsubmit="return confirm('Are you sure?');">
+          <button class="btn mb-2" type="submit" name="task" value="clean-retired">Clean retired</button>
+		  <br>
+		  <span class="supporting-text">Prunes all retired callbacks, but keeps until last day</span>
+	    </form>
+      </div>
+    </div>
+  </div>
+  <div class="col s12 l4 m-4">
+	<div class="card">
+      <div class="card-content">
+        <span class="card-title">Statistics</span>
+	    <h6>Unprocessed callback datapoints</h6>
+		<p>{{ render['stats']['unprocessed_data'] }}</p>
+		<h6>Active callbacks</h6>
+		<p>{{ render['stats']['active_callbacks'] }}</p>
+		<h6>Something</h6>
+		<p>Blah</p>
+      </div>
+    </div>
+  </div>
+</div>
 <div class="divider"></div>
 <div class="row">
   <div class="col s6 l9">

ayta/templates/admin/workers.html

@@ -0,0 +1,47 @@
+{% extends 'material_base.html' %}
+{% block title %}Workers administration page{% endblock %}
+{% block description %}Workers administration page of the AYTA system{% endblock %}
+
+{% block content %}
+<div class="row">
+  <div class="col s12">
+    <h4>Workers administration page</h4>
+  </div>
+</div>
+<div class="divider"></div>
+<div class="row">
+  <div class="col s12">
+    <h5>Options</h5>
+  </div>
+</div>
+<form method="POST">
+      <input title="test-sleep" type="submit" value="test-sleep" name="task">
+</form>
+<div class="divider"></div>
+<div class="row">
+  <div class="col s12">
+    <h6>Current workers</h6>
+    {% for worker in tasks %}
+    <span>{{ worker }}</span>
+    <table class="striped highlight responsive-table" style=" border: 1px solid black;">
+      <thead>
+        <tr>
+          <th>ID</th>
+          <th>Task</th>
+          <th>Time started</th>
+        </tr>
+      </thead>
+      <tbody>
+        {% for task in tasks[worker] %}
+        <tr>
+          <td>{{ task.get('id') }}</td>
+          <td>{{ task.get('type') }}</td>
+          <td>{{ task.get('time_start')|epoch_time }}</td>
+        </tr>
+        {% endfor %}
+      </tbody>
+    </table>
+    {% endfor %}
+  </div>
+</div>
+{% endblock %}

ayta/templates/help.html

@@ -25,12 +25,24 @@
 	  </li>
 	  <li class="collection-item">
 	    <a href="{{ url_for('channel.channel', channelId='UCIcgBZ9hEJxHv6r_jDYOMqg') }}"><span class="title">Unus Annus</span></a>
-        <p>Reason: This channel does not exist. (Self removed)</p>
+        <p>Reason: This channel does not exist.</p>
 	  </li>
 	  <li class="collection-item">
 	    <a href="{{ url_for('channel.channel', channelId='UCz1s8aJYSQuaXJCtEi-VWRA') }}"><span class="title">Dutch Legion</span></a>
         <p>Reason: This account has been terminated due to multiple or severe violations of YouTube's policy prohibiting hate speech.</p>
 	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UC91-8aNaRbp71UMEb_34ryg') }}"><span class="title">RBMK5000</span></a>
+        <p>Reason: This channel does not exist.</p>
+	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UCoPSAT64vfXlulyWd_dPE3Q') }}"><span class="title">Evilfisher2</span></a>
+        <p>Reason: This channel was removed because it violated our Community Guidelines.</p>
+	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UCZXkvavD2YKnFCzCkZ-bNPw') }}"><span class="title">mrabhy</span></a>
+        <p>Reason: This channel was removed because it violated our Community Guidelines.</p>
+	  </li>
     </ul>
   </div>
   <div class="col s12 l6 center-align">
@@ -43,6 +55,22 @@
 	    <a href="{{ url_for('channel.channel', channelId='UCzGdxkzULCa9RlD-Q2EZPXQ') }}"><span class="title">Kalashnikov Group</span></a>
         <p>Reason: This account has been terminated for a violation of YouTube's Terms of Service.</p>
 	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UCtfg1tENiu3SgGMZVduFmTg') }}"><span class="title">FiberNinja</span></a>
+        <p>Reason: This channel was removed because it violated our Community Guidelines.</p>
+	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UCv4VkfbX8YfqodF-4coEEfQ') }}"><span class="title">James Somerton</span></a>
+        <p>Reason: This channel does not exist.</p>
+	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UC8XH9kpilkuss4bVeRZD1kw') }}"><span class="title">Plagued Moth</span></a>
+        <p>Reason: This channel was removed because it violated our Community Guidelines.</p>
+	  </li>
+	  <li class="collection-item">
+	    <a href="{{ url_for('channel.channel', channelId='UCxZTTWP0QN7-ch2wW1QeFwg') }}"><span class="title">CowOfTheSea</span></a>
+        <p>Reason: This channel was removed because it violated our Community Guidelines.</p>
+	  </li>
 	</ul>
   </div>
 </div>

ayta/templates/watch/index.html

@@ -27,7 +27,7 @@
   <div class="col s12 l3">
     <p><b>Video by:</b> <a href="{{ url_for('channel.channel', channelId=render.get('info').get('channel_id')) }}">{{ render.get('info').get('uploader') }}</a></p>
     <p><b>Upload date:</b> {{ render.get('info').get('upload_date')|pretty_time }}</p>
-    <p><b>Archive date:</b> {{ render.get('info').get('epoch')|epoch_time }}</p>
+    <p><b>Archive date:</b> {{ render.get('info').get('epoch')|epoch_date }}</p>
     <p><b>Video length:</b> {{ render.get('info').get('duration')|pretty_duration }}</p>
   </div>
   <div class="col s4 l3 center-align">

requirements.txt

@@ -8,4 +8,5 @@ yt-dlp
 gunicorn
 celery
 sqlalchemy
+requests
 pyjwt[crypto]