mygpo/data/tasks.py

   1 from operator import itemgetter
   2 from datetime import datetime, timedelta
   3
   4 from celery.decorators import periodic_task
   5
   6 from mygpo.data.podcast import calc_similar_podcasts
   7 from mygpo.celery import celery
   8 from mygpo.podcasts.models import Podcast
   9
  10 from celery.utils.log import get_task_logger
  11 logger = get_task_logger(__name__)
  12
  13
  14 @celery.task
  15 def update_podcasts(podcast_urls):
  16     """ Task to update a podcast """
  17     from mygpo.data.feeddownloader import PodcastUpdater
  18     updater = PodcastUpdater()
  19     podcasts = updater.update_queue(podcast_urls)
  20     return list(podcasts)
  21
  22
  23 @celery.task
  24 def update_related_podcasts(podcast, max_related=20):
  25     get_podcast = itemgetter(0)
  26
  27     related = calc_similar_podcasts(podcast)[:max_related]
  28     related = map(get_podcast, related)
  29
  30     for p in related:
  31         podcast.related_podcasts.add(p)
  32
  33
  34 # interval in which podcast updates are scheduled
  35 UPDATE_INTERVAL = timedelta(hours=1)
  36
  37
  38 @periodic_task(run_every=UPDATE_INTERVAL)
  39 def schedule_updates(interval=UPDATE_INTERVAL):
  40     """ Schedules podcast updates that are due within ``interval`` """
  41     now = datetime.utcnow()
  42
  43     # fetch podcasts for which an update is due within the next hour
  44     podcasts = Podcast.objects.all()\
  45                               .next_update_between(now, now+interval)\
  46                               .prefetch_related('urls')\
  47                               .only('pk')
  48
  49     logger.error('Scheduling %d podcasts for update', podcasts.count())
  50     # queue all those podcast updates
  51     for podcast in podcasts:
  52         update_podcasts.delay([podcast.url])