本文目录导读：
我来详细介绍Python脚本校验多环境数据一致性的方法。
基础校验框架
import hashlib
import json
from typing import Dict, List, Any
import requests
from datetime import datetime
class DataConsistencyChecker:
"""数据一致性校验器"""
def __init__(self, config: Dict):
self.config = config
self.results = []
def get_data_from_env(self, env: str, endpoint: str, params: Dict = None) -> Dict:
"""从指定环境获取数据"""
url = self.config[env]['base_url'] + endpoint
headers = self.config[env].get('headers', {})
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
def compare_data(self, data1: Dict, data2: Dict, path: str = "") -> List[Dict]:
"""递归比较两个数据对象的差异"""
differences = []
# 获取所有键
all_keys = set(list(data1.keys()) + list(data2.keys()))
for key in all_keys:
current_path = f"{path}.{key}" if path else key
if key not in data1:
differences.append({
'path': current_path,
'type': 'missing_in_first',
'value': data2[key]
})
elif key not in data2:
differences.append({
'path': current_path,
'type': 'missing_in_second',
'value': data1[key]
})
else:
if isinstance(data1[key], dict) and isinstance(data2[key], dict):
# 递归比较字典
differences.extend(
self.compare_data(data1[key], data2[key], current_path)
)
elif isinstance(data1[key], list) and isinstance(data2[key], list):
# 比较列表
list_diff = self.compare_lists(
data1[key], data2[key], current_path
)
differences.extend(list_diff)
elif data1[key] != data2[key]:
differences.append({
'path': current_path,
'type': 'value_mismatch',
'value1': data1[key],
'value2': data2[key]
})
return differences
def compare_lists(self, list1: List, list2: List, path: str) -> List[Dict]:
"""比较两个列表"""
differences = []
max_len = max(len(list1), len(list2))
for i in range(max_len):
current_path = f"{path}[{i}]"
if i >= len(list1):
differences.append({
'path': current_path,
'type': 'missing_in_first',
'value': list2[i]
})
elif i >= len(list2):
differences.append({
'path': current_path,
'type': 'missing_in_second',
'value': list1[i]
})
else:
if isinstance(list1[i], dict) and isinstance(list2[i], dict):
differences.extend(
self.compare_data(list1[i], list2[i], current_path)
)
elif list1[i] != list2[i]:
differences.append({
'path': current_path,
'type': 'value_mismatch',
'value1': list1[i],
'value2': list2[i]
})
return differences
高级校验方法
import hashlib
import pandas as pd
from typing import Set, Tuple
import deepdiff
class AdvancedConsistencyChecker:
"""高级数据一致性校验器"""
def __init__(self):
self.check_results = []
def checksum_compare(self, data1: str, data2: str) -> bool:
"""使用MD5校验和比较"""
hash1 = hashlib.md5(data1.encode()).hexdigest()
hash2 = hashlib.md5(data2.encode()).hexdigest()
return hash1 == hash2
def deep_diff_compare(self, data1: Dict, data2: Dict,
exclude_paths: List[str] = None) -> Dict:
"""使用deepdiff进行深度比较"""
from deepdiff import DeepDiff
diff = DeepDiff(
data1, data2,
exclude_paths=exclude_paths,
ignore_order=True,
significant_digits=2
)
return diff.to_dict() if diff else {}
def sql_result_compare(self, df1: pd.DataFrame, df2: pd.DataFrame,
key_columns: List[str]) -> List[Dict]:
"""比较两个SQL查询结果"""
differences = []
# 合并数据框
merged = pd.merge(
df1, df2,
on=key_columns,
how='outer',
suffixes=('_env1', '_env2'),
indicator=True
)
# 找出差异
for idx, row in merged.iterrows():
if row['_merge'] == 'left_only':
differences.append({
'type': 'missing_in_env2',
'keys': {col: row[col] for col in key_columns},
'data': row.to_dict()
})
elif row['_merge'] == 'right_only':
differences.append({
'type': 'missing_in_env1',
'keys': {col: row[col] for col in key_columns},
'data': row.to_dict()
})
else:
# 比较非关键列
value_cols = [col for col in merged.columns
if col not in key_columns + ['_merge']]
for col in value_cols:
val1 = row[f"{col}_env1"]
val2 = row[f"{col}_env2"]
if pd.isna(val1) and pd.isna(val2):
continue
if val1 != val2:
differences.append({
'type': 'value_mismatch',
'keys': {col: row[col] for col in key_columns},
'column': col,
'env1_value': val1,
'env2_value': val2
})
return differences
实际应用示例
import yaml
import logging
from typing import List, Optional
from concurrent.futures import ThreadPoolExecutor
class EnvDataValidator:
"""多环境数据校验器"""
def __init__(self, config_path: str):
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
self.setup_logging()
self.checker = DataConsistencyChecker(self.config.get('environments', {}))
def setup_logging(self):
"""设置日志"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('consistency_check.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def validate_single_endpoint(self, endpoint: str,
envs: List[str]) -> Dict:
"""验证单个端点在多个环境的数据一致性"""
results = {
'endpoint': endpoint,
'timestamp': datetime.now().isoformat(),
'checks': []
}
# 获取所有环境的数据
env_data = {}
for env in envs:
try:
data = self.checker.get_data_from_env(env, endpoint)
env_data[env] = data
self.logger.info(f"成功获取 {env} 环境数据: {endpoint}")
except Exception as e:
self.logger.error(f"获取 {env} 环境数据失败: {str(e)}")
return None
# 比较环境数据
for i in range(len(envs)):
for j in range(i+1, len(envs)):
env1, env2 = envs[i], envs[j]
differences = self.checker.compare_data(
env_data[env1], env_data[env2]
)
check_result = {
'environments': f"{env1} vs {env2}",
'status': 'PASS' if not differences else 'FAIL',
'differences': differences[:10], # 只记录前10个差异
'total_differences': len(differences)
}
results['checks'].append(check_result)
if differences:
self.logger.warning(
f"发现 {len(differences)} 个差异在 {env1} 和 {env2} 之间"
)
else:
self.logger.info(f"{env1} 和 {env2} 数据一致")
return results
def batch_validate(self, endpoints: List[str],
envs: List[str]) -> List[Dict]:
"""批量验证多个端点"""
all_results = []
with ThreadPoolExecutor(max_workers=5) as executor:
futures = []
for endpoint in endpoints:
future = executor.submit(
self.validate_single_endpoint, endpoint, envs
)
futures.append(future)
for future in futures:
result = future.result()
if result:
all_results.append(result)
return all_results
def generate_report(self, results: List[Dict], output_file: str):
"""生成校验报告"""
report = {
'summary': {
'total_endpoints': len(results),
'passed': sum(1 for r in results if all(
check['status'] == 'PASS'
for check in r['checks']
)),
'failed': sum(1 for r in results if any(
check['status'] == 'FAIL'
for check in r['checks']
)),
'total_differences': sum(
sum(check['total_differences'] for check in r['checks'])
for r in results
)
},
'details': results,
'generated_at': datetime.now().isoformat()
}
with open(output_file, 'w') as f:
json.dump(report, f, indent=2, default=str)
self.logger.info(f"报告已生成: {output_file}")
return report
# 示例：校验配置
def create_demo_config():
"""创建示例配置"""
config = {
'environments': {
'dev': {
'base_url': 'http://dev-api.example.com',
'headers': {'Authorization': 'Bearer dev-token'}
},
'staging': {
'base_url': 'http://staging-api.example.com',
'headers': {'Authorization': 'Bearer staging-token'}
},
'prod': {
'base_url': 'http://api.example.com',
'headers': {'Authorization': 'Bearer prod-token'}
}
},
'check_settings': {
'ignore_fields': ['timestamp', 'updated_at', 'created_at'],
'tolerance': 0.01, # 数值比较容差
'timeout': 30 # 请求超时时间
}
}
with open('env_config.yaml', 'w') as f:
yaml.dump(config, f)
return config
# 使用示例
if __name__ == "__main__":
# 创建配置
create_demo_config()
# 初始化校验器
validator = EnvDataValidator('env_config.yaml')
# 验证单端点
result = validator.validate_single_endpoint(
'/api/users',
['dev', 'staging', 'prod']
)
if result:
print(f"校验状态: {result['checks']}")
# 批量验证
endpoints = ['/api/users', '/api/orders', '/api/products']
all_results = validator.batch_validate(endpoints, ['dev', 'staging'])
# 生成报告
validator.generate_report(all_results, 'consistency_report.json')
特定场景校验
class SpecificScenarioChecker:
"""特定场景数据一致性校验"""
@staticmethod
def validate_time_series(data1: pd.Series, data2: pd.Series,
tolerance: float = 0.01) -> Dict:
"""校验时间序列数据"""
from sklearn.metrics import mean_squared_error
# 确保时间索引对齐
aligned1, aligned2 = data1.align(data2, join='outer')
# 计算差异指标
mse = mean_squared_error(
aligned1.fillna(0),
aligned2.fillna(0)
)
# 找出差异点
diff_points = []
for idx in aligned1.index:
if pd.notna(aligned1[idx]) and pd.notna(aligned2[idx]):
if abs(aligned1[idx] - aligned2[idx]) > tolerance:
diff_points.append({
'timestamp': idx,
'value1': aligned1[idx],
'value2': aligned2[idx],
'difference': aligned1[idx] - aligned2[idx]
})
return {
'mse': mse,
'total_points': len(aligned1),
'diff_points': len(diff_points),
'is_consistent': mse < tolerance ** 2
}
@staticmethod
def validate_data_quality(data1: pd.DataFrame, data2: pd.DataFrame) -> Dict:
"""验证数据质量一致性"""
quality_checks = {
'null_percentage': {},
'unique_counts': {},
'value_distributions': {}
}
for col in data1.columns:
if col in data2.columns:
# 空值率
null_pct1 = data1[col].isna().sum() / len(data1)
null_pct2 = data2[col].isna().sum() / len(data2)
quality_checks['null_percentage'][col] = {
'env1': null_pct1,
'env2': null_pct2,
'match': abs(null_pct1 - null_pct2) < 0.01
}
# 唯一值统计
unique1 = data1[col].nunique()
unique2 = data2[col].nunique()
quality_checks['unique_counts'][col] = {
'env1': unique1,
'env2': unique2,
'match': unique1 == unique2
}
return quality_checks
使用建议
- 配置管理：使用YAML/JSON配置文件管理环境信息
- 并行处理：使用ThreadPoolExecutor并发获取数据
- 容错处理：单个环境失败不影响整体校验
- 报告生成：生成详细的差异报告便于追踪
- 阈值设置：数值类型设置合理容差
- 排除字段：忽略时间戳等动态字段
这个框架支持多种数据格式和场景,可根据实际需求扩展。
