Implement watcher mode with negative -lookahead values, see help fo…

…r usage

src/cmdargs.py

@@ -29,7 +29,7 @@
 from tagger import valid_extra_tag, valid_playlist_name
 from validators import (
     valid_int, positive_nonzero_int, valid_rating, valid_path, valid_filepath_abs, valid_search_string, valid_proxy, naming_flags,
-    log_level, positive_int, valid_duration,
+    log_level, positive_int, nonzero_int, valid_duration,
 )
 from version import APP_NAME, APP_VERSION
 
@@ -185,7 +185,7 @@ def prepare_arglist_ids(args: Sequence[str]) -> Namespace:
     arggr_start_or_seq.add_argument('-start', metavar='#number', help=HELP_ARG_ID_START, type=positive_nonzero_int)
     arggr_count_or_end.add_argument('-count', metavar='#number', default=1, help=HELP_ARG_ID_COUNT, type=positive_nonzero_int)
     arggr_count_or_end.add_argument('-end', metavar='#number', default=1, help=HELP_ARG_ID_END, type=positive_nonzero_int)
-    par_cmd.add_argument('-lookahead', metavar='#number', default=0, help=HELP_ARG_LOOKAHEAD, type=positive_nonzero_int)
+    par_cmd.add_argument('-lookahead', metavar='#number', default=0, help=HELP_ARG_LOOKAHEAD, type=nonzero_int)
     par_cmd.add_argument('-gpred', '--predict-id-gaps', action=ACTION_STORE_TRUE, help=HELP_ARG_PREDICT_ID_GAPS)
     arggr_start_or_seq.add_argument('-seq', '--use-id-sequence', action=ACTION_STORE_TRUE, help=HELP_ARG_IDSEQUENCE)
 

src/defs.py

@@ -24,6 +24,9 @@
 DOWNLOAD_STATUS_CHECK_TIMER = 60
 DOWNLOAD_QUEUE_STALL_CHECK_TIMER = 30
 DOWNLOAD_CONTINUE_FILE_CHECK_TIMER = 30
+SCAN_CANCEL_KEYSTROKE = 'q'
+LOOKAHEAD_WATCH_RESCAN_DELAY_MIN = 300
+LOOKAHEAD_WATCH_RESCAN_DELAY_MAX = 1800
 
 SCREENSHOTS_COUNT = 20
 FULLPATH_MAX_BASE_LEN = 240
@@ -180,8 +183,10 @@ def __str__(self) -> str:
 HELP_ARG_PAGE_START = 'Start page number. Default is \'1\''
 HELP_ARG_BEGIN_STOP_ID = 'Video id lower / upper bounds filter to only download videos where \'begin_id >= video_id >= stop_id\''
 HELP_ARG_LOOKAHEAD = (
-    'Continue scanning indefinitely after reaching end id until number of non-existing videos encountered in a row'
-    ' reaches this number'
+    f'Continue scanning indefinitely after reaching end id until number of non-existing videos encountered in a row'
+    f' reaches this number.'
+    f' Furthermore, negative value enables watcher mode, periodically re-scanning trailing non-existing videos, this process never finishes'
+    f' on its own but can be interrupted safely by pressing \'{SCAN_CANCEL_KEYSTROKE}\' twice'
 )
 HELP_ARG_PREDICT_ID_GAPS = (
     'Enable ids known to be non-existent prediction. When video is uploaded to the website post id usually gets incremented more than once.'

src/dscanner.py

@@ -7,14 +7,16 @@
 #
 
 from __future__ import annotations
+from asyncio import Task, get_running_loop, CancelledError
 from asyncio.tasks import sleep
 from collections import deque
 from collections.abc import Callable, Coroutine
 from contextlib import suppress
-from typing import Any
+from typing import Any, Optional
 
 from config import Config
-from defs import DownloadResult, QUALITIES
+from defs import DownloadResult, QUALITIES, LOOKAHEAD_WATCH_RESCAN_DELAY_MIN, LOOKAHEAD_WATCH_RESCAN_DELAY_MAX, SCAN_CANCEL_KEYSTROKE
+from input import wait_for_key
 from logger import Log
 from path_util import file_already_exists_arr
 from vinfo import VideoInfo, get_min_max_ids
@@ -42,28 +44,56 @@ def __init__(self, sequence: list[VideoInfo], func: Callable[[VideoInfo], Corout
         self._func = func
         self._seq = deque(sequence)
 
-        self._orig_count = len(self._seq)
+        self._orig_count = len(self._original_sequence)
         self._scan_count = 0
         self._404_counter = 0
+        self._last_non404_id = self._original_sequence[0].id - 1
         self._extra_ids: list[int] = list()
         self._scanned_items = deque[VideoInfo]()
         self._task_finish_callback: Callable[[VideoInfo, DownloadResult], Coroutine[Any, Any, None]] | None = None
 
+        self._sleep_waiter: Optional[Task] = None
+        self._abort_waiter: Optional[Task] = None
+        self._abort = False
+
         self._id_gaps: list[tuple[int, int]] = list()
 
-    def _extend_with_extra(self) -> None:
-        extra_cur = Config.lookahead - self._404_counter
+    def _on_abort(self) -> None:
+        Log.warn('[queue] scanner thread interrupted, finishing pending tasks...')
+        if self._sleep_waiter:
+            self._sleep_waiter.cancel()
+            self._sleep_waiter: Optional[Task] = None
+        self._abort_waiter: Optional[Task] = None
+        self._abort = True
+
+    @staticmethod
+    async def _sleep_task(sleep_time: int) -> None:
+        try:
+            await sleep(float(sleep_time))
+        except CancelledError:
+            pass
+
+    async def _extend_with_extra(self) -> int:
+        watcher_mode = Config.lookahead < 0 and not not self._extra_ids and self._404_counter >= abs(Config.lookahead)
+        extra_cur = (abs(Config.lookahead) - self._404_counter) if not watcher_mode else abs(Config.lookahead)
         if extra_cur > 0:
-            last_id = Config.end_id + len(self._extra_ids)
+            last_id = (Config.end_id + len(self._extra_ids)) if not watcher_mode else self._last_non404_id
             extra_idseq = [(last_id + i + 1) for i in range(extra_cur)]
             extra_vis = [VideoInfo(idi) for idi in extra_idseq]
             minid, maxid = get_min_max_ids(extra_vis)
-            Log.warn(f'[lookahead] extending queue after {last_id:d} with {extra_cur:d} extra ids: {minid:d}-{maxid:d}')
             self._seq.extend(extra_vis)
             self._original_sequence.extend(extra_vis)
             self._extra_ids.extend(extra_idseq)
-
-    async def _at_scan_finish(self, vi: VideoInfo, result: DownloadResult) -> None:
+            if not watcher_mode:
+                Log.warn(f'[lookahead] extending queue after {last_id:d} with {extra_cur:d} extra ids: {minid:d}-{maxid:d}')
+            else:
+                rescan_delay = min(LOOKAHEAD_WATCH_RESCAN_DELAY_MAX, max(abs(Config.lookahead) * 12, LOOKAHEAD_WATCH_RESCAN_DELAY_MIN))
+                Log.warn(f'[watcher] extending queue after {last_id:d} with {extra_cur:d} extra ids: {minid:d}-{maxid:d}'
+                         f' (waiting {rescan_delay:d} seconds before rescan)')
+                return rescan_delay
+        return 0
+
+    async def _at_scan_finish(self, vi: VideoInfo, result: DownloadResult) -> int:
         self._scan_count += 1
         if result in (DownloadResult.FAIL_NOT_FOUND, DownloadResult.FAIL_RETRIES,
                       DownloadResult.FAIL_DELETED, DownloadResult.FAIL_FILTERED_OUTER, DownloadResult.FAIL_SKIPPED):
@@ -84,15 +114,29 @@ async def _at_scan_finish(self, vi: VideoInfo, result: DownloadResult) -> None:
             self._id_gaps.append((vi.id - self._404_counter, vi.id))
 
         self._404_counter = self._404_counter + 1 if result == DownloadResult.FAIL_NOT_FOUND else 0
+        if result != DownloadResult.FAIL_NOT_FOUND:
+            self._last_non404_id = vi.id
         if len(self._seq) == 1 and not not Config.lookahead:
-            self._extend_with_extra()
+            return await self._extend_with_extra()
+        return 0
 
     async def run(self) -> None:
         Log.debug('[queue] scanner thread start')
+        self._abort_waiter = get_running_loop().create_task(wait_for_key(SCAN_CANCEL_KEYSTROKE, self._on_abort))
         while self._seq:
+            if self._abort:
+                self._seq.clear()
+                continue
             result = await self._func(self._seq[0])
-            await self._at_scan_finish(self._seq[0], result)
+            sleep_time = await self._at_scan_finish(self._seq[0], result)
             self._seq.popleft()
+            if sleep_time:
+                self._sleep_waiter = get_running_loop().create_task(self._sleep_task(sleep_time))
+                await self._sleep_waiter
+                self._sleep_waiter = None
+        if self._abort_waiter:
+            self._abort_waiter.cancel()
+            self._abort_waiter = None
         Log.debug('[queue] scanner thread stop: scan complete')
         if self._id_gaps:
             gap_strings = list()

src/input.py

@@ -0,0 +1,66 @@
+# coding=UTF-8
+"""
+Author: trickerer (https://github.com/trickerer, https://github.com/trickerer01)
+"""
+#########################################
+# Original solution by Bharel: https://stackoverflow.com/a/70664652
+#
+
+from asyncio import CancelledError, sleep
+from collections.abc import Callable
+from contextlib import nullcontext, contextmanager
+from platform import system
+
+__all__ = ('wait_for_key',)
+
+if system() == 'Windows':
+    # noinspection PyCompatibility
+    from msvcrt import getwch, kbhit
+
+    set_terminal_raw = nullcontext
+    input_ready = kbhit
+    next_input = getwch
+else:
+    # TODO: test on Linux
+    import sys
+    from functools import partial
+    from select import select
+    from termios import tcgetattr, tcsetattr, TCSADRAIN
+    from tty import setraw
+
+    @contextmanager
+    def set_terminal_raw() -> None:
+        fd = sys.stdin.fileno()
+        old_settings = tcgetattr(fd)
+        try:
+            setraw(sys.stdin.fileno())
+            yield
+        finally:
+            tcsetattr(fd, TCSADRAIN, old_settings)
+
+    def input_ready() -> bool:
+        return select([sys.stdin], [], [], 0) == ([sys.stdin], [], [])
+
+    next_input = partial(sys.stdin.read, 1)
+
+
+async def wait_for_key(key: str, callback: Callable[[], None], *, secondary=False) -> None:
+    try:
+        with set_terminal_raw():
+            ch = ''
+            while ch != key:
+                await sleep(1.0)
+                while input_ready():
+                    ch = next_input()
+            if secondary:
+                while input_ready():
+                    next_input()
+                callback()
+            else:
+                await wait_for_key(key, callback, secondary=True)
+    except CancelledError:
+        pass
+
+#
+#
+#########################################

src/validators.py

@@ -74,11 +74,12 @@ def find_and_resolve_config_conflicts() -> bool:
     return delay_for_message
 
 
-def valid_int(val: str, *, lb: int = None, ub: int = None) -> int:
+def valid_int(val: str, *, lb: int = None, ub: int = None, nonzero=False) -> int:
     try:
         val = int(val)
         assert lb is None or val >= lb
         assert ub is None or val <= ub
+        assert nonzero is False or val != 0
         return val
     except Exception:
         raise ArgumentError
@@ -88,6 +89,10 @@ def positive_int(val: str) -> int:
     return valid_int(val, lb=0)
 
 
+def nonzero_int(val: str) -> int:
+    return valid_int(val, nonzero=True)
+
+
 def positive_nonzero_int(val: str) -> int:
     return valid_int(val, lb=1)
 

src/version.py

@@ -9,7 +9,7 @@
 APP_NAME = 'NM'
 APP_VER_MAJOR = '1'
 APP_VER_SUB = '8'
-APP_REVISION = '461'
+APP_REVISION = '462'
 APP_VERSION = f'{APP_VER_MAJOR}.{APP_VER_SUB}.{APP_REVISION}'
 
 #