diff --git a/qgis_deployment_toolbelt/jobs/generic_job.py b/qgis_deployment_toolbelt/jobs/generic_job.py index c86c5f1b..b41c2395 100644 --- a/qgis_deployment_toolbelt/jobs/generic_job.py +++ b/qgis_deployment_toolbelt/jobs/generic_job.py @@ -18,7 +18,11 @@ from sys import platform as opersys # package -from qgis_deployment_toolbelt.constants import OS_CONFIG, get_qdt_working_directory +from qgis_deployment_toolbelt.constants import ( + OS_CONFIG, + get_qdt_logs_folder, + get_qdt_working_directory, +) from qgis_deployment_toolbelt.exceptions import ( JobOptionBadName, JobOptionBadValue, @@ -60,7 +64,7 @@ def __init__(self) -> None: f"repositories/{getenv('QDT_TMP_RUNNING_SCENARIO_ID', 'default')}" ) self.qdt_plugins_folder = self.qdt_working_folder.joinpath("plugins") - + self.qdt_logs_folder = get_qdt_logs_folder() # destination profiles folder self.qgis_profiles_path: Path = Path(OS_CONFIG.get(opersys).profiles_path) if not self.qgis_profiles_path.exists(): diff --git a/qgis_deployment_toolbelt/jobs/job_autoclean.py b/qgis_deployment_toolbelt/jobs/job_autoclean.py new file mode 100644 index 00000000..6219706c --- /dev/null +++ b/qgis_deployment_toolbelt/jobs/job_autoclean.py @@ -0,0 +1,85 @@ +#! python3 # noqa: E265 + +""" + QDT autocleaner. + + Author: Julien Moura (https://github.com/guts) +""" + + +# ############################################################################# +# ########## Libraries ############# +# ################################## + +# Standard library +import logging +from pathlib import Path + +# package +from qgis_deployment_toolbelt.jobs.generic_job import GenericJob +from qgis_deployment_toolbelt.utils.file_stats import is_file_older_than +from qgis_deployment_toolbelt.utils.trash_or_delete import move_files_to_trash_or_delete + +# ############################################################################# +# ########## Globals ############### +# ################################## + +# logs +logger = logging.getLogger(__name__) + + +# ############################################################################# +# ########## Classes ############### +# ################################## + + +class JobAutoclean(GenericJob): + """ + Job to clean expired QDT resources (logs, plugins archives...) which are older than + a specified frequency. + """ + + ID: str = "qdt-autoclean" + OPTIONS_SCHEMA: dict = { + "delay": { + "type": int, + "required": False, + "default": 730, + "possible_values": range(1, 1000), + "condition": None, + }, + } + + def __init__(self, options: dict) -> None: + """Instantiate the class. + + :param dict options: job options. + """ + super().__init__() + self.options: dict = self.validate_options(options) + + def run(self) -> None: + """Execute job logic.""" + li_files_to_be_deleted: list[Path] = [] + + # clean logs + for log_file in self.qdt_logs_folder.glob("*.log"): + if is_file_older_than( + local_file_path=log_file, + expiration_rotating_hours=self.options.get("delay", 730), + ): + logger.debug(f"Autoclean - LOGS - Outdated file: {log_file}") + li_files_to_be_deleted.append(log_file) + + # clean plugins archives + for plugin_archive in self.qdt_plugins_folder.glob("*.zip"): + if is_file_older_than( + local_file_path=plugin_archive, + expiration_rotating_hours=self.options.get("delay", 730), + ): + logger.debug( + f"Autoclean - PLUGIN ARCHIVE - Outdated file: {plugin_archive}" + ) + li_files_to_be_deleted.append(plugin_archive) + + move_files_to_trash_or_delete(files_to_trash=li_files_to_be_deleted) diff --git a/qgis_deployment_toolbelt/utils/file_stats.py b/qgis_deployment_toolbelt/utils/file_stats.py new file mode 100644 index 00000000..3f5ba74f --- /dev/null +++ b/qgis_deployment_toolbelt/utils/file_stats.py @@ -0,0 +1,79 @@ +#! python3 # noqa: E265 + +"""Some helpers to work with file statistics (dates, etc.). + + Author: Julien Moura (https://github.com/guts) +""" + +# ############################################################################ +# ########## IMPORTS ############# +# ################################ + +# standard library +import logging +from datetime import datetime, timedelta +from pathlib import Path +from sys import platform as opersys +from typing import Literal + +# ############################################################################ +# ########## GLOBALS ############# +# ################################ + +# logs +logger = logging.getLogger(__name__) + +# ############################################################################ +# ########## FUNCTIONS ########### +# ################################ + + +def is_file_older_than( + local_file_path: Path, + expiration_rotating_hours: int = 24, + dt_reference_mode: Literal["auto", "creation", "modification"] = "auto", +) -> bool: + """Check if the creation/modification date of the specified file is older than the \ + mount of hours. + + Args: + local_file_path (Path): path to the file + expiration_rotating_hours (int, optional): number in hours to consider the \ + local file outdated. Defaults to 24. + dt_reference_mode (Literal['auto', 'creation', 'modification'], optional): + reference date type: auto to handle differences between operating systems, + creation for creation date, modification for last modification date. + Defaults to "auto". + + Returns: + bool: True if the creation/modification date of the file is older than the \ + specified number of hours. + """ + # modification date varies depending on operating system: on some systems (like + # Unix) creation date is the time of the last metadata change, and, on others + # (like Windows), is the creation time for path. + if dt_reference_mode == "auto" and opersys == "win32": + dt_reference_mode = "modification" + else: + dt_reference_mode = "creation" + + # get file reference datetime - modification or creation + if dt_reference_mode == "modification": + f_ref_dt = datetime.fromtimestamp(local_file_path.stat().st_mtime) + dt_type = "modified" + else: + f_ref_dt = datetime.fromtimestamp(local_file_path.stat().st_ctime) + dt_type = "created" + + if (datetime.now() - f_ref_dt) < timedelta(hours=expiration_rotating_hours): + logger.debug( + f"{local_file_path} has been {dt_type} less than " + f"{expiration_rotating_hours} hours ago." + ) + return False + else: + logger.debug( + f"{local_file_path} has been {dt_type} more than " + f"{expiration_rotating_hours} hours ago." + ) + return True diff --git a/qgis_deployment_toolbelt/utils/trash_or_delete.py b/qgis_deployment_toolbelt/utils/trash_or_delete.py new file mode 100644 index 00000000..7c91aa5b --- /dev/null +++ b/qgis_deployment_toolbelt/utils/trash_or_delete.py @@ -0,0 +1,88 @@ +#! python3 # noqa: E265 + +""" + QDT autocleaner. + + Author: Julien Moura (https://github.com/guts) +""" + + +# ############################################################################# +# ########## Libraries ############# +# ################################## + +# Standard library +import logging +from pathlib import Path + +# 3rd party library +from send2trash import TrashPermissionError, send2trash + +# ############################################################################# +# ########## Globals ############### +# ################################## + +# logs +logger = logging.getLogger(__name__) + + +# ############################################################################# +# ########## Functions ############# +# ################################## + + +def move_files_to_trash_or_delete( + files_to_trash: list[Path] | Path, + attempt: int = 1, +) -> None: + """Move files to the trash or directly delete them if it's not possible. + + Args: + files_to_trash (list[Path] | Path): list of file paths to move to the trash + attempt (int, optional): attempt (int): attempt count. If attempt < 2, it + tries a single batch operation. If attempt == 2, it works file per file. + Defaults to 1. + """ + # make sure it's a list + if isinstance(files_to_trash, Path): + files_to_trash = [ + files_to_trash, + ] + + # first try a batch + if attempt < 2: + try: + send2trash(paths=files_to_trash) + logger.info(f"{len(files_to_trash)} files have been moved to the trash.") + except Exception as err: + logger.error( + f"Moving {len(files_to_trash)} files to the trash in a single batch " + f"operation failed. Let's try it file per file. Trace: {err}" + ) + move_files_to_trash_or_delete(files_to_trash=files_to_trash, attempt=2) + else: + logger.debug( + f"Moving (or deleting) {len(files_to_trash)} files to trash: " "attempt 2" + ) + for file_to_trash in files_to_trash: + try: + send2trash(paths=file_to_trash) + logger.info(f"{file_to_trash} has been moved to the trash.") + except TrashPermissionError as err: + logger.warning( + f"Unable to move {file_to_trash} to the trash. " + f"Trace: {err}. Let's try to delete it directly." + ) + try: + file_to_trash.unlink(missing_ok=True) + logger.info(f"Deleting directly {file_to_trash} succeeded.") + except Exception as err: + logger.error( + f"An error occurred trying to delete {file_to_trash}. " + f"Trace: {err}" + ) + except Exception as err: + logger.error( + f"An error occurred trying to move {file_to_trash} to trash. " + f"Trace: {err}" + ) diff --git a/requirements/base.txt b/requirements/base.txt index 644dcbc2..07d1ddf2 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -6,4 +6,5 @@ packaging>=20,<24 pyyaml>=5.4,<7 pywin32==306 ; sys_platform == 'win32' requests>=2.31,<3 +send2trash[nativeLib]>=1.8.2,<1.9 typing-extensions>=4,<5 ; python_version < '3.11' diff --git a/tests/dev/dev_files_dates.py b/tests/dev/dev_files_dates.py new file mode 100644 index 00000000..e94a03d7 --- /dev/null +++ b/tests/dev/dev_files_dates.py @@ -0,0 +1,14 @@ +from pathlib import Path +from time import sleep + +test_file = Path(__file__).parent.parent.joinpath("fixtures/tmp/dev_files_dates.txt") +test_file.touch(exist_ok=True) + +sleep(30) + +print( + f"File.\nCreated: {test_file.stat().st_ctime}\nModified: {test_file.stat().st_mtime}" +) + + +test_file.unlink(missing_ok=True) diff --git a/tests/test_utils_file_stats.py b/tests/test_utils_file_stats.py new file mode 100644 index 00000000..f37c0876 --- /dev/null +++ b/tests/test_utils_file_stats.py @@ -0,0 +1,59 @@ +#! python3 # noqa E265 + +""" + Usage from the repo root folder: + + .. code-block:: bash + # for whole tests + python -m unittest tests.test_utils_file_stats + # for specific test + python -m unittest tests.test_utils_file_stats.TestUtilsFileStats.test_created_file_is_not_expired +""" + + +# standard library +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from time import sleep + +# project +from qgis_deployment_toolbelt.__about__ import __title_clean__, __version__ +from qgis_deployment_toolbelt.utils.file_stats import is_file_older_than + +# ############################################################################ +# ########## Classes ############# +# ################################ + + +class TestUtilsFileStats(unittest.TestCase): + """Test package metadata.""" + + def test_created_file_is_not_expired(self): + """Test file creation 'age' is OK.""" + with TemporaryDirectory( + f"{__title_clean__}_{__version__}_not_expired_" + ) as tempo_dir: + tempo_file = Path(tempo_dir, "really_recent_file.txt") + tempo_file.touch() + sleep(3) + self.assertFalse(is_file_older_than(Path(tempo_file))) + + def test_created_file_has_expired(self): + """Test file creation 'age' is too old.""" + with TemporaryDirectory( + prefix=f"{__title_clean__}_{__version__}_expired_" + ) as tempo_dir: + tempo_file = Path(tempo_dir, "not_so_really_recent_file.txt") + tempo_file.touch() + sleep(3) + self.assertTrue( + is_file_older_than(Path(tempo_file), expiration_rotating_hours=0) + ) + + +# ############################################################################ +# ####### Stand-alone run ######## +# ################################ +if __name__ == "__main__": + unittest.main()