Passed
Push — master ( ba6342...71f198 )
by P.R.
07:12
created

BackupScanner.__scan_directory_helper()   B

Complexity

Conditions 8

Size

Total Lines 31
Code Lines 20

Duplication

Lines 31
Ratio 100 %

Importance

Changes 0
Metric Value
cc 8
eloc 20
nop 4
dl 31
loc 31
rs 7.3333
c 0
b 0
f 0
1
"""
2
BackupPC Clone
3
"""
4
import csv
5
import os
6
import shutil
7
from typing import Optional
8
9
from backuppc_clone.Config import Config
10
from backuppc_clone.ProgressBar import ProgressBar
11
from backuppc_clone.helper.BackupInfoScanner import BackupInfoScanner
12
from backuppc_clone.style.BackupPcCloneStyle import BackupPcCloneStyle
13
14
15 View Code Duplication
class BackupScanner:
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
16
    """
17
    Helper class for scanning backup directories.
18
    """
19
20
    # ------------------------------------------------------------------------------------------------------------------
21
    def __init__(self, io: BackupPcCloneStyle):
22
        """
23
        Object constructor.
24
25
        :param BackupPcCloneStyle io: The output style.
26
        """
27
28
        self.__io: BackupPcCloneStyle = io
29
        """
30
        The output style.
31
        """
32
33
        self.__dir_count: int = 0
34
        """
35
        The file count.
36
        """
37
38
        self.__file_count: int = 0
39
        """
40
        The file count.
41
        """
42
43
        self.__entry_seq: int = 0
44
        """
45
        The entry sequence number.
46
        """
47
48
        self.progress: Optional[ProgressBar] = None
49
        """
50
        The progress counter.
51
        """
52
53
    # ------------------------------------------------------------------------------------------------------------------
54
    @property
55
    def dir_count(self) -> int:
56
        """
57
        Returns the number of found directories.
58
59
        :return: int
60
        """
61
        return self.__dir_count
62
63
    # ------------------------------------------------------------------------------------------------------------------
64
    @property
65
    def file_count(self) -> int:
66
        """
67
        Returns the number of found files.
68
69
        :return: int
70
        """
71
        return self.__file_count
72
73
    # ------------------------------------------------------------------------------------------------------------------
74
    def __scan_directory_helper(self, parent_dir: str, dir_name: str, csv_writer: csv.writer) -> None:
75
        """
76
        Scans recursively a list of directories and stores filenames and directories in CSV format.
77
78
        :param str parent_dir: The name of the parent directory.
79
        :param str dir_name: The name of the directory.
80
        :param csv.writer csv_writer: The CSV writer.
81
        """
82
        target_name = os.path.join(parent_dir, dir_name) if dir_name else parent_dir
83
84
        first_file = True
85
        sub_dir_names = []
86
        for entry in os.scandir(target_name):
87
            if entry.is_file():
88
                self.__file_count += 1
89
                if first_file:
90
                    first_file = False
91
                    self.__entry_seq += 1
92
                csv_writer.writerow((self.__entry_seq, entry.inode(), dir_name, entry.name))
93
94
                if entry.name not in ['attrib', 'backupInfo', 'backuppc-clone.csv']:
95
                    self.progress.advance()
96
97
            elif entry.is_dir():
98
                sub_dir_names.append(entry.name)
99
100
        for sub_dir_name in sorted(sub_dir_names):
101
            self.__entry_seq += 1
102
            self.__dir_count += 1
103
            csv_writer.writerow((self.__entry_seq, None, dir_name, sub_dir_name))
104
            self.__scan_directory_helper(parent_dir, os.path.join(dir_name, sub_dir_name), csv_writer)
105
106
    # ------------------------------------------------------------------------------------------------------------------
107
    def scan_directory(self, host: str, backup_no: int, csv_filename: str) -> None:
108
        """
109
        Scans recursively a list of directories and stores filenames and directories in CSV format.
110
111
        :param str host: The host name
112
        :param int backup_no: The backup number.
113
        :param str csv_filename: The filename of the CSV file.
114
        """
115
        self.__dir_count = 0
116
        self.__file_count = 0
117
        self.__entry_seq = 0
118
119
        backup_dir = Config.instance.backup_dir_original(host, backup_no)
120
121
        file_count = int(BackupInfoScanner.get_backup_info(backup_dir, 'nFiles'))
122
        self.progress = ProgressBar(self.__io.output, file_count)
123
124
        with open(csv_filename, 'w') as csv_file:
125
            csv_writer = csv.writer(csv_file)
126
            self.__io.writeln(' Scanning <fso>{}</fso>'.format(backup_dir))
127
            self.__io.writeln('')
128
            self.__scan_directory_helper(backup_dir, '', csv_writer)
129
            self.progress.finish()
130
131
    # ------------------------------------------------------------------------------------------------------------------
132
    def pre_scan_directory(self, host: str, backup_no: int) -> None:
133
        """
134
        Scans recursively a list of directories and stores filenames and directories in CSV format.
135
136
        :param str host: The host name
137
        :param int backup_no: The backup number.
138
        """
139
        self.__dir_count = 0
140
        self.__file_count = 0
141
        self.__entry_seq = 0
142
143
        backup_dir = Config.instance.backup_dir_original(host, backup_no)
144
145
        csv_filename1 = os.path.join(Config.instance.tmp_dir_clone, '.backup-{}-{}.csv'.format(host, backup_no))
146
        csv_filename2 = os.path.join(backup_dir, 'backuppc-clone.csv')
147
148
        file_count = int(BackupInfoScanner.get_backup_info(backup_dir, 'nFiles'))
149
        self.progress = ProgressBar(self.__io.output, file_count)
150
151
        with open(csv_filename1, 'w') as csv_file:
152
            csv_writer = csv.writer(csv_file)
153
            self.__io.writeln(' Scanning <fso>{}</fso>'.format(backup_dir))
154
            self.__io.writeln('')
155
            self.__scan_directory_helper(backup_dir, '', csv_writer)
156
            self.progress.finish()
157
158
        shutil.move(csv_filename1, csv_filename2)
159
        self.__io.writeln('')
160
        self.__io.writeln(' Wrote <fso>{}</fso>'.format(csv_filename2))
161
        self.__io.writeln('')
162
163
# ----------------------------------------------------------------------------------------------------------------------
164