Python脚本如何统计各模块同步成功率

本文目录导读：

1. 基础版本
2. 使用示例
3. 关键功能说明

我来帮你编写一个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}%")

关键功能说明

1. 数据来源灵活：支持CSV、JSON、数据库等多种数据源
2. 时间维度分析：可按小时、天、月统计
3. 失败模式识别：分析失败的时间规律
4. 报告生成：自动生成统计报告

根据你的具体需求选择合适的方法，如果需要处理实时数据流,可以添加异步处理逻辑。