diff --git a/tubearchivist/home/src/download/subscriptions.py b/tubearchivist/home/src/download/subscriptions.py index 67818b84..4cd06323 100644 --- a/tubearchivist/home/src/download/subscriptions.py +++ b/tubearchivist/home/src/download/subscriptions.py @@ -265,3 +265,61 @@ class PlaylistSubscription: missing_videos.append(youtube_id) return missing_videos + + +class SubscriptionScanner: + """add missing videos to queue""" + + def __init__(self): + self.missing_videos = False + + def scan(self): + """scan channels and playlists""" + self.missing_videos = [] + self._notify() + self._scan_channels() + self._scan_playlists() + if not self.missing_videos: + return + + self.add_to_pending() + + def _notify(self): + """set redis notification""" + message = { + "status": "message:rescan", + "level": "info", + "title": "Rescanning channels and playlists.", + "message": "Looking for new videos.", + } + RedisArchivist().set_message("message:rescan", message, expire=True) + + def _scan_channels(self): + """get missing from channels""" + channel_handler = ChannelSubscription() + missing = channel_handler.find_missing() + if not missing: + return + + for vid_id, vid_type in missing: + self.missing_videos.append( + {"type": "video", "vid_type": vid_type, "url": vid_id} + ) + + def _scan_playlists(self): + """get missing from playlists""" + playlist_handler = PlaylistSubscription() + missing = playlist_handler.find_missing() + if not missing: + return + + for i in missing: + self.missing_videos.append( + {"type": "video", "vid_type": VideoTypeEnum.VIDEOS, "url": i} + ) + + def add_to_pending(self): + """add missing videos to pending queue""" + pending_handler = queue.PendingList(youtube_ids=self.missing_videos) + pending_handler.parse_url_list() + pending_handler.add_to_pending() diff --git a/tubearchivist/home/tasks.py b/tubearchivist/home/tasks.py index 23c8d3e4..f17e48c5 100644 --- a/tubearchivist/home/tasks.py +++ b/tubearchivist/home/tasks.py @@ -14,6 +14,7 @@ from home.src.download.queue import PendingList from home.src.download.subscriptions import ( ChannelSubscription, PlaylistSubscription, + SubscriptionScanner, ) from home.src.download.thumbnails import ThumbFilesystem, ThumbValidator from home.src.download.yt_dlp_handler import VideoDownloader @@ -26,6 +27,7 @@ from home.src.index.video_constants import VideoTypeEnum from home.src.ta.config import AppConfig, ReleaseVersion, ScheduleBuilder from home.src.ta.helper import clear_dl_cache from home.src.ta.ta_redis import RedisArchivist, RedisQueue +from home.src.ta.task_manager import TaskManager from home.src.ta.urlparser import Parser CONFIG = AppConfig().config @@ -44,72 +46,37 @@ app.autodiscover_tasks() app.conf.timezone = os.environ.get("TZ") or "UTC" -@shared_task(name="update_subscribed") -def update_subscribed(): +@shared_task(name="update_subscribed", bind=True) +def update_subscribed(self): """look for missing videos and add to pending""" - message = { - "status": "message:rescan", - "level": "info", - "title": "Rescanning channels and playlists.", - "message": "Looking for new videos.", - } - RedisArchivist().set_message("message:rescan", message, expire=True) + manager = TaskManager() + if manager.is_pending(self): + print(f"[task][{self.name}] rescan already running") + message = { + "status": "message:rescan", + "level": "error", + "title": "Rescanning channels and playlists.", + "message": "Rescan already in progress.", + } + RedisArchivist().set_message("message:rescan", message, expire=True) + return - have_lock = False - my_lock = RedisArchivist().get_lock("rescan") - - try: - have_lock = my_lock.acquire(blocking=False) - if have_lock: - channel_handler = ChannelSubscription() - missing_from_channels = channel_handler.find_missing() - playlist_handler = PlaylistSubscription() - missing_from_playlists = playlist_handler.find_missing() - if missing_from_channels or missing_from_playlists: - channel_videos = [ - {"type": "video", "vid_type": vid_type, "url": vid_id} - for vid_id, vid_type in missing_from_channels - ] - playlist_videos = [ - { - "type": "video", - "vid_type": VideoTypeEnum.VIDEOS, - "url": i, - } - for i in missing_from_playlists - ] - pending_handler = PendingList( - youtube_ids=channel_videos + playlist_videos - ) - pending_handler.parse_url_list() - pending_handler.add_to_pending() - - else: - print("Did not acquire rescan lock.") - - finally: - if have_lock: - my_lock.release() + manager.init(self) + SubscriptionScanner().scan() -@shared_task(name="download_pending") -def download_pending(): +@shared_task(name="download_pending", bind=True) +def download_pending(self): """download latest pending videos""" - have_lock = False - my_lock = RedisArchivist().get_lock("downloading") + manager = TaskManager() + if manager.is_pending(self): + print(f"[task][{self.name}] download queue already running") + return - try: - have_lock = my_lock.acquire(blocking=False) - if have_lock: - downloader = VideoDownloader() - downloader.add_pending() - downloader.run_queue() - else: - print("Did not acquire download lock.") - - finally: - if have_lock: - my_lock.release() + manager.init(self) + downloader = VideoDownloader() + downloader.add_pending() + downloader.run_queue() @shared_task(name="download_single")