怎样实现接口异常统计脚本

wen 实用脚本 2026-06-30 289

本文目录导读：

怎样实现接口异常统计脚本

基础版本 - Python实现
实时监控版本
Web API版本（Flask）
配置文件
使用示例

我来为您提供一个完整的接口异常统计脚本实现方案,这个脚本可以监控API接口的异常情况并生成统计报告。

基础版本 - Python实现

日志文件监控脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
接口异常统计脚本
"""
import re
import json
import time
from collections import defaultdict
from datetime import datetime, timedelta
import requests
class APIMonitor:
    def __init__(self, log_file=None, api_endpoint=None):
        self.log_file = log_file
        self.api_endpoint = api_endpoint
        self.stats = {
            'total_requests': 0,
            'success': 0,
            'failed': 0,
            'error_codes': defaultdict(int),
            'response_times': [],
            'api_stats': defaultdict(lambda: {
                'total': 0,
                'success': 0,
                'failed': 0,
                'error_codes': defaultdict(int),
                'avg_response_time': 0
            })
        }
    def parse_log_line(self, line):
        """解析日志行，提取接口信息"""
        # 示例日志格式: [2024-01-15 10:30:45] GET /api/users 200 0.023s
        pattern = r'\[(.*?)\]\s+(\w+)\s+(/\S*)\s+(\d+)\s+([\d.]+)s'
        match = re.match(pattern, line)
        if match:
            return {
                'timestamp': match.group(1),
                'method': match.group(2),
                'endpoint': match.group(3),
                'status_code': int(match.group(4)),
                'response_time': float(match.group(5))
            }
        return None
    def analyze_single_request(self, request_info):
        """分析单个请求"""
        if not request_info:
            return
        self.stats['total_requests'] += 1
        # 更新整体统计
        if request_info['status_code'] < 400:
            self.stats['success'] += 1
        else:
            self.stats['failed'] += 1
            self.stats['error_codes'][request_info['status_code']] += 1
        self.stats['response_times'].append(request_info['response_time'])
        # 更新各接口统计
        api_key = f"{request_info['method']} {request_info['endpoint']}"
        api_stat = self.stats['api_stats'][api_key]
        api_stat['total'] += 1
        api_stat['avg_response_time'] = (
            (api_stat['avg_response_time'] * (api_stat['total'] - 1) + 
             request_info['response_time']) / api_stat['total']
        )
        if request_info['status_code'] < 400:
            api_stat['success'] += 1
        else:
            api_stat['failed'] += 1
            api_stat['error_codes'][request_info['status_code']] += 1
    def analyze_log_file(self, log_file=None):
        """分析日志文件"""
        print(f"开始分析日志文件: {log_file or self.log_file}")
        try:
            with open(log_file or self.log_file, 'r', encoding='utf-8') as f:
                for line in f:
                    request_info = self.parse_log_line(line.strip())
                    self.analyze_single_request(request_info)
            print("日志分析完成！")
        except FileNotFoundError:
            print(f"错误: 日志文件 {log_file or self.log_file} 不存在")
        except Exception as e:
            print(f"错误: 分析日志时发生异常 - {e}")
    def analyze_from_api(self, url=None):
        """从API接口获取数据进行分析"""
        print(f"从API获取数据: {url or self.api_endpoint}")
        try:
            response = requests.get(url or self.api_endpoint)
            response.raise_for_status()
            data = response.json()
            for request_info in data:
                self.analyze_single_request(request_info)
            print("API数据分析完成！")
        except requests.RequestException as e:
            print(f"错误: 请求API失败 - {e}")
    def get_summary_stats(self):
        """获取统计摘要"""
        if self.stats['total_requests'] == 0:
            return "暂无数据"
        success_rate = (self.stats['success'] / self.stats['total_requests']) * 100
        failed_rate = (self.stats['failed'] / self.stats['total_requests']) * 100
        return {
            'total_requests': self.stats['total_requests'],
            'success_rate': f"{success_rate:.2f}%",
            'failed_rate': f"{failed_rate:.2f}%",
            'avg_response_time': f"{sum(self.stats['response_times']) / len(self.stats['response_times']):.3f}s",
            'max_response_time': f"{max(self.stats['response_times']):.3f}s",
            'min_response_time': f"{min(self.stats['response_times']):.3f}s"
        }
    def generate_report(self, format='text'):
        """生成统计报告"""
        if format == 'json':
            return json.dumps(self.stats, indent=2, ensure_ascii=False)
        # 文本格式报告
        report = []
        report.append("=" * 60)
        report.append("接口异常统计报告")
        report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append("=" * 60)
        # 整体统计
        summary = self.get_summary_stats()
        report.append("\n【整体统计】")
        for key, value in summary.items():
            report.append(f"  {key}: {value}")
        # 错误码分布
        if self.stats['error_codes']:
            report.append("\n【错误码分布】")
            for code, count in sorted(self.stats['error_codes'].items()):
                report.append(f"  HTTP {code}: {count} 次 ({count/self.stats['total_requests']*100:.1f}%)")
        # 各接口统计
        report.append("\n【各接口详细统计】")
        for api_key, api_stat in sorted(self.stats['api_stats'].items()):
            success_rate = (api_stat['success'] / api_stat['total']) * 100 if api_stat['total'] > 0 else 0
            report.append(f"\n  API: {api_key}")
            report.append(f"    请求数: {api_stat['total']}")
            report.append(f"    成功率: {success_rate:.1f}%")
            report.append(f"    平均响应时间: {api_stat['avg_response_time']:.3f}s")
            if api_stat['error_codes']:
                report.append("    错误码: " + 
                             ", ".join([f"{code}: {count}次" 
                                       for code, count in api_stat['error_codes'].items()]))
        report.append("\n" + "=" * 60)
        return "\n".join(report)
# 使用示例
if __name__ == "__main__":
    # 1. 从日志文件分析
    monitor = APIMonitor(log_file="api_logs.txt")
    monitor.analyze_log_file()
    # 2. 生成报告
    report = monitor.generate_report()
    print(report)
    # 3. 保存报告到文件
    with open("api_monitor_report.txt", "w", encoding="utf-8") as f:
        f.write(report)

实时监控版本

#!/usr/bin/env python3
"""
实时接口异常监控脚本
"""
import time
import asyncio
from datetime import datetime
from collections import deque
class RealTimeAPIMonitor(APIMonitor):
    def __init__(self, window_size=300):  # 5分钟滑动窗口
        super().__init__()
        self.window_size = window_size
        self.recent_stats = deque(maxlen=1000)  # 保存最近1000次请求
    def record_request(self, endpoint, status_code, response_time):
        """记录实时请求"""
        request_info = {
            'timestamp': time.time(),
            'endpoint': endpoint,
            'status_code': status_code,
            'response_time': response_time
        }
        self.recent_stats.append(request_info)
        self.analyze_single_request({
            'method': 'GET',
            'endpoint': endpoint,
            'status_code': status_code,
            'response_time': response_time
        })
    def check_alerts(self, threshold=0.1, time_window=60):
        """检查是否需要告警"""
        now = time.time()
        start_time = now - time_window
        # 获取指定时间窗口内的请求
        window_requests = [
            r for r in self.recent_stats 
            if r['timestamp'] >= start_time
        ]
        if not window_requests:
            return None
        # 计算异常率
        failed = sum(1 for r in window_requests if r['status_code'] >= 400)
        total = len(window_requests)
        error_rate = failed / total if total > 0 else 0
        if error_rate > threshold:
            return {
                'alert': '异常率过高',
                'time_window': f"{time_window}秒",
                'error_rate': f"{error_rate:.2%}",
                'threshold': f"{threshold:.0%}",
                'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
        return None
    async def continuous_monitor(self, check_interval=5):
        """持续监控"""
        print(f"开始实时监控 (检查间隔: {check_interval}秒)")
        try:
            while True:
                # 模拟接收请求数据
                # 实际应用中这里应该是从消息队列或日志流读取数据
                # 检查告警
                alert = self.check_alerts()
                if alert:
                    print(f"\n⚠️ 告警: {alert}")
                # 输出当前状态
                self.print_status()
                await asyncio.sleep(check_interval)
        except KeyboardInterrupt:
            print("\n\n监控停止")
    def print_status(self):
        """打印当前状态"""
        now = datetime.now().strftime('%H:%M:%S')
        total = self.stats['total_requests']
        if total > 0:
            success_rate = (self.stats['success'] / total) * 100
            print(f"[{now}] 总请求: {total}, 成功率: {success_rate:.1f}%")
# 实时监控使用示例
async def main():
    monitor = RealTimeAPIMonitor()
    # 模拟请求数据
    for i in range(100):
        import random
        endpoint = f"/api/users/{random.randint(1,100)}"
        status_code = random.choices(
            [200, 201, 400, 401, 403, 404, 500],
            weights=[70, 10, 5, 3, 2, 5, 5]
        )[0]
        response_time = random.uniform(0.1, 2.0)
        monitor.record_request(endpoint, status_code, response_time)
        time.sleep(random.uniform(0.1, 0.5))
        # 每10个请求检查一次告警
        if i % 10 == 0:
            alert = monitor.check_alerts()
            if alert:
                print(f"警报: {alert}")
    # 生成最终报告
    print("\n" + monitor.generate_report())
if __name__ == "__main__":
    asyncio.run(main())

Web API版本（Flask）

#!/usr/bin/env python3
"""
接口异常统计Web服务
"""
from flask import Flask, request, jsonify
import json
from datetime import datetime
app = Flask(__name__)
monitor = APIMonitor()
@app.route('/api/record', methods=['POST'])
def record_request():
    """记录请求"""
    data = request.json
    if not data:
        return jsonify({'error': '无效的请求数据'}), 400
    required_fields = ['endpoint', 'status_code', 'response_time']
    for field in required_fields:
        if field not in data:
            return jsonify({'error': f'缺少字段: {field}'}), 400
    monitor.analyze_single_request({
        'method': data.get('method', 'GET'),
        'endpoint': data['endpoint'],
        'status_code': data['status_code'],
        'response_time': data['response_time']
    })
    return jsonify({'status': 'success'})
@app.route('/api/stats', methods=['GET'])
def get_stats():
    """获取统计信息"""
    return jsonify(monitor.get_summary_stats())
@app.route('/api/report', methods=['GET'])
def get_report():
    """获取完整报告"""
    format_type = request.args.get('format', 'json')
    return monitor.generate_report(format=format_type)
@app.route('/api/alerts', methods=['GET'])
def check_alerts():
    """检查告警"""
    threshold = float(request.args.get('threshold', 0.1))
    time_window = int(request.args.get('time_window', 60))
    alert = monitor.check_alerts(threshold=threshold, time_window=time_window)
    if alert:
        return jsonify(alert)
    return jsonify({'status': 'normal'})
if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

配置文件

# config.yaml
monitor:
  log_file: "/var/log/api/access.log"
  check_interval: 60  # 检查间隔（秒）
alerts:
  error_rate_threshold: 0.1  # 10%错误率触发告警
  time_window: 300  # 5分钟窗口
reports:
  output_dir: "/var/reports/"
  save_report: true
  email_report: false
  email_config:
    smtp_server: "smtp.example.com"
    smtp_port: 587
    username: "your-email@example.com"
    password: "your-password"
    recipients: ["admin@example.com"]

使用示例

安装依赖

pip install flask requests pyyaml

基本使用

# 1. 分析日志文件
python api_monitor.py
# 2. 启动Web服务
python api_monitor_web.py
# 3. 发送请求记录
curl -X POST http://localhost:5000/api/record \
  -H "Content-Type: application/json" \
  -d '{"endpoint": "/api/users", "status_code": 200, "response_time": 0.234}'

这个脚本提供了完善的接口异常统计功能,包括：