本文目录导读：
我来帮你编写一个Python脚本来统计各模块的同步成功率。
基础版本
假设数据格式为CSV或JSON,这里提供几种常见场景的解决方案：
从CSV文件统计
import pandas as pd
import csv
from collections import defaultdict
def analyze_sync_from_csv(csv_file_path):
"""
从CSV文件统计同步成功率
CSV格式：module, status, timestamp
status: success/failed
"""
# 方法1：使用pandas
try:
df = pd.read_csv(csv_file_path)
return calculate_success_rate(df)
except:
# 方法2：使用csv模块
data = defaultdict(lambda: {'success': 0, 'failed': 0})
with open(csv_file_path, 'r') as file:
reader = csv.DictReader(file)
for row in reader:
module = row.get('module', 'unknown')
status = row.get('status', '').strip().lower()
if status == 'success':
data[module]['success'] += 1
elif status == 'failed':
data[module]['failed'] += 1
return calculate_stats(data)
def calculate_success_rate(data):
"""计算成功率"""
results = {}
if isinstance(data, pd.DataFrame):
# pandas DataFrame处理
for module in data['module'].unique():
module_data = data[data['module'] == module]
total = len(module_data)
success = len(module_data[module_data['status'] == 'success'])
rate = (success / total * 100) if total > 0 else 0
results[module] = {
'total': total,
'success': success,
'failed': total - success,
'success_rate': round(rate, 2)
}
else:
# dict数据
for module, stats in data.items():
total = stats['success'] + stats['failed']
rate = (stats['success'] / total * 100) if total > 0 else 0
results[module] = {
'total': total,
'success': stats['success'],
'failed': stats['failed'],
'success_rate': round(rate, 2)
}
return results
def print_statistics(results):
"""打印统计结果"""
print(f"{'模块名称':<20} {'总数':<8} {'成功':<8} {'失败':<8} {'成功率':<10}")
print("=" * 54)
for module, stats in sorted(results.items()):
print(f"{module:<20} {stats['total']:<8} {stats['success']:<8} "
f"{stats['failed']:<8} {stats['success_rate']:.2f}%")
# 主函数
def main():
# 假设CSV文件路径
csv_file = "sync_logs.csv"
# 模拟数据
sample_data = [
{"module": "user_sync", "status": "success", "timestamp": "2024-01-01 10:00:00"},
{"module": "user_sync", "status": "failed", "timestamp": "2024-01-01 10:01:00"},
{"module": "order_sync", "status": "success", "timestamp": "2024-01-01 10:02:00"},
{"module": "order_sync", "status": "success", "timestamp": "2024-01-01 10:03:00"},
{"module": "inventory_sync", "status": "failed", "timestamp": "2024-01-01 10:04:00"},
]
# 写入示例数据
import json
with open(csv_file, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=['module', 'status', 'timestamp'])
writer.writeheader()
writer.writerows(sample_data)
# 分析数据
results = analyze_sync_from_csv(csv_file)
print_statistics(results)
if __name__ == "__main__":
main()
从JSON日志文件统计
import json
from datetime import datetime, timedelta
def analyze_sync_from_json(json_file_path, time_range=None):
"""
从JSON日志文件统计同步成功率
JSON格式：[{"module": "xxx", "status": "success/failed", "timestamp": "ISO格式时间"}]
time_range: 可选的时间范围，如 timedelta(hours=24)
"""
with open(json_file_path, 'r') as file:
logs = json.load(file)
stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total': 0})
for log in logs:
# 时间过滤
if time_range:
log_time = datetime.fromisoformat(log.get('timestamp'))
if datetime.now() - log_time > time_range:
continue
module = log.get('module', 'unknown')
status = log.get('status', '').lower()
stats[module]['total'] += 1
if status == 'success':
stats[module]['success'] += 1
else:
stats[module]['failed'] += 1
return calculate_stats(stats)
实时采集统计（从数据库）
import sqlite3
from datetime import datetime
def analyze_sync_from_database(db_path, table_name='sync_logs'):
"""
从SQLite数据库统计同步成功率
数据库表结构：
CREATE TABLE sync_logs (
id INTEGER PRIMARY KEY,
module TEXT,
status TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# 查询每个模块的统计信息
query = """
SELECT
module,
COUNT(*) as total,
SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as success,
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed
FROM {}
GROUP BY module
""".format(table_name)
cursor.execute(query)
rows = cursor.fetchall()
results = {}
for row in rows:
module, total, success, failed = row
rate = (success / total * 100) if total > 0 else 0
results[module] = {
'total': total,
'success': success,
'failed': failed,
'success_rate': round(rate, 2)
}
conn.close()
return results
带时间维度的详细分析
from datetime import datetime, timedelta
class SyncStatisticsAnalyzer:
def __init__(self):
self.sync_data = []
def add_sync_record(self, module, status, timestamp=None):
"""添加同步记录"""
record = {
'module': module,
'status': status,
'timestamp': timestamp or datetime.now()
}
self.sync_data.append(record)
def analyze_by_time_period(self, period='day'):
"""按时间段分析成功率"""
period_stats = defaultdict(lambda: defaultdict(lambda: {'success': 0, 'failed': 0}))
for record in self.sync_data:
if period == 'day':
time_key = record['timestamp'].strftime('%Y-%m-%d')
elif period == 'hour':
time_key = record['timestamp'].strftime('%Y-%m-%d %H:00')
elif period == 'month':
time_key = record['timestamp'].strftime('%Y-%m')
module = record['module']
status = record['status']
if status == 'success':
period_stats[time_key][module]['success'] += 1
else:
period_stats[time_key][module]['failed'] += 1
return self._calculate_period_rates(period_stats)
def identify_failure_patterns(self):
"""识别失败模式"""
failures = [r for r in self.sync_data if r['status'] == 'failed']
patterns = defaultdict(int)
for failure in failures:
hour = failure['timestamp'].hour
patterns[f"hour_{hour}"] += 1
return patterns
def generate_report(self):
"""生成统计报告"""
total_stats = self.analyze_by_time_period('day')
failure_patterns = self.identify_failure_patterns()
report = {
'total_records': len(self.sync_data),
'time_period_stats': total_stats,
'failure_patterns': failure_patterns,
'overall_rate': self._calculate_overall_rate()
}
return report
def _calculate_period_rates(self, period_stats):
"""计算各时段成功率"""
rates = {}
for period, modules in period_stats.items():
rates[period] = {}
for module, stats in modules.items():
total = stats['success'] + stats['failed']
rate = (stats['success'] / total * 100) if total > 0 else 0
rates[period][module] = round(rate, 2)
return rates
def _calculate_overall_rate(self):
"""计算总成功率"""
total = len(self.sync_data)
success = len([r for r in self.sync_data if r['status'] == 'success'])
return round((success / total * 100), 2) if total > 0 else 0
使用示例
# 示例：使用SyncStatisticsAnalyzer
analyzer = SyncStatisticsAnalyzer()
# 添加模拟数据
for i in range(100):
import random
module = random.choice(['user', 'order', 'inventory', 'payment'])
status = 'success' if random.random() > 0.2 else 'failed'
timestamp = datetime.now() - timedelta(hours=random.randint(0, 48))
analyzer.add_sync_record(module, status, timestamp)
# 生成报告
report = analyzer.generate_report()
print(json.dumps(report, indent=2, default=str))
# 获取各模块成功率
module_rates = {}
for period_stats in report['time_period_stats'].values():
for module, rate in period_stats.items():
if module not in module_rates:
module_rates[module] = []
module_rates[module].append(rate)
# 计算平均成功率
for module, rates in module_rates.items():
avg_rate = sum(rates) / len(rates)
print(f"{module} 平均成功率: {avg_rate:.2f}%")
关键功能说明
- 数据来源灵活：支持CSV、JSON、数据库等多种数据源
- 时间维度分析：可按小时、天、月统计
- 失败模式识别：分析失败的时间规律
- 报告生成：自动生成统计报告
根据你的具体需求选择合适的方法，如果需要处理实时数据流,可以添加异步处理逻辑。
