Python Log Analysis Scripts in Practice

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

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

# Generic application log format: 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]:
    """Parse an application log line"""
    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')
    )

# Test
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 format:
# 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]:
    """Parse an Nginx access log line"""
    match = NGINX_PATTERN.match(line.strip())
    if not match:
        return None

    d = match.groupdict()
    # Type conversion
    d['status'] = int(d['status'])
    d['body_bytes_sent'] = int(d['body_bytes_sent']) if d['body_bytes_sent'] != '-' else 0

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

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

    # Status code classification
    d['status_class'] = f"{d['status'] // 100}xx"

    return d

# Batch parse a log file
def load_nginx_logs(filepath: str) -> list:
    """Load and parse an Nginx log file"""
    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"Warning: line {line_no} failed to parse: {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):
    """Base class for log parsers"""
    @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 format log parser"""
    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 format parser"""
    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:
    """Multi-format log parser with automatic format detection"""
    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:
    """Load Nginx logs into a 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

# Usage example
df = nginx_logs_to_dataframe('/var/log/nginx/access.log')
print(f"Total requests: {len(df):,}")
print(f"Time range: {df.index.min()} ~ {df.index.max()}")
print(df.head())

# === Status code distribution ===
print("Status code distribution:")
print(df['status'].value_counts().sort_index())

# === HTTP method distribution ===
print("\nHTTP method distribution:")
print(df['method'].value_counts())

# === Hourly request volume ===
hourly = df.resample('1h').size()
print("\nHourly request volume:")
print(hourly)

# === Top 10 access paths ===
print("\nTop 10 access paths:")
print(df['url'].value_counts().head(10))

# === Top 10 client IPs ===
print("\nTop 10 client IPs:")
print(df['ip'].value_counts().head(10))

# === 4xx/5xx error analysis ===
errors = df[df['status'] >= 400]
print(f"\nError requests: {len(errors)} ({len(errors)/len(df)*100:.1f}%)")
print(errors.groupby('status')['url'].value_counts().head(20))

# === Bandwidth statistics ===
print(f"\nTotal transferred: {df['bytes'].sum() / 1024 / 1024:.2f} MB")
print(f"Average response: {df['bytes'].mean():.0f} bytes")
print(f"Largest response: {df['bytes'].max():.0f} bytes")

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

# === Slow request analysis (if $request_time is available) ===
# Assuming the log includes a response time field
# slow_requests = df[df['request_time'] > 2.0]
# print(f"\nSlow requests (>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

# Aggregate by different time granularities
def time_series_analysis(df: pd.DataFrame):
    """Time series analysis"""
    # Per-minute request volume
    per_minute = df.resample('1min').size()

    # Hourly status code distribution
    hourly_status = df.groupby([
        df.index.floor('1h'),
        'status_class'
    ]).size().unstack(fill_value=0)

    # Daily traffic trends
    daily = df.resample('1D').agg({
        'status': 'count',
        'bytes': 'sum',
        'ip': 'nunique'
    })
    daily.columns = ['requests', 'total_bytes', 'unique_ips']

    # Calculate day-over-day change
    daily['requests_change'] = daily['requests'].pct_change() * 100

    # Weekday vs weekend comparison
    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 average requests: {weekday_avg:,.0f}")
    print(f"Weekend average requests: {weekend_avg:,.0f}")
    print(f"Weekend/weekday ratio: {weekend_avg/weekday_avg:.2%}")

    return daily

daily_stats = time_series_analysis(df)

# Multi-dimensional cross analysis
def pivot_analysis(df: pd.DataFrame):
    """Multi-dimensional pivot analysis"""

    # Status code × path cross-tab
    # Extract path (strip query parameters)
    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("Path × Status code cross-tab:")
    print(pivot.head(20))

    # Hourly × status code heatmap data
    hourly_status = pd.pivot_table(
        df,
        values='ip',
        index=df.index.hour,
        columns='status_class',
        aggfunc='count',
        fill_value=0
    )
    print("\nHour × Status code:")
    print(hourly_status)

    # Client IP × access path
    ip_path = pd.crosstab(df['ip'], df['path'])
    # Find IPs with single-path access (likely scanners)
    single_path_ips = ip_path[ip_path.sum(axis=1) > 100].sum(axis=1)
    print(f"\nHigh-frequency IPs (>100 requests): {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:
    """Detect traffic anomalies"""
    # Aggregate by time window
    traffic = df.resample(window).size().to_frame('count')

    # Rolling statistics
    rolling_mean = traffic['count'].rolling(window=24, min_periods=1).mean()
    rolling_std = traffic['count'].rolling(window=24, min_periods=1).std()

    # Z-Score anomaly detection
    traffic['z_score'] = (traffic['count'] - rolling_mean) / rolling_std

    # Flag anomalies
    traffic['is_anomaly'] = traffic['z_score'].abs() > 3

    # Detect traffic drops (possible service outage)
    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"Detected {len(anomalies)} anomaly points:")
        for ts, row in anomalies.iterrows():
            direction = "spike" if row['z_score'] > 0 else "drop"
            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:
    """Detect error rate anomalies"""
    df['is_error'] = df['status'] >= 500

    # Calculate error rate per time window
    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
    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"Detected {len(anomalies)} error rate 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:
    """Detect suspicious IP behavior"""
    suspicious = []

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

        # 1. High frequency access
        if len(group) > 500:
            reasons.append(f"High frequency ({len(group)} requests)")

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

        # 3. Scanning behavior (many 404s)
        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"Scanning ({unique_404_paths} unique 404 paths)")

        # 4. Accessing sensitive paths
        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 path access ({sensitive_hits} hits)")

        # 5. Burst traffic (many requests in a short period)
        if len(group) > 1:
            time_span = (group.index.max() - group.index.min()).total_seconds()
            if time_span > 0:
                rate = len(group) / time_span  # requests/second
                if rate > 10:
                    reasons.append(f"Burst traffic ({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

# Run detection
suspicious = detect_suspicious_ips(df)
if not suspicious.empty:
    print(f"\nDetected {len(suspicious)} suspicious IPs:")
    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:
    """Sliding window anomaly detection"""
    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:
    """Generate a log analysis report"""
    report = {}

    # Basic info
    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,
    }

    # Status code distribution
    report['status_codes'] = df['status'].value_counts().to_dict()

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

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

    # Hourly trend
    report['hourly_trend'] = df.resample('1h').size().to_dict()

    # Error request statistics
    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

# Formatted output
def print_report(report: dict):
    """Print a formatted report"""
    print("=" * 60)
    print("Nginx Log Analysis Report")
    print("=" * 60)

    s = report['summary']
    print(f"\nTotal requests: {s['total_requests']:,}")
    print(f"Time range: {s['time_range']}")
    print(f"Unique IPs: {s['unique_ips']:,}")
    print(f"Unique paths: {s['unique_paths']:,}")
    print(f"Total bandwidth: {s['total_bandwidth_mb']:.2f} MB")

    print(f"\n--- Status Code Distribution ---")
    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--- Error Statistics ---")
    e = report['errors']
    print(f"  Total errors: {e['total']:,} ({e['rate']:.2%})")
    if e['top_error_paths']:
        print(f"  Top error paths:")
        for path, count in list(e['top_error_paths'].items())[:5]:
            print(f"    {count:>6,}  {path[:60]}")

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

    print(f"\n--- Top 10 Client IPs ---")
    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:
    """Real-time log processing: continuous monitoring like tail -f"""

    def __init__(self, filepath: str, parser_func):
        self.filepath = filepath
        self.parser = parser_func
        self.stats_window = deque(maxlen=300)  # 5-minute sliding window

    def follow(self):
        """Continuously read new log entries"""
        with open(self.filepath, 'r', errors='replace') as f:
            # Move to end of file
            f.seek(0, 2)
            print(f"Monitoring {self.filepath}...")

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

    def _process_line(self, line: str):
        """Process a single log line"""
        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', ''))
        })

        # Real-time error alerting
        status = entry.get('status', 0)
        if status >= 500:
            print(f"[{now:%H:%M:%S}] 5xx error: {status} {entry.get('url', '')} from {entry.get('ip', '')}")

    def get_realtime_stats(self) -> dict:
        """Get real-time statistics"""
        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):
        """Output statistics every 10 seconds"""
        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

# Usage example
processor = LogTailProcessor('/var/log/nginx/access.log', parse_nginx_log)
# Statistics thread
stats_thread = threading.Thread(target=processor.stats_loop, daemon=True)
stats_thread.start()
# Main thread follows the log
processor.follow()

import threading
from queue import Queue
from collections import defaultdict

class MultiFileLogProcessor:
    """Parallel processing of multiple log files"""

    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):
        """Tail a single log file"""
        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):
        """Process log entries from the queue"""
        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):
        """Periodically output reports"""
        while True:
            time.sleep(60)
            print(f"\n--- {datetime.now():%H:%M:%S} Statistics ---")
            for source, stats in self.stats.items():
                rate = stats['errors'] / stats['count'] if stats['count'] > 0 else 0
                print(f"  {source}: {stats['count']} requests, {rate:.1%} error rate")

    def run(self, files: dict):
        """Start the processor"""
        # File tailing threads
        for source, path in files.items():
            t = threading.Thread(
                target=self.tail_file,
                args=(path, source),
                daemon=True
            )
            t.start()

        # Processing thread
        threading.Thread(target=self.process_loop, daemon=True).start()
        # Reporting thread
        threading.Thread(target=self.report_loop, daemon=True).start()

        # Keep main thread alive
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            print("\nStopping monitoring...")

# Usage
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):
    """Terminal bar chart"""
    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):
    """Terminal time series sparkline"""
    if series.empty:
        return

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

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

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

    # Sample to specified width
    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):
    """Generate an HTML format log analysis report"""
    hourly = df.resample('1h').size()
    status_dist = df['status'].value_counts().sort_index()

    html = f"""<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>Log Analysis Report - {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 Log Analysis Report</h1>
    <p>Generated: {datetime.now():%Y-%m-%d %H:%M:%S}</p>

    <div class="card">
        <h2>Overview</h2>
        <div class="metric">
            <div class="value">{len(df):,}</div>
            <div class="label">Total Requests</div>
        </div>
        <div class="metric">
            <div class="value">{df['ip'].nunique():,}</div>
            <div class="label">Unique IPs</div>
        </div>
        <div class="metric">
            <div class="value">{df['bytes'].sum() / 1024 / 1024:.1f} MB</div>
            <div class="label">Total Bandwidth</div>
        </div>
        <div class="metric">
            <div class="value">{(df['status'] >= 400).mean():.1%}</div>
            <div class="label">Error Rate</div>
        </div>
    </div>

    <div class="card">
        <h2>Status Code Distribution</h2>
        <table>
            <tr><th>Status</th><th>Count</th><th>Percentage</th><th>Distribution</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 Access Paths</h2>
        <table>
            <tr><th>Path</th><th>Requests</th><th>Percentage</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"Report generated: {output_path}")

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

def count_lines_fast(filepath: str) -> int:
    """Fast line counting (mmap + regex)"""
    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:
    """Parse a specific region of a file"""
    results = []
    with open(filepath, 'r', errors='replace') as f:
        f.seek(start)
        # If not at file start, skip the first line (may be incomplete)
        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:
    """Multi-process parallel parsing of large log files"""
    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

# Usage example
# results = parallel_parse('/var/log/nginx/access.log', num_workers=8)

def analyze_large_file_streaming(filepath: str):
    """Stream-process large files, avoiding loading everything into memory"""
    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']

    # Keep only 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 Log Inspection Script
Features: parse logs, statistical analysis, anomaly detection, report generation
Usage: 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 log inspector"""

    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):
        """Parse the log file"""
        print(f"Parsing log file: {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 lines: {total:,}")
        print(f"  Successfully parsed: {parsed:,} ({parsed/total*100:.1f}%)")
        return self

    def analyze(self):
        """Run analysis"""
        if not self.entries:
            print("No data to analyze")
            return self

        total = len(self.entries)

        # Status code statistics
        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 statistics
        ip_counts = Counter(e['ip'] for e in self.entries)

        # Path statistics
        path_counts = Counter(e['url'].split('?')[0] for e in self.entries)

        # Bandwidth statistics
        total_bytes = sum(e['bytes'] for e in self.entries)

        # Suspicious IP detection
        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"High frequency ({count})")
            if error_rate > 0.5:
                reasons.append(f"High error rate ({error_rate:.0%})")
            if not_found_rate > 0.3:
                reasons.append(f"Scanning ({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):
        """Print the report"""
        s = self.stats
        print("\n" + "=" * 60)
        print("Nginx Log Inspection Report")
        print("=" * 60)

        print(f"\nTime range: {s['time_range']['start']} ~ {s['time_range']['end']}")
        print(f"Total requests: {s['total_requests']:,}")
        print(f"Unique IPs: {s['unique_ips']:,}")
        print(f"Unique paths: {s['unique_paths']:,}")
        print(f"Total bandwidth: {s['total_bandwidth_mb']:.2f} MB")
        print(f"Error rate: {s['error_rate']:.2%}")
        print(f"5xx errors: {s['server_error_count']}")

        print(f"\nStatus code distribution:")
        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 IPs:")
        for ip, count in list(s['top_ips'].items())[:10]:
            print(f"  {count:>8,}  {ip}")

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

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

    def save_json(self, output_path: str):
        """Save as 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"\nReport saved: {output_path}")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("Usage: 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])