Python日志分析与告警脚本
分析Nginx/Apache访问日志，统计PV/UV、热门页面、错误状态码、异常IP，超过阈值自动告警。
详细内容
#!/usr/bin/env python3
# Python日志分析与告警脚本
# 使用方法: python3 log_analyzer.py /var/log/nginx/access.log
import re
import sys
from collections import Counter, defaultdict
from datetime import datetime
# 日志格式正则 (Nginx combined格式)
LOG_PATTERN = r'(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) (\S+) \S+" (\d{3}) (\d+) "([^"]*)" "([^"]*)"'
def parse_log(log_file):
"""解析日志文件"""
records = []
with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
for line in f:
match = re.match(LOG_PATTERN, line)
if match:
records.append({
'ip': match.group(1),
'time': match.group(2),
'method': match.group(3),
'path': match.group(4),
'status': int(match.group(5)),
'size': int(match.group(6)),
'referer': match.group(7),
'user_agent': match.group(8)
})
return records
def analyze(records):
"""分析日志数据"""
print('=' * 50)
print(' 日志分析报告')
print('=' * 50)
print(f'总请求数: {len(records)}')
# 1. PV/UV统计
ips = set(r['ip'] for r in records)
print(f'独立IP数(UV): {len(ips)}')
print(f'平均每IP请求: {len(records)/len(ips):.1f}')
# 2. 状态码统计
status_counter = Counter(r['status'] for r in records)
print('\n【状态码分布】')
for status, count in sorted(status_counter.items()):
percent = count / len(records) * 100
bar = '█' * int(percent / 2)
print(f' {status}: {count:6d} ({percent:5.1f}%) {bar}')
# 3. 错误请求(4xx/5xx)
errors = [r for r in records if r['status'] >= 400]
if errors:
print(f'\n【错误请求TOP10】')
error_paths = Counter(r['path'] for r in errors)
for path, count in error_paths.most_common(10):
print(f' {count:4d}次 {path[:60]}')
# 4. 热门页面
print('\n【热门页面TOP10】')
path_counter = Counter(r['path'] for r in records)
for path, count in path_counter.most_common(10):
print(f' {count:4d}次 {path[:60]}')
# 5. 高频IP(可能是爬虫或攻击)
print('\n【高频IP TOP10】')
ip_counter = Counter(r['ip'] for r in records)
for ip, count in ip_counter.most_common(10):
print(f' {count:4d}次 {ip}')
# 6. 异常检测
print('\n【异常检测】')
alerts = []
# 5xx错误率超过5%
server_errors = sum(1 for r in records if r['status'] >= 500)
error_rate = server_errors / len(records) * 100
if error_rate > 5:
alerts.append(f'⚠️ 5xx错误率过高: {error_rate:.1f}%')
# 单IP请求超过1000次
for ip, count in ip_counter.most_common(5):
if count > 1000:
alerts.append(f'⚠️ 高频IP: {ip} 请求{count}次')
if alerts:
for alert in alerts:
print(f' {alert}')
else:
print(' ✅ 未发现异常')
return {
'total': len(records),
'uv': len(ips),
'errors': len(errors),
'alerts': alerts
}
if __name__ == '__main__':
if len(sys.argv) < 2:
print('使用方法: python3 log_analyzer.py <日志文件路径>')
print('示例: python3 log_analyzer.py /var/log/nginx/access.log')
sys.exit(1)
log_file = sys.argv[1]
print(f'正在解析日志: {log_file}')
records = parse_log(log_file)
if not records:
print('未解析到有效日志记录')
sys.exit(1)
result = analyze(records)
# 有告警时返回非0退出码(可配合监控)
if result['alerts']:
sys.exit(1)