本文目录导读：
我可以为您提供一个基础的代理访问记录分析脚本方案,以下是一个Python脚本示例，用于分析常见的代理日志格式：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
代理访问日志分析脚本
支持的日志格式：Squid, Nginx反向代理, Apache等常见格式
"""
import re
import sys
import os
from datetime import datetime
from collections import Counter, defaultdict
import argparse
class ProxyLogAnalyzer:
"""代理日志分析器"""
def __init__(self, log_file):
self.log_file = log_file
self.total_requests = 0
self.successful_requests = 0
self.failed_requests = 0
self.status_codes = Counter()
self.source_ips = Counter()
self.requested_urls = Counter()
self.user_agents = Counter()
self.hourly_requests = defaultdict(int)
self.error_logs = []
def parse_line(self, line):
"""解析单行日志（支持多种格式）"""
# 尝试匹配标准组合日志格式 (Nginx/Apache)
pattern = r'(\d+\.\d+\.\d+\.\d+)\s+.*?\s+.*?\s+\[(.*?)\]\s+"(.*?)"\s+(\d+)\s+(\d+)\s+"(.*?)"\s+"(.*?)"'
match = re.match(pattern, line)
if not match:
# 尝试匹配Squid日志格式
pattern = r'(\d+\.\d+\.\d+\.\d+)\s+.*?\s+.*?\s+(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2})\s+(\w+)\s+(.*?)\s+(.*?)\s+(\d+)\s+(\d+)\s+(\d+)'
match = re.match(pattern, line)
if not match:
return None
try:
if len(match.groups()) == 7: # Nginx格式
ip = match.group(1)
timestamp = match.group(2)
request = match.group(3)
status = int(match.group(4))
size = int(match.group(5))
referer = match.group(6)
user_agent = match.group(7)
else: # Squid格式
ip = match.group(1)
timestamp = match.group(2)
method = match.group(3)
url = match.group(4)
user_agent = match.group(5)
status = int(match.group(6))
size = int(match.group(7))
request = f"{method} {url}"
# 解析时间
dt = datetime.strptime(timestamp, '%d/%b/%Y:%H:%M:%S %z')
return {
'ip': ip,
'timestamp': dt,
'request': request,
'status': status,
'size': size,
'user_agent': user_agent[:100] if user_agent else 'unknown'
}
except (ValueError, IndexError) as e:
self.error_logs.append((line, str(e)))
return None
def analyze(self):
"""分析日志文件"""
try:
with open(self.log_file, 'r', encoding='utf-8', errors='ignore') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
parsed = self.parse_line(line)
if not parsed:
continue
self.total_requests += 1
# 统计状态码
self.status_codes[parsed['status']] += 1
if parsed['status'] < 400:
self.successful_requests += 1
else:
self.failed_requests += 1
# 统计IP
self.source_ips[parsed['ip']] += 1
# 统计URL
self.requested_urls[parsed['request']] += 1
# 统计User-Agent
ua = parsed['user_agent']
self.user_agents[ua] += 1
# 按小时统计
hour = parsed['timestamp'].hour
self.hourly_requests[hour] += 1
except FileNotFoundError:
print(f"错误：文件 {self.log_file} 不存在")
sys.exit(1)
except PermissionError:
print(f"错误：没有权限读取文件 {self.log_file}")
sys.exit(1)
except Exception as e:
print(f"读取文件时出错: {e}")
sys.exit(1)
def generate_report(self):
"""生成分析报告"""
if self.total_requests == 0:
print("未找到有效的日志记录")
return
print("\n" + "="*60)
print("代理访问日志分析报告")
print("="*60)
# 基本信息
print(f"\n📊 基本统计")
print(f" - 总请求数: {self.total_requests}")
print(f" - 成功请求(2xx/3xx): {self.successful_requests} ({self.successful_requests/self.total_requests*100:.1f}%)")
print(f" - 失败请求(4xx/5xx): {self.failed_requests} ({self.failed_requests/self.total_requests*100:.1f}%)")
# 状态码分布
print(f"\n📈 状态码分布")
for code, count in self.status_codes.most_common():
print(f" {code}: {count}次 ({count/self.total_requests*100:.1f}%)")
# 来源IP Top 10
print(f"\n🌐 来源IP Top 10")
for ip, count in self.source_ips.most_common(10):
print(f" {ip}: {count}次")
# 请求URL Top 10
print(f"\n🔗 请求URL Top 10")
for url, count in self.requested_urls.most_common(10):
print(f" {url[:80]}: {count}次")
# 用户代理 Top 10
print(f"\n🤖 用户代理 Top 10")
for ua, count in self.user_agents.most_common(10):
print(f" {ua[:50]}: {count}次")
# 小时分布
print(f"\n⏰ 小时请求分布")
for hour in range(24):
count = self.hourly_requests.get(hour, 0)
bar = '█' * min(count // 10, 50)
print(f" {hour:02d}:00 - {hour:02d}:59: {count}次 {bar}")
# 解析错误
if self.error_logs:
print(f"\n⚠️ 解析错误记录 ({len(self.error_logs)}条)")
for line, error in self.error_logs[:5]:
print(f" 行内容: {line[:80]}...")
print(f" 错误: {error}")
print("\n" + "="*60)
def main():
parser = argparse.ArgumentParser(
description='代理访问日志分析工具',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
使用示例:
python log_analyzer.py access.log
python log_analyzer.py /var/log/squid/access.log
python log_analyzer.py proxy.log -f csv
"""
)
parser.add_argument('logfile', help='代理日志文件路径')
parser.add_argument('-f', '--format', choices=['text', 'csv'],
default='text', help='输出格式 (默认: text)')
args = parser.parse_args()
print(f"开始分析日志文件: {args.logfile}")
print(f"日志格式: 自动检测")
analyzer = ProxyLogAnalyzer(args.logfile)
analyzer.analyze()
analyzer.generate_report()
if __name__ == "__main__":
main()
使用说明
基本使用
# 分析代理日志文件 python proxy_log_analyzer.py /var/log/squid/access.log # 指定日志格式（目前自动检测） python proxy_log_analyzer.py access.log -f text
支持的日志格式
- Nginx/Apache 标准组合日志格式
- Squid 代理日志格式
- 可扩展支持其他格式
分析功能
| 功能 | 说明 |
|---|---|
| 总请求数 | 统计总请求量 |
| 成功/失败率 | 按HTTP状态码分类 |
| 状态码分布 | 各状态码占比 |
| IP统计 | 访问来源Top 10 |
| URL统计 | 请求路径Top 10 |
| User-Agent | 客户端类型分布 |
| 小时分布 | 24小时请求热力图 |
| 错误日志 | 解析失败记录 |
扩展用法
实时监控
# 使用tail -f 监控实时日志
import subprocess
log_analyzer = ProxyLogAnalyzer("access.log")
process = subprocess.Popen(['tail', '-f', 'access.log'],
stdout=subprocess.PIPE)
while True:
line = process.stdout.readline().decode()
if line:
log_analyzer.parse_line(line)
异常检测
# 检测异常IP行为
from collections import defaultdict
class AnomalyDetector:
def detect_ddos(self, analyzer, threshold=100):
"""检测潜在的DDoS攻击"""
for ip, count in analyzer.source_ips.items():
if count > threshold:
print(f"警告: IP {ip} 请求次数异常 ({count}次)")
def detect_scanning(self, analyzer, threshold=10):
"""检测端口扫描"""
# 分析404错误分布
error_ips = defaultdict(list)
for log in analyzer.error_logs:
error_ips[log[0]].append(log[1])
# 进一步分析...
注意事项
- 日志格式：不同代理服务器日志格式不同，可能需要调整解析规则
- 文件大小：处理大文件时考虑使用流式处理
- 性能优化：对于GB级日志，建议使用分批处理
- 隐私保护：注意IP等敏感信息的处理合规性
此脚本提供了一个基础框架,您可以根据实际需求添加：
- 地理IP查询
- 用户行为分析
- 安全威胁检测
- 实时告警功能
- 数据库存储分析结果
