2651086069

Build Type

push

github

Committed by mborsetti

Commit Message

Version 3.10.3rc0

Run Details

1394 of 2002 branches covered (69.63%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 2 files covered. (100.0%)

426 existing lines in 8 files now uncovered.

3273 of 4049 relevant lines covered (80.83%)

6.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

56.44

/webchanges/cli.py

#!/usr/bin/env python3

"""Module containing the entry point: the function main()."""

# See config module for the command line arguments.

# The code below is subject to the license contained in the LICENSE file, which is part of the source code.

from __future__ import annotations

import logging
import os
import shutil
import signal
import subprocess
import sys
import warnings
from pathlib import Path, PurePath
from typing import List, Optional

import platformdirs

from . import __copyright__, __docs_url__, __min_python_version__, __project_name__, __version__
from .config import CommandConfig
from .util import get_new_version_number

# Ignore signal SIGPIPE ("broken pipe") for stdout (see https://github.com/thp/urlwatch/issues/77)
if sys.platform != 'win32':  # Windows does not have signal.SIGPIPE
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)  # type: ignore[attr-defined]  # not defined in Windows

logger = logging.getLogger(__name__)


def python_version_warning() -> None:
    """Check if we're running on the minimum Python version supported and if so print and issue a pending deprecation
    warning."""
    if sys.version_info[0:2] == __min_python_version__:
        current_minor_version = '.'.join(str(n) for n in sys.version_info[0:2])
        next_minor_version = f'{__min_python_version__[0]}.{__min_python_version__[1] + 1}'
        warning = (
            f'Support for Python {current_minor_version} will be ending three years from the date Python '
            f'{next_minor_version} was released'
        )
        print(f'WARNING: {warning}\n')
        PendingDeprecationWarning(warning)


def migrate_from_legacy(
    legacy_package: str, config_file: Path, jobs_file: Path, hooks_file: Path, cache_file: Path
) -> None:
    """Check for existence of legacy files for configuration, jobs and Python hooks and migrate them (i.e. make a copy
    to new folder and/or name). Original files are not deleted.

    :param legacy_package: The name of the legacy package to migrate (e.g. urlwatch).
    :param config_file: The Path to the configuration file.
    :param jobs_file: The Path to the jobs file.
    :param hooks_file: The Path to the hooks file.
    :param cache_file: The Path to the snapshot database file.
    """
    lg_project_path = Path.home().joinpath(f'.{legacy_package}')
    lg_config_file = lg_project_path.joinpath(f'{legacy_package}.yaml')
    lg_urls_file = lg_project_path.joinpath('urls.yaml')
    lg_hooks_file = lg_project_path.joinpath('hooks.py')
    lg_cache_path = platformdirs.user_cache_path(legacy_package)
    lg_cache_file = lg_cache_path.joinpath('cache.db')
    for old_file, new_file in zip(
        (lg_config_file, lg_urls_file, lg_hooks_file, lg_cache_file), (config_file, jobs_file, hooks_file, cache_file)
    ):
        if old_file.is_file() and not new_file.is_file():
            new_file.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(old_file, new_file)
            logger.warning(f"Copied {legacy_package} '{old_file}' file to {__project_name__} '{new_file}'.")
            logger.warning(f"You can safely delete '{old_file}'.")


def setup_logger(verbose: Optional[int] = None) -> None:
    """Set up the logger.

    :param verbose: the verbosity level (1 = INFO, 2 = ERROR).
    """
    import platform

    log_level = None
    if verbose is not None:
        if verbose >= 2:
            log_level = 'DEBUG'
        elif verbose == 1:
            log_level = 'INFO'

    if not verbose:
        sys.tracebacklimit = 0

    logging.basicConfig(format='%(asctime)s %(module)s[%(thread)s] %(levelname)s: %(message)s', level=log_level)
    logger.info(f'{__project_name__}: {__version__} {__copyright__}')
    logger.info(
        f'{platform.python_implementation()}: {platform.python_version()} '
        f'{platform.python_build()} {platform.python_compiler()}'
    )
    logger.info(f'System: {platform.platform()}')


def locate_jobs_files(filename: Path, default_path: Path, ext: Optional[str] = None) -> List[Path]:
    """Searches for file both as specified and in the default directory, then retries with 'ext' extension if defined.

    :param filename: The filename.
    :param default_path: The default directory.
    :param ext: The extension, e.g. '.yaml', to add for searching if first scan fails.

    :returns: The filename, either original or one with path where found and/or extension.
    """
    search_filenames = [filename]

    # if ext is given, iterate both on raw filename and the filename with ext if different
    if ext and filename.suffix != ext:
        search_filenames.append(filename.with_suffix(ext))

    # try as given
    for file in search_filenames:
        # https://stackoverflow.com/questions/56311703/globbing-absolute-paths-with-pathlib
        file_list = list(Path(file.anchor).glob(str(file.relative_to(file.anchor))))
        if any(f.is_file() for f in file_list):
            return file_list

        # no directory specified (and not in current one): add default one
        if not file.is_absolute() and not Path(file).parent == Path.cwd():
            file_list = list(default_path.glob(str(file)))
            if any(f.is_file() for f in file_list):
                return file_list

    # no matches found
    return [filename]


def locate_storage_file(filename: Path, default_path: Path, ext: Optional[str] = None) -> Path:
    """Searches for file both as specified and in the default directory, then retries with 'ext' extension if defined.

    :param filename: The filename.
    :param default_path: The default directory.
    :param ext: The extension, e.g. '.yaml', to add for searching if first scan fails.

    :returns: The filename, either original or one with path where found and/or extension.
    """
    search_filenames = [filename]

    # if ext is given, iterate both on raw filename and the filename with ext if different
    if ext and filename.suffix != ext:
        search_filenames.append(filename.with_suffix(ext))

    for file in search_filenames:
        # return if found
        if file.is_file():
            return file

        # no directory specified (and not in current one): add default one
        if file.parent == PurePath('.'):
            new_file = default_path.joinpath(file)
            if new_file.is_file():
                return new_file

    # no matches found
    return filename


def first_run(command_config: CommandConfig) -> None:
    """Create configuration and jobs files.

    :param command_config: the CommandConfig containing the command line arguments selected.
    """
    if not command_config.config_file.is_file():
        command_config.config_file.parent.mkdir(parents=True, exist_ok=True)
        from .storage import YamlConfigStorage

        YamlConfigStorage.write_default_config(command_config.config_file)
        print(f'Created default config file at {command_config.config_file}')
        if not command_config.edit_config:
            print(f'> Edit it with {__project_name__} --edit-config')
    if not any(f.is_file() for f in command_config.jobs_files):
        command_config.jobs_files[0].parent.mkdir(parents=True, exist_ok=True)
        command_config.jobs_files[0].write_text(f'# {__project_name__} jobs file. See {__docs_url__}\n')
        command_config.edit = True
        print(f'Created default jobs file at {command_config.jobs_files}')
        if not command_config.edit:
            print(f'> Edit it with {__project_name__} --edit')


def handle_unitialized_actions(urlwatch_config: CommandConfig) -> None:
    """Handles CLI actions that do not require all classes etc. to be initialized.  For speed purposes."""

    def _exit(arg: object) -> None:
        logger.info(f'Exiting with exit code {arg}')
        sys.exit(arg)

    def print_new_version() -> int:
        """Will print alert message if a newer version is found on PyPi."""
        print(f'{__project_name__} {__version__}', end='')
        new_release = get_new_version_number(timeout=2)
        if new_release:
            print(
                f'. Release version {new_release} is available; we recommend updating using e.g. '
                f"'pip install -U {__project_name__}'."
            )
            return 0
        elif new_release == '':
            print('. You are running the latest release.')
            return 0
        else:
            print('. Error contacting PyPI to determine the latest release.')
            return 1

    def playwright_install_chrome() -> int:  # pragma: no cover
        """
        Replicates playwright.___main__.main() function, which is called by the playwright executable, in order to
        install the browser executable.

        :return: Playwright's executable return code.
        """
        try:
            from playwright._impl._driver import compute_driver_executable
        except ImportError:
            raise ImportError('Python package playwright is not installed; cannot install the Chrome browser') from None

        driver_executable = compute_driver_executable()
        env = os.environ.copy()
        env['PW_CLI_TARGET_LANG'] = 'python'
        cmd = [str(driver_executable), 'install', 'chrome']
        logger.info(f"Running playwright CLI: {' '.join(cmd)}")
        completed_process = subprocess.run(cmd, env=env, capture_output=True, text=True)
        if completed_process.returncode:
            print(completed_process.stderr)
            return completed_process.returncode
        if completed_process.stdout:
            logger.info(f'Success! Output of Playwright CLI: {completed_process.stdout}')
        return 0

    if urlwatch_config.check_new:
        _exit(print_new_version())

    if urlwatch_config.install_chrome:  # pragma: no cover
        _exit(playwright_install_chrome())


def main() -> None:  # pragma: no cover
    """The entry point run when __name__ == '__main__'.

    Contains all the high-level logic to instantiate all classes that run the program.

    :raises NotImplementedError: If a `--database-engine` is specified that is not supported.
    :raises RuntimeError: If `--database-engine redis` is selected but `--cache` with a redis URI is not provided.
    """
    # Make sure that PendingDeprecationWarning are displayed from all modules (otherwise only those in __main__ are)
    warnings.filterwarnings('default', category=PendingDeprecationWarning)

    # Issue deprecation warning if running on minimum version supported
    python_version_warning()

    # Path where the config, jobs and hooks files are located
    if sys.platform != 'win32':
        config_path = platformdirs.user_config_path(__project_name__)  # typically ~/.config/{__project_name__}
    else:
        config_path = Path.home().joinpath('Documents').joinpath(__project_name__)

    # Path where the database is located; typically ~/.cache/{__project_name__}
    # or %LOCALAPPDATA%\{__project_name__}\{__project_name__}\Cache
    cache_path = platformdirs.user_cache_path(__project_name__)

    # Default config, jobs, hooks and cache (database) files
    default_config_file = config_path.joinpath('config.yaml')
    default_jobs_file = config_path.joinpath('jobs.yaml')
    default_hooks_file = config_path.joinpath('hooks.py')
    default_cache_file = cache_path.joinpath('cache.db')

    # Check for and if found migrate legacy (urlwatch 2.25) files
    migrate_from_legacy('urlwatch', default_config_file, default_jobs_file, default_hooks_file, default_cache_file)

    # Parse command line arguments
    command_config = CommandConfig(
        sys.argv[1:],
        __project_name__,
        config_path,
        default_config_file,
        default_jobs_file,
        default_hooks_file,
        default_cache_file,
    )

    # Set up the logger to verbose if needed
    setup_logger(command_config.verbose)

    # For speed, run these here
    handle_unitialized_actions(command_config)

    # Only now we can load other modules (so that logging is enabled)
    from .command import UrlwatchCommand
    from .main import Urlwatch
    from .storage import (
        CacheDirStorage,
        CacheRedisStorage,
        CacheSQLite3Storage,
        CacheStorage,
        YamlConfigStorage,
        YamlJobsStorage,
    )

    # Locate config, job and hooks files
    command_config.config_file = locate_storage_file(command_config.config_file, command_config.config_path, '.yaml')
    command_config.jobs_files = locate_jobs_files(command_config.jobs_def_file, command_config.config_path, '.yaml')
    command_config.hooks_file = locate_storage_file(command_config.hooks_file, command_config.config_path, '.py')

    # Check for first run
    if command_config.config_file == default_config_file and not Path(command_config.config_file).is_file():
        first_run(command_config)

    # Setup config file API
    config_storage = YamlConfigStorage(command_config.config_file)  # storage.py

    # Setup database API
    if command_config.database_engine == 'sqlite3':
        cache_storage: CacheStorage = CacheSQLite3Storage(
            command_config.cache, command_config.max_snapshots
        )  # storage.py
    elif any(str(command_config.cache).startswith(prefix) for prefix in ('redis://', 'rediss://')):
        cache_storage = CacheRedisStorage(command_config.cache)  # storage.py
    elif command_config.database_engine == 'redis':
        raise RuntimeError("A redis URI (starting with 'redis://' or 'rediss://' needs to be specified with --cache")
    elif command_config.database_engine == 'textfiles':
        cache_storage = CacheDirStorage(command_config.cache)  # storage.py
    elif command_config.database_engine == 'minidb':
        from .storage_minidb import CacheMiniDBStorage  # legacy code imported only if needed (requires minidb)

        cache_storage = CacheMiniDBStorage(command_config.cache)  # storage.py
    else:
        raise NotImplementedError(f'Database engine {command_config.database_engine} not implemented')

    # Setup jobs file API
    jobs_storage = YamlJobsStorage(command_config.jobs_files)  # storage.py

    # Setup 'urlwatch'
    urlwatcher = Urlwatch(command_config, config_storage, cache_storage, jobs_storage)  # main.py
    urlwatch_command = UrlwatchCommand(urlwatcher)  # command.py

    # Run 'urlwatch', starting with processing command line arguments
    urlwatch_command.run()


if __name__ == '__main__':
    main()

1	#!/usr/bin/env python3
2
3	"""Module containing the entry point: the function main()."""	8✔
4
5	# See config module for the command line arguments.
6
7	# The code below is subject to the license contained in the LICENSE file, which is part of the source code.
8
9	from __future__ import annotations	8✔
10
11	import logging	8✔
12	import os	8✔
13	import shutil	8✔
14	import signal	8✔
15	import subprocess	8✔
16	import sys	8✔
17	import warnings	8✔
18	from pathlib import Path, PurePath	8✔
19	from typing import List, Optional	8✔
20
21	import platformdirs	8✔
22
23	from . import __copyright__, __docs_url__, __min_python_version__, __project_name__, __version__	8✔
24	from .config import CommandConfig	8✔
25	from .util import get_new_version_number	8✔
26
27	# Ignore signal SIGPIPE ("broken pipe") for stdout (see https://github.com/thp/urlwatch/issues/77)
28	if sys.platform != 'win32': # Windows does not have signal.SIGPIPE	8!
29	signal.signal(signal.SIGPIPE, signal.SIG_DFL) # type: ignore[attr-defined] # not defined in Windows	8✔
30
31	logger = logging.getLogger(__name__)	8✔
32
33
34	def python_version_warning() -> None:	8✔
35	"""Check if we're running on the minimum Python version supported and if so print and issue a pending deprecation
36	warning."""
37	if sys.version_info[0:2] == __min_python_version__:	8✔
38	current_minor_version = '.'.join(str(n) for n in sys.version_info[0:2])	2✔
39	next_minor_version = f'{__min_python_version__[0]}.{__min_python_version__[1] + 1}'	2✔
40	warning = (	2✔
41	f'Support for Python {current_minor_version} will be ending three years from the date Python '
42	f'{next_minor_version} was released'
43	)
44	print(f'WARNING: {warning}\n')	2✔
45	PendingDeprecationWarning(warning)	2✔
46
47
48	def migrate_from_legacy(	8✔
49	legacy_package: str, config_file: Path, jobs_file: Path, hooks_file: Path, cache_file: Path
50	) -> None:
51	"""Check for existence of legacy files for configuration, jobs and Python hooks and migrate them (i.e. make a copy
52	to new folder and/or name). Original files are not deleted.
53
54	:param legacy_package: The name of the legacy package to migrate (e.g. urlwatch).
55	:param config_file: The Path to the configuration file.
56	:param jobs_file: The Path to the jobs file.
57	:param hooks_file: The Path to the hooks file.
58	:param cache_file: The Path to the snapshot database file.
59	"""
60	lg_project_path = Path.home().joinpath(f'.{legacy_package}')	8✔
61	lg_config_file = lg_project_path.joinpath(f'{legacy_package}.yaml')	8✔
62	lg_urls_file = lg_project_path.joinpath('urls.yaml')	8✔
63	lg_hooks_file = lg_project_path.joinpath('hooks.py')	8✔
64	lg_cache_path = platformdirs.user_cache_path(legacy_package)	8✔
65	lg_cache_file = lg_cache_path.joinpath('cache.db')	8✔
66	for old_file, new_file in zip(	8✔
67	(lg_config_file, lg_urls_file, lg_hooks_file, lg_cache_file), (config_file, jobs_file, hooks_file, cache_file)
68	):
69	if old_file.is_file() and not new_file.is_file():	8!
70	new_file.parent.mkdir(parents=True, exist_ok=True)	×
71	shutil.copyfile(old_file, new_file)	×
72	logger.warning(f"Copied {legacy_package} '{old_file}' file to {__project_name__} '{new_file}'.")	×
73	logger.warning(f"You can safely delete '{old_file}'.")	×
74
75
76	def setup_logger(verbose: Optional[int] = None) -> None:	8✔
77	"""Set up the logger.
78
79	:param verbose: the verbosity level (1 = INFO, 2 = ERROR).
80	"""
81	import platform	8✔
82
83	log_level = None	8✔
84	if verbose is not None:	8!
85	if verbose >= 2:	×
86	log_level = 'DEBUG'	×
87	elif verbose == 1:	×
88	log_level = 'INFO'	×
89
90	if not verbose:	8!
91	sys.tracebacklimit = 0	8✔
92
93	logging.basicConfig(format='%(asctime)s %(module)s[%(thread)s] %(levelname)s: %(message)s', level=log_level)	8✔
94	logger.info(f'{__project_name__}: {__version__} {__copyright__}')	8✔
95	logger.info(	8✔
96	f'{platform.python_implementation()}: {platform.python_version()} '
97	f'{platform.python_build()} {platform.python_compiler()}'
98	)
99	logger.info(f'System: {platform.platform()}')	8✔
100
101
102	def locate_jobs_files(filename: Path, default_path: Path, ext: Optional[str] = None) -> List[Path]:	8✔
103	"""Searches for file both as specified and in the default directory, then retries with 'ext' extension if defined.
104
105	:param filename: The filename.
106	:param default_path: The default directory.
107	:param ext: The extension, e.g. '.yaml', to add for searching if first scan fails.
108
109	:returns: The filename, either original or one with path where found and/or extension.
110	"""
UNCOV 111	search_filenames = [filename]	×
112
113	# if ext is given, iterate both on raw filename and the filename with ext if different
UNCOV 114	if ext and filename.suffix != ext:	×
UNCOV 115	search_filenames.append(filename.with_suffix(ext))	×
116
117	# try as given
UNCOV 118	for file in search_filenames:	×
119	# https://stackoverflow.com/questions/56311703/globbing-absolute-paths-with-pathlib
UNCOV 120	file_list = list(Path(file.anchor).glob(str(file.relative_to(file.anchor))))	×
UNCOV 121	if any(f.is_file() for f in file_list):	×
UNCOV 122	return file_list	×
123
124	# no directory specified (and not in current one): add default one
UNCOV 125	if not file.is_absolute() and not Path(file).parent == Path.cwd():	×
UNCOV 126	file_list = list(default_path.glob(str(file)))	×
UNCOV 127	if any(f.is_file() for f in file_list):	×
UNCOV 128	return file_list	×
129
130	# no matches found
UNCOV 131	return [filename]	×
132
133
134	def locate_storage_file(filename: Path, default_path: Path, ext: Optional[str] = None) -> Path:	8✔
135	"""Searches for file both as specified and in the default directory, then retries with 'ext' extension if defined.
136
137	:param filename: The filename.
138	:param default_path: The default directory.
139	:param ext: The extension, e.g. '.yaml', to add for searching if first scan fails.
140
141	:returns: The filename, either original or one with path where found and/or extension.
142	"""
143	search_filenames = [filename]	8✔
144
145	# if ext is given, iterate both on raw filename and the filename with ext if different
146	if ext and filename.suffix != ext:	8!
147	search_filenames.append(filename.with_suffix(ext))	8✔
148
149	for file in search_filenames:	8✔
150	# return if found
151	if file.is_file():	8!
UNCOV 152	return file	×
153
154	# no directory specified (and not in current one): add default one
155	if file.parent == PurePath('.'):	8!
156	new_file = default_path.joinpath(file)	8✔
157	if new_file.is_file():	8!
158	return new_file	×
159
160	# no matches found
161	return filename	8✔
162
163
164	def first_run(command_config: CommandConfig) -> None:	8✔
165	"""Create configuration and jobs files.
166
167	:param command_config: the CommandConfig containing the command line arguments selected.
168	"""
169	if not command_config.config_file.is_file():	8!
170	command_config.config_file.parent.mkdir(parents=True, exist_ok=True)	8✔
171	from .storage import YamlConfigStorage	8✔
172
173	YamlConfigStorage.write_default_config(command_config.config_file)	8✔
174	print(f'Created default config file at {command_config.config_file}')	8✔
175	if not command_config.edit_config:	8!
176	print(f'> Edit it with {__project_name__} --edit-config')	8✔
177	if not any(f.is_file() for f in command_config.jobs_files):	8!
178	command_config.jobs_files[0].parent.mkdir(parents=True, exist_ok=True)	8✔
179	command_config.jobs_files[0].write_text(f'# {__project_name__} jobs file. See {__docs_url__}\n')	8✔
180	command_config.edit = True	8✔
181	print(f'Created default jobs file at {command_config.jobs_files}')	8✔
182	if not command_config.edit:	8!
UNCOV 183	print(f'> Edit it with {__project_name__} --edit')	×
184
185
186	def handle_unitialized_actions(urlwatch_config: CommandConfig) -> None:	8✔
187	"""Handles CLI actions that do not require all classes etc. to be initialized. For speed purposes."""
188
UNCOV 189	def _exit(arg: object) -> None:	×
UNCOV 190	logger.info(f'Exiting with exit code {arg}')	×
UNCOV 191	sys.exit(arg)	×
192
UNCOV 193	def print_new_version() -> int:	×
194	"""Will print alert message if a newer version is found on PyPi."""
UNCOV 195	print(f'{__project_name__} {__version__}', end='')	×
UNCOV 196	new_release = get_new_version_number(timeout=2)	×
UNCOV 197	if new_release:	×
UNCOV 198	print(	×
199	f'. Release version {new_release} is available; we recommend updating using e.g. '
200	f"'pip install -U {__project_name__}'."
201	)
UNCOV 202	return 0	×
UNCOV 203	elif new_release == '':	×
UNCOV 204	print('. You are running the latest release.')	×
UNCOV 205	return 0	×
206	else:
UNCOV 207	print('. Error contacting PyPI to determine the latest release.')	×
UNCOV 208	return 1	×
209
210	def playwright_install_chrome() -> int: # pragma: no cover
211	"""
212	Replicates playwright.___main__.main() function, which is called by the playwright executable, in order to
213	install the browser executable.
214
215	:return: Playwright's executable return code.
216	"""
217	try:
218	from playwright._impl._driver import compute_driver_executable
219	except ImportError:
220	raise ImportError('Python package playwright is not installed; cannot install the Chrome browser') from None
221
222	driver_executable = compute_driver_executable()
223	env = os.environ.copy()
224	env['PW_CLI_TARGET_LANG'] = 'python'
225	cmd = [str(driver_executable), 'install', 'chrome']
226	logger.info(f"Running playwright CLI: {' '.join(cmd)}")
227	completed_process = subprocess.run(cmd, env=env, capture_output=True, text=True)
228	if completed_process.returncode:
229	print(completed_process.stderr)
230	return completed_process.returncode
231	if completed_process.stdout:
232	logger.info(f'Success! Output of Playwright CLI: {completed_process.stdout}')
233	return 0
234
UNCOV 235	if urlwatch_config.check_new:	×
UNCOV 236	_exit(print_new_version())	×
237
238	if urlwatch_config.install_chrome: # pragma: no cover
239	_exit(playwright_install_chrome())
240
241
242	def main() -> None: # pragma: no cover
243	"""The entry point run when __name__ == '__main__'.
244
245	Contains all the high-level logic to instantiate all classes that run the program.
246
247	:raises NotImplementedError: If a `--database-engine` is specified that is not supported.
248	:raises RuntimeError: If `--database-engine redis` is selected but `--cache` with a redis URI is not provided.
249	"""
250	# Make sure that PendingDeprecationWarning are displayed from all modules (otherwise only those in __main__ are)
251	warnings.filterwarnings('default', category=PendingDeprecationWarning)
252
253	# Issue deprecation warning if running on minimum version supported
254	python_version_warning()
255
256	# Path where the config, jobs and hooks files are located
257	if sys.platform != 'win32':
258	config_path = platformdirs.user_config_path(__project_name__) # typically ~/.config/{__project_name__}
259	else:
260	config_path = Path.home().joinpath('Documents').joinpath(__project_name__)
261
262	# Path where the database is located; typically ~/.cache/{__project_name__}
263	# or %LOCALAPPDATA%\{__project_name__}\{__project_name__}\Cache
264	cache_path = platformdirs.user_cache_path(__project_name__)
265
266	# Default config, jobs, hooks and cache (database) files
267	default_config_file = config_path.joinpath('config.yaml')
268	default_jobs_file = config_path.joinpath('jobs.yaml')
269	default_hooks_file = config_path.joinpath('hooks.py')
270	default_cache_file = cache_path.joinpath('cache.db')
271
272	# Check for and if found migrate legacy (urlwatch 2.25) files
273	migrate_from_legacy('urlwatch', default_config_file, default_jobs_file, default_hooks_file, default_cache_file)
274
275	# Parse command line arguments
276	command_config = CommandConfig(
277	sys.argv[1:],
278	__project_name__,
279	config_path,
280	default_config_file,
281	default_jobs_file,
282	default_hooks_file,
283	default_cache_file,
284	)
285
286	# Set up the logger to verbose if needed
287	setup_logger(command_config.verbose)
288
289	# For speed, run these here
290	handle_unitialized_actions(command_config)
291
292	# Only now we can load other modules (so that logging is enabled)
293	from .command import UrlwatchCommand
294	from .main import Urlwatch
295	from .storage import (
296	CacheDirStorage,
297	CacheRedisStorage,
298	CacheSQLite3Storage,
299	CacheStorage,
300	YamlConfigStorage,
301	YamlJobsStorage,
302	)
303
304	# Locate config, job and hooks files
305	command_config.config_file = locate_storage_file(command_config.config_file, command_config.config_path, '.yaml')
306	command_config.jobs_files = locate_jobs_files(command_config.jobs_def_file, command_config.config_path, '.yaml')
307	command_config.hooks_file = locate_storage_file(command_config.hooks_file, command_config.config_path, '.py')
308
309	# Check for first run
310	if command_config.config_file == default_config_file and not Path(command_config.config_file).is_file():
311	first_run(command_config)
312
313	# Setup config file API
314	config_storage = YamlConfigStorage(command_config.config_file) # storage.py
315
316	# Setup database API
317	if command_config.database_engine == 'sqlite3':
318	cache_storage: CacheStorage = CacheSQLite3Storage(
319	command_config.cache, command_config.max_snapshots
320	) # storage.py
321	elif any(str(command_config.cache).startswith(prefix) for prefix in ('redis://', 'rediss://')):
322	cache_storage = CacheRedisStorage(command_config.cache) # storage.py
323	elif command_config.database_engine == 'redis':
324	raise RuntimeError("A redis URI (starting with 'redis://' or 'rediss://' needs to be specified with --cache")
325	elif command_config.database_engine == 'textfiles':
326	cache_storage = CacheDirStorage(command_config.cache) # storage.py
327	elif command_config.database_engine == 'minidb':
328	from .storage_minidb import CacheMiniDBStorage # legacy code imported only if needed (requires minidb)
329
330	cache_storage = CacheMiniDBStorage(command_config.cache) # storage.py
331	else:
332	raise NotImplementedError(f'Database engine {command_config.database_engine} not implemented')
333
334	# Setup jobs file API
335	jobs_storage = YamlJobsStorage(command_config.jobs_files) # storage.py
336
337	# Setup 'urlwatch'
338	urlwatcher = Urlwatch(command_config, config_storage, cache_storage, jobs_storage) # main.py
339	urlwatch_command = UrlwatchCommand(urlwatcher) # command.py
340
341	# Run 'urlwatch', starting with processing command line arguments
342	urlwatch_command.run()
343
344
345	if __name__ == '__main__':
346	main()

mborsetti / webchanges / 2651086069

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous