Compare commits
	
		
			No commits in common. "develop" and "main" have entirely different histories.
		
	
	
		
	
		
							
								
								
									
										4
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										4
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							| @ -8,6 +8,4 @@ data | ||||
| dist | ||||
| docs | ||||
| 
 | ||||
| *.log | ||||
| *.err* | ||||
| *.swp | ||||
| *.log | ||||
| @ -126,7 +126,6 @@ def main(): | ||||
|     logging.info( | ||||
|         f"total optimization ({optimized}/{len(result.orig)}): {percent:.2f}% -> {size:.2f} Mb"  # noqa | ||||
|     ) | ||||
|     result.check_errors() | ||||
| 
 | ||||
| 
 | ||||
| if __name__ == "__main__": | ||||
|  | ||||
| @ -4,7 +4,7 @@ import signal | ||||
| import subprocess | ||||
| import time | ||||
| from concurrent.futures import ProcessPoolExecutor, as_completed | ||||
| from dataclasses import dataclass, field | ||||
| from dataclasses import dataclass | ||||
| from typing import Optional | ||||
| 
 | ||||
| from tqdm import tqdm | ||||
| @ -15,9 +15,7 @@ __all__ = ["ImgOptimizer", "OptimizerResult"] | ||||
| 
 | ||||
| 
 | ||||
| # TODO(rmanach): add argument to set the size or leave it empty for loseless optim | ||||
| def _jpeg_optim( | ||||
|     dest_dir: str, file: File | ||||
| ) -> tuple["File", Optional["File"], str | None]: | ||||
| def _jpeg_optim(dest_dir: str, file: File) -> tuple["File", Optional["File"]] | None: | ||||
|     """ | ||||
|     Optimize the `file` with `jpegoptim` and put the result in | ||||
|     `dest_dir` directory keeping file path. | ||||
| @ -37,26 +35,22 @@ def _jpeg_optim( | ||||
|     try: | ||||
|         _ = subprocess.run(cmd, shell=True, check=True) | ||||
|     except subprocess.CalledProcessError as e: | ||||
|         logging.debug("error while running command: %s, err: %s", cmd, e.stderr or "?") | ||||
|         return file, None, f"command: {cmd} failed, err: {e.stderr}" | ||||
|         logging.error("error while running command: %s, err: %s", cmd, e.stderr or "?") | ||||
|         return None | ||||
|     except Exception: | ||||
|         logging.debug("unexpected error while running command: %s", cmd, exc_info=True) | ||||
|         return ( | ||||
|             file, | ||||
|             None, | ||||
|             f"command: {cmd} failed unexpectedly, set debug mode for more details", | ||||
|         ) | ||||
|         logging.error("unexpected error while running command: %s", cmd, exc_info=True) | ||||
|         return None | ||||
| 
 | ||||
|     try: | ||||
|         file_optim = File.from_directory(dest_dir, file.name) | ||||
|     except Exception as e: | ||||
|         logging.debug("unable to get file: %s after optimization: %s", file, e) | ||||
|         return file, None, None | ||||
|         return file, None | ||||
| 
 | ||||
|     return file, file_optim, None | ||||
|     return file, file_optim | ||||
| 
 | ||||
| 
 | ||||
| def _optim(dest_dir: str, file: File) -> tuple["File", Optional["File"], str | None]: | ||||
| def _optim(dest_dir: str, file: File) -> tuple["File", Optional["File"]] | None: | ||||
|     """ | ||||
|     Entry point of `file` optimization selection the handler. | ||||
|     NOTE: Must be launched in separated process. | ||||
| @ -65,27 +59,7 @@ def _optim(dest_dir: str, file: File) -> tuple["File", Optional["File"], str | N | ||||
|     signal.signal(signal.SIGINT, signal.SIG_IGN) | ||||
|     if file.mimetype == FileImgMimetype.JPEG.value: | ||||
|         return _jpeg_optim(dest_dir, file) | ||||
|     return file, None, f"mimetype: {file.mimetype} not supported for optimization" | ||||
| 
 | ||||
| 
 | ||||
| @dataclass(slots=True, frozen=True) | ||||
| class OptimizerErrors: | ||||
|     _data: dict[str, str] = field(default_factory=dict) | ||||
| 
 | ||||
|     def save(self): | ||||
|         if not len(self._data): | ||||
|             return | ||||
| 
 | ||||
|         with open("optimg.err.csv", "w") as f: | ||||
|             f.write("filepath;err\n") | ||||
|             for filepath, error in self._data.items(): | ||||
|                 f.write(f"{filepath};{error}\n") | ||||
| 
 | ||||
|     def has_errors(self) -> bool: | ||||
|         return len(self._data) > 0 | ||||
| 
 | ||||
|     def add(self, filepath: str, error: str): | ||||
|         self._data[filepath] = error | ||||
|     return None | ||||
| 
 | ||||
| 
 | ||||
| @dataclass(slots=True, frozen=True) | ||||
| @ -99,7 +73,6 @@ class OptimizerResult: | ||||
|     orig: FileGroup | ||||
|     opti: FileGroup | ||||
|     optimized: int | ||||
|     err: OptimizerErrors | ||||
| 
 | ||||
|     def stats(self) -> tuple[int, float, float]: | ||||
|         """ | ||||
| @ -112,13 +85,6 @@ class OptimizerResult: | ||||
|         size = self.orig._size - self.opti._size | ||||
|         return (self.optimized, percent, size) | ||||
| 
 | ||||
|     def check_errors(self): | ||||
|         if self.err.has_errors(): | ||||
|             logging.error( | ||||
|                 "some errors detected during optimization, check 'optimg.err.csv' for details"  # noqa | ||||
|             ) | ||||
|             self.err.save() | ||||
| 
 | ||||
| 
 | ||||
| @dataclass(slots=True, frozen=True) | ||||
| class ImgOptimizer: | ||||
| @ -128,8 +94,8 @@ class ImgOptimizer: | ||||
| 
 | ||||
|     Example: | ||||
|     ```python | ||||
|         optimizer = ImgOptimizer() | ||||
|         res = optimizer.optimize(file_group) | ||||
|         optimizer = ImgOptimizer("mypath") | ||||
|         optimizer.optimize() | ||||
|     ``` | ||||
|     """ | ||||
| 
 | ||||
| @ -168,24 +134,20 @@ class ImgOptimizer: | ||||
| 
 | ||||
|             file_group_optim = FileGroup() | ||||
|             optimized = 0 | ||||
|             errors = OptimizerErrors() | ||||
|             for f in tqdm( | ||||
|                 as_completed(futures), total=len(futures), desc="Optimizing..." | ||||
|             ): | ||||
|                 match f.result(): | ||||
|                     case (orig, None, err): | ||||
|                         logging.debug(f"optim error for file: {orig}, err: {err}") | ||||
|                         file_group_optim.add(orig) | ||||
|                         errors.add(orig.path, err) | ||||
|                     case (orig, None, None): | ||||
|                         logging.debug(f"no optimization for file: {orig}") | ||||
|                         file_group_optim.add(orig) | ||||
|                     case (orig, opti, None): | ||||
|                         optimized += 1 | ||||
|                         logging.debug( | ||||
|                             f"optimization for file: {orig} -> {(1 - (opti.size / orig.size)) * 100:.2f}%"  # noqa | ||||
|                         ) | ||||
|                         file_group_optim.add(opti) | ||||
|                 if (res := f.result()) and res is not None: | ||||
|                     match res: | ||||
|                         case (orig, None): | ||||
|                             logging.debug(f"no optimization for file: {orig}") | ||||
|                             file_group_optim.add(orig) | ||||
|                         case (orig, opti): | ||||
|                             optimized += 1 | ||||
|                             logging.debug( | ||||
|                                 f"optimization for file: {orig} -> {(1 - (opti.size / orig.size)) * 100:.2f}%"  # noqa | ||||
|                             ) | ||||
|                             file_group_optim.add(opti) | ||||
| 
 | ||||
|         logging.info(f"optimization finished in {time.perf_counter() - start:.2f}s") | ||||
|         return OptimizerResult(file_group, file_group_optim, optimized, errors) | ||||
|         return OptimizerResult(file_group, file_group_optim, optimized) | ||||
|  | ||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user