Python 日志分析脚本实战

import re
from datetime import datetime
from typing import NamedTuple, Optional

class LogEntry(NamedTuple):
    timestamp: datetime
    level: str
    message: str
    source: Optional[str] = None

# 通用应用日志格式：2026-07-10 14:32:01 [ERROR] [app.payment] Payment failed: order=12345
APP_LOG_PATTERN = re.compile(
    r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+'
    r'\[(?P<level>DEBUG|INFO|WARN|WARNING|ERROR|FATAL)\]\s+'
    r'(?:\[(?P<source>[\w.]+)\]\s+)?'
    r'(?P<message>.+)'
)

def parse_app_log(line: str) -> Optional[LogEntry]:
    """解析应用日志行"""
    match = APP_LOG_PATTERN.match(line.strip())
    if not match:
        return None
    return LogEntry(
        timestamp=datetime.strptime(match.group('timestamp'), '%Y-%m-%d %H:%M:%S'),
        level=match.group('level'),
        message=match.group('message'),
        source=match.group('source')
    )

# 测试
log_line = '2026-07-10 14:32:01 [ERROR] [app.payment] Payment failed: order=12345, amount=99.00'
entry = parse_app_log(log_line)
print(entry)
# LogEntry(timestamp=datetime.datetime(2026, 7, 10, 14, 32, 1), level='ERROR',
#          message='Payment failed: order=12345, amount=99.00', source='app.payment')

import re
from typing import Optional, Dict
from urllib.parse import urlparse, parse_qs

# Nginx combined 格式：
# 192.168.1.1 - - [10/Jul/2026:14:32:01 +0800] "GET /api/users?page=1 HTTP/1.1" 200 1234
# "https://example.com/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"

NGINX_PATTERN = re.compile(
    r'(?P<remote_addr>\S+)\s+'
    r'\S+\s+\S+\s+'  # identd, user
    r'\[(?P<time_local>[^\]]+)\]\s+'
    r'"(?P<method>\S+)\s+(?P<url>\S+)\s+(?P<protocol>[^"]+)"\s+'
    r'(?P<status>\d{3})\s+'
    r'(?P<body_bytes_sent>\S+)\s+'
    r'"(?P<referer>[^"]*)"\s+'
    r'"(?P<user_agent>[^"]*)"'
)

def parse_nginx_log(line: str) -> Optional[Dict]:
    """解析 Nginx 访问日志"""
    match = NGINX_PATTERN.match(line.strip())
    if not match:
        return None

    d = match.groupdict()
    # 类型转换
    d['status'] = int(d['status'])
    d['body_bytes_sent'] = int(d['body_bytes_sent']) if d['body_bytes_sent'] != '-' else 0

    # 解析时间：10/Jul/2026:14:32:01 +0800
    dt = datetime.strptime(d['time_local'], '%d/%b/%Y:%H:%M:%S %z')
    d['datetime'] = dt

    # 解析 URL
    parsed_url = urlparse(d['url'])
    d['path'] = parsed_url.path
    d['query'] = parse_qs(parsed_url.query)
    d['query_string'] = parsed_url.query

    # 状态码分类
    d['status_class'] = f"{d['status'] // 100}xx"

    return d

# 批量解析日志文件
def load_nginx_logs(filepath: str) -> list:
    """加载并解析 Nginx 日志文件"""
    logs = []
    with open(filepath, 'r', encoding='utf-8', errors='replace') as f:
        for line_no, line in enumerate(f, 1):
            entry = parse_nginx_log(line)
            if entry is None:
                print(f"警告: 第 {line_no} 行解析失败: {line.strip()[:80]}")
                continue
            logs.append(entry)
    return logs

import re
from abc import ABC, abstractmethod
from typing import Optional, Dict, List

class LogParser(ABC):
    """日志解析器基类"""
    @abstractmethod
    def parse(self, line: str) -> Optional[Dict]:
        pass

    @abstractmethod
    def name(self) -> str:
        pass

class NginxAccessParser(LogParser):
    def name(self): return "nginx_access"
    PATTERN = re.compile(
        r'(?P<ip>\S+).*?\[(?P<time>[^\]]+)\].*?'
        r'"(?P<method>\S+)\s+(?P<path>\S+).*?"\s+'
        r'(?P<status>\d+)\s+(?P<bytes>\S+)'
    )
    def parse(self, line: str) -> Optional[Dict]:
        m = self.PATTERN.search(line)
        if not m:
            return None
        d = m.groupdict()
        d['status'] = int(d['status'])
        d['bytes'] = int(d['bytes']) if d['bytes'] != '-' else 0
        return d

class JsonLogParser(LogParser):
    """JSON 格式日志解析器"""
    def name(self): return "json"
    def parse(self, line: str) -> Optional[Dict]:
        import json
        try:
            return json.loads(line.strip())
        except json.JSONDecodeError:
            return None

class SyslogParser(LogParser):
    """Syslog 格式解析器"""
    def name(self): return "syslog"
    PATTERN = re.compile(
        r'(?P<month>\w{3})\s+(?P<day>\d+)\s+'
        r'(?P<time>\d{2}:\d{2}:\d{2})\s+'
        r'(?P<host>\S+)\s+'
        r'(?P<process>[\w\[\]-]+):\s+(?P<message>.+)'
    )
    def parse(self, line: str) -> Optional[Dict]:
        m = self.PATTERN.match(line)
        if not m:
            return None
        return m.groupdict()

class MultiFormatParser:
    """多格式日志解析器：自动检测格式"""
    def __init__(self):
        self.parsers: List[LogParser] = [
            JsonLogParser(),
            NginxAccessParser(),
            SyslogParser(),
        ]

    def parse(self, line: str) -> Optional[Dict]:
        for parser in self.parsers:
            result = parser.parse(line)
            if result is not None:
                result['_parser'] = parser.name()
                return result
        return None

    def parse_file(self, filepath: str) -> List[Dict]:
        results = []
        with open(filepath, 'r', errors='replace') as f:
            for line in f:
                parsed = self.parse(line)
                if parsed:
                    results.append(parsed)
        return results

import pandas as pd
import re
from datetime import datetime

def nginx_logs_to_dataframe(filepath: str) -> pd.DataFrame:
    """将 Nginx 日志加载为 DataFrame"""
    NGINX_PATTERN = re.compile(
        r'(?P<ip>\S+)\s+\S+\s+\S+\s+'
        r'\[(?P<time>[^\]]+)\]\s+'
        r'"(?P<method>\S+)\s+(?P<url>\S+)\s+\S+"\s+'
        r'(?P<status>\d{3})\s+(?P<bytes>\S+)\s+'
        r'"(?P<referer>[^"]*)"\s+'
        r'"(?P<ua>[^"]*)"'
    )

    records = []
    with open(filepath, 'r', errors='replace') as f:
        for line in f:
            m = NGINX_PATTERN.match(line.strip())
            if m:
                d = m.groupdict()
                d['status'] = int(d['status'])
                d['bytes'] = int(d['bytes']) if d['bytes'] != '-' else 0
                d['datetime'] = datetime.strptime(
                    d['time'], '%d/%b/%Y:%H:%M:%S %z'
                )
                records.append(d)

    df = pd.DataFrame(records)
    if not df.empty:
        df = df.set_index('datetime')
        df['status_class'] = (df['status'] // 100).astype(str) + 'xx'

    return df

# 使用示例
df = nginx_logs_to_dataframe('/var/log/nginx/access.log')
print(f"总请求数: {len(df):,}")
print(f"时间范围: {df.index.min()} ~ {df.index.max()}")
print(df.head())

# === 状态码分布 ===
print("状态码分布:")
print(df['status'].value_counts().sort_index())

# === HTTP 方法分布 ===
print("\nHTTP 方法分布:")
print(df['method'].value_counts())

# === 每小时请求量 ===
hourly = df.resample('1h').size()
print("\n每小时请求量:")
print(hourly)

# === Top 10 访问路径 ===
print("\nTop 10 访问路径:")
print(df['url'].value_counts().head(10))

# === Top 10 客户端 IP ===
print("\nTop 10 客户端 IP:")
print(df['ip'].value_counts().head(10))

# === 4xx/5xx 错误分析 ===
errors = df[df['status'] >= 400]
print(f"\n错误请求: {len(errors)} ({len(errors)/len(df)*100:.1f}%)")
print(errors.groupby('status')['url'].value_counts().head(20))

# === 带宽统计 ===
print(f"\n总传输: {df['bytes'].sum() / 1024 / 1024:.2f} MB")
print(f"平均响应: {df['bytes'].mean():.0f} bytes")
print(f"最大响应: {df['bytes'].max():.0f} bytes")

# === User-Agent 分析 ===
print("\nTop 5 User-Agent:")
print(df['ua'].value_counts().head(5))

# === 慢请求分析（如果有 $request_time）===
# 假设日志中包含响应时间字段
# slow_requests = df[df['request_time'] > 2.0]
# print(f"\n慢请求 (>2s): {len(slow_requests)}")
# print(slow_requests.groupby('url')['request_time'].agg(['mean', 'max', 'count'])
#       .sort_values('mean', ascending=False).head(10))

import pandas as pd

# 按不同时间粒度聚合
def time_series_analysis(df: pd.DataFrame):
    """时间序列分析"""
    # 每分钟请求量
    per_minute = df.resample('1min').size()

    # 每小时状态码分布
    hourly_status = df.groupby([
        df.index.floor('1h'),
        'status_class'
    ]).size().unstack(fill_value=0)

    # 每天访问趋势
    daily = df.resample('1D').agg({
        'status': 'count',
        'bytes': 'sum',
        'ip': 'nunique'
    })
    daily.columns = ['requests', 'total_bytes', 'unique_ips']

    # 计算环比变化
    daily['requests_change'] = daily['requests'].pct_change() * 100

    # 工作日 vs 周末对比
    daily['day_of_week'] = daily.index.day_name()
    daily['is_weekend'] = daily.index.dayofweek >= 5

    weekend_avg = daily[daily['is_weekend']]['requests'].mean()
    weekday_avg = daily[~daily['is_weekend']]['requests'].mean()

    print(f"工作日平均请求: {weekday_avg:,.0f}")
    print(f"周末平均请求: {weekend_avg:,.0f}")
    print(f"周末/工作日比: {weekend_avg/weekday_avg:.2%}")

    return daily

daily_stats = time_series_analysis(df)

# 多维交叉分析
def pivot_analysis(df: pd.DataFrame):
    """多维透视分析"""

    # 状态码 × 路径 交叉表
    # 提取路径（去掉查询参数）
    df['path'] = df['url'].str.split('?').str[0]

    pivot = pd.pivot_table(
        df,
        values='ip',
        index='path',
        columns='status_class',
        aggfunc='count',
        fill_value=0,
        margins=True
    )
    print("路径 × 状态码 交叉表:")
    print(pivot.head(20))

    # 每小时 × 状态码热力图数据
    hourly_status = pd.pivot_table(
        df,
        values='ip',
        index=df.index.hour,
        columns='status_class',
        aggfunc='count',
        fill_value=0
    )
    print("\n小时 × 状态码:")
    print(hourly_status)

    # 客户端 IP × 访问路径
    ip_path = pd.crosstab(df['ip'], df['path'])
    # 找出访问路径单一的 IP（可能是扫描器）
    single_path_ips = ip_path[ip_path.sum(axis=1) > 100].sum(axis=1)
    print(f"\n高频访问IP (>100次): {len(single_path_ips)}")

    return pivot

pivot_analysis(df)

import numpy as np
import pandas as pd

def detect_traffic_anomalies(df: pd.DataFrame, window: str = '5min') -> pd.DataFrame:
    """检测流量异常"""
    # 按时间窗口聚合
    traffic = df.resample(window).size().to_frame('count')

    # 滚动统计
    rolling_mean = traffic['count'].rolling(window=24, min_periods=1).mean()
    rolling_std = traffic['count'].rolling(window=24, min_periods=1).std()

    # Z-Score 异常检测
    traffic['z_score'] = (traffic['count'] - rolling_mean) / rolling_std

    # 标记异常
    traffic['is_anomaly'] = traffic['z_score'].abs() > 3

    # 检测流量骤降（可能是服务不可用）
    traffic['pct_change'] = traffic['count'].pct_change()
    traffic['traffic_drop'] = traffic['pct_change'] < -0.5

    anomalies = traffic[traffic['is_anomaly'] | traffic['traffic_drop']]

    if not anomalies.empty:
        print(f"检测到 {len(anomalies)} 个异常点:")
        for ts, row in anomalies.iterrows():
            direction = "激增" if row['z_score'] > 0 else "骤降"
            print(f"  {ts}: {direction} (count={row['count']}, z={row['z_score']:.2f})")

    return traffic

def detect_error_rate_anomalies(df: pd.DataFrame, window: str = '5min') -> pd.DataFrame:
    """检测错误率异常"""
    df['is_error'] = df['status'] >= 500

    # 按时间窗口计算错误率
    window_stats = df.resample(window).agg(
        total=('status', 'count'),
        errors=('is_error', 'sum')
    )
    window_stats['error_rate'] = window_stats['errors'] / window_stats['total']

    # 基线错误率
    baseline_error_rate = window_stats['error_rate'].rolling(
        window=48, min_periods=1
    ).median()

    window_stats['is_anomaly'] = (
        window_stats['error_rate'] > baseline_error_rate * 3
    ) & (window_stats['errors'] > 5)

    anomalies = window_stats[window_stats['is_anomaly']]
    if not anomalies.empty:
        print(f"检测到 {len(anomalies)} 个错误率异常:")
        for ts, row in anomalies.iterrows():
            print(f"  {ts}: error_rate={row['error_rate']:.2%} "
                  f"(baseline={baseline_error_rate[ts]:.2%})")

    return window_stats

from collections import defaultdict, Counter
from datetime import timedelta

def detect_suspicious_ips(df: pd.DataFrame) -> pd.DataFrame:
    """检测可疑 IP 行为"""
    suspicious = []

    for ip, group in df.groupby('ip'):
        reasons = []

        # 1. 高频访问
        if len(group) > 500:
            reasons.append(f"高频访问({len(group)}次)")

        # 2. 高错误率
        error_rate = (group['status'] >= 400).mean()
        if error_rate > 0.5 and len(group) > 10:
            reasons.append(f"高错误率({error_rate:.0%})")

        # 3. 扫描行为（大量 404）
        not_found_rate = (group['status'] == 404).mean()
        if not_found_rate > 0.3 and len(group) > 20:
            unique_404_paths = group[group['status'] == 404]['url'].nunique()
            reasons.append(f"扫描行为({unique_404_paths}个404路径)")

        # 4. 访问敏感路径
        sensitive_patterns = ['/admin', '/wp-login', '/.env', '/phpmyadmin', '/config']
        sensitive_hits = group['url'].apply(
            lambda u: any(p in u.lower() for p in sensitive_patterns)
        ).sum()
        if sensitive_hits > 0:
            reasons.append(f"访问敏感路径({sensitive_hits}次)")

        # 5. 突发流量（短时间内大量请求）
        if len(group) > 1:
            time_span = (group.index.max() - group.index.min()).total_seconds()
            if time_span > 0:
                rate = len(group) / time_span  # 请求/秒
                if rate > 10:
                    reasons.append(f"突发流量({rate:.1f} req/s)")

        if reasons:
            suspicious.append({
                'ip': ip,
                'total_requests': len(group),
                'error_rate': error_rate,
                'reasons': '; '.join(reasons),
                'first_seen': group.index.min(),
                'last_seen': group.index.max()
            })

    result = pd.DataFrame(suspicious)
    if not result.empty:
        result = result.sort_values('total_requests', ascending=False)

    return result

# 运行检测
suspicious = detect_suspicious_ips(df)
if not suspicious.empty:
    print(f"\n检测到 {len(suspicious)} 个可疑 IP:")
    print(suspicious[['ip', 'total_requests', 'reasons']].head(20).to_string())

def sliding_window_anomaly(
    df: pd.DataFrame,
    metric: str = 'status',
    window_size: int = 100,
    threshold: float = 3.0
) -> list:
    """滑动窗口异常检测"""
    anomalies = []

    values = df[metric].values
    n = len(values)

    for i in range(window_size, n):
        window = values[i - window_size:i]
        current = values[i]

        mean = np.mean(window)
        std = np.std(window)

        if std > 0:
            z_score = abs((current - mean) / std)
            if z_score > threshold:
                anomalies.append({
                    'timestamp': df.index[i],
                    'value': current,
                    'z_score': z_score,
                    'window_mean': mean,
                    'window_std': std
                })

    return anomalies

def generate_log_report(df: pd.DataFrame) -> dict:
    """生成日志分析报告"""
    report = {}

    # 基本信息
    report['summary'] = {
        'total_requests': len(df),
        'time_range': f"{df.index.min()} ~ {df.index.max()}",
        'unique_ips': df['ip'].nunique(),
        'unique_paths': df['url'].nunique(),
        'total_bandwidth_mb': df['bytes'].sum() / 1024 / 1024,
    }

    # 状态码分布
    report['status_codes'] = df['status'].value_counts().to_dict()

    # Top 路径
    report['top_paths'] = df['url'].value_counts().head(20).to_dict()

    # Top IP
    report['top_ips'] = df['ip'].value_counts().head(20).to_dict()

    # 每小时趋势
    report['hourly_trend'] = df.resample('1h').size().to_dict()

    # 错误请求统计
    errors = df[df['status'] >= 400]
    report['errors'] = {
        'total': len(errors),
        'rate': len(errors) / len(df) if len(df) > 0 else 0,
        'by_status': errors['status'].value_counts().to_dict(),
        'top_error_paths': errors['url'].value_counts().head(10).to_dict(),
    }

    return report

# 格式化输出
def print_report(report: dict):
    """打印格式化报告"""
    print("=" * 60)
    print("Nginx 日志分析报告")
    print("=" * 60)

    s = report['summary']
    print(f"\n总请求数: {s['total_requests']:,}")
    print(f"时间范围: {s['time_range']}")
    print(f"独立 IP: {s['unique_ips']:,}")
    print(f"独立路径: {s['unique_paths']:,}")
    print(f"总带宽: {s['total_bandwidth_mb']:.2f} MB")

    print(f"\n--- 状态码分布 ---")
    for status, count in sorted(report['status_codes'].items()):
        pct = count / s['total_requests'] * 100
        bar = '█' * int(pct / 2)
        print(f"  {status}: {count:>8,} ({pct:5.1f}%) {bar}")

    print(f"\n--- 错误统计 ---")
    e = report['errors']
    print(f"  错误总数: {e['total']:,} ({e['rate']:.2%})")
    if e['top_error_paths']:
        print(f"  Top 错误路径:")
        for path, count in list(e['top_error_paths'].items())[:5]:
            print(f"    {count:>6,}  {path[:60]}")

    print(f"\n--- Top 10 访问路径 ---")
    for path, count in list(report['top_paths'].items())[:10]:
        print(f"  {count:>8,}  {path[:60]}")

    print(f"\n--- Top 10 客户端 IP ---")
    for ip, count in list(report['top_ips'].items())[:10]:
        print(f"  {count:>8,}  {ip}")

import time
from pathlib import Path
from collections import deque, defaultdict
from datetime import datetime, timedelta

class LogTailProcessor:
    """实时日志处理：类似 tail -f 的持续监控"""

    def __init__(self, filepath: str, parser_func):
        self.filepath = filepath
        self.parser = parser_func
        self.stats_window = deque(maxlen=300)  # 5分钟滑动窗口

    def follow(self):
        """持续读取新增日志"""
        with open(self.filepath, 'r', errors='replace') as f:
            # 移动到文件末尾
            f.seek(0, 2)
            print(f"开始监控 {self.filepath}...")

            while True:
                line = f.readline()
                if line:
                    self._process_line(line)
                else:
                    time.sleep(0.1)

    def _process_line(self, line: str):
        """处理单行日志"""
        entry = self.parser(line)
        if entry is None:
            return

        now = datetime.now()
        self.stats_window.append({
            'time': now,
            'status': entry.get('status', 0),
            'ip': entry.get('ip', ''),
            'path': entry.get('url', entry.get('path', ''))
        })

        # 实时错误告警
        status = entry.get('status', 0)
        if status >= 500:
            print(f"[{now:%H:%M:%S}] 5xx 错误: {status} {entry.get('url', '')} from {entry.get('ip', '')}")

    def get_realtime_stats(self) -> dict:
        """获取实时统计"""
        now = datetime.now()
        one_min_ago = now - timedelta(minutes=1)

        recent = [s for s in self.stats_window if s['time'] > one_min_ago]

        if not recent:
            return {'rpm': 0, 'error_rate': 0}

        total = len(recent)
        errors = sum(1 for s in recent if s['status'] >= 400)

        return {
            'rpm': total,
            'error_rate': errors / total,
            'unique_ips': len(set(s['ip'] for s in recent)),
        }

    def stats_loop(self):
        """每 10 秒输出一次统计"""
        while True:
            time.sleep(10)
            stats = self.get_realtime_stats()
            print(f"[{datetime.now():%H:%M:%S}] "
                  f"RPM={stats['rpm']} "
                  f"ErrorRate={stats['error_rate']:.1%} "
                  f"IPs={stats['unique_ips']}")

import threading

# 使用示例
processor = LogTailProcessor('/var/log/nginx/access.log', parse_nginx_log)
# 统计线程
stats_thread = threading.Thread(target=processor.stats_loop, daemon=True)
stats_thread.start()
# 主线程跟随日志
processor.follow()

import threading
from queue import Queue
from collections import defaultdict

class MultiFileLogProcessor:
    """多日志文件并行处理"""

    def __init__(self):
        self.log_queue = Queue(maxsize=10000)
        self.stats = defaultdict(lambda: {'count': 0, 'errors': 0})

    def tail_file(self, filepath: str, source: str):
        """跟随单个日志文件"""
        with open(filepath, 'r', errors='replace') as f:
            f.seek(0, 2)
            while True:
                line = f.readline()
                if line:
                    self.log_queue.put((source, line))
                else:
                    time.sleep(0.1)

    def process_loop(self):
        """处理队列中的日志"""
        while True:
            source, line = self.log_queue.get()
            entry = parse_nginx_log(line)
            if entry:
                self.stats[source]['count'] += 1
                if entry['status'] >= 400:
                    self.stats[source]['errors'] += 1
            self.log_queue.task_done()

    def report_loop(self):
        """定期输出报告"""
        while True:
            time.sleep(60)
            print(f"\n--- {datetime.now():%H:%M:%S} 统计 ---")
            for source, stats in self.stats.items():
                rate = stats['errors'] / stats['count'] if stats['count'] > 0 else 0
                print(f"  {source}: {stats['count']} 请求, {rate:.1%} 错误率")

    def run(self, files: dict):
        """启动处理器"""
        # 文件尾随线程
        for source, path in files.items():
            t = threading.Thread(
                target=self.tail_file,
                args=(path, source),
                daemon=True
            )
            t.start()

        # 处理线程
        threading.Thread(target=self.process_loop, daemon=True).start()
        # 报告线程
        threading.Thread(target=self.report_loop, daemon=True).start()

        # 保持主线程运行
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            print("\n停止监控...")

# 使用
processor = MultiFileLogProcessor()
processor.run({
    'nginx': '/var/log/nginx/access.log',
    'app': '/var/log/myapp/app.log',
})

def plot_terminal_bar(data: dict, title: str = "", width: int = 40):
    """终端柱状图"""
    if not data:
        return

    max_val = max(data.values())
    print(f"\n{title}")
    print("-" * (width + 20))

    for label, value in sorted(data.items(), key=lambda x: -x[1]):
        bar_len = int(value / max_val * width) if max_val > 0 else 0
        bar = '█' * bar_len
        print(f"  {str(label):>15} │{bar:<{width}} │ {value:>8,}")

def plot_terminal_timeseries(series, title: str = "", width: int = 60):
    """终端时间序列 sparkline"""
    if series.empty:
        return

    values = series.values
    max_val = max(values)
    min_val = min(values)

    # sparkline 字符
    chars = '▁▂▃▄▅▆▇█'

    print(f"\n{title}")
    print(f"  Max: {max_val:.0f}  Min: {min_val:.0f}")

    # 采样到指定宽度
    step = max(1, len(values) // width)
    sampled = values[::step]

    line = ''
    for v in sampled:
        if max_val == min_val:
            idx = 4
        else:
            idx = int((v - min_val) / (max_val - min_val) * 7)
        line += chars[idx]

    print(f"  {series.index[0]:%H:%M} │{line}")
    print(f"  {series.index[-1]:%H:%M} │")

def generate_html_report(df: pd.DataFrame, output_path: str):
    """生成 HTML 格式日志分析报告"""
    hourly = df.resample('1h').size()
    status_dist = df['status'].value_counts().sort_index()

    html = f"""<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>日志分析报告 - {datetime.now():%Y-%m-%d %H:%M}</title>
    <style>
        body {{ font-family: -apple-system, sans-serif; margin: 40px; background: #f5f5f5; }}
        .card {{ background: white; border-radius: 8px; padding: 20px; margin: 20px 0;
                 box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        h1 {{ color: #333; }}
        .metric {{ display: inline-block; margin: 10px 20px; text-align: center; }}
        .metric .value {{ font-size: 2em; font-weight: bold; color: #2196F3; }}
        .metric .label {{ color: #666; font-size: 0.9em; }}
        table {{ border-collapse: collapse; width: 100%; }}
        th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
        th {{ background: #f8f8f8; }}
        .bar {{ background: #4CAF50; height: 20px; border-radius: 3px; }}
        .bar.error {{ background: #f44336; }}
    </style>
</head>
<body>
    <h1>Nginx 日志分析报告</h1>
    <p>生成时间: {datetime.now():%Y-%m-%d %H:%M:%S}</p>

    <div class="card">
        <h2>概览</h2>
        <div class="metric">
            <div class="value">{len(df):,}</div>
            <div class="label">总请求</div>
        </div>
        <div class="metric">
            <div class="value">{df['ip'].nunique():,}</div>
            <div class="label">独立IP</div>
        </div>
        <div class="metric">
            <div class="value">{df['bytes'].sum() / 1024 / 1024:.1f} MB</div>
            <div class="label">总带宽</div>
        </div>
        <div class="metric">
            <div class="value">{(df['status'] >= 400).mean():.1%}</div>
            <div class="label">错误率</div>
        </div>
    </div>

    <div class="card">
        <h2>状态码分布</h2>
        <table>
            <tr><th>状态码</th><th>数量</th><th>占比</th><th>分布</th></tr>
            {''.join(f'''
            <tr>
                <td>{status}</td>
                <td>{count:,}</td>
                <td>{count/len(df)*100:.1f}%</td>
                <td><div class="bar {'error' if status >= 400 else ''}"
                     style="width: {count/len(df)*100*2}%"></div></td>
            </tr>''' for status, count in status_dist.items())}
        </table>
    </div>

    <div class="card">
        <h2>Top 20 访问路径</h2>
        <table>
            <tr><th>路径</th><th>请求数</th><th>占比</th></tr>
            {''.join(f'''
            <tr>
                <td>{path}</td>
                <td>{count:,}</td>
                <td>{count/len(df)*100:.1f}%</td>
            </tr>''' for path, count in df['url'].value_counts().head(20).items())}
        </table>
    </div>
</body>
</html>"""

    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(html)
    print(f"报告已生成: {output_path}")

import mmap
import re
from concurrent.futures import ProcessPoolExecutor, as_completed
import os

def count_lines_fast(filepath: str) -> int:
    """快速统计行数（mmap + 正则）"""
    with open(filepath, 'r+b') as f:
        mm = mmap.mmap(f.fileno(), 0)
        count = 0
        while mm.readline():
            count += 1
        mm.close()
        return count

def parse_chunk(filepath: str, start: int, end: int) -> list:
    """解析文件指定区域"""
    results = []
    with open(filepath, 'r', errors='replace') as f:
        f.seek(start)
        # 如果不是文件开头，跳过第一行（可能不完整）
        if start > 0:
            f.readline()
        while f.tell() < end:
            line = f.readline()
            if not line:
                break
            entry = parse_nginx_log(line)
            if entry:
                results.append(entry)
    return results

def parallel_parse(filepath: str, num_workers: int = 4) -> list:
    """多进程并行解析大日志文件"""
    file_size = os.path.getsize(filepath)
    chunk_size = file_size // num_workers

    chunks = []
    for i in range(num_workers):
        start = i * chunk_size
        end = (i + 1) * chunk_size if i < num_workers - 1 else file_size
        chunks.append((filepath, start, end))

    all_results = []
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = [executor.submit(parse_chunk, *chunk) for chunk in chunks]
        for future in as_completed(futures):
            all_results.extend(future.result())

    return all_results

# 使用示例
# results = parallel_parse('/var/log/nginx/access.log', num_workers=8)

def analyze_large_file_streaming(filepath: str):
    """流式处理大文件，避免全部加载到内存"""
    from collections import defaultdict

    status_counts = defaultdict(int)
    ip_counts = defaultdict(int)
    path_counts = defaultdict(int)
    total_bytes = 0
    total_requests = 0

    with open(filepath, 'r', errors='replace') as f:
        for line in f:
            entry = parse_nginx_log(line)
            if not entry:
                continue

            total_requests += 1
            status_counts[entry['status']] += 1
            ip_counts[entry['ip']] += 1
            path_counts[entry['url']] += 1
            total_bytes += entry['bytes']

    # 只保留 Top N
    top_ips = dict(sorted(ip_counts.items(), key=lambda x: -x[1])[:20])
    top_paths = dict(sorted(path_counts.items(), key=lambda x: -x[1])[:20])

    return {
        'total_requests': total_requests,
        'total_bytes': total_bytes,
        'status_distribution': dict(status_counts),
        'top_ips': top_ips,
        'top_paths': top_paths,
    }

#!/usr/bin/env python3
"""
Nginx 日志巡检脚本
功能：解析日志、统计分析、异常检测、生成报告
用法：python3 log_inspector.py /var/log/nginx/access.log
"""

import re
import sys
import json
from datetime import datetime, timedelta
from collections import defaultdict, Counter
from pathlib import Path

class NginxLogInspector:
    """Nginx 日志巡检器"""

    NGINX_PATTERN = re.compile(
        r'(?P<ip>\S+)\s+\S+\s+\S+\s+'
        r'\[(?P<time>[^\]]+)\]\s+'
        r'"(?P<method>\S+)\s+(?P<url>\S+)\s+\S+"\s+'
        r'(?P<status>\d{3})\s+(?P<bytes>\S+)\s+'
        r'"(?P<referer>[^"]*)"\s+'
        r'"(?P<ua>[^"]*)"'
    )

    def __init__(self, filepath: str):
        self.filepath = filepath
        self.entries = []
        self.stats = {}

    def parse(self):
        """解析日志文件"""
        print(f"解析日志文件: {self.filepath}")
        with open(self.filepath, 'r', errors='replace') as f:
            total = 0
            parsed = 0
            for line in f:
                total += 1
                m = self.NGINX_PATTERN.match(line.strip())
                if m:
                    d = m.groupdict()
                    d['status'] = int(d['status'])
                    d['bytes'] = int(d['bytes']) if d['bytes'] != '-' else 0
                    try:
                        d['datetime'] = datetime.strptime(
                            d['time'], '%d/%b/%Y:%H:%M:%S %z'
                        )
                        self.entries.append(d)
                        parsed += 1
                    except ValueError:
                        pass

        print(f"  总行数: {total:,}")
        print(f"  解析成功: {parsed:,} ({parsed/total*100:.1f}%)")
        return self

    def analyze(self):
        """执行分析"""
        if not self.entries:
            print("无数据可分析")
            return self

        total = len(self.entries)

        # 状态码统计
        status_counts = Counter(e['status'] for e in self.entries)
        error_count = sum(c for s, c in status_counts.items() if s >= 400)
        server_error_count = sum(c for s, c in status_counts.items() if s >= 500)

        # IP 统计
        ip_counts = Counter(e['ip'] for e in self.entries)

        # 路径统计
        path_counts = Counter(e['url'].split('?')[0] for e in self.entries)

        # 带宽统计
        total_bytes = sum(e['bytes'] for e in self.entries)

        # 可疑 IP 检测
        suspicious = []
        for ip, count in ip_counts.most_common(100):
            ip_entries = [e for e in self.entries if e['ip'] == ip]
            error_rate = sum(1 for e in ip_entries if e['status'] >= 400) / len(ip_entries)
            not_found_rate = sum(1 for e in ip_entries if e['status'] == 404) / len(ip_entries)
            reasons = []
            if count > 500:
                reasons.append(f"高频({count})")
            if error_rate > 0.5:
                reasons.append(f"高错误率({error_rate:.0%})")
            if not_found_rate > 0.3:
                reasons.append(f"扫描({not_found_rate:.0%} 404)")
            if reasons:
                suspicious.append({'ip': ip, 'count': count, 'reasons': ', '.join(reasons)})

        self.stats = {
            'total_requests': total,
            'unique_ips': len(ip_counts),
            'unique_paths': len(path_counts),
            'total_bandwidth_mb': total_bytes / 1024 / 1024,
            'error_rate': error_count / total,
            'server_error_count': server_error_count,
            'status_distribution': dict(status_counts.most_common()),
            'top_ips': dict(ip_counts.most_common(20)),
            'top_paths': dict(path_counts.most_common(20)),
            'suspicious_ips': suspicious[:10],
            'time_range': {
                'start': min(e['datetime'] for e in self.entries).isoformat(),
                'end': max(e['datetime'] for e in self.entries).isoformat(),
            }
        }
        return self

    def print_report(self):
        """打印报告"""
        s = self.stats
        print("\n" + "=" * 60)
        print("Nginx 日志巡检报告")
        print("=" * 60)

        print(f"\n时间范围: {s['time_range']['start']} ~ {s['time_range']['end']}")
        print(f"总请求: {s['total_requests']:,}")
        print(f"独立IP: {s['unique_ips']:,}")
        print(f"独立路径: {s['unique_paths']:,}")
        print(f"总带宽: {s['total_bandwidth_mb']:.2f} MB")
        print(f"错误率: {s['error_rate']:.2%}")
        print(f"5xx 错误: {s['server_error_count']}")

        print(f"\n状态码分布:")
        for status, count in sorted(s['status_distribution'].items()):
            pct = count / s['total_requests'] * 100
            print(f"  {status}: {count:>8,} ({pct:5.1f}%)")

        print(f"\nTop 10 IP:")
        for ip, count in list(s['top_ips'].items())[:10]:
            print(f"  {count:>8,}  {ip}")

        print(f"\nTop 10 路径:")
        for path, count in list(s['top_paths'].items())[:10]:
            print(f"  {count:>8,}  {path[:50]}")

        if s['suspicious_ips']:
            print(f"\n⚠ 可疑 IP:")
            for item in s['suspicious_ips']:
                print(f"  {item['ip']:>16}  {item['count']:>6}  {item['reasons']}")

    def save_json(self, output_path: str):
        """保存为 JSON"""
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(self.stats, f, indent=2, ensure_ascii=False, default=str)
        print(f"\n报告已保存: {output_path}")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("用法: python3 log_inspector.py <nginx_access.log> [output.json]")
        sys.exit(1)

    inspector = NginxLogInspector(sys.argv[1])
    inspector.parse().analyze().print_report()

    if len(sys.argv) > 2:
        inspector.save_json(sys.argv[2])