Python日志分析实战：从杂乱日志中提取关键信息的6个技巧

import re
from collections import Counter

log = """
192.168.1.100 - - [25/Aug/2026:10:15:30 +0800] "GET /api/users HTTP/1.1" 500 152 "-" "Python-requests/2.28"
192.168.1.101 - - [25/Aug/2026:10:15:31 +0800] "POST /api/orders HTTP/1.1" 200 341 "-" "Mozilla/5.0"
192.168.1.100 - - [25/Aug/2026:10:15:32 +0800] "GET /api/products HTTP/1.1" 502 0 "-" "Python-requests/2.28"
192.168.1.102 - - [25/Aug/2026:10:15:33 +0800] "GET /health HTTP/1.1" 200 2 "-" "curl/7.88"
"""

# 提取状态码
pattern = r'"(\d{3})'
codes = re.findall(pattern, log)
print(Counter(codes))
# Counter({'200': 2, '500': 1, '502': 1})

import pandas as pd
import re

# 读取日志
df = pd.read_csv('nginx.log', sep=' ', header=None, 
                  names=['ip','x','user','time','request','status','size','ref','ua'])

# 解析状态码和请求路径
df['status'] = df['request'].str.extract(r'" (\d{3})')
df['path'] = df['request'].str.extract(r'"[A-Z]+ (.+?) ')

# 统计各状态码数量
print(df['status'].value_counts())

# 只看5xx错误
errors = df[df['status'].str.startswith('5')]
print(f"5xx错误共{len(errors)}条")
print(errors['path'].value_counts().head(5))

import pandas as pd

df = pd.read_csv('app.log', parse_dates=['timestamp'])

# 按小时聚合，统计错误数
hourly_errors = df[df['level'] == 'ERROR'].groupby(
    df['timestamp'].dt.hour
).size()

print(hourly_errors)
# timestamp
# 10    23
# 11    156   ← 峰值
# 12     45

import pandas as pd
from openai import OpenAI

client = OpenAI()
df = pd.read_csv('app.log')

error_msgs = df[df['level'] == 'ERROR']['message'].tolist()

# 批量让LLM分类
batch = '\n'.join(error_msgs[:20])
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": f"""请对以下日志错误进行分类统计，输出格式：类别:数量
{batch}"""
    }]
)
print(response.choices[0].message.content)

import pandas as pd
import numpy as np

df = pd.read_csv('app.log', parse_dates=['timestamp'])

# 按5分钟聚合错误数
error_counts = df[df['level'] == 'ERROR'].groupby(
    pd.Grouper(key='timestamp', freq='5min')
).size()

# 计算7个窗口前的均值和标准差
mean = error_counts.rolling(7, min_periods=1).mean()
std = error_counts.rolling(7, min_periods=1).std().fillna(0)

# 超过均值+2倍标准差，视为异常
anomalies = error_counts[error_counts > mean + 2 * std]

print(f"检测到{len(anomalies)}个异常时段：")
print(anomalies)

import pandas as pd

# 读取不同日志
nginx = pd.read_csv('nginx.log', sep=' ', header=None,
                    names=['ip','x','user','time','request','status','size','ref','ua'],
                    parse_dates=['time'])

app = pd.read_csv('app.log', parse_dates=['timestamp'])

# 关联：Nginx 500对应的时间段，应用日志里有什么
error_times = nginx[nginx['status'] == '500']['time']
app_in_window = app[
    app['timestamp'].between(error_times.min(), error_times.max())
]

print("500错误期间应用层日志：")
print(app_in_window[['timestamp', 'level', 'message']].head(10))

#!/usr/bin/env python3
"""一键日志分析工具"""
import pandas as pd
import re
from collections import Counter

def analyze_log(filepath):
    df = pd.read_csv(filepath)
    
    # 1. 状态码统计
    print("=== 状态码分布 ===")
    if 'status' in df.columns:
        print(df['status'].value_counts())
    
    # 2. 错误类型统计
    print("\n=== 错误分布 ===")
    errors = df[df['level'].str.contains('ERROR|FATAL', case=False, na=False)]
    print(f"共{len(errors)}条错误")
    
    # 3. 按时间聚合
    if 'timestamp' in df.columns:
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        print("\n=== 按小时错误统计 ===")
        print(errors.groupby(errors['timestamp'].dt.hour).size())
    
    # 4. Top错误消息
    print("\n=== 高频错误消息 TOP5 ===")
    if 'message' in df.columns:
        print(errors['message'].value_counts().head(5))

if __name__ == '__main__':
    analyze_log('app.log')

python3 analyze_log.py app.log