chore: Update backup-data.py to execute rotation even if the backup has already been done but the number of files is over the limit

src/common/cli/CLI.py

@@ -72,6 +72,10 @@ class CLI(ApiCaller):
 
         assert isinstance(self.__variables, dict), "Failed to get variables from database"
 
+        tz = getenv("TZ")
+        if tz:
+            self.__variables["TZ"] = tz
+
         self.__integration = get_integration()
         self.__use_redis = self.__get_variable("USE_REDIS", "no") == "yes"
         self.__redis = None

src/common/core/backup/bwcli/list.py

@@ -3,7 +3,6 @@
 from datetime import datetime
 from os.path import join, sep
 from sys import exit as sys_exit, path as sys_path
-from traceback import format_exc
 
 deps_path = join(sep, "usr", "share", "bunkerweb", "core", "backup")
 if deps_path not in sys_path:
@@ -17,19 +16,19 @@ try:
     if backups:
         message = f"Found {len(backups)} backup{'s' if len(backups) > 1 else ''} in {BACKUP_DIR} :"
         # Show a table with the backups details
-        message += "\n+------------+---------------------+"
-        message += "\n| Database   |        Date         |"
-        message += "\n+------------+---------------------+"
+        message += "\n+------------+--------------------------+"
+        message += "\n|  Database  |           Date           |"
+        message += "\n+------------+--------------------------+"
         for backup in backups:
             database = backup.name.split("-")[1]
-            date = datetime.strptime("-".join(backup.stem.split("-")[2:]), "%Y-%m-%d_%H-%M-%S")
-            message += f"\n| {database:<10} | {date.strftime('%Y/%m/%d %H:%M:%S %Z')} |"
-        message += "\n+------------+---------------------+"
+            date = datetime.strptime("-".join(backup.stem.split("-")[2:]), "%Y-%m-%d_%H-%M-%S").astimezone()
+            message += f"\n| {database:<10} | {date.strftime('%Y/%m/%d %H:%M:%S %Z'):<24} |"
+        message += "\n+------------+--------------------------+"
     else:
         message = f"No backup found in {BACKUP_DIR}"
     LOGGER.info(message)
 except SystemExit as se:
     sys_exit(se.code)
-except:
-    LOGGER.error(f"Error while executing backup list command :\n{format_exc()}")
+except BaseException as e:
+    LOGGER.error(f"Error while executing backup list command: {e}")
     sys_exit(1)

src/common/core/backup/bwcli/restore.py

@@ -5,7 +5,6 @@ from datetime import datetime
 from os.path import join, sep
 from pathlib import Path
 from sys import exit as sys_exit, path as sys_path
-from traceback import format_exc
 
 deps_path = join(sep, "usr", "share", "bunkerweb", "core", "backup")
 if deps_path not in sys_path:
@@ -58,8 +57,8 @@ try:
     restore_database(backup_file, db)
 except SystemExit as se:
     status = se.code
-except:
-    LOGGER.error(f"Error while executing backup restore command :\n{format_exc()}")
+except BaseException as e:
+    LOGGER.error(f"Error while executing backup restore command: {e}")
     status = 1
 finally:
     DB_LOCK_FILE.unlink(missing_ok=True)

src/common/core/backup/bwcli/save.py

@@ -5,13 +5,12 @@ from datetime import datetime
 from os.path import join, sep
 from pathlib import Path
 from sys import exit as sys_exit, path as sys_path
-from traceback import format_exc
 
 deps_path = join(sep, "usr", "share", "bunkerweb", "core", "backup")
 if deps_path not in sys_path:
     sys_path.append(deps_path)
 
-from utils import acquire_db_lock, backup_database, BACKUP_DIR, DB_LOCK_FILE, LOGGER
+from utils import acquire_db_lock, backup_database, BACKUP_DIR, DB_LOCK_FILE, LOGGER, update_cache_file
 
 status = 0
 
@@ -22,7 +21,13 @@ try:
     parser = ArgumentParser(description="BunkerWeb's backup plugin save command line interface")
 
     # Optional directory argument
-    parser.add_argument("-d", "--directory", default=BACKUP_DIR, type=str, help="directory where to save the backup")
+    parser.add_argument(
+        "-d",
+        "--directory",
+        default=BACKUP_DIR,
+        type=str,
+        help="directory where to save the backup, default is the one defined in the setting BACKUP_DIRECTORY",
+    )
 
     # Parse args
     args = parser.parse_args()
@@ -39,11 +44,14 @@ try:
         LOGGER.info(f"Creating directory {directory} as it does not exist")
         directory.mkdir(parents=True, exist_ok=True)
 
-    backup_database(datetime.now(), backup_dir=directory)
+    db = backup_database(datetime.now(), backup_dir=directory)
+
+    if directory == BACKUP_DIR:
+        update_cache_file(db, directory)
 except SystemExit as se:
     status = se.code
-except:
-    LOGGER.error(f"Error while executing backup save command :\n{format_exc()}")
+except BaseException as e:
+    LOGGER.error(f"Error while executing backup save command: {e}")
     status = 1
 finally:
     DB_LOCK_FILE.unlink(missing_ok=True)

src/common/core/backup/jobs/backup-data.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 from datetime import datetime, timedelta
-from json import dumps, loads
+from json import loads
 from os import getenv, sep
 from os.path import join
 from pathlib import Path
@@ -14,7 +14,7 @@ for deps_path in [join(sep, "usr", "share", "bunkerweb", *paths) for paths in ((
 
 from logger import setup_logger  # type: ignore
 from jobs import Job  # type: ignore
-from utils import backup_database
+from utils import backup_database, update_cache_file
 
 LOGGER = setup_logger("BACKUP", getenv("LOG_LEVEL", "INFO"))
 status = 0
@@ -33,7 +33,7 @@ try:
     last_backup = loads(JOB.get_cache("backup.json") or "{}")
     last_backup_date = last_backup.get("date", None)
     if last_backup_date:
-        last_backup_date = datetime.fromisoformat(last_backup_date)
+        last_backup_date = datetime.fromisoformat(last_backup_date).astimezone()
 
     current_time = datetime.now()
     backup_period = getenv("BACKUP_SCHEDULE", "daily")
@@ -43,15 +43,17 @@ try:
         "monthly": timedelta(weeks=4).total_seconds(),
     }
 
+    already_done = last_backup_date and last_backup_date.timestamp() + PERIOD_STAMPS[backup_period] > current_time.timestamp()
     backup_rotation = int(getenv("BACKUP_ROTATION", "7"))
 
-    # Get all backup files in the directory
-    backup_files = backup_dir.glob("backup-*.zip")
+    sorted_files = []
+    if already_done:
 
-    # Sort the backup files by name
-    sorted_files = sorted(backup_files)
+        # Get all backup files in the directory
+        backup_files = backup_dir.glob("backup-*.zip")
 
-    already_done = last_backup_date and last_backup_date.timestamp() + PERIOD_STAMPS[backup_period] > current_time.timestamp()
+        # Sort the backup files by name
+        sorted_files = sorted(backup_files)
 
     if len(sorted_files) <= backup_rotation and already_done:
         LOGGER.info(f"Backup already done within the last {backup_period} period, skipping backup ...")
@@ -66,6 +68,12 @@ try:
 
         backup_database(current_time, JOB.db, backup_dir)
 
+        # Get all backup files in the directory
+        backup_files = backup_dir.glob("backup-*.zip")
+
+        # Sort the backup files by name
+        sorted_files = sorted(backup_files)
+
     # Check if the number of backup files exceeds the rotation limit
     if len(sorted_files) > backup_rotation:
         # Calculate the number of files to remove
@@ -76,12 +84,7 @@ try:
             LOGGER.warning(f"Removing old backup file: {file}, as the rotation limit has been reached ...")
             file.unlink()
 
-    backup_files = sorted([file.name for file in backup_dir.glob("backup-*.zip")])
-
-    cached, err = JOB.cache_file("backup.json", dumps({"date": current_time.isoformat(), "files": backup_files}, indent=2).encode())
-    if not cached:
-        LOGGER.error(f"Failed to cache backup.json :\n{err}")
-        status = 2
+    update_cache_file(JOB.db, backup_dir)
 except SystemExit as e:
     status = e.code
 except:

src/common/core/backup/utils.py

@@ -38,6 +38,21 @@ def acquire_db_lock():
     DB_LOCK_FILE.touch()
 
 
+def update_cache_file(db: Database, backup_dir: Path) -> str:
+    """Update the cache file in the database."""
+    backup_data = loads(db.get_job_cache_file("backup-data", "backup.json") or "{}")
+    backup_data["files"] = sorted([file.name for file in backup_dir.glob("backup-*.zip")])
+    content = dumps(backup_data, indent=2).encode()
+    checksum = bytes_hash(content)
+    err = db.upsert_job_cache(None, "backup.json", content, job_name="backup-data", checksum=checksum)
+    if err:
+        LOGGER.error(f"Failed to update the backup.json cache file: {err}")
+        return err
+
+    LOGGER.info("Backup cache file updated successfully")
+    return ""
+
+
 def backup_database(current_time: datetime, db: Database = None, backup_dir: Path = BACKUP_DIR) -> Database:
     """Backup the database."""
     db = db or Database(LOGGER)
@@ -103,14 +118,6 @@ def backup_database(current_time: datetime, db: Database = None, backup_dir: Pat
 
     backup_file.chmod(0o600)
 
-    backup_data = loads(db.get_job_cache_file("backup-data", "backup.json") or "{}")
-    backup_data["files"] = sorted([file.name for file in backup_dir.glob("backup-*.zip")])
-    content = dumps(backup_data, indent=2).encode()
-    checksum = bytes_hash(content)
-    err = db.upsert_job_cache(None, "backup.json", content, job_name="backup-data", checksum=checksum)
-    if err:
-        LOGGER.error(f"Failed to update the backup.json cache file: {err}")
-
     LOGGER.info(f"💾 Backup {backup_file.name} created successfully in {backup_dir}")
     return db