Compress large user-uploaded images. Args: url_or_path: Image path or URL. max_size: Longest edge of the compressed image in pixels. quality: JPEG output quality in the range 1-100. Returns: The compressed image path. Returns the original path if compression
(
url_or_path: str,
max_size: int = IMAGE_COMPRESS_DEFAULT_MAX_SIZE,
quality: int = IMAGE_COMPRESS_DEFAULT_QUALITY,
)
| 1543 | |
| 1544 | |
| 1545 | async def compress_image( |
| 1546 | url_or_path: str, |
| 1547 | max_size: int = IMAGE_COMPRESS_DEFAULT_MAX_SIZE, |
| 1548 | quality: int = IMAGE_COMPRESS_DEFAULT_QUALITY, |
| 1549 | ) -> str: |
| 1550 | """Compress large user-uploaded images. |
| 1551 | |
| 1552 | Args: |
| 1553 | url_or_path: Image path or URL. |
| 1554 | max_size: Longest edge of the compressed image in pixels. |
| 1555 | quality: JPEG output quality in the range 1-100. |
| 1556 | |
| 1557 | Returns: |
| 1558 | The compressed image path. Returns the original path if compression |
| 1559 | fails or the source does not need compression. |
| 1560 | """ |
| 1561 | max_size = max(int(max_size), 1) |
| 1562 | quality = min(max(int(quality), 1), 100) |
| 1563 | optimize = IMAGE_COMPRESS_DEFAULT_OPTIMIZE |
| 1564 | min_file_size_bytes = int(IMAGE_COMPRESS_DEFAULT_MIN_FILE_SIZE_MB * 1024 * 1024) |
| 1565 | image_source: bytes | Path | None = None |
| 1566 | |
| 1567 | def _exceeds_max_size(source: bytes | Path) -> bool: |
| 1568 | try: |
| 1569 | fp = io.BytesIO(source) if isinstance(source, bytes) else source |
| 1570 | with PILImage.open(fp) as opened_img: |
| 1571 | return max(opened_img.size) > max_size |
| 1572 | except Exception: # noqa: BLE001 |
| 1573 | return False |
| 1574 | |
| 1575 | # Skip compression for remote images and return the original value. |
| 1576 | if url_or_path.startswith("http"): |
| 1577 | return url_or_path |
| 1578 | elif url_or_path.startswith("data:image"): |
| 1579 | _header, encoded = url_or_path.split(",", 1) |
| 1580 | image_source = _decode_base64_payload( |
| 1581 | encoded, |
| 1582 | error_message="invalid image data URI payload", |
| 1583 | ) |
| 1584 | if len(image_source) < min_file_size_bytes and not _exceeds_max_size( |
| 1585 | image_source |
| 1586 | ): |
| 1587 | return url_or_path |
| 1588 | else: |
| 1589 | local_path = Path(url_or_path) |
| 1590 | if not local_path.exists(): |
| 1591 | return url_or_path |
| 1592 | if local_path.stat().st_size < min_file_size_bytes and not _exceeds_max_size( |
| 1593 | local_path |
| 1594 | ): |
| 1595 | return url_or_path |
| 1596 | image_source = local_path |
| 1597 | |
| 1598 | if image_source is None: |
| 1599 | return url_or_path |
| 1600 | |
| 1601 | temp_dir = Path(get_astrbot_temp_path()) |
| 1602 | temp_dir.mkdir(parents=True, exist_ok=True) |
no test coverage detected