Skip to content

base operation

Bases: ABC

Abstract class for file operations

Source code in file_operations/file_operation.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class FileOperation(ABC):
    """Abstract class for file operations"""
    def __init__(self, settings: AppSettings, **kwargs):
        self.settings: AppSettings = settings
        self.command: str = kwargs.get("command", "operation")
        self.sleep: float = kwargs.get('sleep', settings.sleep)
        self.repeat: bool = kwargs.get('repeat', settings.repeat)
        self.files_for_task: Tuple[Union[Path]] = tuple()
        self.pattern: tuple = kwargs.get('pattern', settings.pattern)
        self.src: str = kwargs.get('src', '')
        self.dst: str = kwargs.get('dst', '')
        self.source_directory = Path(self.src)
        self.target_directory = self.dst
        self.stop: bool = False

        # -----логування-----
        log_file = self.command
        log_level = kwargs.get("log_level", settings.log_level)
        self.log_path = kwargs.get("log_path", settings.log_path)

        self.logger = LoggerConfigurator.setup(
            name=self.__class__.__name__,
            log_level=log_level,
            log_path=Path(self.log_path) / f"{log_file}.log" if self.log_path else None
        )
        self.logger.info(f"Started with parameters: {kwargs}")


    def get_files(self, source_directory: Path, pattern: Union[Tuple[str], Tuple[str, ...]]) -> Tuple[Path]:
        """Get files from source directory that match a set of patterns"""
        files = set()

        for p in pattern:
            current_pattern_files = source_directory.glob(f"*{p}*")
            files.update(current_pattern_files)

        files_for_task = tuple(files)
        self.logger.debug(f"Total files_for_task: {len(files_for_task)}")
        return files_for_task

    def check_source_directory(self) -> None:
        """Check if source directory is valid"""
        if not self.source_directory.exists():
            # print(f"[ERROR] Source path '{self.src}' does not exist.")
            self.logger.error(f"Source path '{self.src}' does not exist.")
            raise FileNotFoundError(f"Source path '{self.src}' does not exist.")

    def check_directories(self) -> None:
        """Check if source directory is valid and if target directory exists.
        If target directory does not exist - create it."""
        self.check_source_directory()
        self.target_directory.mkdir(parents=True, exist_ok=True)


    def run(self) -> None:
        """Run the file operation"""
        self.check_directories()
        while True:
            try:
                self.files_for_task = self.get_files(source_directory=self.source_directory, pattern=self.pattern)

                if len(self.files_for_task) == 0 and self.repeat:
                    self.logger.info(f"No files found for task'{self.pattern}'. Wait for {self.sleep} seconds...")
                    time.sleep(self.sleep)
                    continue

                self.do_task()

            except KeyboardInterrupt:
                self.stop = True
                self.logger.info(f"Ctrl+C pressed, stopping...")

            if self.stop or not self.repeat:
                self.logger.info(f"Finished\n{'-' * 10}\n")
                break

    @staticmethod
    @abstractmethod
    def add_arguments(settings: AppSettings, parser: argparse.ArgumentParser) -> None:
        """Add specific arguments for operation"""
        pass

    @abstractmethod
    def do_task(self):
        """Abstract method to do a task of a file operation"""
        pass

    @property
    def sleep(self):
        return self._sleep

    @sleep.setter
    def sleep(self, value):
        self._sleep = int(value)

    @property
    def pattern(self):
        return self._pattern

    @pattern.setter
    def pattern(self, value):
        if isinstance(value, str):
            self._pattern = (value, )
        else:
            self._pattern = value

    @property
    def stop(self) -> bool:
        return self.__stop

    @stop.setter
    def stop(self, value):
        self.__stop = value

    @property
    def target_directory(self):
        return self._target_directory

    @target_directory.setter
    def target_directory(self, value: Union[Path, str, None]) -> None:
        if value is None:
            self._target_directory = self.source_directory
        elif isinstance(value, Path):
            self._target_directory = value
        elif isinstance(value, str):
            self._target_directory = Path(value)
        else:
            self.logger.error(f"Target directory '{value}' is not valid. Got type '{type(value)}'")
            raise TypeError(f"Target directory '{value}' is not valid. Got type '{type(value)}'")

add_arguments(settings, parser) abstractmethod staticmethod

Add specific arguments for operation

Source code in file_operations/file_operation.py 
88
89
90
91
92
@staticmethod
@abstractmethod
def add_arguments(settings: AppSettings, parser: argparse.ArgumentParser) -> None:
    """Add specific arguments for operation"""
    pass

check_directories()

Check if source directory is valid and if target directory exists. If target directory does not exist - create it.

Source code in file_operations/file_operation.py 
59
60
61
62
63
def check_directories(self) -> None:
    """Check if source directory is valid and if target directory exists.
    If target directory does not exist - create it."""
    self.check_source_directory()
    self.target_directory.mkdir(parents=True, exist_ok=True)

check_source_directory()

Check if source directory is valid

Source code in file_operations/file_operation.py 
52
53
54
55
56
57
def check_source_directory(self) -> None:
    """Check if source directory is valid"""
    if not self.source_directory.exists():
        # print(f"[ERROR] Source path '{self.src}' does not exist.")
        self.logger.error(f"Source path '{self.src}' does not exist.")
        raise FileNotFoundError(f"Source path '{self.src}' does not exist.")

do_task() abstractmethod

Abstract method to do a task of a file operation

Source code in file_operations/file_operation.py 
94
95
96
97
@abstractmethod
def do_task(self):
    """Abstract method to do a task of a file operation"""
    pass

get_files(source_directory, pattern)

Get files from source directory that match a set of patterns

Source code in file_operations/file_operation.py 
40
41
42
43
44
45
46
47
48
49
50
def get_files(self, source_directory: Path, pattern: Union[Tuple[str], Tuple[str, ...]]) -> Tuple[Path]:
    """Get files from source directory that match a set of patterns"""
    files = set()

    for p in pattern:
        current_pattern_files = source_directory.glob(f"*{p}*")
        files.update(current_pattern_files)

    files_for_task = tuple(files)
    self.logger.debug(f"Total files_for_task: {len(files_for_task)}")
    return files_for_task

run()

Run the file operation

Source code in file_operations/file_operation.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def run(self) -> None:
    """Run the file operation"""
    self.check_directories()
    while True:
        try:
            self.files_for_task = self.get_files(source_directory=self.source_directory, pattern=self.pattern)

            if len(self.files_for_task) == 0 and self.repeat:
                self.logger.info(f"No files found for task'{self.pattern}'. Wait for {self.sleep} seconds...")
                time.sleep(self.sleep)
                continue

            self.do_task()

        except KeyboardInterrupt:
            self.stop = True
            self.logger.info(f"Ctrl+C pressed, stopping...")

        if self.stop or not self.repeat:
            self.logger.info(f"Finished\n{'-' * 10}\n")
            break