Feature: add a job to autoclean QDT expired resources (logs, plugins archives...)

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():

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)

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

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}"
+                )

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'

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)

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()