Coverage for /var/devmt/py/utils4_1.8.0/utils4/filesys.py: 100%

70 statements  

« prev     ^ index     » next       coverage.py v7.6.9, created at 2025-08-26 21:39 +0100

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3""" 

4:Purpose: This module contains tests and utilities relating to files and the 

5 filesystem. 

6 

7:Developer: J Berendt 

8:Email: development@s3dev.uk 

9 

10:Comments: n/a 

11 

12:Example: 

13 

14 Example for comparing two files:: 

15 

16 >>> from utils4 import filesys 

17 

18 >>> filesys.compare_files(file1='/path/to/file1.txt', 

19 file2='/path/to/file2.txt') 

20 True 

21 

22 

23 If the files are expected to have *different* line endings, yet the 

24 contents are otherwise expected to be the same, pass the ``contents_only`` 

25 argument as ``True``; as this will skip the file signature test:: 

26 

27 >>> from utils4 import filesys 

28 

29 >>> filesys.compare_files(file1='/path/to/file1.txt', 

30 file2='/path/to/file2.txt', 

31 contents_only=True) 

32 True 

33 

34""" 

35# pylint: disable=invalid-name 

36 

37import os 

38import shutil 

39import stat 

40from glob import glob 

41from utils4.reporterror import reporterror 

42try: 

43 from natsort import natsorted 

44 _IMP_NATSORT = True 

45except ImportError: 

46 # Built-in sorting will be used instead. 

47 _IMP_NATSORT = False 

48 

49_SIZE = 16*1024 # 16 KiB 

50 

51 

52def compare_files(file1: str, 

53 file2: str, 

54 encoding: str='utf-8', 

55 contents_only: bool=False, 

56 sig_only: bool=False) -> bool: 

57 """Test if two files are the same. 

58 

59 This method is *modelled* after the built-in :func:`~filecmp.cmp` function, 

60 yet has been modified to *ignore* line endings. Meaning, if two files have 

61 the same signature and the contents are the same, except for the line 

62 endings, a result of True is returned. 

63 

64 Args: 

65 file1 (str): Full path to a file to be tested. 

66 file2 (str): Full path to a file to be tested. 

67 encoding (str, optional): Encoding to be used when reading the files. 

68 Defaults to 'utf-8'. 

69 contents_only (bool, optional): Only compare the file contents, do not 

70 test the signatures. This is useful if the line endings are 

71 expected to be different, as a file with DOS line endings will be 

72 marginally larger than a file with UNIX line endings; meaning 

73 the file signature test will *fail*. Defaults to False. 

74 sig_only (bool, optional): Only compare the file signatures. The files' 

75 contents are *not* compared. Defaults to False. 

76 

77 :Tests: 

78 If any of the following tests fail, a value of False is returned 

79 immediately, and no further tests are conducted. 

80 

81 The following tests are conducted, given default function parameters: 

82 

83 - Test both files are 'regular' files. 

84 - Test the files have the same size (in bytes), they are both regular 

85 files and their inode mode is the same. 

86 - Test the contents are the same; ignoring line endings. 

87 

88 Returns: 

89 bool: True if *all* tests pass, indicating the files are the same; 

90 otherwise False. 

91 

92 """ 

93 if contents_only: 

94 return _compare_content(file1=file1, file2=file2, encoding=encoding) 

95 sig1 = _sig(file1) 

96 sig2 = _sig(file2) 

97 if sig1[1] != stat.S_IFREG | sig2[1] != stat.S_IFREG: 

98 return False 

99 if sig_only: 

100 # Only compare signatures. 

101 return sig1 == sig2 

102 if sig1 != sig2: 

103 # Shortcut to bypass file content compare. 

104 return False 

105 return _compare_content(file1=file1, file2=file2, encoding=encoding) 

106 

107def dirsplit(path: str, 

108 nfiles: int, 

109 pattern: str='*', 

110 pairs: bool=False, 

111 repl: tuple=(None,)) -> bool: 

112 """Move all files from a single directory into (n) sub-directories. 

113 

114 Args: 

115 path (str): Full path to the source files. Additionally, all files 

116 will be moved into sub-directories in this path. 

117 nfiles (int): Number of source files to be moved into each directory. 

118 pattern (str, optional): A shell-style wildcard pattern used for 

119 collecting the source files. For example: ``*.csv``. 

120 Defaults to '*'. 

121 pairs (bool, optional): Are the files in pairs?. If True, the ``repl`` 

122 argument is used to replace a sub-string of the source file with 

123 that of the paired file, so each file pair is moved into the same 

124 directory. Defaults to False. 

125 repl (tuple, optional): A tuple containing the old and new replacement 

126 strings. This argument is only in effect if the ``pairs`` argument 

127 is True. Defaults to (None,). 

128 

129 For example:: 

130 

131 ('_input.csv', '_output.txt') 

132 

133 Raises: 

134 FileNotFoundError: If the input file path does not exist. 

135 

136 Returns: 

137 bool: True if the operation completes, otherwise False. 

138 

139 """ 

140 if not os.path.exists(path): 

141 raise FileNotFoundError('The requested path does not exist.') 

142 success = False 

143 try: 

144 # Setup. 

145 files = [f for f in glob(os.path.join(path, pattern)) if os.path.isfile(f)] 

146 files = natsorted(files) if _IMP_NATSORT else sorted(files) 

147 total = len(files) 

148 i = nfiles 

149 dirnum = 0 

150 # File iterator. 

151 for idx, file in enumerate(files, 1): 

152 # Define the (next) copy-to directory and create it. 

153 if i >= nfiles: 

154 i = 0 

155 dirnum += 1 

156 dirnam = str(dirnum).zfill(2) 

157 dirpath = os.path.join(path, dirnam) 

158 if not os.path.exists(dirpath): 

159 os.mkdir(path=dirpath) 

160 # Copy source file. 

161 base = os.path.basename(file) 

162 dst = os.path.join(path, dirnam, base) 

163 print(f'Moving {idx} of {total}: {base} -> {dirnam}') 

164 shutil.move(src=file, dst=dst) 

165 _file_move_test(fpath=dst) 

166 if pairs: 

167 # Copy paired file. 

168 base2 = base.replace(*repl) 

169 dst2 = os.path.join(path, dirnam, base2) 

170 print(rf'\t\-- {base2} -> {dirnam}') 

171 shutil.move(src=os.path.join(path, base2), dst=dst2) 

172 _file_move_test(fpath=dst2) 

173 i += 1 

174 success = True 

175 except FileNotFoundError as ferr: # progma nocover (cannot test) 

176 # Designed to catch / print file move errors from _file_move_test(). 

177 print(ferr) 

178 except Exception as err: 

179 reporterror(err) 

180 return success 

181 

182def _compare_content(file1: str, file2: str, encoding: str='utf-8') -> bool: 

183 """Compare the content of each file. 

184 

185 Args: 

186 file1 (str): Full path to a file to be tested. 

187 file2 (str): Full path to a file to be tested. 

188 encoding (str, optional): Encoding to be used when reading the files. 

189 Defaults to 'utf-8'. 

190 

191 This function short-circuits once a difference is found and immediately 

192 returns False. 

193 

194 Returns: 

195 bool: True if the file contents are the same, otherwise False. 

196 

197 """ 

198 with open(file1, 'r', encoding=encoding) as f1, open(file2, 'r', encoding=encoding) as f2: 

199 while True: 

200 data1 = f1.read(_SIZE) 

201 data2 = f2.read(_SIZE) 

202 if data1 != data2: 

203 return False 

204 # Both files have reached EOF and are the same. 

205 if not data1 and not data2: 

206 return True 

207 

208def _file_move_test(fpath: str) -> bool: 

209 """Test a file exists. 

210 

211 This method is used to verify the subject file was moved successfully. 

212 

213 Args: 

214 fpath (str): File path to be tested. 

215 

216 Raises: 

217 FileNotFoundError: If the subject file does not exist. 

218 

219 Returns: 

220 bool: True if the file was moved successfully, otherwise False. 

221 

222 """ 

223 if not os.path.exists(fpath): 

224 msg = ('\nThe following file was not copied successfully. Processing aborted.\n' 

225 f'-- {fpath}\n') 

226 raise FileNotFoundError(msg) 

227 return True 

228 

229def _sig(file: str) -> tuple: 

230 """Build a tuple containing elements of a file's signature. 

231 

232 Args: 

233 file (str): Full path to the file to be tested. 

234 

235 Returns: 

236 tuple: A tuple containing elements of the file's signature, as:: 

237 

238 (file size, file type, inode mode) 

239 

240 """ 

241 st = os.stat(file) 

242 return (st.st_size, stat.S_IFMT(st.st_mode), st.st_mode)