—— 解析 nginx 日志+统计+可视化
"""nginx combined 日志格式：
$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent"
示例：
192.168.1.1 - - [10/Mar/2024:13:55:36 +0800] "GET /api/users HTTP/1.1" 200 1234 "-" "Mozilla/5.0"
10.0.0.5 - admin [10/Mar/2024:13:56:01 +0800] "POST /api/login HTTP/1.1" 401 95 "-" "curl/7.88"
"""
import re
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class NginxLogEntry:
ip: str
time: datetime
method: str
path: str
status: int
size: int
referer: str
user_agent: str
# 解析正则
NGINX_PATTERN = re.compile(
r'(?P<ip>\S+) - (?P<user>\S+) '
r'\[(?P<time>[^\]]+)\] '
r'"(?P<method>\S+) (?P<path>\S+) \S+" '
r'(?P<status>\d+) (?P<size>\d+) '
r'"(?P<referer>[^"]*)" '
r'"(?P<agent>[^"]*)"'
)
def parse_nginx_line(line):
"""解析单行 nginx 日志"""
m = NGINX_PATTERN.match(line.strip())
if not m:
return None
d = m.groupdict()
try:
time = datetime.strptime(d["time"], "%d/%b/%Y:%H:%M:%S %z")
except ValueError:
time = None
return NginxLogEntry(
ip=d["ip"],
time=time,
method=d["method"],
path=d["path"],
status=int(d["status"]),
size=int(d["size"]),
referer=d["referer"],
user_agent=d["agent"],
)
def parse_nginx_file(filepath):
"""解析 nginx 日志文件"""
entries = []
with open(filepath) as f:
for line in f:
entry = parse_nginx_line(line)
if entry:
entries.append(entry)
print(f"解析完成: {len(entries)} 条有效日志")
return entriesfrom collections import Counter
from typing import List
class LogAnalyzer:
"""日志分析器"""
def __init__(self, entries: List[NginxLogEntry]):
self.entries = entries
def top_ips(self, n=10):
"""访问量 Top IP"""
counter = Counter(e.ip for e in self.entries)
return counter.most_common(n)
def top_paths(self, n=10):
"""访问量 Top 路径"""
counter = Counter(e.path for e in self.entries)
return counter.most_common(n)
def status_distribution(self):
"""状态码分布"""
counter = Counter(e.status for e in self.entries)
return dict(sorted(counter.items()))
def hourly_distribution(self):
"""按小时分布"""
counter = Counter(e.time.hour for e in self.entries if e.time)
return dict(sorted(counter.items()))
def error_rate(self):
"""错误率（4xx + 5xx）"""
total = len(self.entries)
errors = sum(1 for e in self.entries if e.status >= 400)
return errors / total if total > 0 else 0
def avg_response_size(self):
"""平均响应大小"""
if not self.entries:
return 0
return sum(e.size for e in self.entries) / len(self.entries)
def total_traffic(self):
"""总流量"""
return sum(e.size for e in self.entries)
def slow_paths(self, size_threshold=100000, n=10):
"""大响应路径（可能是慢接口）"""
big = [e for e in self.entries if e.size > size_threshold]
counter = Counter(e.path for e in big)
return counter.most_common(n)
def summary(self):
"""生成汇总报告"""
return {
"total_requests": len(self.entries),
"unique_ips": len(set(e.ip for e in self.entries)),
"error_rate": f"{self.error_rate()*100:.2f}%",
"avg_size": f"{self.avg_response_size():.0f} bytes",
"total_traffic": f"{self.total_traffic()/1024/1024:.1f} MB",
"top_ips": self.top_ips(5),
"top_paths": self.top_paths(5),
"status_dist": self.status_distribution(),
}class AnomalyDetector:
"""日志异常检测"""
def __init__(self, entries: List[NginxLogEntry]):
self.entries = entries
def detect_status_anomalies(self, threshold=0.1):
"""检测异常状态码"""
status_counter = Counter(e.status for e in self.entries)
total = len(self.entries)
anomalies = []
for status, count in status_counter.items():
ratio = count / total
if status >= 500 and ratio > threshold:
anomalies.append({
"type": "高5xx比例",
"status": status,
"count": count,
"ratio": f"{ratio*100:.1f}%",
})
if status == 404 and ratio > 0.3:
anomalies.append({
"type": "高404比例",
"status": status,
"count": count,
"ratio": f"{ratio*100:.1f}%",
})
return anomalies
def detect_ip_anomalies(self, threshold=100):
"""检测可疑 IP（访问量异常高）"""
counter = Counter(e.ip for e in self.entries)
return [{"ip": ip, "count": count}
for ip, count in counter.most_common()
if count > threshold]
def detect_traffic_spike(self, window_minutes=5, spike_threshold=3.0):
"""检测流量尖峰"""
from itertools import groupby
# 按分钟分组
by_minute = {}
for e in self.entries:
if e.time:
minute_key = e.time.strftime("%Y-%m-%d %H:%M")
by_minute[minute_key] = by_minute.get(minute_key, 0) + 1
if not by_minute:
return []
# 计算平均值和标准差
counts = list(by_minute.values())
avg = sum(counts) / len(counts)
spikes = []
for minute, count in sorted(by_minute.items()):
if count > avg * spike_threshold:
spikes.append({
"minute": minute,
"count": count,
"avg": f"{avg:.1f}",
"spike_ratio": f"{count/avg:.1f}x",
})
return spikes#!/usr/bin/env python3
"""第33课 日志分析验证"""
import re
from collections import Counter
from datetime import datetime
# 模拟日志数据
SAMPLE_LOGS = [
'192.168.1.1 - - [10/Mar/2024:13:55:36 +0800] "GET /api/users HTTP/1.1" 200 1234 "-" "Mozilla/5.0"',
'10.0.0.5 - - [10/Mar/2024:13:56:01 +0800] "POST /api/login HTTP/1.1" 401 95 "-" "curl/7.88"',
'192.168.1.1 - - [10/Mar/2024:13:57:12 +0800] "GET /api/users HTTP/1.1" 200 2345 "-" "Mozilla/5.0"',
'10.0.0.5 - - [10/Mar/2024:13:58:00 +0800] "GET /static/app.js HTTP/1.1" 200 54321 "-" "Chrome/120"',
'172.16.0.1 - - [10/Mar/2024:13:59:30 +0800] "GET /api/data HTTP/1.1" 500 12 "-" "Python/3.12"',
]
NGINX_PATTERN = re.compile(
r'(?P<ip>\S+) - (?P<user>\S+) '
r'\[(?P<time>[^\]]+)\] '
r'"(?P<method>\S+) (?P<path>\S+) \S+" '
r'(?P<status>\d+) (?P<size>\d+) '
r'"(?P<referer>[^"]*)" '
r'"(?P<agent>[^"]*)"'
)
def test_parse():
entries = []
for line in SAMPLE_LOGS:
m = NGINX_PATTERN.match(line)
assert m is not None, f"Failed to parse: {line}"
d = m.groupdict()
entries.append(d)
assert len(entries) == 5
assert entries[0]["ip"] == "192.168.1.1"
assert entries[0]["status"] == "200"
print("✅ 日志解析测试通过")
def test_counter_stats():
entries = []
for line in SAMPLE_LOGS:
m = NGINX_PATTERN.match(line)
entries.append(m.groupdict())
ip_counter = Counter(e["ip"] for e in entries)
assert ip_counter["192.168.1.1"] == 2
status_counter = Counter(e["status"] for e in entries)
assert status_counter["200"] == 3
path_counter = Counter(e["path"] for e in entries)
assert path_counter["/api/users"] == 2
print("✅ 统计分析测试通过")
def test_error_detection():
entries = []
for line in SAMPLE_LOGS:
m = NGINX_PATTERN.match(line)
entries.append(m.groupdict())
errors = [e for e in entries if int(e["status"]) >= 400]
assert len(errors) == 2 # 401 + 500
print("✅ 异常检测测试通过")
if __name__ == "__main__":
test_parse()
test_counter_stats()
test_error_detection()
print("\n🎉 第33课全部验证通过！")from datetime import datetime
def generate_report(analysis_data: dict, output_path: str):
"""生成 HTML 日志分析报告"""
html = f"""<!DOCTYPE html>
<html><head><meta charset="UTF-8">
<title>日志分析报告</title>
<style>
body {{ font-family: sans-serif; margin: 20px; background: #0f172a; color: #e2e8f0; }}
h1 {{ color: #34d399; }} h2 {{ color: #34d399; }}
table {{ border-collapse: collapse; width: 100%; }}
th, td {{ border: 1px solid #334155; padding: 8px; }}
th {{ background: #1e293b; color: #34d399; }}
.metric {{ display: inline-block; background: #1e293b; border-radius: 8px; padding: 15px; margin: 5px; }}
.metric .value {{ font-size: 2em; color: #34d399; }}
</style></head><body>
<h1>日志分析报告</h1>
<p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<h2>核心指标</h2>
<div class="metric"><div class="value">{analysis_data.get('total', 0):,}</div><div>总请求数</div></div>
</body></html>"""
with open(output_path, "w") as f:
f.write(html)
print(f"报告已生成: {output_path}")import re
from collections import Counter, deque
from datetime import datetime
from typing import Callable, Optional
class LiveLogMonitor:
"""实时日志监控器"""
def __init__(self, alert_rules=None, window_size=100):
self.rules = alert_rules or []
self.window = deque(maxlen=window_size)
self.counters = Counter()
self.on_alert: Optional[Callable] = None
def process_line(self, line: str):
self.window.append(line)
self.counters["total"] += 1
status_match = re.search(r"\s(\d{3})\s", line)
if status_match:
status = int(status_match.group(1))
self.counters[f"status_{status // 100}xx"] += 1
if status >= 500:
self._check_alert("5xx", status)
def _check_alert(self, metric, value):
for rule in self.rules:
if rule["metric"] == metric and value > rule["threshold"]:
if self.on_alert:
self.on_alert({"metric": metric, "value": value})
def get_error_rate(self):
total = self.counters.get("total", 0)
if total == 0: return 0
errors = self.counters.get("status_4xx", 0) + self.counters.get("status_5xx", 0)
return errors / total * 100import re
from dataclasses import dataclass
# ===== Apache 日志 =====
APACHE_PATTERN = re.compile(
r'(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) (\S+) \S+" (\d+) (\d+)'
)
# ===== Python logging 日志 =====
PYTHON_LOG_PATTERN = re.compile(
r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+) - (\w+) - (.+)'
)
@dataclass
class PythonLogEntry:
timestamp: str
level: str
message: str
def parse_python_log(content: str):
entries = []
for line in content.strip().split("\n"):
m = PYTHON_LOG_PATTERN.match(line)
if m:
entries.append(PythonLogEntry(
timestamp=m.group(1),
level=m.group(2),
message=m.group(3),
))
return entries
# ===== JSON 日志 =====
import json
def parse_json_logs(content: str):
entries = []
for line in content.strip().split("\n"):
try:
entry = json.loads(line)
entries.append(entry)
except json.JSONDecodeError:
pass
return entries
# ===== Syslog =====
SYSLOG_PATTERN = re.compile(
r'(\w+\s+\d+\s+\d+:\d+:\d+)\s+(\S+)\s+(\S+?):\s+(.*)'
)
def parse_syslog(content: str):
entries = []
for line in content.strip().split("\n"):
m = SYSLOG_PATTERN.match(line)
if m:
entries.append({
"timestamp": m.group(1),
"host": m.group(2),
"process": m.group(3),
"message": m.group(4),
})
return entries