From 0fbbd362979751e617ae91f522536f0b9ed28f0d Mon Sep 17 00:00:00 2001 From: Roland Date: Fri, 31 May 2024 23:54:00 -0600 Subject: [PATCH] Creating perpetual funciton which checks if the internet is accessible. This is used to prevent downloads from starting when no internet exists. --- main.py | 37 +++++++++++++++++--- modules/videos/downloader.py | 2 +- modules/videos/models.py | 7 ++-- modules/videos/video/lib.py | 64 ++++++++++++++++++----------------- wrolpi/common.py | 15 ++++++++ wrolpi/downloader.py | 5 +++ wrolpi/files/lib.py | 41 ++++++++++++++++++++++ wrolpi/files/test/test_lib.py | 27 +++++++++++++-- wrolpi/flags.py | 14 ++++++-- wrolpi/status.py | 2 +- wrolpi/vars.py | 3 ++ 11 files changed, 171 insertions(+), 46 deletions(-) diff --git a/main.py b/main.py index ed2f2ecb..a8c900f2 100755 --- a/main.py +++ b/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') diff --git a/modules/videos/downloader.py b/modules/videos/downloader.py index dc020868..79a83cd2 100755 --- a/modules/videos/downloader.py +++ b/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', diff --git a/modules/videos/models.py b/modules/videos/models.py index 5ae54269..922c99e5 100644 --- a/modules/videos/models.py +++ b/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) diff --git a/modules/videos/video/lib.py b/modules/videos/video/lib.py index 079ff7a7..6502266f 100644 --- a/modules/videos/video/lib.py +++ b/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)) diff --git a/wrolpi/common.py b/wrolpi/common.py index d01bb4d6..de12d714 100644 --- a/wrolpi/common.py +++ b/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 diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index 52d179b8..bd34d3fb 100644 --- a/wrolpi/downloader.py +++ b/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...') diff --git a/wrolpi/files/lib.py b/wrolpi/files/lib.py index cde4d6ed..9ae2163a 100644 --- a/wrolpi/files/lib.py +++ b/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) diff --git a/wrolpi/files/test/test_lib.py b/wrolpi/files/test/test_lib.py index a7a3747b..61263d93 100644 --- a/wrolpi/files/test/test_lib.py +++ b/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) diff --git a/wrolpi/flags.py b/wrolpi/flags.py index ca8d6778..8a9aeba6 100644 --- a/wrolpi/flags.py +++ b/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 diff --git a/wrolpi/status.py b/wrolpi/status.py index feccefdf..5c625378 100755 --- a/wrolpi/status.py +++ b/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: diff --git a/wrolpi/vars.py b/wrolpi/vars.py index 3e7a686e..328d2204 100644 --- a/wrolpi/vars.py +++ b/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', ''))