Performance and cleanup

ayta/blueprints/admin.py

@@ -2,7 +2,7 @@ from flask import Blueprint, render_template, request, redirect, url_for, flash,
 from ..nosql import get_nosql
 from ..dlp import checkChannelId, getChannelInfo
 from ..decorators import login_required
-from ..tasks import test_sleep, websub_subscribe_callback, websub_unsubscribe_callback, video_download
+from ..tasks import test_sleep, websub_subscribe_callback, websub_unsubscribe_callback, video_download, video_queue, playlist_to_queue
 from datetime import datetime
 from secrets import token_urlsafe
 
@@ -30,28 +30,35 @@ def channels():
     generic = {}
 
     if request.method == 'POST':
-        channelId = request.form.get('channel_id', None)
-        originalName = request.form.get('original_name', None)
-        addedDate = request.form.get('added_date', None)
+        task = request.form.get('task', None)
+        
+        if task == 'subscribe-websub':
+            channelId = request.form.get('channel_id', None)
+            originalName = request.form.get('original_name', None)
+            addedDate = request.form.get('added_date', None)
 
-        ### add some validation
-        
-        addedDate = datetime.strptime(addedDate, '%Y-%m-%d')
-        
-        if checkChannelId(channelId) is False:
-            channelId, originalName = getChannelInfo(channelId, ('channel_id', 'uploader'))
-        
-        if not get_nosql().insert_new_channel(channelId, originalName, addedDate):
-            flash('Error inserting new channel, you probably made a mistake somewhere')
-            return redirect(url_for('admin.channels'))
+            ### add some validation
+            
+            addedDate = datetime.strptime(addedDate, '%Y-%m-%d')
+            
+            if checkChannelId(channelId) is False:
+                channelId, originalName = getChannelInfo(channelId, ('channel_id', 'uploader'))
+            
+            if not get_nosql().insert_new_channel(channelId, originalName, addedDate):
+                flash('Error inserting new channel, you probably made a mistake somewhere')
+                return redirect(url_for('admin.channels'))
 
-        return redirect(url_for('admin.channel', channelId=channelId))
+            return redirect(url_for('admin.channel', channelId=channelId))
+            
+        elif task == 'playlist-queue':
+            task = playlist_to_queue.delay()
+            flash(f'Task playlist-queue has been queued: {task.id}')
 
     generic['currentDate'] = datetime.utcnow()
     channelIds = get_nosql().list_all_channels()
     
     for channelId in channelIds:
-        channels[channelId] = get_nosql().get_channel_info(channelId)
+        channels[channelId] = get_nosql().get_channel_info(channelId, limited=True)
         channels[channelId]['video_count'] = get_nosql().get_channel_videos_count(channelId)
         
     return render_template('admin/channels.html', channels=channels, generic=generic)
@@ -208,6 +215,12 @@ def queue():
         elif task == 'empty-queue':
             get_nosql().queue_emptyQueue()
             flash(f'Queue has been emptied')
+            
+        elif task == 'queue-run-once':
+            value = int(value) if value.isdigit() else 1
+            for x in range(value):
+                task =  video_queue.delay()
+                flash(f'Task has been started on the oldest queued item: {task.id}')
         
         return redirect(url_for('admin.queue'))
         
@@ -245,7 +258,7 @@ def users():
     return render_template('admin/users.html', users=users)
     
 @bp.route('/workers', methods=['GET', 'POST'])
-#@login_required
+@login_required
 def workers():
     if request.method == 'POST':
         task = request.form.get('task', None)
@@ -255,4 +268,5 @@ def workers():
     celery = current_app.extensions.get('celery')
     
     tasks = celery.control.inspect().active()
-    return render_template('admin/workers.html', tasks=tasks)
+    reserved = celery.control.inspect().reserved()
+    return render_template('admin/workers.html', tasks=tasks, reserved=reserved)

ayta/blueprints/channel.py

@@ -11,7 +11,7 @@ def base():
     channelIds = get_nosql().list_all_channels()
     
     for channelId in channelIds:
-        channel = get_nosql().get_channel_info(channelId)
+        channel = get_nosql().get_channel_info(channelId, limited=True)
         channel['video_count'] = get_nosql().get_channel_videos_count(channelId)
         channels.append(channel)
         

ayta/nosql.py

@@ -168,8 +168,12 @@ class Mango:
             ids.append(video['id'])
         return tuple(ids)
  
-    def get_channel_info(self, channelId):
-        return self.channels.find_one({'id': channelId})
+    def get_channel_info(self, channelId, limited=False):
+        projection = {}
+        if limited:
+            projection['playlist'] = 0
+            
+        return self.channels.find_one({'id': channelId}, projection)
         
         
     def update_channel_key(self, channelId, key, value):
@@ -411,13 +415,16 @@ class Mango:
     ##########################################
         
     def queue_insertQueue(self, videoId, endpointId=None):
-        # if no document exists
-        if not self.download_queue.count_documents({'id': videoId}) >= 1:
-            self.download_queue.insert_one({'id': videoId, 'endpoint': endpointId, 'created_time': current_time(object=True), 'status': 'queued'}).inserted_id
-            return True
-            
-        # key already in queue
-        return False
+        # 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
+        return self.download_queue.insert_one({'id': videoId, 'endpoint': endpointId, 'created_time': current_time(object=True), 'status': 'queued'}).inserted_id
         
     def queue_deleteQueue(self, videoId):
         if self.download_queue.delete_one({'id': videoId}):
@@ -429,7 +436,22 @@ class Mango:
         
     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'])
+                continue
+            self.download_queue.update_one({'id': queueItem['id']}, {'$set': {'status': 'working'}})
+            return queueItem
+        
 ##########################################
 #           HELPER FUNCTIONS             #
 ##########################################
@@ -453,7 +475,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

@@ -4,6 +4,7 @@ class OIDC():
     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.
+    The nonce and state store is in memory, so only one instance can be used at a time until central key caching is implemented.
     """
     def __init__(self, app=None):
         self.states = {}
@@ -151,7 +152,6 @@ class OIDC():
                                  
         # 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 by matching the nonce in the signed key

ayta/tasks.py

@@ -24,6 +24,26 @@ def video_download(videoId):
         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):
@@ -144,6 +164,19 @@ def websub_renew_expiring(hours=6):
         if count >= 256:
             break
 
+@shared_task()
+def playlist_to_queue():
+    from .nosql import get_nosql
+    
+    channels = get_nosql().list_all_channels(active=True)
+    
+    for channel in channels:
+        info = get_nosql().get_channel_info(channel)
+        
+        for item in info['playlist']['entries']:
+            videoId = item['id']
+            get_nosql().queue_insertQueue(videoId, 'Playlist mirroring')
+
 ##########################################
 #              TASK MODULES              #
 ##########################################

ayta/templates/admin/channels.html

@@ -15,6 +15,18 @@
   </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">
+          <button class="btn mb-2 green" type="submit" name="task" value="playlist-queue">Playlist to Queue</button>
+          <br>
+          <span class="supporting-text">Forcerun playlist to queue task</span>
+        </form>
+      </div>
+    </div>
+  </div>
   <div class="col s12 l4 m-4">
 	<div class="card">
       <div class="card-content">
@@ -38,7 +50,7 @@
               });
             </script>
 		  </div>
-          <button class="btn mt-4" type="submit" name="action" value="add_channel">Add</button>
+          <button class="btn mt-4" type="submit" name="task" value="add_channel">Add</button>
 	    </form>
       </div>
     </div>

ayta/templates/admin/queue.html

@@ -11,67 +11,72 @@
 <div class="divider"></div>
 <div class="row">
   <div class="col s12">
-	<h5>Options</h5>
+    <h5>Options</h5>
   </div>
 </div>
 <div class="row">
   <div class="col s12 l4 m-4">
-	<div class="card">
+    <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?');">
+        <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?');">
+          <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>
-		
+          <br>
+          <span class="supporting-text">Prunes all deactivated endpoints, but keeps last 3 days</span>
+        </form>
+        <form class="mt-4 input-field" method="post" onsubmit="return confirm('Are you sure?');">
+          <input type="number" style="width: 80px" value="1" name="value" min="1" max="99">
+          <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">
       <div class="card-content">
         <span class="card-title">Create new endpoint</span>
-	    <form method="post">
-		  <div class="row">
+        <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">
+              <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>
+              <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>
+        </form>
       </div>
     </div>
   </div>
   <div class="col s12 l4 m-4">
-	<div class="card">
+    <div class="card">
       <div class="card-content">
         <span class="card-title">Queue manually</span>
-	    <form method="post">
-		  <div class="row">
+        <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">
+              <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>
+                <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>
+        </form>
       </div>
     </div>
   </div>
@@ -79,10 +84,10 @@
 <div class="divider"></div>
 <div class="row">
   <div class="col s6 l9">
-	<h5>Registered endpoints</h5>
+    <h5>Registered endpoints</h5>
   </div>
   <div class="col s6 l3 m-4 input-field">
-	<input id="filter_query" type="text">
+    <input id="filter_query" type="text">
     <label for="filter_query">Filter results</label>
   </div>
 </div>
@@ -91,30 +96,30 @@
     <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>
+          <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 %}
+        {% 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>
+          <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 %}
+        {% endfor %}
       </tbody>
     </table>
   </div>
@@ -122,10 +127,10 @@
 <div class="divider"></div>
 <div class="row">
   <div class="col s6 l9">
-	<h5>Queued ID's</h5>
+    <h5>Queued ID's</h5>
   </div>
   <div class="col s6 l3 m-4 input-field">
-	<input id="filter_query" type="text">
+    <input id="filter_query" type="text">
     <label for="filter_query">Filter results</label>
   </div>
 </div>
@@ -134,28 +139,33 @@
     <table class="striped highlight responsive-table">
       <thead>
         <tr>
-		  <th>Actions</th>
-		  <th>id</th>
-		  <th>endpoint</th>
-		  <th>status</th>
-		  <th>created_time</th>
+          <th>Actions</th>
+          <th>id</th>
+          <th>endpoint</th>
+          <th>status</th>
+          <th>created_time</th>
         </tr>
       </thead>
       <tbody>
-	    {% for id in queue %}
+        {% 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>
+          <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') == 'working' %}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 %}
+        {% endfor %}
       </tbody>
     </table>
   </div>

ayta/templates/admin/workers.html

@@ -19,8 +19,40 @@
 </form>
 <div class="divider"></div>
 <div class="row">
-  <div class="col s12">
-    <h6>Current workers</h6>
+  <div class="col s12 m-4">
+    <h5>Reserved tasks per worker</h5>
+    <p>Usually 4 tasks per worker</p>
+	{% if reserved is none %}
+	<h6>No workers with reserved tasks, are there any workers with stuck tasks or are they even online?</h6>
+	{% else %}
+    {% for worker in reserved %}
+    <span>{{ worker }}</span>
+    <table class="striped highlight responsive-table" style=" border: 1px solid black;">
+      <thead>
+        <tr>
+          <th>ID</th>
+          <th>Task</th>
+          <th>Arguments</th>
+        </tr>
+      </thead>
+      <tbody>
+        {% for task in reserved[worker] %}
+        <tr>
+          <td>{{ task.get('id') }}</td>
+          <td>{{ task.get('name') }}</td>
+          <td>{{ task.get('args') }} {{ task.get('kwargs') }}</td>
+        </tr>
+        {% endfor %}
+      </tbody>
+    </table>
+    {% endfor %}
+	{% endif %}
+  </div>
+  <div class="col s12 m-4">
+    <h5>Current workers and processing tasks</h5>
+	{% if tasks is none %}
+	<h6>No workers with running tasks, are there any workers with stuck tasks or are they even online?</h6>
+	{% else %}
     {% for worker in tasks %}
     <span>{{ worker }}</span>
     <table class="striped highlight responsive-table" style=" border: 1px solid black;">
@@ -35,13 +67,14 @@
         {% for task in tasks[worker] %}
         <tr>
           <td>{{ task.get('id') }}</td>
-          <td>{{ task.get('type') }}</td>
+          <td>{{ task.get('name') }}</td>
           <td>{{ task.get('time_start')|epoch_time }}</td>
         </tr>
         {% endfor %}
       </tbody>
     </table>
     {% endfor %}
+	{% endif %}
   </div>
 </div>
 {% endblock %}

requirements.txt

@@ -3,6 +3,8 @@
 flask
 flask-caching
 flask-limiter
+flask-sqlalchemy
+flask-migrate
 pymongo
 yt-dlp
 gunicorn