Python脚本如何压缩缓存存储文件体积

wen python案例 2026-06-26 238

本文目录导读：

基础文件压缩方法
缓存数据压缩存储
图片缓存压缩
缓存清理策略
完整示例：缓存管理类
最佳实践建议

我来介绍几种Python压缩缓存文件的方法：

基础文件压缩方法

使用 `gzip` 压缩单个文件

import gzip
import shutil
import os
def compress_file(input_file, output_file=None):
    """压缩单个文件"""
    if output_file is None:
        output_file = input_file + '.gz'
    with open(input_file, 'rb') as f_in:
        with gzip.open(output_file, 'wb') as f_out:
            shutil.copyfileobj(f_in, f_out)
    # 可选：删除原文件
    # os.remove(input_file)
    return output_file
# 使用示例
compress_file('cache/data.json')

使用 `zipfile` 压缩多个文件

import zipfile
import os
def compress_to_zip(source_dir, output_file='cache.zip'):
    """将目录压缩为zip文件"""
    with zipfile.ZipFile(output_file, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for root, dirs, files in os.walk(source_dir):
            for file in files:
                file_path = os.path.join(root, file)
                arcname = os.path.relpath(file_path, os.path.dirname(source_dir))
                zipf.write(file_path, arcname)
    return output_file

缓存数据压缩存储

使用 pickle + gzip 压缩序列化数据

import pickle
import gzip
def save_compressed_cache(data, filepath):
    """保存压缩的缓存数据"""
    with gzip.open(filepath, 'wb') as f:
        pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)
def load_compressed_cache(filepath):
    """加载压缩的缓存数据"""
    with gzip.open(filepath, 'rb') as f:
        return pickle.load(f)
# 使用示例
cache_data = {'key': 'value', 'list': [1, 2, 3]}
save_compressed_cache(cache_data, 'cache/compressed_cache.pkl.gz')
loaded_data = load_compressed_cache('cache/compressed_cache.pkl.gz')

使用 lzma 提供更好的压缩率

import lzma
import json
def compress_with_lzma(data, filepath):
    """使用 LZMA 算法压缩数据"""
    compressed = lzma.compress(json.dumps(data).encode('utf-8'))
    with open(filepath, 'wb') as f:
        f.write(compressed)
def decompress_with_lzma(filepath):
    """解压 LZMA 压缩的数据"""
    with open(filepath, 'rb') as f:
        compressed = f.read()
    return json.loads(lzma.decompress(compressed).decode('utf-8'))

图片缓存压缩

使用 Pillow 压缩图片

from PIL import Image
import os
def compress_image(input_path, output_path=None, quality=85):
    """压缩图片文件"""
    if output_path is None:
        output_path = input_path
    img = Image.open(input_path)
    # 转换为RGB模式（去除alpha通道）
    if img.mode in ('RGBA', 'LA', 'P'):
        img = img.convert('RGB')
    # 保存并压缩
    img.save(output_path, 'JPEG', optimize=True, quality=quality)
    return output_path
# 批量压缩目录中的图片
def compress_images_in_directory(directory, quality=85):
    for filename in os.listdir(directory):
        if filename.lower().endswith(('.png', '.jpg', '.jpeg', '.gif')):
            filepath = os.path.join(directory, filename)
            compress_image(filepath, quality=quality)

缓存清理策略

自动清理过期缓存

import time
import os
import shutil
class CacheCleaner:
    def __init__(self, cache_dir, max_size_mb=100, expire_hours=24):
        self.cache_dir = cache_dir
        self.max_size_bytes = max_size_mb * 1024 * 1024
        self.expire_seconds = expire_hours * 3600
    def clean_expired(self):
        """清理过期缓存"""
        now = time.time()
        for filename in os.listdir(self.cache_dir):
            filepath = os.path.join(self.cache_dir, filename)
            if os.path.isfile(filepath):
                # 检查文件是否过期
                if now - os.path.getmtime(filepath) > self.expire_seconds:
                    os.remove(filepath)
                    print(f"删除过期文件: {filename}")
    def clean_by_size(self):
        """清理超出大小限制的缓存"""
        files = []
        total_size = 0
        # 获取所有文件及其大小
        for filename in os.listdir(self.cache_dir):
            filepath = os.path.join(self.cache_dir, filename)
            if os.path.isfile(filepath):
                size = os.path.getsize(filepath)
                files.append((filepath, size, os.path.getmtime(filepath)))
                total_size += size
        # 按修改时间排序（最旧的在前）
        files.sort(key=lambda x: x[2])
        # 删除最旧的文件，直到总大小低于限制
        while total_size > self.max_size_bytes and files:
            filepath, size, _ = files.pop(0)
            os.remove(filepath)
            total_size -= size
            print(f"删除缓存文件: {os.path.basename(filepath)}")
# 使用示例
cleaner = CacheCleaner('cache/', max_size_mb=50, expire_hours=48)
cleaner.clean_expired()
cleaner.clean_by_size()

完整示例：缓存管理类

import hashlib
import json
import gzip
import os
import time
class CompressedCache:
    def __init__(self, cache_dir='.cache', compress_level=9):
        self.cache_dir = cache_dir
        self.compress_level = compress_level
        os.makedirs(cache_dir, exist_ok=True)
    def _get_cache_path(self, key):
        """获取缓存文件路径"""
        hash_key = hashlib.md5(key.encode()).hexdigest()
        return os.path.join(self.cache_dir, f'{hash_key}.cache')
    def set(self, key, value, ttl=None):
        """设置缓存"""
        cache_data = {
            'value': value,
            'expire_time': time.time() + ttl if ttl else None,
            'created': time.time()
        }
        cache_path = self._get_cache_path(key)
        with gzip.open(cache_path, 'wb', compresslevel=self.compress_level) as f:
            f.write(json.dumps(cache_data).encode())
    def get(self, key):
        """获取缓存"""
        cache_path = self._get_cache_path(key)
        if not os.path.exists(cache_path):
            return None
        try:
            with gzip.open(cache_path, 'rb') as f:
                cache_data = json.loads(f.read().decode())
            # 检查是否过期
            if cache_data['expire_time'] and time.time() > cache_data['expire_time']:
                os.remove(cache_path)
                return None
            return cache_data['value']
        except Exception:
            return None
    def clear(self):
        """清空所有缓存"""
        for filename in os.listdir(self.cache_dir):
            filepath = os.path.join(self.cache_dir, filename)
            if filename.endswith('.cache'):
                os.remove(filepath)
    def get_stats(self):
        """获取缓存统计信息"""
        total_size = 0
        file_count = 0
        for filename in os.listdir(self.cache_dir):
            filepath = os.path.join(self.cache_dir, filename)
            if filename.endswith('.cache'):
                total_size += os.path.getsize(filepath)
                file_count += 1
        return {
            'file_count': file_count,
            'total_size_mb': total_size / (1024 * 1024),
            'cache_dir': self.cache_dir
        }
# 使用示例
cache = CompressedCache()
cache.set('user_123', {'name': 'Alice', 'data': [1,2,3]}, ttl=3600)
data = cache.get('user_123')
print(cache.get_stats())