Small concurrency and logging fix

Dockerfile

@@ -1,8 +1,7 @@
-FROM python:3-alpine
+FROM python:3.12-alpine
-RUN apk update && apk add python3-dev gcc libc-dev libffi-dev
 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/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
 from datetime import datetime
 from secrets import token_urlsafe
 
@@ -190,14 +190,16 @@ def queue():
             flash(f'Cleaned retired endpoints')
             
         elif task == 'manual-queue':
-            direct = request.form.get('direct', None)
+            if not get_nosql().check_exists(value):
-            
+                direct = request.form.get('direct', None)
-            if direct:
+                if direct:
-                task = video_download.delay(value)
+                    task = video_download.delay(value)
-                flash(f"Started task {task.id}")
+                    flash(f"Started task {task.id}")
+                else:
+                    get_nosql().queue_insertQueue(value, 'webui')
+                    flash(f'Added to queue: {value}')
             else:
-                get_nosql().queue_insertQueue(value, 'webui')
+                flash(f'This video ID already exists in the archive: {value}')
-                flash(f'Added to queue: {value}')
             
         elif task == 'delete-queue':
             get_nosql().queue_deleteQueue(value)
@@ -205,7 +207,11 @@ def queue():
             
         elif task == 'empty-queue':
             get_nosql().queue_emptyQueue()
-            flash(f'Queue has  been emptied')
+            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.queue'))
         

ayta/nosql.py

@@ -411,13 +411,17 @@ class Mango:
     ##########################################
         
     def queue_insertQueue(self, videoId, endpointId=None):
-        # if no document exists
+        # if already queued
-        if not self.download_queue.count_documents({'id': videoId}) >= 1:
+        if 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 False
-            return True
+        
-            
+        # if already in archive
-        # key already in queue
+        if self.check_exists(videoId):
-        return False
+            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}):
@@ -429,7 +433,21 @@ 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'])
+            self.download_queue.update_one({'id': queueItem['id']}, {'$set': {'status': 'working'}})
+            return queueItem
+        
 ##########################################
 #           HELPER FUNCTIONS             #
 ##########################################
@@ -453,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/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):

ayta/templates/admin/queue.html

@@ -11,67 +11,71 @@
 <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>
+          <br>
-		  <span class="supporting-text">Removes all queued ids</span>
+          <span class="supporting-text">Removes all queued ids</span>
-	    </form>
+        </form>
-		<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" type="submit" name="task" value="clean-retired">Clean retired</button>
-		  <br>
+          <br>
-		  <span class="supporting-text">Prunes all deactivated endpoints, but keeps last 3 days</span>
+          <span class="supporting-text">Prunes all deactivated endpoints, but keeps last 3 days</span>
-	    </form>
+        </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">
       <div class="card-content">
         <span class="card-title">Create new endpoint</span>
-	    <form method="post">
+        <form method="post">
-		  <div class="row">
+          <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>
+              <span class="supporting-text">Leaving this empty will create a random secure string</span>
-	        </div>
+            </div>
-			<div class="col s12 input-field">
+            <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>
+              <span class="supporting-text">Description for the endpoint for better administration</span>
-	        </div>
+            </div>
           <button class="btn mt-4" type="submit" name="task" value="add-endpoint">Create</button>
-		  </div>
+          </div>
-	    </form>
+        </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">
+        <form method="post">
-		  <div class="row">
+          <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>
+              <span class="supporting-text">Must be a valid Youtube video ID</span>
-	        </div>
+            </div>
-			<div class="col s12 mt-5 input-field">
+            <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>
+                <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>
+                <span class="supporting-text">Queue up or start directly</span>
-		      </div>
+              </div>
-	        </div>
+            </div>
           <button class="btn mt-4" type="submit" name="task" value="manual-queue">Queue</button>
-		  </div>
+          </div>
-	    </form>
+        </form>
       </div>
     </div>
   </div>
@@ -79,10 +83,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 +95,30 @@
     <table class="striped highlight responsive-table">
       <thead>
         <tr>
-		  <th>Actions</th>
+          <th>Actions</th>
-		  <th>id</th>
+          <th>id</th>
-		  <th>description</th>
+          <th>description</th>
-		  <th>status</th>
+          <th>status</th>
-		  <th>created_time</th>
+          <th>created_time</th>
-		  <th>retired_time</th>
+          <th>retired_time</th>
         </tr>
       </thead>
       <tbody>
-	    {% for endpoint in endpoints %}
+        {% for endpoint in endpoints %}
         <tr class="filterable">
-		  <td>
+          <td>
-		    <form method="post">
+            <form method="post">
-			  <input type="text" value="{{ endpoint.get('id') }}" name="value" hidden>
+              <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>
+              <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>
+            </form>
-		  </td>
+          </td>
-		  <td>{{ endpoint.get('id') }}</td>
+          <td>{{ endpoint.get('id') }}</td>
-		  <td>{{ endpoint.get('description') }}</td>
+          <td>{{ endpoint.get('description') }}</td>
-		  <td>{{ endpoint.get('status') }}</td>
+          <td>{{ endpoint.get('status') }}</td>
-		  <td>{{ endpoint.get('created_time') }}</td>
+          <td>{{ endpoint.get('created_time') }}</td>
-		  <td>{{ endpoint.get('retired_time') }}</td>
+          <td>{{ endpoint.get('retired_time') }}</td>
         </tr>
-		{% endfor %}
+        {% endfor %}
       </tbody>
     </table>
   </div>
@@ -122,10 +126,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 +138,33 @@
     <table class="striped highlight responsive-table">
       <thead>
         <tr>
-		  <th>Actions</th>
+          <th>Actions</th>
-		  <th>id</th>
+          <th>id</th>
-		  <th>endpoint</th>
+          <th>endpoint</th>
-		  <th>status</th>
+          <th>status</th>
-		  <th>created_time</th>
+          <th>created_time</th>
         </tr>
       </thead>
       <tbody>
-	    {% for id in queue %}
+        {% for id in queue %}
         <tr class="filterable">
-		  <td>
+          <td>
-		    <form method="post">
+            <form method="post">
-			  <input type="text" value="{{ id.get('id') }}" name="value" hidden>
+              <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>
+              <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>
-		  </td>
+            <form method="post">
-		  <td>{{ id.get('id') }}</td>
+              <input type="text" value="{{ id.get('id') }}" name="value" hidden>
-		  <td>{{ id.get('endpoint') }}</td>
+              <button class="btn-small waves-effect waves-light" type="submit" name="task" value="run-download" title="Run download task" disabled>⏩</button>
-		  <td>{{ id.get('status') }}</td>
+			  <!-- This function fill not work until the download queue and video download process is rewritten -->
-		  <td>{{ id.get('created_time') }}</td>
+            </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/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">
@@ -45,6 +57,18 @@
 	  </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>

requirements.txt

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