1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
| import zlib import gzip import json from abc import ABC, abstractmethod from typing import Protocol
class CompressionStrategy(Protocol): """压缩策略接口"""
def compress(self, data: str) -> bytes: ...
def decompress(self, data: bytes) -> str: ...
class NoCompression: """无压缩策略"""
def compress(self, data: str) -> bytes: return data.encode('utf-8')
def decompress(self, data: bytes) -> str: return data.decode('utf-8')
class ZlibCompression: """Zlib 压缩策略"""
def compress(self, data: str) -> bytes: return zlib.compress(data.encode('utf-8'))
def decompress(self, data: bytes) -> str: return zlib.decompress(data).decode('utf-8')
class GzipCompression: """Gzip 压缩策略"""
def compress(self, data: str) -> bytes: return gzip.compress(data.encode('utf-8'))
def decompress(self, data: bytes) -> str: return gzip.decompress(data).decode('utf-8')
class JSONCompression: """JSON 格式化压缩策略"""
def compress(self, data: str) -> bytes: obj = json.loads(data) return gzip.compress(json.dumps(obj).encode('utf-8'))
def decompress(self, data: bytes) -> str: return gzip.decompress(data).decode('utf-8')
class DataProcessor: """数据处理器上下文"""
def __init__(self, strategy: CompressionStrategy): self.strategy = strategy
def set_strategy(self, strategy: CompressionStrategy) -> None: self.strategy = strategy
def save_data(self, data: str, filename: str) -> None: """保存数据(应用压缩)""" compressed = self.strategy.compress(data) with open(filename, 'wb') as f: f.write(compressed) print(f"数据已保存到 {filename}")
def load_data(self, filename: str) -> str: """加载数据(应用解压)""" with open(filename, 'rb') as f: compressed = f.read() return self.strategy.decompress(compressed)
def main(): data = json.dumps({"name": "Alice", "age": 30, "city": "Beijing"})
processor = DataProcessor(NoCompression()) processor.save_data("data_plain.txt", data)
processor.set_strategy(ZlibCompression()) processor.save_data("data_zlib.bin", data) print(f"解压: {processor.load_data('data_zlib.bin')}")
processor.set_strategy(GzipCompression()) processor.save_data("data_gzip.bin", data) print(f"解压: {processor.load_data('data_gzip.bin')}")
if __name__ == "__main__": main()
|