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 | dist | ||||||
| docs | docs | ||||||
| 
 | 
 | ||||||
| *.log | *.log | ||||||
| *.err* |  | ||||||
| *.swp |  | ||||||
| @ -126,7 +126,6 @@ def main(): | |||||||
|     logging.info( |     logging.info( | ||||||
|         f"total optimization ({optimized}/{len(result.orig)}): {percent:.2f}% -> {size:.2f} Mb"  # noqa |         f"total optimization ({optimized}/{len(result.orig)}): {percent:.2f}% -> {size:.2f} Mb"  # noqa | ||||||
|     ) |     ) | ||||||
|     result.check_errors() |  | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
| if __name__ == "__main__": | if __name__ == "__main__": | ||||||
|  | |||||||
| @ -4,7 +4,7 @@ import signal | |||||||
| import subprocess | import subprocess | ||||||
| import time | import time | ||||||
| from concurrent.futures import ProcessPoolExecutor, as_completed | from concurrent.futures import ProcessPoolExecutor, as_completed | ||||||
| from dataclasses import dataclass, field | from dataclasses import dataclass | ||||||
| from typing import Optional | from typing import Optional | ||||||
| 
 | 
 | ||||||
| from tqdm import tqdm | 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 | # TODO(rmanach): add argument to set the size or leave it empty for loseless optim | ||||||
| def _jpeg_optim( | def _jpeg_optim(dest_dir: str, file: File) -> tuple["File", Optional["File"]] | None: | ||||||
|     dest_dir: str, file: File |  | ||||||
| ) -> tuple["File", Optional["File"], str | None]: |  | ||||||
|     """ |     """ | ||||||
|     Optimize the `file` with `jpegoptim` and put the result in |     Optimize the `file` with `jpegoptim` and put the result in | ||||||
|     `dest_dir` directory keeping file path. |     `dest_dir` directory keeping file path. | ||||||
| @ -37,26 +35,22 @@ def _jpeg_optim( | |||||||
|     try: |     try: | ||||||
|         _ = subprocess.run(cmd, shell=True, check=True) |         _ = subprocess.run(cmd, shell=True, check=True) | ||||||
|     except subprocess.CalledProcessError as e: |     except subprocess.CalledProcessError as e: | ||||||
|         logging.debug("error while running command: %s, err: %s", cmd, e.stderr or "?") |         logging.error("error while running command: %s, err: %s", cmd, e.stderr or "?") | ||||||
|         return file, None, f"command: {cmd} failed, err: {e.stderr}" |         return None | ||||||
|     except Exception: |     except Exception: | ||||||
|         logging.debug("unexpected error while running command: %s", cmd, exc_info=True) |         logging.error("unexpected error while running command: %s", cmd, exc_info=True) | ||||||
|         return ( |         return None | ||||||
|             file, |  | ||||||
|             None, |  | ||||||
|             f"command: {cmd} failed unexpectedly, set debug mode for more details", |  | ||||||
|         ) |  | ||||||
| 
 | 
 | ||||||
|     try: |     try: | ||||||
|         file_optim = File.from_directory(dest_dir, file.name) |         file_optim = File.from_directory(dest_dir, file.name) | ||||||
|     except Exception as e: |     except Exception as e: | ||||||
|         logging.debug("unable to get file: %s after optimization: %s", file, 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. |     Entry point of `file` optimization selection the handler. | ||||||
|     NOTE: Must be launched in separated process. |     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) |     signal.signal(signal.SIGINT, signal.SIG_IGN) | ||||||
|     if file.mimetype == FileImgMimetype.JPEG.value: |     if file.mimetype == FileImgMimetype.JPEG.value: | ||||||
|         return _jpeg_optim(dest_dir, file) |         return _jpeg_optim(dest_dir, file) | ||||||
|     return file, None, f"mimetype: {file.mimetype} not supported for optimization" |     return None | ||||||
| 
 |  | ||||||
| 
 |  | ||||||
| @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 |  | ||||||
| 
 | 
 | ||||||
| 
 | 
 | ||||||
| @dataclass(slots=True, frozen=True) | @dataclass(slots=True, frozen=True) | ||||||
| @ -99,7 +73,6 @@ class OptimizerResult: | |||||||
|     orig: FileGroup |     orig: FileGroup | ||||||
|     opti: FileGroup |     opti: FileGroup | ||||||
|     optimized: int |     optimized: int | ||||||
|     err: OptimizerErrors |  | ||||||
| 
 | 
 | ||||||
|     def stats(self) -> tuple[int, float, float]: |     def stats(self) -> tuple[int, float, float]: | ||||||
|         """ |         """ | ||||||
| @ -112,13 +85,6 @@ class OptimizerResult: | |||||||
|         size = self.orig._size - self.opti._size |         size = self.orig._size - self.opti._size | ||||||
|         return (self.optimized, percent, 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) | @dataclass(slots=True, frozen=True) | ||||||
| class ImgOptimizer: | class ImgOptimizer: | ||||||
| @ -128,8 +94,8 @@ class ImgOptimizer: | |||||||
| 
 | 
 | ||||||
|     Example: |     Example: | ||||||
|     ```python |     ```python | ||||||
|         optimizer = ImgOptimizer() |         optimizer = ImgOptimizer("mypath") | ||||||
|         res = optimizer.optimize(file_group) |         optimizer.optimize() | ||||||
|     ``` |     ``` | ||||||
|     """ |     """ | ||||||
| 
 | 
 | ||||||
| @ -168,24 +134,20 @@ class ImgOptimizer: | |||||||
| 
 | 
 | ||||||
|             file_group_optim = FileGroup() |             file_group_optim = FileGroup() | ||||||
|             optimized = 0 |             optimized = 0 | ||||||
|             errors = OptimizerErrors() |  | ||||||
|             for f in tqdm( |             for f in tqdm( | ||||||
|                 as_completed(futures), total=len(futures), desc="Optimizing..." |                 as_completed(futures), total=len(futures), desc="Optimizing..." | ||||||
|             ): |             ): | ||||||
|                 match f.result(): |                 if (res := f.result()) and res is not None: | ||||||
|                     case (orig, None, err): |                     match res: | ||||||
|                         logging.debug(f"optim error for file: {orig}, err: {err}") |                         case (orig, None): | ||||||
|                         file_group_optim.add(orig) |                             logging.debug(f"no optimization for file: {orig}") | ||||||
|                         errors.add(orig.path, err) |                             file_group_optim.add(orig) | ||||||
|                     case (orig, None, None): |                         case (orig, opti): | ||||||
|                         logging.debug(f"no optimization for file: {orig}") |                             optimized += 1 | ||||||
|                         file_group_optim.add(orig) |                             logging.debug( | ||||||
|                     case (orig, opti, None): |                                 f"optimization for file: {orig} -> {(1 - (opti.size / orig.size)) * 100:.2f}%"  # noqa | ||||||
|                         optimized += 1 |                             ) | ||||||
|                         logging.debug( |                             file_group_optim.add(opti) | ||||||
|                             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") |         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