Python利用PIL进行图片压缩

from pil import image, imagefilter
import io
 
 
def compress_image(input_path, output_path, max_size_kb, quality=85, step=5):
    """
    压缩图片到指定大小以下，并尽可能保证清晰度。
    :param input_path: 输入图片路径
    :param output_path: 输出图片路径
    :param max_size_kb: 目标大小（单位：kb）
    :param quality: 初始压缩质量（默认85）
    :param step: 每次降低质量的步长（默认5）
    """
    # 打开图片
    img = image.open(input_path)
 
    # 将图片转换为rgb模式（如果是rgba或其他模式）
    if img.mode in ('rgba', 'la'):
        img = img.convert('rgb')
 
    # 创建一个字节流对象
    img_byte_arr = io.bytesio()
 
    # 保存图片到字节流，初始质量为quality
    img.save(img_byte_arr, format='jpeg', quality=quality)
 
    # 获取当前图片大小
    current_size = len(img_byte_arr.getvalue()) / 1024  # 转换为kb
 
    # 如果图片大小超过最大限制，逐步降低质量
    while current_size > max_size_kb and quality > 10:
        quality -= step
        img_byte_arr = io.bytesio()
        img.save(img_byte_arr, format='jpeg', quality=quality)
        current_size = len(img_byte_arr.getvalue()) / 1024
 
    # 如果图片大小仍然超过最大限制，调整图片尺寸
    while current_size > max_size_kb:
        width, height = img.size
        # 计算缩放比例，确保图片大小接近目标大小
        scale_factor = (max_size_kb / current_size) ** 0.5
        new_width = int(width * scale_factor)
        new_height = int(height * scale_factor)
        img = img.resize((new_width, new_height), image.resampling.lanczos)
        img_byte_arr = io.bytesio()
        img.save(img_byte_arr, format='jpeg', quality=quality)
        current_size = len(img_byte_arr.getvalue()) / 1024
 
    # 对图片进行锐化处理
    img = img.filter(imagefilter.sharpen)
 
    # 保存压缩后的图片
    with open(output_path, 'wb') as f:
        img.save(f, format='jpeg', quality=quality)
 
    print(f"压缩后的图片大小: {current_size:.2f} kb")
 
 
input_image = r"e:\桌面\2.jpg"
output_image = r"e:\桌面\2-1.jpg"
target_size_kb = 35  # 自定义目标大小，例如30kb
compress_image(input_image, output_image, max_size_kb=target_size_kb)

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# main.py

from pil import image
from io import bytesio
from os import system
from os import listdir
from os import remove
from os.path import getsize
from os.path import exists


__author__ = 'one-ccs'

"""
功能: 把 ".\old" 目录下的所有图片压缩并填充至 200kb 后存放在 ".\new" 目录下.
"""


settings = {
    'loadpath': r'.\old',  # 待压缩图片路径
    'savepath': r'.\new',  # 保存路径
    'size': 200,
    'quality': 90
}


class imagefixcompressor():

    def __init__(self) -> none:
        self.imgbytes = none
        self.imgpath = none
        self.name = none
        self.oldsize = 0
        self.newsize = 0
        self.savesize = 0

        self.format = 'jpeg'
        self.targetsize = 200 # 目标大小 (kb)
        self.quality = 90     # 压缩比率 若压缩后图片大于目标大小应减小该值

    def split_path(self, path:str='') -> tuple:
        """
        提取 path 中的路径与文件名.
        """
        if not isinstance(path, str):
            raise valueerror(f'参数 "path" 的数据类型应该为 "str", 但是传入了 "{type(path)}" 类型.')

        # 判断是否是以 '/' '\' 作为目录分隔符
        flag = path[::-1].find('/')
        if flag == -1:
            flag = path[::-1].find('\\')
            if flag == -1:
                raise valueerror(f'参数 "path" 的数据类型应该为 "str", 但是传入了 "{type(path)}" 类型.')

        name = path[-flag:]
        path = path[:-flag]

        return (path, name)

    def full_path(self, path) -> str:
        return fr'{path}\{self.name}'

    def open(self, imgpath:str) -> none:
        """
        打开图像文件

        :参数 imgpath: 图片文件路径.
        """
        try:
            _, self.name = self.split_path(imgpath)
            self.oldsize = getsize(imgpath)
            self.image = image.open(imgpath)

            print(f'打开: "{imgpath}" 成功; 大小: {self.oldsize / 1024:.2f} kb.')
        except exception as e:
            print(f'错误: 文件 "{imgpath}" 打开失败; 原因: "{e}".')

    def show(self) -> none:
        self.image.show()

    def compress(self, format='jpeg', quality=none) -> none:
        if format == 'png' or format == 'png':
            self.format = 'png'
        if quality:
            self.quality = quality

        self.image.tobytes()
        self.imagebuffer = bytesio()
        self.image.save(self.imagebuffer, format=self.format, quality=self.quality)

    def save(self, savepath:str, cover:bool=false, name=none) -> none:
        if cover:
            mode = 'wb+'
        else:
            mode = 'rb+'

        try:
            self.newsize = self.imagebuffer.tell() / 1024
            # ~ 补齐大小
            for i in range(0, self.targetsize * 1024 - self.imagebuffer.tell()):
                self.imagebuffer.write(b'\x00')

            with open(self.full_path(savepath), mode) as fb:
                fb.write(self.imagebuffer.getvalue())

            self.savesize = getsize(self.full_path(savepath)) / 1024
            print(fr'保存: "{self.full_path(savepath)}" 成功; 压缩大小: {self.newsize:.2f} kb; 保存大小: {self.savesize:.2f} kb.')
        except exception as e:
            print(f'错误: "{self.name}" 保存失败; 原因: "{e}".')

    def save2(self, savepath:str, cover:bool=false, name=none) -> none:
        if cover:
            if exists(self.full_path(savepath)):
                remove(self.full_path(savepath))
        else:
            print(f'异常: 文件 "{savepath}" 已存在, 已放弃保存.')
            return

        system('@echo off')
        system(f'fsutil file createnew {self.full_path(savepath)} {self.targetsize * 1024}')

        try:
            with open(self.full_path(savepath), 'rb+') as fb:
                fb.write(self.imagebuffer.getvalue())
                self.newsize = self.imagebuffer.tell() / 1024
            self.savesize = getsize(self.full_path(savepath)) / 1024
            print(fr'保存: "{self.full_path(savepath)}" 成功; 压缩大小: {self.newsize:.2f} kb; 保存大小: {self.savesize:.2f} kb.')
        except exception as e:
            print(f'错误: "{self.name}" 保存失败; 原因: "{e}".')


def main(args):
    compressor = imagefixcompressor()

    oldimgpaths = listdir(settings['loadpath'])

    for oldimgpath in oldimgpaths:
        fullpath = f"{settings['loadpath']}\{oldimgpath}"

        compressor.open(fullpath)
        compressor.compress()
        compressor.save(settings['savepath'], cover=true)
        # ~ return

    return 0

if __name__ == '__main__':
    import sys
    sys.exit(main(sys.argv))