Coverage for src/ezcompiler/utils/_zip_utils.py: 94.51%
134 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 15:46 +0000
1# ///////////////////////////////////////////////////////////////
2# ZIP_UTILS - ZIP archive operation utilities
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7ZIP utilities - ZIP archive operation utilities for EzCompiler.
9This module provides utility functions for creating, extracting, and managing
10ZIP archives used in the EzCompiler project, with support for progress tracking
11and logging.
12"""
14from __future__ import annotations
16# ///////////////////////////////////////////////////////////////
17# IMPORTS
18# ///////////////////////////////////////////////////////////////
19# Standard library imports
20import stat
21import sys
22import zipfile
23from collections.abc import Callable
24from pathlib import Path
26# Local imports
27from ..shared.exceptions.utils import (
28 ZipCreationError,
29 ZipExtractionError,
30 ZipFileCorruptedError,
31 ZipFileNotFoundError,
32 ZipPathError,
33)
34from ._file_utils import FileUtils
36# ///////////////////////////////////////////////////////////////
37# CLASSES
38# ///////////////////////////////////////////////////////////////
41class ZipUtils:
42 """
43 ZIP archive operations utility class.
45 Provides static methods for creating, extracting, and managing
46 ZIP archives used in the EzCompiler project. Supports compression,
47 progress tracking, and detailed logging.
49 Class Variables:
50 _ezpl: Cached Ezpl instance for logging
51 _logger: Cached logger instance
52 _printer: Cached printer instance
54 Example:
55 >>> ZipUtils.create_zip_archive("./source", "./output.zip")
56 >>> ZipUtils.extract_zip_archive("./output.zip", "./extracted")
57 """
59 # Utils layer should not initialize logging directly
60 # Logging is handled by the service and interface layers
62 # ////////////////////////////////////////////////
63 # ARCHIVE CREATION METHODS
64 # ////////////////////////////////////////////////
66 @staticmethod
67 def create_zip_archive(
68 source_path: str | Path,
69 output_path: str | Path,
70 compression: int = zipfile.ZIP_DEFLATED,
71 include_hidden: bool = False,
72 progress_callback: Callable[[str, int], None] | None = None,
73 ) -> None:
74 """
75 Create a ZIP archive from a file or directory.
77 Args:
78 source_path: Path to the source file or directory
79 output_path: Path where the ZIP file will be created
80 compression: Compression method (default: ZIP_DEFLATED)
81 include_hidden: Whether to include hidden files (default: False)
82 progress_callback: Optional callback for progress updates (file, progress)
84 Raises:
85 FileOperationError: If ZIP creation fails
87 Note:
88 Progress callback receives (filename: str, progress: int) where
89 progress is a percentage from 0 to 100.
90 """
91 try:
92 source = Path(source_path)
93 output = Path(output_path)
95 if not source.exists():
96 raise ZipPathError(f"Source path does not exist: {source_path}")
98 # Ensure output directory exists
99 FileUtils.ensure_parent_directory_exists(output)
101 # Create the ZIP file
102 with zipfile.ZipFile(output, "w", compression) as zipf:
103 if source.is_file():
104 # Add single file
105 arcname = source.name
106 zipf.write(source, arcname)
107 if progress_callback:
108 progress_callback(str(source), 100)
110 elif source.is_dir(): 110 ↛ exitline 110 didn't jump to the function exit
111 # Add directory recursively with progress tracking
112 total_files = sum(1 for _ in source.rglob("*") if _.is_file())
113 processed_files = 0
115 for file_path in source.rglob("*"):
116 if file_path.is_file(): 116 ↛ 115line 116 didn't jump to line 115 because the condition on line 116 was always true
117 # Skip hidden files if not requested
118 if not include_hidden and ZipUtils._is_hidden_file(
119 file_path
120 ):
121 continue
123 # Calculate relative path for archive
124 arcname = file_path.relative_to(source)
125 zipf.write(file_path, arcname)
127 processed_files += 1
128 if progress_callback:
129 progress = int((processed_files / total_files) * 100)
130 progress_callback(str(file_path), progress)
132 except Exception as e:
133 raise ZipCreationError(
134 f"Failed to create ZIP archive {output_path}: {e}"
135 ) from e
137 # ////////////////////////////////////////////////
138 # ARCHIVE EXTRACTION METHODS
139 # ////////////////////////////////////////////////
141 @staticmethod
142 def extract_zip_archive(
143 zip_path: str | Path,
144 extract_path: str | Path,
145 password: str | None = None,
146 progress_callback: Callable[[str, int], None] | None = None,
147 ) -> None:
148 """
149 Extract a ZIP archive to a directory.
151 Args:
152 zip_path: Path to the ZIP file
153 extract_path: Directory where files will be extracted
154 password: Optional password for encrypted archives (default: None)
155 progress_callback: Optional callback for progress updates (file, progress)
157 Raises:
158 FileOperationError: If extraction fails
160 Note:
161 Progress callback receives (filename: str, progress: int) where
162 progress is a percentage from 0 to 100.
163 """
164 try:
165 zip_file = Path(zip_path)
166 extract_dir = Path(extract_path)
168 if not zip_file.exists():
169 raise ZipFileNotFoundError(f"ZIP file does not exist: {zip_path}")
171 if not zip_file.is_file():
172 raise ZipPathError(f"Path is not a file: {zip_path}")
174 # Create extraction directory
175 FileUtils.create_directory_if_not_exists(extract_dir)
177 # Extract the ZIP file
178 with zipfile.ZipFile(zip_file, "r") as zipf:
179 # Set password if provided
180 if password: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 zipf.setpassword(password.encode("utf-8"))
183 # Get list of files for progress tracking
184 file_list = zipf.namelist()
185 total_files = len(file_list)
187 for index, file_info in enumerate(zipf.filelist):
188 try:
189 zipf.extract(file_info, extract_dir)
191 if progress_callback:
192 progress = int(((index + 1) / total_files) * 100)
193 progress_callback(file_info.filename, progress)
195 except Exception as e:
196 raise ZipExtractionError(
197 f"Failed to extract file '{file_info.filename}' from ZIP archive {zip_path}: {e}"
198 ) from e
200 except Exception as e:
201 raise ZipExtractionError(
202 f"Failed to extract ZIP archive {zip_path}: {e}"
203 ) from e
205 # ////////////////////////////////////////////////
206 # ARCHIVE INFORMATION METHODS
207 # ////////////////////////////////////////////////
209 @staticmethod
210 def list_zip_contents(zip_path: str | Path) -> list[str]:
211 """
212 List the contents of a ZIP archive.
214 Args:
215 zip_path: Path to the ZIP file
217 Returns:
218 list[str]: List of file names in the archive
220 Raises:
221 ZipFileCorruptedError: If listing fails
222 """
223 try:
224 zip_file = Path(zip_path)
226 if not zip_file.exists():
227 raise ZipFileNotFoundError(f"ZIP file does not exist: {zip_path}")
229 with zipfile.ZipFile(zip_file, "r") as zipf:
230 return zipf.namelist()
232 except Exception as e:
233 raise ZipFileCorruptedError(
234 f"Failed to list ZIP contents {zip_path}: {e}"
235 ) from e
237 @staticmethod
238 def get_zip_info(zip_path: str | Path) -> dict:
239 """
240 Get information about a ZIP archive.
242 Args:
243 zip_path: Path to the ZIP file
245 Returns:
246 dict: Dictionary with ZIP information including:
247 - file_count: Number of files in the archive
248 - total_size: Total uncompressed size in bytes
249 - compressed_size: Total compressed size in bytes
250 - compression_ratio: Compression ratio as percentage
251 - files: List of file names in the archive
253 Raises:
254 ZipFileCorruptedError: If getting info fails
255 """
256 try:
257 zip_file = Path(zip_path)
259 if not zip_file.exists():
260 raise ZipFileNotFoundError(f"ZIP file does not exist: {zip_path}")
262 with zipfile.ZipFile(zip_file, "r") as zipf:
263 info = zipf.infolist()
265 total_size = sum(file_info.file_size for file_info in info)
266 compressed_size = sum(file_info.compress_size for file_info in info)
268 return {
269 "file_count": len(info),
270 "total_size": total_size,
271 "compressed_size": compressed_size,
272 "compression_ratio": (
273 (1 - compressed_size / total_size) * 100
274 if total_size > 0
275 else 0
276 ),
277 "files": [file_info.filename for file_info in info],
278 }
280 except Exception as e:
281 raise ZipFileCorruptedError(
282 f"Failed to get ZIP info {zip_path}: {e}"
283 ) from e
285 @staticmethod
286 def is_valid_zip(zip_path: str | Path) -> bool:
287 """
288 Check if a file is a valid ZIP archive.
290 Args:
291 zip_path: Path to the file to check
293 Returns:
294 bool: True if file is a valid ZIP archive, False otherwise
295 """
296 try:
297 zip_file = Path(zip_path)
299 if not zip_file.exists() or not zip_file.is_file():
300 return False
302 with zipfile.ZipFile(zip_file, "r") as zipf:
303 # Try to read the ZIP structure to test integrity
304 zipf.testzip()
305 return True
307 except Exception:
308 return False
310 # ////////////////////////////////////////////////
311 # FILE MODIFICATION METHODS
312 # ////////////////////////////////////////////////
314 @staticmethod
315 def add_file_to_zip(
316 zip_path: str | Path,
317 file_path: str | Path,
318 arcname: str | None = None,
319 ) -> None:
320 """
321 Add a single file to an existing ZIP archive.
323 Args:
324 zip_path: Path to the ZIP file
325 file_path: Path to the file to add
326 arcname: Name of the file in the archive (default: file name)
328 Raises:
329 ZipCreationError: If adding file fails
330 """
331 try:
332 zip_file = Path(zip_path)
333 file_to_add = Path(file_path)
335 if not file_to_add.exists():
336 raise ZipPathError(f"File to add does not exist: {file_path}")
338 if not file_to_add.is_file(): 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 raise ZipPathError(f"Path is not a file: {file_path}")
341 # Use file name if arcname not specified
342 if arcname is None:
343 arcname = file_to_add.name
345 with zipfile.ZipFile(zip_file, "a") as zipf:
346 zipf.write(file_to_add, arcname)
348 except Exception as e:
349 raise ZipCreationError(f"Failed to add file to ZIP {zip_path}: {e}") from e
351 @staticmethod
352 def remove_file_from_zip(zip_path: str | Path, file_name: str) -> None:
353 """
354 Remove a file from a ZIP archive.
356 Args:
357 zip_path: Path to the ZIP file
358 file_name: Name of the file to remove from the archive
360 Raises:
361 ZipCreationError: If removal fails
363 Note:
364 Creates a temporary file during removal process.
365 """
366 try:
367 zip_file = Path(zip_path)
369 if not zip_file.exists():
370 raise ZipFileNotFoundError(f"ZIP file does not exist: {zip_path}")
372 # Create a temporary ZIP without the file
373 temp_zip = zip_file.with_suffix(".tmp.zip")
375 with (
376 zipfile.ZipFile(zip_file, "r") as zipf_read,
377 zipfile.ZipFile(temp_zip, "w") as zipf_write,
378 ):
379 for item in zipf_read.infolist():
380 if item.filename != file_name:
381 zipf_write.writestr(item, zipf_read.read(item.filename))
383 # Replace original with temporary
384 zip_file.unlink()
385 temp_zip.rename(zip_file)
387 except Exception as e:
388 raise ZipCreationError(
389 f"Failed to remove file from ZIP {zip_path}: {e}"
390 ) from e
392 # ////////////////////////////////////////////////
393 # PRIVATE UTILITY METHODS
394 # ////////////////////////////////////////////////
396 @staticmethod
397 def _is_hidden_file(file_path: Path) -> bool:
398 """
399 Check if a file is hidden (starts with dot or has hidden attribute).
401 Args:
402 file_path: Path to the file
404 Returns:
405 bool: True if file is hidden, False otherwise
407 Note:
408 On Windows, checks both filename and FILE_ATTRIBUTE_HIDDEN.
409 On Unix, checks if filename starts with dot.
410 """
411 # Check if filename starts with dot
412 if file_path.name.startswith("."):
413 return True
415 # Check hidden attribute on Windows
416 if sys.platform == "win32": 416 ↛ 417line 416 didn't jump to line 417 because the condition on line 416 was never true
417 return bool(
418 file_path.stat().st_file_attributes & stat.FILE_ATTRIBUTE_HIDDEN
419 )
420 return False