本文目录导读：
我来帮你编写一个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']})")
选择建议
- 方案1：适合简单的任务去重，基于内容特征
- 方案2：适合持久化存储，与数据库配合
- 方案3：适合时间敏感的同步任务
- 方案4：综合方案，功能最完整
根据你的具体需求选择合适的实现方案，如果需要更多定制化的功能,欢迎继续提问！
