Schedule indexing tasks using huey

tubesync/sync/tasks.py

@@ -28,7 +28,7 @@ from background_task import background
 from background_task.exceptions import InvalidTaskError
 from background_task.models import Task, CompletedTask
 from django_huey import db_periodic_task, db_task, task as huey_task # noqa
-from huey import CancelExecution
+from huey import CancelExecution, crontab as huey_crontab
 from common.huey import dynamic_retry
 from common.logger import log
 from common.errors import ( BgTaskWorkerError, DownloadFailedException,
@@ -36,7 +36,7 @@ from common.errors import ( BgTaskWorkerError, DownloadFailedException,
                             NoThumbnailException, )
 from common.utils import (  django_queryset_generator as qs_gen,
                             remove_enclosed, seconds_to_timestr, )
-from .choices import Val, TaskQueue
+from .choices import Val, IndexSchedule, TaskQueue
 from .models import Source, Media, MediaServer, Metadata
 from .utils import get_remote_image, resize_image_to_height, filter_response
 from .youtube import YouTubeError
@@ -222,6 +222,38 @@ def save_model(instance):
     time.sleep(random.expovariate(arg))
 
 
+@db_periodic_task(
+    huey_crontab(minute=59, strict=True,),
+    priority=100,
+    expires=30*60,
+)
+def schedule_indexing():
+    now = timezone.now()
+    next_hour = now + timezone.timedelta(hours=1, minutes=1)
+    qs = Source.objects.filter(
+        index_schedule__gt=Val(IndexSchedule.NEVER),
+    )
+    for source in qs:
+        previous_run = now - timezone.timedelta(
+            seconds=source.index_schedule
+        )
+        skip_source = (
+            not source.is_active or
+            source.target_schedule >= next_hour or
+            source.last_crawl >= previous_run
+        )
+        if skip_source:
+            continue
+        vn_fmt = _('Index media from source "{}"')
+        index_source_task(
+            str(source.pk),
+            repeat=0,
+            schedule=600,
+            verbose_name=vn_fmt.format(source.name),
+        )
+    pass
+
+
 def schedule_media_servers_update():
     # Schedule a task to update media servers
     log.info('Scheduling media server updates')