fuck

ayta/__init__.py

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

ayta/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, video_queue
+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 == 'add_channel':
+            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)
@@ -156,6 +163,10 @@ def reports():
             get_nosql().close_report(value)
             flash(f'Report closed {value}')
             return redirect(url_for('admin.reports'))
+        elif task == 'clean-closed':
+            get_nosql().report_clean()
+            flash(f'Cleaned closed reports older than 30 days')
+            return redirect(url_for('admin.reports'))
 
     reports = get_nosql().list_reports()
    

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):
@@ -243,6 +247,10 @@ class Mango:
     def close_report(self, _id):
         _id = ObjectId(_id)
         return self.reports.update_one({'_id': _id}, {'$set': {'status': 'closed', 'closing_time': current_time(object=True)}})
+        
+    def report_clean(self, keep=30):
+        days = self.datetime.utcnow() - self.timedelta(days=keep)
+        self.reports.delete_many({'status': 'closed', 'closing_time': {'$lt': days}})
 
     ##########################################
     #           RUNLOG FUNCTIONS             #
@@ -420,8 +428,7 @@ class Mango:
             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
+        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}):
@@ -434,8 +441,8 @@ 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_setFailed(self, videoId, reason=None):
+        return self.download_queue.update_one({'id': videoId}, {'$set': {'status': 'failed', 'fail_reason': reason}})
         
     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."""
@@ -445,6 +452,7 @@ class Mango:
                 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
         

ayta/tasks.py

@@ -21,8 +21,8 @@ def video_download(videoId):
     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
+        return (False, process.stdout)
+    return (True, None)
     
 @shared_task()
 def video_queue():
@@ -38,11 +38,13 @@ def video_queue():
     else:
         return None
     
-    if video_download(videoId):
+    status, reason = video_download(videoId)
+    
+    if status:
         get_nosql().queue_deleteQueue(videoId)
         return True
     else:
-        get_nosql().queue_setFailed(videoId)
+        get_nosql().queue_setFailed(videoId, reason)
         return False
 
 @shared_task()
@@ -164,6 +166,28 @@ def websub_renew_expiring(hours=6):
         if count >= 256:
             break
 
+@shared_task()
+def playlist_to_queue():
+    from .nosql import get_nosql
+    import random
+    from datetime import datetime, timedelta
+    
+    pivot = datetime.utcnow() - timedelta(days=3)  # calculates 3 days before now
+    channels = list(get_nosql().list_all_channels(active=True))
+    random.shuffle(channels)  # randomize channelId order because otherwise the queue will follow the channel order as well
+    
+    for channel in channels:
+        info = get_nosql().get_channel_info(channel)
+        
+        # if last_run not set or last_run is older than the pivot (indicating it has not been updated)
+        if not info.get('last_run') or info.get('last_run') < pivot:
+            # skip channel
+            continue
+        
+        for item in info['playlist']['entries']:
+            videoId = item['id']
+            get_nosql().queue_insertQueue(videoId, 'Playlist mirroring')
+
 ##########################################
 #              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

@@ -144,6 +144,7 @@
           <th>endpoint</th>
           <th>status</th>
           <th>created_time</th>
+          <th>fail_reason</th>
         </tr>
       </thead>
       <tbody>
@@ -164,6 +165,7 @@
           <td>{{ id.get('endpoint') }}</td>
           <td>{{ id.get('status') }}</td>
           <td>{{ id.get('created_time') }}</td>
+          <td><textarea class="info">{{ id.get('fail_reason') }}</textarea></td>
         </tr>
         {% endfor %}
       </tbody>

ayta/templates/admin/workers.html

@@ -22,6 +22,9 @@
   <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;">
@@ -43,9 +46,13 @@
       </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;">
@@ -67,6 +74,7 @@
       </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