|
| 1 | +from shutil import copytree, SameFileError, make_archive |
| 2 | +from datetime import datetime |
| 3 | +import os |
| 4 | +import contextlib |
| 5 | +from pathlib import Path |
| 6 | +import logzero |
| 7 | +from logzero import logger |
| 8 | + |
| 9 | + |
| 10 | +class Archiver: |
| 11 | + """Archive folders and files.""" |
| 12 | + |
| 13 | + def __init__(self, source_dir, archive_dir): |
| 14 | + """ |
| 15 | + Initialize logger and archive parameters. |
| 16 | +
|
| 17 | + Args: |
| 18 | + source_dir (str or Path): The directory to copy or archive. |
| 19 | + archive_dir (str or Path): The destination directory for the archive. |
| 20 | + """ |
| 21 | + self.source_dir = Path(source_dir).resolve() |
| 22 | + self.archive_dir = Path(archive_dir).resolve() |
| 23 | + |
| 24 | + if not self.source_dir.exists() or not self.source_dir.is_dir(): |
| 25 | + raise ValueError(f"Source directory '{self.source_dir}' does not exist or is not a directory.") |
| 26 | + logger.debug(f"Source directory: {self.source_dir}") |
| 27 | + |
| 28 | + if not self.archive_dir.parent.exists(): |
| 29 | + raise ValueError(f"Archive directory parent '{self.archive_dir.parent}' does not exist.") |
| 30 | + logger.debug(f"Archive directory: {self.archive_dir}") |
| 31 | + |
| 32 | + # Configure log file |
| 33 | + logzero.logfile(f"archive_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log") |
| 34 | + logger.info("Archive initialized.") |
| 35 | + |
| 36 | + def archive(self): |
| 37 | + """ |
| 38 | + Perform the archive by copying the source directory to the archive directory |
| 39 | + and then creating a zip file of the archived folder. |
| 40 | +
|
| 41 | + Raises: |
| 42 | + SameFileError: If the source and destination are the same. |
| 43 | + OSError: For other filesystem-related errors. |
| 44 | + """ |
| 45 | + logger.info("Starting archive process...") |
| 46 | + logger.info(f"Source: {self.source_dir}") |
| 47 | + logger.info(f"Destination: {self.archive_dir}") |
| 48 | + |
| 49 | + # Step 1: Copy the source directory to the archive directory |
| 50 | + try: |
| 51 | + copytree(self.source_dir, self.archive_dir) |
| 52 | + logger.info(f"Archive folder created successfully: {self.source_dir} -> {self.archive_dir}") |
| 53 | + except SameFileError as e: |
| 54 | + logger.error(f"Source and destination are the same: {e}") |
| 55 | + raise |
| 56 | + except FileExistsError: |
| 57 | + logger.warning(f"Archive destination already exists: {self.archive_dir}") |
| 58 | + except OSError as e: |
| 59 | + logger.error(f"Failed to complete archive: {e}") |
| 60 | + raise |
| 61 | + |
| 62 | + # Create a zip file for the archived folder |
| 63 | + try: |
| 64 | + zip_path = make_archive( |
| 65 | + base_name=str(self.archive_dir), # The base name of the archive |
| 66 | + format="zip", # Archive format |
| 67 | + root_dir=str(self.archive_dir), # Root directory to archive |
| 68 | + ) |
| 69 | + logger.info(f"Archive zipped successfully: {zip_path}") |
| 70 | + except OSError as e: |
| 71 | + logger.error(f"Failed to zip the archive: {e}") |
| 72 | + raise |
| 73 | + |
| 74 | + return zip_path |
| 75 | + |
| 76 | + @staticmethod |
| 77 | + def get_archive_dir(base_dir, project_name): |
| 78 | + """ |
| 79 | + Generate a timestamped archive directory path. |
| 80 | +
|
| 81 | + Args: |
| 82 | + base_dir (str or Path): The base directory where archives are stored. |
| 83 | + project_name (str): The name of the project being archived. |
| 84 | +
|
| 85 | + Returns: |
| 86 | + Path: A path to the timestamped archive directory. |
| 87 | + """ |
| 88 | + base_dir = Path(base_dir).resolve() |
| 89 | + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| 90 | + archive_dir = base_dir / f"{project_name}_archive_{timestamp}" |
| 91 | + logger.debug(f"Generated archive directory: {archive_dir}") |
| 92 | + return archive_dir |
0 commit comments