Python脚本如何过滤无需同步冗余数据

wen python案例 2026-06-24 239

本文目录导读：

基于哈希值去重
基于时间戳过滤
基于版本号过滤
完整的数据同步过滤系统
实用工具函数

我来介绍几种Python过滤冗余数据的常用方法：

基于哈希值去重

import hashlib
import json
class DataDeduplicator:
    def __init__(self):
        self.seen_hashes = set()
    def get_data_hash(self, data):
        """生成数据哈希值"""
        # 将数据转换为JSON字符串再计算哈希
        if isinstance(data, (dict, list)):
            data_str = json.dumps(data, sort_keys=True)
        else:
            data_str = str(data)
        return hashlib.md5(data_str.encode()).hexdigest()
    def is_duplicate(self, data):
        """检查数据是否重复"""
        data_hash = self.get_data_hash(data)
        if data_hash in self.seen_hashes:
            return True
        self.seen_hashes.add(data_hash)
        return False
    def filter_duplicates(self, data_list):
        """过滤重复数据"""
        return [data for data in data_list if not self.is_duplicate(data)]
# 使用示例
dedup = DataDeduplicator()
data_list = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    {"id": 1, "name": "Alice"},  # 重复数据
]
filtered_data = dedup.filter_duplicates(data_list)
print(filtered_data)

基于时间戳过滤

from datetime import datetime, timedelta
import time
class TimeBasedFilter:
    def __init__(self, time_window_seconds=3600):
        self.time_window = timedelta(seconds=time_window_seconds)
        self.last_sync_times = {}
    def should_sync(self, data_id, data_timestamp):
        """
        判断是否需要同步
        Args:
            data_id: 数据唯一标识
            data_timestamp: 数据时间戳
        """
        if data_id not in self.last_sync_times:
            self.last_sync_times[data_id] = data_timestamp
            return True
        last_sync = self.last_sync_times[data_id]
        time_diff = data_timestamp - last_sync
        # 如果时间差大于时间窗，才需要同步
        if timedelta(seconds=abs(time_diff.total_seconds())) > self.time_window:
            self.last_sync_times[data_id] = data_timestamp
            return True
        return False
# 使用示例
filter = TimeBasedFilter(time_window_seconds=300)  # 5分钟窗口
now = datetime.now()
data_items = [
    ("file1", now - timedelta(minutes=10)),  # 需要同步
    ("file1", now - timedelta(minutes=2)),   # 不需要同步
    ("file2", now - timedelta(minutes=30)),  # 需要同步
]
for data_id, timestamp in data_items:
    if filter.should_sync(data_id, timestamp):
        print(f"需要同步: {data_id}")
    else:
        print(f"跳过同步: {data_id}")

基于版本号过滤

class VersionBasedFilter:
    def __init__(self):
        self.latest_versions = {}
    def update_version(self, data_id, new_version):
        """更新数据版本"""
        self.latest_versions[data_id] = new_version
    def is_newer_version(self, data_id, version):
        """检查是否为更新版本"""
        if data_id not in self.latest_versions:
            return True
        return version > self.latest_versions[data_id]
class DataItem:
    def __init__(self, id, version, content):
        self.id = id
        self.version = version
        self.content = content
# 使用示例
version_filter = VersionBasedFilter()
datasets = [
    DataItem("doc1", 1, "版本1内容"),
    DataItem("doc1", 2, "版本2内容"),  # 新版本
    DataItem("doc1", 1, "旧版本内容"),  # 旧版本，应过滤
]
for dataset in datasets:
    if version_filter.is_newer_version(dataset.id, dataset.version):
        print(f"同步数据: {dataset.id} 版本 {dataset.version}")
        version_filter.update_version(dataset.id, dataset.version)
    else:
        print(f"跳过旧版本: {dataset.id} 版本 {dataset.version}")

完整的数据同步过滤系统

import hashlib
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
class SyncFilter:
    """同步过滤系统"""
    def __init__(self):
        self.seen_hashes = set()  # 数据哈希
        self.latest_versions = {}  # 版本号
        self.last_sync_times = {}  # 同步时间
        self.change_log = {}  # 变更日志
    def filter_by_hash(self, data: Any) -> bool:
        """基于哈希值过滤"""
        data_hash = self._compute_hash(data)
        if data_hash in self.seen_hashes:
            return False
        self.seen_hashes.add(data_hash)
        return True
    def filter_by_version(self, data_id: str, version: int) -> bool:
        """基于版本号过滤"""
        current_version = self.latest_versions.get(data_id, -1)
        if version <= current_version:
            return False
        self.latest_versions[data_id] = version
        return True
    def filter_by_time(self, data_id: str, timestamp: datetime, 
                      window_minutes: int = 60) -> bool:
        """基于时间戳过滤"""
        last_sync = self.last_sync_times.get(data_id)
        if last_sync:
            time_diff = (timestamp - last_sync).total_seconds() / 60
            if time_diff < window_minutes:
                return False
        self.last_sync_times[data_id] = timestamp
        return True
    def filter_by_content(self, old_content: Any, new_content: Any) -> bool:
        """基于内容变化过滤"""
        return old_content != new_content
    def multi_filter(self, data: Dict) -> bool:
        """
        多重过滤，任何一条过滤条件通过即认为需要同步
        Returns:
            True 表示需要同步，False 表示可以跳过
        """
        data_id = data.get('id')
        version = data.get('version', 0)
        timestamp = data.get('timestamp', datetime.now())
        content = data.get('content')
        # 基于哈希值过滤（内容完全相同）
        if not self.filter_by_hash(content):
            return False
        # 基于版本号过滤
        if not self.filter_by_version(data_id, version):
            return False
        # 基于时间过滤（默认1小时内不重复同步）
        if not self.filter_by_time(data_id, timestamp):
            # 但如果版本有更新，仍然需要同步
            if not self._has_version_update(data_id, version):
                return False
        return True
    def _compute_hash(self, data: Any) -> str:
        """计算数据哈希"""
        if isinstance(data, (dict, list)):
            data_str = json.dumps(data, sort_keys=True)
        else:
            data_str = str(data)
        return hashlib.md5(data_str.encode()).hexdigest()
    def _has_version_update(self, data_id: str, version: int) -> bool:
        """检查版本是否有更新"""
        current = self.latest_versions.get(data_id, -1)
        return version > current
    def log_sync(self, data_id: str, action: str, details: Optional[Dict] = None):
        """记录同步日志"""
        if data_id not in self.change_log:
            self.change_log[data_id] = []
        self.change_log[data_id].append({
            'action': action,
            'timestamp': datetime.now(),
            'details': details or {}
        })
# 使用示例
def main():
    sync_filter = SyncFilter()
    # 模拟数据流
    mock_data = [
        {
            'id': 'user_001',
            'version': 1,
            'timestamp': datetime(2024, 1, 1, 10, 0),
            'content': {'name': 'Alice', 'email': 'alice@example.com'}
        },
        # 相同内容的重复数据
        {
            'id': 'user_001',
            'version': 1,
            'timestamp': datetime(2024, 1, 1, 10, 30),
            'content': {'name': 'Alice', 'email': 'alice@example.com'}
        },
        # 新版本数据
        {
            'id': 'user_001',
            'version': 2,
            'timestamp': datetime(2024, 1, 1, 11, 0),
            'content': {'name': 'Alice', 'email': 'alice_new@example.com'}
        },
    ]
    for i, data in enumerate(mock_data, 1):
        print(f"\n数据 {i}:")
        print(f"  内容: {data['content']}")
        if sync_filter.multi_filter(data):
            print("  ✅ 需要同步")
            sync_filter.log_sync(data['id'], 'sync', 
                               {'version': data['version']})
        else:
            print("  ❌ 跳过同步（冗余数据）")
if __name__ == "__main__":
    main()

实用工具函数

def batch_filter_with_progress(data_list, filter_func, batch_size=100):
    """
    批量过滤并显示进度
    Args:
        data_list: 数据列表
        filter_func: 过滤函数
        batch_size: 批处理大小
    """
    total = len(data_list)
    result = []
    for i in range(0, total, batch_size):
        batch = data_list[i:i + batch_size]
        progress = min(i + batch_size, total)
        print(f"处理进度: {progress}/{total}")
        for data in batch:
            if filter_func(data):
                result.append(data)
    return result
# 使用示例
def smart_filter(data):
    """智能过滤规则"""
    # 过滤掉无效数据
    if not data.get('id') or not data.get('content'):
        return False
    # 过滤掉空内容
    if not data['content'].strip():
        return False
    # 过滤掉特定类型的数据
    if data.get('type') == 'temporary':
        return False
    return True
data_list = [
    {'id': '1', 'content': 'valid data'},
    {'id': '2', 'content': ''},  # 空内容
    {'type': 'temporary'},  # 临时数据
    {'id': '3', 'content': 'another valid data'},
]
filtered = batch_filter_with_progress(data_list, smart_filter)
print(f"\n过滤结果: {filtered}")

选择哪种过滤方法取决于你的具体场景：