本文目录导读：
我来为您提供一个完整的接口异常统计脚本实现方案,这个脚本可以监控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}'
这个脚本提供了完善的接口异常统计功能,包括：
- 日志文件分析
- 实时监控
- Web API接口
- 告警机制
- 多种报告格式
您可以根据实际需求调整和扩展这些功能。
