11import os
2- import subprocess
3- import datetime
4- from dataclasses import dataclass , field
5- from typing import List
6- from tqdm import tqdm
72import sys
8- import time
9- import getpass
103import logging
11- from src .size_calculator import SizeCalculator
12- from src .old_delete import BackupDeletionManager
13- import hashlib
14- import tarfile
15-
16-
17- @dataclass
18- class BackupManager :
19- backup_folder : str = os .path .expanduser ("~/Documents/backup-for-cloud/" )
20- dirs_file_path : str = "dirs_to_backup.txt"
21- current_date : str = datetime .datetime .now ().strftime ("%d-%m-%Y" )
22- backup_file_path : str = field (init = False )
23- ignore_file_path : str = os .path .expanduser ("ignore.txt" )
24-
25- def __post_init__ (self ):
26- self .backup_file_path = os .path .expanduser (
27- f"{ self .backup_folder } /{ self .current_date } .tar.xz"
28- )
29-
30- def check_backup_exist (self ) -> bool :
31- """Check if a backup file already exists for today"""
32- return os .path .isfile (self .backup_file_path )
33-
34- def backup_directories (self ) -> bool :
35- """Backup the directories listed in dirs_to_backup.txt to a compressed file"""
36-
37- # Read in the directories to backup from the file
38- dirs_to_backup : List [str ] = []
39- with open (self .dirs_file_path , "r" , encoding = "utf-8" ) as file :
40- for line in file :
41- directory = line .strip ()
42- if directory :
43- dirs_to_backup .append (directory )
44-
45- # Read and expand paths in the ignore file
46- ignore_paths = []
47- if os .path .isfile (self .ignore_file_path ):
48- with open (self .ignore_file_path , "r" , encoding = "utf-8" ) as file :
49- for line in file :
50- ignore_path = line .strip ()
51- if ignore_path :
52- ignore_paths .append (os .path .expanduser (ignore_path ))
53-
54- # Generate the exclude options for the tar command
55- exclude_options = " " .join ([f"--exclude={ path } " for path in ignore_paths ])
56-
57- # Only backup files on the same filesystem as the backup folder
58- filesystem_option = "--one-file-system"
59-
60- # Expand the user's home directory for each directory to backup
61- dir_paths = [os .path .expanduser (path ) for path in dirs_to_backup ]
62-
63- # Calculate the total size using SizeCalculator
64- size_calculator = SizeCalculator (self .dirs_file_path , self .ignore_file_path )
65- total_size_bytes = size_calculator .calculate_total_backup_size (
66- dirs_to_backup , size_calculator .ignore_list
67- )
68-
69- # Convert the total size to MB and GiB
70- total_size_mb = total_size_bytes / (1024 * 1024 )
71- total_size_gib = total_size_bytes / (1024 * 1024 * 1024 )
72-
73- print (f"Total size: { total_size_mb :.2f} MB / { total_size_gib :.2f} GiB" )
74-
75- # Get the number of CPU threads for xz compression
76- cpu_threads = os .cpu_count () - 1
77- print (f"CPU threads - 1: { cpu_threads } " )
78-
79- # Create the tar command
80- os_cmd = (
81- f"tar -cf - { filesystem_option } { exclude_options } { ' ' .join (dir_paths )} | "
82- f"xz --threads={ cpu_threads } > { self .backup_file_path } "
83- )
84-
85- # Run the tar command and update the progress bar
86- try :
87- proc = subprocess .Popen (
88- os_cmd , shell = True , stdout = subprocess .PIPE , stderr = subprocess .PIPE
89- )
90- pbar = tqdm (
91- total = total_size_bytes ,
92- unit = "B" ,
93- unit_scale = True ,
94- desc = "Processing" ,
95- dynamic_ncols = True ,
96- )
97-
98- while proc .poll () is None :
99- if os .path .exists (self .backup_file_path ):
100- current_size = os .path .getsize (self .backup_file_path )
101- pbar .update (
102- current_size - pbar .n
103- ) # Update the progress bar with the difference
104- time .sleep (0.1 ) # Sleep briefly to avoid too frequent polling
105-
106- proc .wait ()
107- pbar .close ()
108-
109- if proc .returncode != 0 :
110- raise subprocess .CalledProcessError (proc .returncode , os_cmd )
111-
112- print ("Backup completed successfully" )
113- return True
114- except (
115- subprocess .CalledProcessError ,
116- FileNotFoundError ,
117- PermissionError ,
118- OSError ,
119- ValueError ,
120- ) as error :
121- print (f"Error backing up files: { type (error ).__name__ } - { error } " )
122- return False
123- except KeyboardInterrupt :
124- print ("Backup cancelled" )
125- sys .exit (0 )
126-
127- def list_backup_files (self , extension : str = ".tar.xz" ) -> List [str ]:
128- """List all backup files with the specified extension in the backup directory"""
129- try :
130- files = [f for f in os .listdir (self .backup_folder ) if f .endswith (extension )]
131- if not files :
132- print ("No backup files found." )
133- return []
134- for i , file in enumerate (files , start = 1 ):
135- print (f"{ i } . { file } " )
136- return files
137- except Exception as error :
138- print (f"Error listing backup files: { type (error ).__name__ } - { error } " )
139- return []
140-
141- def extract_backup (self , file_to_extract : str ) -> bool :
142- """Extract the backup file to the specified directory"""
143- date_str = os .path .basename (file_to_extract ).split ("." )[0 ]
144- extract_to = os .path .join (self .backup_folder , f"{ date_str } -extracted" )
145- if not os .path .exists (extract_to ):
146- os .makedirs (extract_to )
147-
148- def filter_function (tarinfo , path ):
149- # Customize the tarinfo object here if needed
150- return tarinfo
151-
152- try :
153- with tarfile .open (file_to_extract , "r:xz" ) as tar :
154- tar .extractall (path = extract_to , filter = filter_function )
155- print (f"Backup extracted to { extract_to } " )
156- return True
157- except (tarfile .TarError , FileNotFoundError , PermissionError ) as error :
158- print (f"Error extracting backup: { type (error ).__name__ } - { error } " )
159- return False
160- except KeyboardInterrupt :
161- print ("Extraction cancelled" )
162- sys .exit (0 )
163-
164-
165- @dataclass
166- class EncryptionManager (BackupManager ):
167- decrypt_file_path : str = os .path .expanduser (
168- "~/Documents/backup-for-cloud/decrypted.tar.xz"
169- )
170-
171- def encrypt_backup (self ) -> bool :
172- """Encrypt the backup file with openssl command"""
173- password = getpass .getpass (prompt = "Enter encryption password: " )
174-
175- # The encrypted backup file will be named with the current date
176- file_to_encrypt = os .path .join (
177- self .backup_folder , f"{ self .current_date } .tar.xz.enc"
178- )
179-
180- # Encrypt the backed up file with openssl command
181- encrypt_cmd = [
182- "openssl" ,
183- "aes-256-cbc" ,
184- "-a" ,
185- "-salt" ,
186- "-pbkdf2" ,
187- "-in" ,
188- self .backup_file_path ,
189- "-out" ,
190- file_to_encrypt ,
191- "-pass" ,
192- f"pass:{ password } " ,
193- ]
194-
195- try :
196- subprocess .run (encrypt_cmd , check = True )
197- logging .info ("Encryption completed successfully" )
198- return True
199- except subprocess .CalledProcessError as error :
200- logging .error (f"Error encrypting file: { error } " )
201- return False
202- except KeyboardInterrupt :
203- logging .info ("Encryption cancelled" )
204- sys .exit (0 )
205-
206- def decrypt (self , file_to_decrypt : str ) -> bool :
207- """Decrypt the backup file"""
208- password = getpass .getpass (prompt = "Enter decryption password: " )
209-
210- # Decrypt the backup file with openssl command
211- decrypt_cmd = [
212- "openssl" ,
213- "aes-256-cbc" ,
214- "-d" ,
215- "-a" ,
216- "-salt" ,
217- "-pbkdf2" ,
218- "-in" ,
219- file_to_decrypt ,
220- "-out" ,
221- self .decrypt_file_path ,
222- "-pass" ,
223- f"pass:{ password } " ,
224- ]
225-
226- try :
227- subprocess .run (decrypt_cmd , check = True )
228- time .sleep (1 )
229- logging .info ("Decryption completed successfully" )
230- return True
231- except subprocess .CalledProcessError as error :
232- logging .error (f"Error decrypting file: { error } " )
233- return False
234- except KeyboardInterrupt :
235- logging .info ("Decryption cancelled" )
236- sys .exit (0 )
237-
238- def verify_decrypt_file (self , file_to_decrypt : str ):
239- """Verify the decrypted file is the same as the original file"""
240- original_file_path = file_to_decrypt [:- 4 ]
241-
242- # Compute the SHA256 checksum of the decrypted file
243- hasher = hashlib .sha256 ()
244- with open (self .decrypt_file_path , "rb" ) as f :
245- while True :
246- data = f .read (65536 )
247- if not data :
248- break
249- hasher .update (data )
250- actual_checksum = hasher .hexdigest ()
251-
252- # Compute the SHA256 checksum of the original file
253- hasher_original = hashlib .sha256 ()
254- with open (original_file_path , "rb" ) as f :
255- while True :
256- data = f .read (65536 )
257- if not data :
258- break
259- hasher_original .update (data )
260- expected_checksum = hasher_original .hexdigest ()
261-
262- # Compare the checksums
263- if actual_checksum == expected_checksum :
264- logging .info ("File integrity verified: checksums match" )
265- else :
266- logging .error ("File integrity check failed: checksums do not match" )
267-
4+ from src .backup_manager import BackupManager
2685
2696# Configure logging
2707logging .basicConfig (
@@ -277,8 +14,12 @@ def main():
27714
27815 # Classes
27916 backup_manager = BackupManager ()
280- encryption_manager = EncryptionManager ()
281- deletion_manager = BackupDeletionManager ()
17+
18+ if not os .path .isfile (backup_manager .config_file_path ):
19+ backup_manager .ask_inputs ()
20+ backup_manager .save_credentials ()
21+ else :
22+ backup_manager .load_credentials ()
28223
28324 # Create a loop that will run until the user enters 6 to exit
28425 while True :
@@ -314,7 +55,7 @@ def main():
31455 else :
31556 print ("Backup failed." )
31657 elif choice == 2 :
317- encryption_manager .encrypt_backup ()
58+ backup_manager .encrypt_backup ()
31859 elif choice == 3 :
31960 # List all encrypted files
32061 print ("=====================================" )
@@ -331,10 +72,10 @@ def main():
33172 file_to_decrypt = os .path .join (
33273 backup_manager .backup_folder , files [choice - 1 ]
33374 )
334- encryption_manager .decrypt (file_to_decrypt )
335- encryption_manager .verify_decrypt_file (file_to_decrypt )
75+ backup_manager .decrypt (file_to_decrypt )
76+ backup_manager .verify_decrypt_file (file_to_decrypt )
33677 elif choice == 4 :
337- deletion_manager .delete_old_backups ()
78+ backup_manager .delete_old_backups ()
33879 elif choice == 5 :
33980 # List all tar.xz files
34081 print ("=====================================" )
0 commit comments