Python脚本如何过滤重复批次同步任务

wen python案例 2026-06-25 260

本文目录导读：

方案1：基于任务特征去重
方案2：基于数据库操作去重
方案3：基于时间窗口去重
方案4：综合去重管理器
选择建议

我来帮你编写一个Python脚本来过滤重复的批次同步任务,以下是几种常见的实现方案：

方案1：基于任务特征去重

import hashlib
import json
from typing import Dict, List, Any
class BatchSyncFilter:
    def __init__(self):
        self.seen_tasks = set()
        self.unique_tasks = []
    def generate_task_signature(self, task: Dict) -> str:
        """
        生成任务的唯一签名
        """
        # 创建任务特征的规范表示
        task_data = {
            'source': task.get('source'),
            'target': task.get('target'),
            'type': task.get('sync_type'),
            'filters': task.get('filters'),
            'schedule': task.get('schedule')
        }
        # 序列化为JSON并生成Hash
        task_json = json.dumps(task_data, sort_keys=True)
        return hashlib.md5(task_json.encode()).hexdigest()
    def filter_duplicates(self, tasks: List[Dict]) -> List[Dict]:
        """
        过滤重复任务
        """
        self.seen_tasks = set()
        self.unique_tasks = []
        for task in tasks:
            signature = self.generate_task_signature(task)
            if signature not in self.seen_tasks:
                self.seen_tasks.add(signature)
                self.unique_tasks.append(task)
        return self.unique_tasks
# 使用示例
if __name__ == "__main__":
    # 模拟重复的批量同步任务
    sync_tasks = [
        {
            "id": 1,
            "source": "db1",
            "target": "db2",
            "sync_type": "incremental",
            "filters": {"date": "2024-01-01"},
            "schedule": "daily"
        },
        {
            "id": 2,  # 重复任务（内容相同，ID不同）
            "source": "db1",
            "target": "db2",
            "sync_type": "incremental",
            "filters": {"date": "2024-01-01"},
            "schedule": "daily"
        },
        {
            "id": 3,
            "source": "db3",
            "target": "db4",
            "sync_type": "full",
            "filters": {},
            "schedule": "hourly"
        }
    ]
    filter_instance = BatchSyncFilter()
    unique_tasks = filter_instance.filter_duplicates(sync_tasks)
    print(f"原始任务数: {len(sync_tasks)}")
    print(f"去重后任务数: {len(unique_tasks)}")
    print("唯一任务:")
    for task in unique_tasks:
        print(f"  - ID: {task['id']}, Source: {task['source']}")

方案2：基于数据库操作去重

import sqlite3
from datetime import datetime
class DatabaseBatchFilter:
    def __init__(self, db_path: str = "sync_tasks.db"):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
        self.create_tables()
    def create_tables(self):
        """创建任务表"""
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS sync_tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                source TEXT NOT NULL,
                target TEXT NOT NULL,
                sync_type TEXT,
                schedule TEXT,
                created_at TIMESTAMP,
                UNIQUE(source, target, sync_type, schedule)
            )
        ''')
        self.conn.commit()
    def insert_unique_task(self, task: Dict) -> bool:
        """
        插入任务，返回是否成功（False表示重复）
        """
        try:
            self.cursor.execute('''
                INSERT OR IGNORE INTO sync_tasks 
                (source, target, sync_type, schedule, created_at)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                task['source'],
                task['target'],
                task.get('sync_type'),
                task.get('schedule'),
                datetime.now()
            ))
            self.conn.commit()
            return self.cursor.rowcount > 0
        except sqlite3.IntegrityError:
            return False
    def batch_insert_unique(self, tasks: List[Dict]) -> Dict[str, int]:
        """
        批量插入去重
        """
        stats = {"success": 0, "duplicates": 0, "failed": 0}
        for task in tasks:
            if self.insert_unique_task(task):
                stats["success"] += 1
            else:
                stats["duplicates"] += 1
        return stats
    def get_all_tasks(self) -> List[Dict]:
        """获取所有任务"""
        self.cursor.execute("SELECT * FROM sync_tasks")
        columns = [desc[0] for desc in self.cursor.description]
        return [dict(zip(columns, row)) for row in self.cursor.fetchall()]
# 使用示例
db_filter = DatabaseBatchFilter()
tasks = [
    {"source": "db1", "target": "db2", "sync_type": "incremental", "schedule": "daily"},
    {"source": "db1", "target": "db2", "sync_type": "incremental", "schedule": "daily"},  # 重复
    {"source": "db3", "target": "db4", "sync_type": "full", "schedule": "hourly"}
]
stats = db_filter.batch_insert_unique(tasks)
print(f"统计: 成功={stats['success']}, 重复={stats['duplicates']}, 失败={stats['failed']}")

方案3：基于时间窗口去重

from datetime import datetime, timedelta
from collections import defaultdict
class TimeWindowBatchFilter:
    def __init__(self, time_window_minutes: int = 5):
        self.time_window = timedelta(minutes=time_window_minutes)
        self.task_history = defaultdict(list)
    def is_within_time_window(self, task_key: str, current_time: datetime) -> bool:
        """检查是否在时间窗口内"""
        if task_key in self.task_history:
            last_time = self.task_history[task_key][-1]
            return (current_time - last_time) <= self.time_window
        return False
    def filter_by_timestamp(self, tasks: List[Dict]) -> List[Dict]:
        """
        基于时间戳过滤重复任务
        """
        unique_tasks = []
        current_time = datetime.now()
        for task in tasks:
            # 生成任务键
            task_key = f"{task['source']}:{task['target']}:{task.get('sync_type', 'unknown')}"
            if not self.is_within_time_window(task_key, current_time):
                unique_tasks.append(task)
                self.task_history[task_key].append(current_time)
        return unique_tasks
# 使用示例
time_filter = TimeWindowBatchFilter(time_window_minutes=10)
# 模拟带有时间戳的同步任务
recent_tasks = [
    {"source": "db1", "target": "db2", "sync_type": "incremental", "timestamp": "2024-01-01 10:00:00"},
    {"source": "db1", "target": "db2", "sync_type": "incremental", "timestamp": "2024-01-01 10:05:00"},  # 可能重复
    {"source": "db1", "target": "db2", "sync_type": "incremental", "timestamp": "2024-01-01 10:15:00"},  # 可接受
]
filtered_tasks = time_filter.filter_by_timestamp(recent_tasks)
print(f"过滤后任务数: {len(filtered_tasks)}")

方案4：综合去重管理器

import hashlib
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
class BatchSyncDeduplicator:
    """综合去重管理器"""
    def __init__(self, 
                 use_content_hash: bool = True,
                 use_timestamp: bool = True,
                 time_window_minutes: int = 5):
        self.use_content_hash = use_content_hash
        self.use_timestamp = use_timestamp
        self.time_window = timedelta(minutes=time_window_minutes)
        self.content_db = set()
        self.timestamp_db = {}
        self.stats = {"total": 0, "removed": 0, "kept": 0}
    def calculate_content_hash(self, task: Dict) -> str:
        """计算内容哈希"""
        # 移除动态字段（如ID、时间戳）
        static_fields = {k: v for k, v in task.items() 
                        if k not in ['id', 'timestamp', 'created_at']}
        return hashlib.sha256(
            json.dumps(static_fields, sort_keys=True).encode()
        ).hexdigest()
    def check_and_add(self, task: Dict) -> bool:
        """
        检查并添加任务
        返回True表示任务唯一，False表示重复
        """
        self.stats["total"] += 1
        # 1. 内容去重
        if self.use_content_hash:
            content_hash = self.calculate_content_hash(task)
            if content_hash in self.content_db:
                self.stats["removed"] += 1
                return False
            self.content_db.add(content_hash)
        # 2. 时间窗口去重
        if self.use_timestamp:
            task_key = f"{task.get('source')}:{task.get('target')}"
            current_time = datetime.now()
            if task_key in self.timestamp_db:
                last_time = self.timestamp_db[task_key]
                if (current_time - last_time) <= self.time_window:
                    self.stats["removed"] += 1
                    return False
            self.timestamp_db[task_key] = current_time
        self.stats["kept"] += 1
        return True
    def process_batch(self, tasks: List[Dict]) -> List[Dict]:
        """处理批量任务"""
        unique_tasks = []
        for task in tasks:
            if self.check_and_add(task):
                unique_tasks.append(task)
        return unique_tasks
    def reset(self):
        """重置去重器状态"""
        self.content_db.clear()
        self.timestamp_db.clear()
        self.stats = {"total": 0, "removed": 0, "kept": 0}
    def get_statistics(self) -> Dict:
        """获取去重统计信息"""
        return {
            **self.stats,
            "removal_rate": f"{self.stats['removed'] / max(self.stats['total'], 1) * 100:.2f}%"
        }
# 使用示例
if __name__ == "__main__":
    # 创建去重器
    deduplicator = BatchSyncDeduplicator(
        use_content_hash=True,
        use_timestamp=True,
        time_window_minutes=5
    )
    # 模拟重复的同步任务
    batch_tasks = [
        {"source": "api_v1", "target": "elasticsearch", "sync_type": "incremental", "data": {"users": True}},
        {"source": "api_v1", "target": "elasticsearch", "sync_type": "incremental", "data": {"users": True}},  # 内容重复
        {"source": "api_v2", "target": "elasticsearch", "sync_type": "full", "data": {"products": True}},
        {"source": "api_v2", "target": "elasticsearch", "sync_type": "full", "data": {"products": True}},  # 内容重复
        {"source": "api_v3", "target": "mongodb", "sync_type": "incremental", "data": {"orders": True}},
    ]
    # 处理任务
    unique = deduplicator.process_batch(batch_tasks)
    # 输出结果
    print(f"原始任务数: {len(batch_tasks)}")
    print(f"唯一任务数: {len(unique)}")
    print(f"去重统计: {deduplicator.get_statistics()}")
    print("\n保留的任务:")
    for task in unique:
        print(f"  - {task['source']} -> {task['target']} ({task['sync_type']})")