wartana/xp-probackup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Scope backup cleanup to specific job directories for safety and pass the job name from configuration.

Browse Source
This commit is contained in:
wwartana
2026-01-30 14:25:34 +08:00
parent 018ea9172a
commit 69e5990db0
1 changed files with 21 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
backup_manager.py
View File

@@ -219,21 +219,30 @@ class BackupStrategy(ABC):
def is_cancelled(self): def is_cancelled(self):
return self._cancel_flag return self._cancel_flag
def cleanup_old_backups(self, dest, retention_days): def cleanup_old_backups(self, dest, retention_days, job_name=None):
if retention_days <= 0: if retention_days <= 0:
return return
self.log("Menjalankan pembersihan (Retensi: {} hari)...".format(retention_days)) # SAFETY FIX: Scope cleanup to the specific Job Name folder
if job_name:
safe_job_name = "".join(c for c in job_name if c.isalnum() or c in (' ', '.', '_', '-')).strip()
scan_root = os.path.join(dest, safe_job_name)
else:
# If no job name provided, DO NOT scan the root dest to prevent deleting user files
# self.log("Warning: Nama pekerjaan tidak ada. Skip pembersihan demi keamanan.")
return
if not os.path.exists(scan_root):
return
self.log("Menjalankan pembersihan di {} (Retensi: {} hari)...".format(scan_root, retention_days))
now = time.time() now = time.time()
cutoff = now - (retention_days * 86400) cutoff = now - (retention_days * 86400)
if not os.path.exists(dest):
return
count = 0 count = 0
try: try:
# Recursive scan for YYYY/MM/DD structure # Recursive scan for YYYY/MM/DD structure
for root, dirs, files in os.walk(dest): for root, dirs, files in os.walk(scan_root):
for filename in files: for filename in files:
filepath = os.path.join(root, filename) filepath = os.path.join(root, filename)
try: try:
@@ -247,7 +256,7 @@ class BackupStrategy(ABC):
self.log("Gagal menghapus {}: {}".format(filename, e)) self.log("Gagal menghapus {}: {}".format(filename, e))
# Optional: Remove empty folders after cleanup # Optional: Remove empty folders after cleanup
for root, dirs, files in os.walk(dest, topdown=False): for root, dirs, files in os.walk(scan_root, topdown=False):
for name in dirs: for name in dirs:
d_path = os.path.join(root, name) d_path = os.path.join(root, name)
try: try:
@@ -376,14 +385,11 @@ class FileBackup(BackupStrategy):
dst_file = os.path.join(current_dest_dir, file) dst_file = os.path.join(current_dest_dir, file)
shutil.copy2(src_file, dst_file) shutil.copy2(src_file, dst_file)
processed_files += 1 # Retention
if total_files > 0: retention_days = config.get('retention_days', 0)
self.progress(int(processed_files / total_files * 100)) if retention_days > 0:
job_name = config.get('name')
self.log("Penyalinan berhasil diselesaikan.") self.cleanup_old_backups(dest, retention_days, job_name)
# Cleanup
self.cleanup_old_backups(dest, config.get('retention_days', 0))
self.progress(100) self.progress(100)