Creating perpetual funciton which checks if the internet is accessible. This is used to prevent downloads from starting when no internet exists.

main.py

@@ -12,12 +12,12 @@ from wrolpi import root_api  # noqa
 from wrolpi import tags
 from wrolpi.api_utils import api_app
 from wrolpi.common import logger, check_media_directory, set_log_level, limit_concurrent, \
-    cancel_refresh_tasks, cancel_background_tasks, get_wrolpi_config
+    cancel_refresh_tasks, cancel_background_tasks, get_wrolpi_config, can_connect_to_server
 from wrolpi.contexts import attach_shared_contexts, reset_shared_contexts, initialize_configs_contexts
 from wrolpi.dates import Seconds
 from wrolpi.downloader import import_downloads_config, download_manager, \
     perpetual_download_worker
-from wrolpi.vars import PROJECT_DIR, DOCKERIZED
+from wrolpi.vars import PROJECT_DIR, DOCKERIZED, INTERNET_SERVER
 from wrolpi.version import get_version_string
 
 logger = logger.getChild('wrolpi-main')
@@ -218,6 +218,7 @@ async def start_single_tasks(app: Sanic):
     # Start perpetual tasks.  DO NOT AWAIT!
     app.add_task(perpetual_check_db_is_up_worker())  # noqa
     app.add_task(perpetual_download_worker())  # noqa
+    app.add_task(perpetual_have_internet_worker())  # noqa
 
     await app.dispatch('wrolpi.periodic.start_video_missing_comments_download')
     await app.dispatch('wrolpi.periodic.bandwidth')
@@ -299,10 +300,28 @@ async def perpetual_check_db_is_up_worker():
         logger.info('periodic_check_db_is_up was cancelled...')
 
 
+async def perpetual_have_internet_worker():
+    try:
+        while True:
+            try:
+                if can_connect_to_server(INTERNET_SERVER):
+                    flags.have_internet.set()
+                    # Check hourly once we have internet.
+                    await asyncio.sleep(float(Seconds.hour))
+                else:
+                    # Check every minute until the internet is back.
+                    flags.have_internet.clear()
+                    await asyncio.sleep(float(Seconds.minute))
+            except Exception as e:
+                logger.error('Failed to check if internet is up', exc_info=e)
+
+            await asyncio.sleep(10)
+    except asyncio.CancelledError:
+        logger.info('perpetual_check_have_internet_worker was cancelled...')
+
+
 @api_app.signal('wrolpi.periodic.start_video_missing_comments_download')
 async def periodic_start_video_missing_comments_download():
-    logger.debug('periodic_start_video_missing_comments_download is running')
-
     from modules.videos.video.lib import get_missing_videos_comments
 
     async with flags.refresh_complete.wait_for():
@@ -314,9 +333,17 @@ async def periodic_start_video_missing_comments_download():
 
     # Fetch comments for videos every hour.
     if download_manager.disabled.is_set() or download_manager.stopped.is_set():
+        logger.debug('Waiting for downloads to be enabled before downloading comments...')
         await asyncio.sleep(10)
+    elif not flags.have_internet.is_set():
+        logger.debug('Waiting for internet before downloading comments...')
+        await asyncio.sleep(30)
     else:
-        await get_missing_videos_comments()
+        try:
+            await get_missing_videos_comments()
+        except Exception as e:
+            logger.error('Failed to get missing video comments', exc_info=e)
+
         await asyncio.sleep(int(Seconds.hour))
 
     await api_app.dispatch('wrolpi.periodic.start_video_missing_comments_download')

modules/videos/downloader.py

@@ -304,7 +304,7 @@ class VideoDownloader(Downloader, ABC):
             file_name_format = '%(uploader)s_%(upload_date)s_%(id)s_%(title)s.%(ext)s'
             cmd = (
                 str(YT_DLP_BIN),
-                '-cw',  # Continue downloads, do not clobber existing files.
+                '-c',  # Continue downloads
                 '-f', PREFERRED_VIDEO_FORMAT,
                 '--match-filter', '!is_live',  # Do not attempt to download Live videos.
                 '--write-subs',

modules/videos/models.py

@@ -9,10 +9,10 @@ from sqlalchemy.orm.collections import InstrumentedList
 
 from modules.videos.errors import UnknownVideo, UnknownChannel
 from wrolpi.captions import read_captions
-from wrolpi.common import Base, ModelHelper, logger, get_media_directory, background_task, truncate_object_bytes
+from wrolpi.common import Base, ModelHelper, logger, get_media_directory, background_task
 from wrolpi.db import get_db_curs, get_db_session, optional_session
 from wrolpi.downloader import Download, download_manager
-from wrolpi.files.lib import refresh_files, split_path_stem_and_suffix
+from wrolpi.files.lib import refresh_files, split_path_stem_and_suffix, replace_file
 from wrolpi.files.models import FileGroup
 from wrolpi.media_path import MediaPathType
 from wrolpi.tags import Tag, TagFile
@@ -487,7 +487,8 @@ class Video(ModelHelper, Base):
     def replace_info_json(self, info_json: dict):
         """Replace the info json file with the new json dict.  Handles adding new info_json file, if necessary."""
         info_json_path = self.info_json_path or self.video_path.with_suffix('.info.json')
-        info_json_path.write_text(json.dumps(info_json, indent=2))
+        info_json = json.dumps(info_json, indent=2)
+        replace_file(info_json_path, info_json, missing_ok=True)
 
         if not self.info_json_path:
             self.file_group.append_files(info_json_path)

modules/videos/video/lib.py

@@ -222,6 +222,8 @@ async def get_missing_videos_comments(limit: int = VIDEO_COMMENTS_FETCH_COUNT):
         download_manager.add_to_skip_list(video_url)
 
     for url in video_urls:
+        logger.info(f'Getting comments for: {url}')
+
         # Sleep to catch cancel.
         await asyncio.sleep(0)
 
@@ -241,42 +243,42 @@ async def get_missing_videos_comments(limit: int = VIDEO_COMMENTS_FETCH_COUNT):
                 download_manager.add_to_skip_list(url)
                 continue
 
-        tries = 5
-        info = None
-        while tries > 0:
-            try:
-                tries -= 1
-                # Get info json about the video.  Retry until we have a list of comments.
-                info = download_video_info_json(url)
-                if info and isinstance(info.get('comments'), list):
-                    break
-            except Exception as e:
-                if 'no longer available' in str(e):
-                    logger.error(f'Giving up on downloading comments because video does not exist: {url=}', exc_info=e)
-                    break
-                if 'Comments are turned off' in str(e):
-                    logger.error(f'Giving up on downloading comments because comments are disabled: {url=}', exc_info=e)
-                    break
-                if 'Private video' in str(e):
-                    logger.error(f'Giving up on downloading comments because video is private: {url=}', exc_info=e)
-                    break
-                if 'Video unavailable' in str(e):
-                    logger.error(f'Giving up on downloading comments because video is unavailable: {url=}', exc_info=e)
-                    break
-                logger.error(f'Got error when attempting to download video comments: {url=}', exc_info=e)
-                await asyncio.sleep(random.randint(1, 5))
+        try:
+            # Get info json about the video.  Retry until we have a list of comments.
+            info = download_video_info_json(url)
 
-        if not info or not isinstance(info.get('comments'), list):
-            logger.error(f'Never got comments for video: {url=}')
-            add_video_to_skip_list(url)
+            if not info or not isinstance(info.get('comments'), list):
+                logger.error(f'Unable to get comments for video: {url=}')
+                add_video_to_skip_list(url)
+                continue
+        except Exception as e:
+            if 'no longer available' in str(e):
+                logger.error(f'Giving up on downloading comments because video does not exist: {url=}', exc_info=e)
+            elif 'Comments are turned off' in str(e):
+                logger.error(f'Giving up on downloading comments because comments are disabled: {url=}', exc_info=e)
+            elif 'Private video' in str(e):
+                logger.error(f'Giving up on downloading comments because video is private: {url=}', exc_info=e)
+            elif 'Video unavailable' in str(e):
+                logger.error(f'Giving up on downloading comments because video is unavailable: {url=}', exc_info=e)
+            elif ' members ' in str(e):
+                logger.error(f'Giving up on downloading comments because video is members-only: {url=}', exc_info=e)
+            else:
+                logger.error(f'Got error when attempting to download video comments: {url=}', exc_info=e)
             continue
 
-        logger.debug(f'Got {len(info["comments"])} comments for Video: {url=}')
+        logger.info(f'Got {len(info["comments"])} comments for Video: {url=}')
 
         with get_db_session(commit=True) as session:
-            for video in session.query(Video).join(FileGroup).filter_by(url=url).all():
-                video.replace_info_json(info)
-                video.have_comments = True
+            videos = list(session.query(Video).join(FileGroup).filter_by(url=url).all())
+            for video in videos:
+                if video.video_path.is_file():
+                    video.replace_info_json(info)
+                    video.have_comments = True
+                else:
+                    logger.error(f'Attempting to replace comments for non-existent video!  {video}')
+                    if len(videos) == 1:
+                        download_manager.add_to_skip_list(video.url)
+                    video.comments_failed = True
 
         # Sleep a random amount of time, so we don't spam.
         await asyncio.sleep(random.randint(2, 20))

wrolpi/common.py

@@ -10,6 +10,7 @@ import os
 import pathlib
 import re
 import shutil
+import socket
 import string
 import sys
 import tempfile
@@ -1791,3 +1792,17 @@ def get_title_from_html(html: str, url: str = None) -> str:
         return soup.title.string.strip()
     except Exception:  # noqa
         logger_.debug(f'Unable to extract title {url}')
+
+
+def can_connect_to_server(hostname: str) -> bool:
+    """Return True only if `hostname` responds on port 80."""
+    # Thanks https://stackoverflow.com/questions/20913411/test-if-an-internet-connection-is-present-in-python
+    try:
+        # connect to the host -- tells us if the host is actually
+        # reachable
+        conn = socket.create_connection((hostname, 53))
+        conn.close()
+        return True
+    except Exception:
+        pass  # We ignore any errors, returning False
+    return False

wrolpi/downloader.py

@@ -51,6 +51,11 @@ async def perpetual_download_worker():
                 logger.warning('DownloadManager stopped, quitting...')
                 return
 
+            if not flags.have_internet.is_set():
+                logger.warning('DownloadManager waiting for internet...')
+                await asyncio.sleep(10)
+                continue
+
             await download_manager.do_downloads()
         except asyncio.CancelledError:
             logger.info('perpetual_download cancelled...')

wrolpi/files/lib.py

@@ -1298,6 +1298,47 @@ async def rename_file(path: pathlib.Path, new_name: str) -> pathlib.Path:
     return new_path
 
 
+def replace_file(file: pathlib.Path | str, contents: str | bytes, missing_ok: bool = False):
+    """Rename `file` to a temporary path, write contents to `file`, then, delete temporary file only if successful.
+    The original file will be restored if any of these steps fail."""
+    file = pathlib.Path(file)
+
+    if missing_ok is False and not file.is_file():
+        raise FileNotFoundError(f'Cannot replace non-existent file: {str(file)}')
+
+    temporary_path = pathlib.Path(str(file) + '.tmp')
+    if temporary_path.exists():
+        raise RuntimeError(f'Cannot replace file, temporary path already exists!  {str(temporary_path)}')
+
+    if not contents:
+        raise RuntimeError('Refusing to replace file with empty contents')
+
+    try:
+        if file.exists():
+            file.rename(temporary_path)
+        mode = 'wb' if isinstance(contents, bytes) else 'wt'
+        with file.open(mode) as fp:
+            fp.write(contents)
+        if file.exists() and temporary_path.exists():
+            # New contents have been written, delete the old file.
+            temporary_path.unlink()
+    except Exception as e:
+        logger.error(f'Failed to replace file: {str(file)}', exc_info=e)
+        if temporary_path.exists():
+            # Move original file back.
+            if file.exists():
+                file.unlink()
+            temporary_path.rename(file)
+        elif file.exists():
+            logger.error(f'Replacing file, but original file still exists: {str(file)}')
+        else:
+            logger.critical('Neither original, nor temporary file exist.  I am so sorry.')
+        raise
+    finally:
+        if file.is_file() and file.stat().st_size > 0 and temporary_path.is_file():
+            temporary_path.unlink()
+
+
 async def rename_directory(directory: pathlib.Path, new_name: str) -> pathlib.Path:
     """Rename a directory.  This is done by moving all files into the new directory, and removing the old directory."""
     new_directory = directory.with_name(new_name)

wrolpi/files/test/test_lib.py

@@ -839,7 +839,8 @@ async def test_move_deep_directory(test_async_client, test_session, test_directo
 
 
 @pytest.mark.asyncio
-async def test_move_directory(test_async_client, test_session, test_directory, make_files_structure, assert_directories):
+async def test_move_directory(test_async_client, test_session, test_directory, make_files_structure,
+                              assert_directories):
     """A Directory record is deleted when it's directory is deleted."""
     make_files_structure({
         'foo/one.txt': 'one',
@@ -890,7 +891,8 @@ async def test_move_error(test_session, test_directory, make_files_structure, vi
 
 
 @pytest.mark.asyncio
-async def test_file_group_move(test_async_client, test_session, make_files_structure, test_directory, video_bytes, srt_text):
+async def test_file_group_move(test_async_client, test_session, make_files_structure, test_directory, video_bytes,
+                               srt_text):
     """Test FileGroup's move method"""
     video, srt = make_files_structure({
         'video.mp4': video_bytes,
@@ -1049,3 +1051,24 @@ async def test_file_search_date_range(test_async_client, test_session, example_p
     files, total = lib.search_files('', 10, 0, to_year=2022)
     assert total == 1
     assert files[0]['id'] == pdf.id
+
+
+def test_replace_file(test_directory):
+    file = test_directory / 'foo.txt'
+    file.write_text('old contents')
+    assert file.read_text() == 'old contents'
+
+    lib.replace_file(file, 'new contents')
+    assert file.read_text() == 'new contents'
+    assert not (test_directory / 'foo.txt.tmp').exists()
+
+    lib.replace_file(file, b'new bytes')
+    assert file.read_bytes() == b'new bytes'
+    assert not (test_directory / 'foo.txt.tmp').exists()
+
+    # Default behavior is to refuse to replace a non-existent file.
+    with pytest.raises(FileNotFoundError):
+        lib.replace_file('does not exist', 'foo')
+
+    # Non-existent file can be overwritten.
+    lib.replace_file('does not exist', 'foo', missing_ok=True)

wrolpi/flags.py

@@ -114,6 +114,8 @@ outdated_zims = Flag('outdated_zims', store_db=True)
 kiwix_restart = Flag('kiwix_restart')
 # Map Importing
 map_importing = Flag('map_importing')
+# Used to disable or enable downloading when Internet is down.
+have_internet = Flag('have_internet')
 
 # Steps of refreshing.
 refresh_counting = Flag('refresh_counting')
@@ -130,10 +132,16 @@ def get_flags() -> dict:
     """Return a list of all Flags which are set."""
     flags = dict(
         db_up=db_up.is_set(),
-        refreshing=refreshing.is_set(),
-        refresh_complete=refresh_complete.is_set(),
-        outdated_zims=outdated_zims.is_set(),
         kiwix_restart=kiwix_restart.is_set(),
+        map_importing=map_importing.is_set(),
+        outdated_zims=outdated_zims.is_set(),
+        refresh_cleanup=refresh_cleanup.is_set(),
+        refresh_complete=refresh_complete.is_set(),
+        refresh_counting=refresh_counting.is_set(),
+        refresh_discovery=refresh_discovery.is_set(),
+        refresh_indexing=refresh_indexing.is_set(),
+        refresh_modeling=refresh_modeling.is_set(),
+        refreshing=refreshing.is_set(),
     )
     return flags
 

wrolpi/status.py

@@ -12,7 +12,7 @@ from typing import List, Optional, Tuple
 
 from wrolpi.api_utils import api_app
 from wrolpi.cmd import which
-from wrolpi.common import logger, limit_concurrent, get_warn_once
+from wrolpi.common import logger, get_warn_once
 from wrolpi.dates import now
 
 try:

wrolpi/vars.py

@@ -44,6 +44,9 @@ DOCKERIZED = truthy_arg(os.environ.get('DOCKER', ''))
 # tests are running
 PYTEST = 'pytest' in sys.modules
 
+# Used to test if the Internet is up.
+INTERNET_SERVER = "one.one.one.one"
+
 # running on circlci
 CIRCLECI = truthy_arg(os.environ.get('CIRCLECI', ''))