日志自动采集、分析与报表

Python 办公自动化专题 · 从日志中挖掘价值并生成可视化报告

专题：Python 自动化办公系统学习

关键词：Python, 自动化办公, 日志采集, 日志分析, 日志监控, 数据分析, 报表, 可视化, Python自动化

一、日志管理概述

日志是系统和应用程序运行状态的"黑匣子"，记录着每一刻发生的关键事件、错误信息和用户行为轨迹。在现代化运维和开发流程中，日志管理已经从单纯的故障排查工具演变为数据驱动决策的重要组成部分。无论是Web服务器的访问日志、应用程序的运行日志、数据库的查询日志，还是操作系统的事件日志，其中都蕴含着大量可挖掘的价值信息。一个成熟的日志管理系统通常包含采集、传输、存储、解析、分析和可视化等完整链条，这正是自动化办公需要掌握的核心技能之一。

常见的日志格式各有特点。Apache/Nginx访问日志以标准的组合日志格式（Combined Log Format）记录每次HTTP请求，包含客户端IP、时间戳、请求方法、URL、状态码、响应大小和User-Agent等字段。系统日志（syslog）则采用RFC 5424标准格式，包含设施代码、严重级别、时间戳、主机名和应用进程信息。应用程序日志的格式最为灵活，有的采用纯文本分隔结构，有的采用JSON这样的结构化格式以便于程序化解析。企业级应用还可能使用W3C扩展格式或自定义的二进制格式。

日志处理的标准流程可以概括为五个阶段：采集（从各种源头获取原始日志）、传输（将日志安全可靠地送达处理中心）、解析（将非结构化的文本转化为结构化数据）、分析（运用统计和机器学习方法提取有价值信息）和呈现（通过报表和仪表盘直观展示分析结果）。Python生态中有丰富的库可以支撑整个流程，从内置的logging模块到loguru、filetail、watchdog、pandas、matplotlib等，形成了一个完整的日志自动化处理工具链。

import logging
import json
import re
from datetime import datetime
from pathlib import Path

# 配置基础日志系统（既写文件又输出到控制台）
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    handlers=[
        logging.FileHandler('app.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger('LogManager')
logger.info('日志管理系统初始化完成')

# 模拟生成一条Apache风格的访问日志
def generate_access_log(ip, method, path, status, size, ua):
    now = datetime.now().strftime('%d/%b/%Y:%H:%M:%S +0800')
    return f'{ip} - - [{now}] "{method} {path} HTTP/1.1" {status} {size} "{ua}"'

log_entry = generate_access_log(
    ip='192.168.1.100', method='GET', path='/api/users',
    status=200, size=1532,
    ua='Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
)
print(log_entry)
# 输出: 192.168.1.100 - - [05/May/2026:14:30:22 +0800] "GET /api/users HTTP/1.1" 200 1532 "Mozilla/5.0 ..."

二、日志采集

日志采集是整个日志处理流水线的第一环，也是最关键的环节。采集策略决定了日志数据的完整性、实时性和可靠性。在生产环境中，日志可能分布在多台服务器上，格式各不相同，且持续不断地产生新数据。Python提供了多种方式来实现高效的日志采集，既可以模拟Linux的tail -f命令实现文件尾随读取，也可以使用watchdog库监控文件系统事件，还可以通过SSH协议远程采集分布式系统中的日志文件。

文件尾随是最常用的实时日志采集方式，其核心原理是持续跟踪文件的变化，当新内容被写入时立即读取。Python实现tail -f功能通常有两种方法：一是使用文件的tell指针和read方法周期性地检查新内容；二是利用操作系统的文件事件通知机制。for watchdog库则提供了跨平台的文件系统监控能力，可以监听文件的创建、修改、删除和移动事件，在监控目录发生变化时触发回调函数处理新产生的日志文件。对于已经归档的旧日志，glob模块配合文件通配符可以实现批量读取和处理。

远程日志采集场景下，paramiko库提供了SSH协议的Python实现，可以安全地连接到远程服务器执行命令或传输文件。结合SFTP协议，可以定时拉取远程服务器上的日志文件到本地进行处理。对于压缩格式的日志文件，gzip和bz2模块可以直接读取.gz和.bz2格式的压缩文件，无需手动解压到磁盘。实际项目中通常采用多策略组合的方式：对于关键业务系统使用实时tail监控，对于非关键系统采用定时批量采集，对于历史归档日志则按需加载。

# 实现一个类 tail -f 功能的实时日志读取器
import time
from pathlib import Path

class LogTailer:
    def __init__(self, filepath, encoding='utf-8'):
        self.filepath = Path(filepath)
        self.encoding = encoding
        self._file = None
        self._pos = 0

    def start(self):
        if not self.filepath.exists():
            raise FileNotFoundError(f'日志文件不存在: {self.filepath}')
        self._file = open(self.filepath, 'r', encoding=self.encoding)
        self._file.seek(0, 2)  # 移动到文件末尾
        self._pos = self._file.tell()

    def read_new_lines(self):
        if not self._file:
            return []
        line = self._file.readline()
        if line and line.endswith('\n'):
            return [line.rstrip('\n\r')]
        return []

    def follow(self, interval=0.5):
        """持续跟踪文件新增内容，生成器模式"""
        self.start()
        try:
            while True:
                lines = self.read_new_lines()
                if lines:
                    yield lines
                else:
                    time.sleep(interval)
        finally:
            self.close()

    def close(self):
        if self._file:
            self._file.close()
            self._file = None

# 使用示例
# tailer = LogTailer('/var/log/nginx/access.log')
# for batch in tailer.follow():
#     for line in batch:
#         print(f'新日志: {line}')

# 使用glob批量读取旧日志文件
import glob
import gzip

def batch_read_logs(pattern='logs/*.log*'):
    """批量读取日志文件，支持.gz压缩格式"""
    files = sorted(glob.glob(pattern))
    total_lines = 0

    for fpath in files:
        if fpath.endswith('.gz'):
            opener = gzip.open
        else:
            opener = open

        with opener(fpath, 'rt', encoding='utf-8') as f:
            for line in f:
                total_lines += 1
                yield line.strip()

    print(f'共处理 {len(files)} 个文件，{total_lines} 行日志')

# 使用示例
# for line in batch_read_logs('access.log*'):
#     process(line)

# 使用watchdog监控日志目录
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class LogFileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('.log'):
            print(f'日志文件变更: {event.src_path}')

observer = Observer()
handler = LogFileHandler()
observer.schedule(handler, path='/var/log/nginx', recursive=False)
observer.start()

三、日志解析

原始的日志文件本质上是非结构化的纯文本，要从中提取有价值的信息，必须进行解析和结构化处理。日志解析是将符合某种格式规则的文本行拆分为有意义的字段和结构化数据的过程。Python提供了强大的正则表达式（re模块）来处理复杂的文本匹配模式，同时也支持直接解析JSON、XML等结构化格式。对于CSV格式或固定分隔符的日志，csv模块和str.split方法都是轻量而高效的选择。

Apache和Nginx访问日志是最典型的需要解析的日志类型。标准Apache组合日志格式包含IP地址、身份标识、用户、时间戳、HTTP请求行、状态码、响应大小、Referer和User-Agent等字段。通过精心构造的正则表达式，可以一次性提取所有这些字段并转换为结构化的字典格式。对于JSON格式的应用程序日志，Python的json.loads方法可以一步到位地将日志行解析为Python字典，配合嵌套字段的取值操作非常方便。

在处理自定义格式日志时，灵活运用正则表达式的命名分组功能是提高代码可读性的关键。例如使用(?P<ip>\S+)这样的语法，可以让匹配结果直接生成字段名与值对应的字典。对于非标准或格式不规范的日志，可以采用多模式匹配策略，依次尝试不同的正则规则，直到找到能成功匹配的模式为止。解析性能方面，正则表达式预编译（re.compile）可以显著提高批量处理大量日志行时的执行效率。实际项目中，建议将解析逻辑封装为独立的函数或类，便于单元测试和维护。

import re
from datetime import datetime

# Apache/Nginx 组合日志格式解析正则
APACHE_LOG_PATTERN = re.compile(r'''
    ^(?P<ip>\S+)                    # 客户端IP
    \s+                                # 分隔符
    (?P<ident>\S+)                  # 身份标识（通常是-）
    \s+                                # 分隔符
    (?P<user>\S+)                   # 用户名（通常是-）
    \s+\[                             # 时间戳开始
    (?P<time>[^\]]+)                # 时间戳内容
    \]\s+                              # 时间戳结束
    "(?P<method>\S+)               # HTTP方法
    \s+                                # 空格
    (?P<path>\S+)                   # 请求路径
    \s+                                # 空格
    (?P<protocol>\S+)"              # HTTP协议版本
    \s+                                # 分隔符
    (?P<status>\d+)                 # HTTP状态码
    \s+                                # 分隔符
    (?P<size>\S+)                   # 响应大小（- 表示0）
    \s*                                # 可选空格
    "(?P<referer>[^"]*)"            # Referer
    \s+                                # 分隔符
    "(?P<ua>[^"]*)"                # User-Agent
    $
''', re.VERBOSE)

def parse_apache_log(line):
    """解析单条Apache/Nginx访问日志，返回结构化字典"""
    match = APACHE_LOG_PATTERN.match(line)
    if not match:
        return None

    data = match.groupdict()

    # 转换状态码为整数
    data['status'] = int(data['status'])

    # 转换响应大小（- 表示0）
    data['size'] = 0 if data['size'] == '-' else int(data['size'])

    # 解析时间戳
    try:
        data['datetime'] = datetime.strptime(
            data['time'], '%d/%b/%Y:%H:%M:%S %z'
        )
    except ValueError:
        data['datetime'] = None

    return data

# 测试解析
sample = '192.168.1.1 - - [05/May/2026:10:15:30 +0800] "POST /api/login HTTP/1.1" 200 832 "https://example.com" "Mozilla/5.0"'
parsed = parse_apache_log(sample)
print(parsed['ip'], parsed['method'], parsed['path'], parsed['status'])
# 输出: 192.168.1.1 POST /api/login 200

# 解析JSON格式的结构化日志
import json

def parse_json_log(line):
    """解析JSON格式日志"""
    try:
        record = json.loads(line)
        return {
            'timestamp': record.get('@timestamp'),
            'level': record.get('level', record.get('severity')),
            'logger': record.get('logger_name'),
            'message': record.get('message'),
            'exception': record.get('exception'),
            'thread': record.get('thread_name'),
        }
    except json.JSONDecodeError:
        return None

# 解析CSV格式日志（逗号/管道符/制表符分隔）
import csv
from io import StringIO

def parse_csv_logs(raw_text, delimiter=','):
    reader = csv.DictReader(StringIO(raw_text), delimiter=delimiter)
    return [row for row in reader]

# 灵活的自定义格式解析器（多模式匹配）
import re

class FlexibleLogParser:
    """支持多种日志格式的灵活解析器"""

    def __init__(self):
        self.patterns = [
            ('apache', APACHE_LOG_PATTERN),
            ('json', None),  # 特殊处理
            ('syslog', re.compile(
                r'^<(?P<facility>\d+)>(?P<month>\w{3})\s+'
                r'(?P<day>\d+)\s+(?P<time>\S+)\s+'
                r'(?P<host>\S+)\s+(?P<app>[^:]+):\s+'
                r'(?P<message>.*)'
            )),
            ('custom', re.compile(
                r'\[(?P<level>\w+)\]\s+'
                r'(?P<timestamp>\d{4}-\d{2}-\d{2}.*?)'
                r'\s+(?P<message>.*)'
            )),
        ]

    def parse(self, line):
        line = line.strip()
        if not line:
            return None

        # 尝试JSON解析
        if line.startswith('{'):
            result = parse_json_log(line)
            if result:
                return result

        # 尝试正则匹配
        for name, pattern in self.patterns:
            if pattern is None:
                continue
            match = pattern.match(line)
            if match:
                result = match.groupdict()
                result['_format'] = name
                return result

        return {'_format': 'unknown', 'raw': line}

四、日志统计分析

日志在经过采集和解析之后，最重要的工作就是从海量的记录中提取出有价值的统计信息。日志统计分析的核心任务包括频率统计（如每分钟请求数、各状态码的出现次数）、时间趋势分析（如按小时/天/周聚合的指标变化）、IP分析（如访问来源分布、高频IP识别）、用户行为分析（如页面访问路径、会话时长分布）等。python的pandas库是进行日志统计分析最强大的工具，它提供了DataFrame数据结构和丰富的聚合函数，可以轻松完成分组、计数、排序、透视等操作。

频率统计是最基础也最常用的分析维度。通过对日志中的某个字段（如状态码、请求路径、客户端IP）进行分组计数，可以快速了解系统的运行状况。例如，按照小时统计请求量可以识别流量高峰时段，按照状态码分布可以评估系统健康度（4xx和5xx的比例是重要的服务健康指标）。时间趋势分析则将日志按时间维度聚合，生成时间序列数据，用于观察系统负载的变化规律、识别周期性模式、发现异常波动。

IP分析在安全审计和流量分析中尤为重要。通过统计每个IP的请求次数、请求间隔和访问的资源路径，可以识别出恶意爬虫、DDoS攻击或异常扫描行为。用户行为分析更进一步，通过追踪同一用户的请求序列（通常基于Cookie或Session ID），可以还原用户的访问路径，分析页面停留时间和转化漏斗。这些分析结果对于系统优化、安全防护和业务决策都具有重要参考价值。在实际项目中，建议将统计结果导出为CSV或Excel文件，便于进一步的离线分析和跨团队共享。

import pandas as pd
from collections import Counter
from datetime import datetime

# 示例：对解析后的日志列表进行统计分析
def analyze_logs(parsed_logs):
    """对解析后的日志进行全方位统计分析"""
    df = pd.DataFrame(parsed_logs)

    # 1. 状态码分布
    status_dist = df['status'].value_counts().sort_index()
    print('=== 状态码分布 ===')
    for status, count in status_dist.items():
        print(f'  {status}: {count} 次 ({count/len(df)*100:.1f}%)')

    # 2. 错误率统计
    error_rate = (df['status'] >= 400).mean() * 100
    print(f'\n错误率（4xx+5xx）: {error_rate:.2f}%')

    # 3. 最频繁请求路径 Top 10
    top_paths = df['path'].value_counts().head(10)
    print('\n=== 最频繁请求路径 Top 10 ===')
    for path, count in top_paths.items():
        print(f'  {count:6d}  {path}')

    # 4. 请求方法分布
    method_dist = df['method'].value_counts()
    print('\n=== 请求方法分布 ===')
    for method, count in method_dist.items():
        print(f'  {method}: {count}')

    return df

# 时间趋势分析：按小时统计请求量
def time_series_analysis(df):
    """按小时统计请求量并检测异常峰值"""
    df['hour'] = df['datetime'].dt.floor('H')
    hourly_counts = df.groupby('hour').size().reset_index(name='count')

    mean_val = hourly_counts['count'].mean()
    std_val = hourly_counts['count'].std()
    threshold = mean_val + 3 * std_val

    # 标记异常时段
    hourly_counts['anomaly'] = hourly_counts['count'] > threshold
    anomalies = hourly_counts[hourly_counts['anomaly']]

    print(f'每小时平均请求: {mean_val:.0f} (标准差: {std_val:.0f})')
    print(f'异常阈值: {threshold:.0f}')
    if not anomalies.empty:
        print('异常时段:')
        for _, row in anomalies.iterrows():
            print(f'  {row["hour"]}: {row["count"]} 请求')

    return hourly_counts

# IP分析与访问者统计
def ip_analysis(df, top_n=20):
    """分析IP访问特征，识别异常行为"""
    ip_stats = df.groupby('ip').agg(
        request_count=('status', 'count'),
        error_count=('status', lambda x: (x >= 400).sum()),
        unique_paths=('path', 'nunique'),
        total_bytes=('size', 'sum'),
    ).reset_index()

    ip_stats['error_rate'] = (ip_stats['error_count'] / ip_stats['request_count'] * 100).round(2)
    ip_stats = ip_stats.sort_values('request_count', ascending=False)

    # 标记可疑IP：高请求量 + 高错误率
    ip_stats['suspicious'] = (
        (ip_stats['error_rate'] > 50) &
        (ip_stats['request_count'] > ip_stats['request_count'].median())
    )

    print(f'=== 访问量 Top {top_n} IP ===')
    for _, row in ip_stats.head(top_n).iterrows():
        flag = ' [可疑]' if row['suspicious'] else ''
        print(f'  {row["ip"]:20s} {row["request_count"]:6d} 次, 错误率 {row["error_rate"]:5.1f}%{flag}')

    return ip_stats

五、异常检测

异常检测是日志分析的进阶应用，目的是从海量正常的日志记录中自动发现那些"与众不同"的模式和事件。传统的异常检测方法依赖人工设定固定阈值，例如当5xx错误率超过5%时触发告警。但更智能的异常检测需要基于历史数据动态调整阈值，识别出统计意义上的异常波动。在Python中，可以综合运用多种方法：基于统计学的Z-Score和移动平均法、基于规则的模式匹配、以及简单的时间序列分析方法（如季节性分解）。

错误率监控是日常运维中最基础的异常检测场景。通过计算每分钟或每小时的错误请求占比，并与历史基线进行比较，可以及时发现服务质量的突然下降。异常模式识别则更进一步，不仅关注数量变化，还关注异常序列的特征，例如短时间内同一个IP反复触发404（可能为扫描攻击）、大量500错误集中出现在某个特定接口（可能为程序缺陷）、或者请求量在某分钟突然飙升（可能为流量攻击）。

阈值告警系统通常采用分级策略：WARNING级别用于预警异常趋势，可以结合移动平均线的偏离度来判断；CRITICAL级别用于应急响应，当错误率突破严重阈值或某个关键接口不可用时立即触发。趋势变化检测还可以使用环比和同比分析，与前一周期或去年同期的数据进行对比。所有检测出的异常事件都应记录到异常事件表中，包含时间、类型、严重程度、相关上下文和原始日志行等信息，便于后续分析和复盘。

import numpy as np

class ErrorRateMonitor:
    """错误率监控与异常检测器"""

    def __init__(self, window_size=60, z_threshold=3.0):
        self.window_size = window_size  # 滑动窗口大小（分钟）
        self.z_threshold = z_threshold  # Z-Score异常阈值
        self.history = []  # 历史错误率记录

    def detect(self, current_error_rate):
        """基于Z-Score检测当前错误率是否异常"""
        self.history.append(current_error_rate)

        # 保留最近的N个数据
        if len(self.history) > self.window_size:
            self.history.pop(0)

        if len(self.history) < 10:  # 数据不足时不做判断
            return False, 0.0

        mean = np.mean(self.history[:-1])  # 排除当前值计算历史均值
        std = np.std(self.history[:-1])

        if std == 0:
            return False, 0.0

        z_score = (current_error_rate - mean) / std
        is_anomaly = abs(z_score) > self.z_threshold

        return is_anomaly, z_score

# 使用示例
monitor = ErrorRateMonitor(window_size=30, z_threshold=3.0)
for rate in [0.02, 0.03, 0.01, 0.25, 0.02]:
    is_anom, z = monitor.detect(rate)
    print(f'错误率 {rate:.2%} -> 异常: {is_anom}, Z-Score: {z:.2f}')

# 异常模式识别：检测特定异常序列
def detect_abnormal_patterns(parsed_logs):
    """检测多种异常模式"""
    alerts = []

    # 模式1: 短时间内同一IP大量404
    ip_404 = Counter()
    for log in parsed_logs:
        if log.get('status') == 404:
            ip_404[log['ip']] += 1

    for ip, cnt in ip_404.most_common(5):
        if cnt >= 50:
            alerts.append({
                'type': 'SCAN_ATTACK',
                'ip': ip,
                'count': cnt,
                'message': f'IP {ip} 在分析窗口内产生 {cnt} 次404，疑似扫描攻击',
                'severity': 'CRITICAL',
            })

    # 模式2: 接口错误率飙升
    path_errors = Counter()
    path_total = Counter()
    for log in parsed_logs:
        path = log.get('path', '')
        path_total[path] += 1
        if int(log.get('status', 200)) >= 500:
            path_errors[path] += 1

    for path in path_errors:
        if path_total[path] >= 10:
            err_rate = path_errors[path] / path_total[path]
            if err_rate > 0.2:
                alerts.append({
                    'type': 'HIGH_ERROR_RATE',
                    'path': path,
                    'error_rate': f'{err_rate:.0%}',
                    'message': f'接口 {path} 错误率达 {err_rate:.0%}',
                    'severity': 'WARNING',
                })

    return alerts

# 趋势变化检测（环比/同比）
def trend_change_detection(current_hour_count, yesterday_hour_count, last_week_hour_count):
    """检测请求量趋势变化"""
    findings = []

    # 环比：与昨天同一小时对比
    if yesterday_hour_count > 0:
        change = (current_hour_count - yesterday_hour_count) / yesterday_hour_count
        if abs(change) > 0.3:
            findings.append({
                'type': '环比变化',
                'change': f'{change:+.1%}',
            })

    # 同比：与上周同一小时对比
    if last_week_hour_count > 0:
        yoy = (current_hour_count - last_week_hour_count) / last_week_hour_count
        if abs(yoy) > 0.5:
            findings.append({
                'type': '同比变化',
                'change': f'{yoy:+.1%}',
            })

    return findings

六、可视化报表

日志分析的最终目的是将复杂的统计数据转化为直观的可视化报告，让决策者和管理者能够一目了然地掌握系统运行状况。Python拥有丰富的数据可视化生态系统，其中matplotlib和seaborn是最常用的图表生成库。matplotlib提供了从折线图、柱状图、饼图到热力图、箱线图的完整图表类型，而seaborn则在此基础上提供了更美观的默认样式和更高层次的统计图表接口。

HTML报表是最灵活的可视化输出形式。通过Python生成包含CSS样式、JavaScript图表库（如ECharts）和统计数据的HTML页面，可以实现交互式的数据探索体验。报告中可以包含请求量趋势折线图、状态码分布饼图、IP地理分布地图、响应时间热力图等多种图表，并支持鼠标悬停提示、图表缩放、数据筛选等交互功能。对于需要邮件发送或存档的场景，matplotlib可以直接输出静态图片嵌入到邮件正文或PDF文件中。

Excel自动报表在企业办公场景中需求广泛。利用openpyxl库，可以将统计数据和图表直接嵌入到Excel工作簿中，生成专业的周报/月报。openpyxl支持在Excel中创建柱状图、折线图、饼图等，并且可以精确控制图表的样式、位置和数据源。结合smtplib库，还可以实现定时邮件发送功能，让报表在每天固定时间自动推送到相关人员的邮箱。在实际项目中，建议将报表生成逻辑抽象为模板化的函数，通过参数配置灵活调整报表的时间范围、维度和图表类型，以适应不同场景的需求。

import matplotlib
matplotlib.use('Agg')  # 无界面后端，适用于服务器
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import seaborn as sns
from io import BytesIO
import base64

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei']
plt.rcParams['axes.unicode_minus'] = False

def generate_trend_chart(hourly_data, output_path='trend.png'):
    """生成请求量趋势折线图"""
    fig, ax = plt.subplots(figsize=(14, 6))

    hours = [row['hour'] for row in hourly_data]
    counts = [row['count'] for row in hourly_data]
    anomalies = [row for row in hourly_data if row.get('anomaly')]

    # 主趋势线
    ax.plot(hours, counts, color='#2196F3', linewidth=2, marker='o',
            markersize=4, label='请求量')
    ax.fill_between(hours, counts, alpha=0.1, color='#2196F3')

    # 标注异常点
    if anomalies:
        ax.scatter(
            [a['hour'] for a in anomalies],
            [a['count'] for a in anomalies],
            color='#f44336', s=100, zorder=5, label='异常点'
        )

    ax.set_xlabel('时间', fontsize=12)
    ax.set_ylabel('请求量', fontsize=12)
    ax.set_title('请求量时间趋势图（按小时）', fontsize=14, fontweight='bold')
    ax.legend()
    ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d %H:00'))
    plt.xticks(rotation=45)
    plt.tight_layout()

    plt.savefig(output_path, dpi=150, bbox_inches='tight')
    plt.close()
    print(f'趋势图已保存: {output_path}')

# 生成HTML报表（带内嵌图片）
def fig_to_base64(fig):
    """将matplotlib图形转换为base64编码的HTML内嵌图片"""
    buf = BytesIO()
    fig.savefig(buf, format='png', dpi=120, bbox_inches='tight')
    buf.seek(0)
    img_b64 = base64.b64encode(buf.read()).decode('utf-8')
    plt.close(fig)
    return f'data:image/png;base64,{img_b64}'

def generate_html_report(df, stats, output_path):
    """生成HTML格式的日志分析报表"""
    # 生成图表
    fig, axes = plt.subplots(1, 2, figsize=(16, 6))

    # 状态码分布饼图
    status_counts = df['status'].value_counts()
    colors = {200: '#4CAF50', 301: '#FF9800', 302: '#FF9800',
              404: '#F44336', 500: '#9C27B0'}
    pie_colors = [colors.get(s, '#607D8B') for s in status_counts.index]
    axes[0].pie(status_counts.values, labels=status_counts.index.astype(str),
               autopct='%1.1f%%', colors=pie_colors, startangle=90)
    axes[0].set_title('状态码分布')

    # 请求方法分布柱状图
    method_counts = df['method'].value_counts()
    axes[1].bar(method_counts.index, method_counts.values, color='#2196F3')
    axes[1].set_title('请求方法分布')
    axes[1].set_ylabel('次数')

    plt.tight_layout()
    chart_img = fig_to_base64(fig)

    # 构建HTML内容
    html = f'''<!DOCTYPE html>
<html>
<head><meta charset="utf-8">
<title>日志分析报表</title>
<style>
  body {{ font-family: 'Microsoft YaHei', sans-serif; max-width: 1000px; margin: 20px auto; padding: 20px; }}
  h1 {{ color: #2e7d32; }}
  .stats {{ display: grid; grid-template-columns: repeat(4, 1fr); gap: 15px; margin: 20px 0; }}
  .stat-card {{ background: #f5f5f5; padding: 15px; border-radius: 8px; text-align: center; }}
  .stat-card .num {{ font-size: 2em; font-weight: bold; color: #2e7d32; }}
  img {{ max-width: 100%; }}
</style></head>
<body>
<h1>日志分析报表</h1>
<p>生成时间: {datetime.now().strftime("%Y-%m-%d %H:%M")}</p>
<div class="stats">
  <div class="stat-card"><div class="num">{stats["total_requests"]}</div><div>总请求数</div></div>
  <div class="stat-card"><div class="num">{stats["error_rate"]:.1f}%</div><div>错误率</div></div>
  <div class="stat-card"><div class="num">{stats["unique_ips"]}</div><div>独立IP</div></div>
  <div class="stat-card"><div class="num">{stats["peak_qps"]}</div><div>峰值QPS</div></div>
</div>
<h2>图表分析</h2>
<img src="{chart_img}" alt="统计图表">
</body></html>'''

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

# 使用openpyxl生成Excel报表（含图表）
from openpyxl import Workbook
from openpyxl.chart import BarChart, Reference, PieChart
from openpyxl.styles import Font, PatternFill, Alignment

def generate_excel_report(df, output_path):
    wb = Workbook()
    ws = wb.active
    ws.title = '日志分析报表'

    # 写入标题
    ws['A1'] = '日志分析报告'
    ws['A1'].font = Font(size=16, bold=True, color='2E7D32')
    ws.merge_cells('A1:E1')

    # 写入状态码分布数据
    ws['A3'] = '状态码'
    ws['B3'] = '次数'
    for i, (code, count) in enumerate(df['status'].value_counts().items(), start=4):
        ws[f'A{i}'] = code
        ws[f'B{i}'] = count

    # 添加柱状图
    chart = BarChart()
    chart.type = 'col'
    chart.title = '状态码分布'
    data = Reference(ws, min_col=2, min_row=3, max_row=4 + len(df['status'].value_counts()))
    cats = Reference(ws, min_col=1, min_row=4, max_row=4 + len(df['status'].value_counts()))
    chart.add_data(data, titles_from_data=True)
    chart.set_categories(cats)
    ws.add_chart(chart, 'D3')

    wb.save(output_path)
    print(f'Excel报表已生成: {output_path}')

七、日志告警

日志告警是连接日志分析与运维响应的关键桥梁。当系统出现异常时，如果不能及时通知到相关人员，日志分析的价值将大打折扣。一个成熟的告警系统需要具备基于规则的灵活配置能力、多级别的触发机制、多渠道的通知方式以及告警抑制和升级机制。Python的灵活性使得构建这样一个告警系统变得相对简单，可以通过配置文件和策略模式轻松调整告警规则和通知方式。

基于规则的告警是最常用也是最可靠的方式。每条规则包含触发条件（如错误率超过5%）、持续时长（如连续5分钟超过阈值）、告警级别（INFO/WARNING/CRITICAL）和通知渠道。通知渠道的选择应根据告警的紧急程度来定：CRITICAL级别的告警应通过企业微信或钉钉机器人实时推送，同时发送短信或电话通知；WARNING级别的告警可以通过邮件发送；INFO级别的告警则记录到告警日志中即可。告警抑制机制可以避免在短时间内重复发送相同内容的告警，通常采用时间窗口去重的策略。

告警升级机制是为防止告警被忽略而设计的。如果一条CRITICAL告警在指定时间内未被确认或解决，系统会自动将其上报给更高层级的管理者或运维团队。在实际实现中，可以使用责任链模式或状态机模式管理告警的生命周期。告警状态通常包括：触发（Firing）、确认（Acknowledged）、处理中（Processing）、已解决（Resolved）和已忽略（Ignored）。每条告警的处理过程应该被完整记录，形成可追溯的事件时间线，用于事后分析和流程改进。

# 告警规则引擎定义
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from datetime import datetime, timedelta
import hashlib

class AlertRule:
    """告警规则定义"""

    def __init__(self, name, condition_func, level='WARNING',
                 channels=None, cooldown_minutes=15):
        self.name = name
        self.condition_func = condition_func  # 输入 stats, 返回 bool
        self.level = level  # INFO / WARNING / CRITICAL
        self.channels = channels or ['email']
        self.cooldown = timedelta(minutes=cooldown_minutes)
        self.last_triggered = datetime.min  # 上次触发时间（用于抑制）

    def should_trigger(self, stats):
        now = datetime.now()
        # 检查冷却时间
        if now - self.last_triggered < self.cooldown:
            return False
        # 执行条件判断
        if self.condition_func(stats):
            self.last_triggered = now
            return True
        return False

# 邮件通知发送器
class EmailNotifier:
    def __init__(self, smtp_host, smtp_port, username, password, recipients):
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
        self.recipients = recipients if isinstance(recipients, list) else [recipients]

    def send(self, subject, body, level='WARNING'):
        msg = MIMEMultipart()
        msg['From'] = self.username
        msg['To'] = ','.join(self.recipients)
        msg['Subject'] = Header(f'[{level}] {subject}', 'utf-8')

        html_body = f'''<html><body>
<h2>{subject}</h2>
<p>级别: {level}</p>
<p>时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p>
<pre>{body}</pre>
</body></html>'''
        msg.attach(MIMEText(html_body, 'html', 'utf-8'))

        with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
            server.starttls()
            server.login(self.username, self.password)
            server.send_message(msg)

# 企业微信机器人通知
import requests

def send_wechat_alert(webhook_url, content, level='WARNING'):
    color_map = {'INFO': 'info', 'WARNING': 'warning', 'CRITICAL': 'error'}
    payload = {
        'msgtype': 'markdown',
        'markdown': {
            'content': f'## [{level}] 日志告警通知\n> {content}\n> 时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
        }
    }
    try:
        resp = requests.post(webhook_url, json=payload, timeout=10)
        resp.raise_for_status()
        print('企业微信告警已发送')
    except Exception as e:
        print(f'发送企业微信告警失败: {e}')

# 告警引擎：整合规则、通知和抑制机制
class AlertEngine:
    """告警引擎，管理规则注册、触发与通知"""

    def __init__(self):
        self.rules = []
        self.notifiers = {}
        self.alert_history = []
        self.alert_id_set = set()

    def add_rule(self, rule):
        self.rules.append(rule)

    def register_notifier(self, name, notifier):
        self.notifiers[name] = notifier

    def evaluate(self, stats):
        """评估所有规则并触发告警"""
        for rule in self.rules:
            if rule.should_trigger(stats):
                alert_id = hashlib.md5(
                    f'{rule.name}{datetime.now().strftime("%Y%m%d%H")}'.encode()
                ).hexdigest()[:8]

                # 告警抑制：同一规则同一小时内不重复告警
                if alert_id in self.alert_id_set:
                    continue
                self.alert_id_set.add(alert_id)

                alert_record = {
                    'id': alert_id,
                    'rule': rule.name,
                    'level': rule.level,
                    'time': datetime.now(),
                    'stats': stats,
                }
                self.alert_history.append(alert_record)

                # 发送通知
                for channel in rule.channels:
                    notifier = self.notifiers.get(channel)
                    if notifier:
                        notifier.send(
                            subject=f'告警: {rule.name}',
                            body=f'规则 {rule.name} 触发 {rule.level} 告警\n当前状态: {stats}',
                            level=rule.level,
                        )

                print(f'[告警] {rule.level} {rule.name}')

八、归档与轮转

日志文件的增长速度往往超出预期，一个中等规模的Web应用每天可能产生数百MB甚至数GB的日志数据。如果不加管理，磁盘空间很快就会被耗尽，同时查询旧日志的效率也会急剧下降。日志归档与轮转正是为了解决这些问题而设计的。Python的logging模块内置了RotatingFileHandler和TimedRotatingFileHandler，分别支持基于文件大小和基于时间的日志轮转策略，可以自动完成日志切割、压缩和清理工作。

基于文件大小的轮转策略（RotatingFileHandler）在日志文件达到指定大小时进行切割，例如设置为100MB，当日志文件超过100MB时自动重命名为app.log.1，并创建一个新的app.log继续写入。可以设置保留的文件数量，超过数量的旧日志会被自动删除。基于时间的轮转策略（TimedRotatingFileHandler）则按照固定的时间间隔进行切割，支持按秒、分、时、天、周和月轮转。对于生产环境，建议同时使用两种策略：按天轮转并使用gzip压缩，保留最近30天的日志用于在线查询，超过30天的自动归档到冷存储。

冷热数据分离是一种经济高效的日志存储策略。热数据（最近7天）存放在高速磁盘上，支持快速查询和分析。温数据（8-30天）存放在普通存储上，按天压缩存档。冷数据（超过30天）转移到低成本的对象存储（如阿里云OSS、AWS S3）或归档存储中，仅在需要审计或事故调查时才恢复。自动清理策略同样重要，需要根据合规要求和实际需求设定合理的保留期限。对于金融、医疗等受监管行业，日志保留期限可能有明确的法律要求，在设置清理策略时必须予以考虑。

# 使用RotatingFileHandler实现基于文件大小的日志轮转
import logging
from logging.handlers import RotatingFileHandler

def setup_rotating_logger(name, log_file, max_bytes=10*1024*1024, backup_count=5):
    """配置基于大小轮转的日志记录器"""
    logger = logging.getLogger(name)

    # 创建RotatingFileHandler
    handler = RotatingFileHandler(
        log_file,
        mode='a',
        maxBytes=max_bytes,    # 单个文件最大10MB
        backupCount=backup_count,  # 保留5个备份文件
        encoding='utf-8',
    )

    formatter = logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

    return logger

# 使用示例
logger = setup_rotating_logger('AppLogger', 'app.log', max_bytes=5*1024*1024, backup_count=10)
# 当日志达到5MB时自动轮转，产生 app.log.1, app.log.2 ... app.log.10

# 基于时间的日志轮转+自动压缩
from logging.handlers import TimedRotatingFileHandler
import gzip
import shutil
from pathlib import Path

def setup_timed_rotating_logger(name, log_file, when='midnight', backup_count=30):
    """配置基于时间轮转的日志记录器（每天午夜轮转）"""
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)

    handler = TimedRotatingFileHandler(
        log_file,
        when=when,           # 'midnight' 表示每天午夜轮转
        interval=1,
        backupCount=backup_count,  # 保留30天
        encoding='utf-8',
        utc=False,
    )

    formatter = logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    return logger

# 压缩旧日志文件
def compress_old_logs(log_dir, days_old=7):
    """压缩超过指定天数的日志文件"""
    log_path = Path(log_dir)
    now = datetime.now()
    compressed_count = 0

    for fpath in log_path.glob('*.log.*'):
        if fpath.suffix == '.gz':
            continue  # 已压缩跳过

        mtime = datetime.fromtimestamp(fpath.stat().st_mtime)
        age_days = (now - mtime).days

        if age_days >= days_old:
            gz_path = fpath.with_suffix(f'{fpath.suffix}.gz')
            with open(fpath, 'rb') as f_in:
                with gzip.open(gz_path, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            fpath.unlink()  # 删除原文件
            compressed_count += 1

    print(f'已压缩 {compressed_count} 个旧日志文件')

# 冷热数据分离与自动清理
class LogArchiveManager:
    """日志归档管理器：实现冷热数据分离和自动清理"""

    def __init__(self, hot_dir, cold_dir, retention_days):
        self.hot_dir = Path(hot_dir)
        self.cold_dir = Path(cold_dir)
        self.retention_days = retention_days
        self.cold_dir.mkdir(parents=True, exist_ok=True)

    def archive_old_logs(self, days_threshold=7):
        """将超过指定天数的日志转移到冷存储"""
        now = datetime.now()
        moved = 0

        for fpath in self.hot_dir.glob('*.log*'):
            mtime = datetime.fromtimestamp(fpath.stat().st_mtime)
            age_days = (now - mtime).days

            if age_days >= days_threshold:
                dest = self.cold_dir / fpath.name
                # 压缩后移动到冷存储
                gz_dest = dest.with_suffix(f'{dest.suffix}.gz')
                with open(fpath, 'rb') as f_in:
                    with gzip.open(gz_dest, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)
                fpath.unlink()
                moved += 1

        print(f'已归档 {moved} 个日志文件到冷存储')

    def cleanup_expired(self):
        """删除超过保留期限的日志文件"""
        now = datetime.now()
        removed = 0

        for fpath in list(self.hot_dir.glob('*')) + list(self.cold_dir.glob('*')):
            mtime = datetime.fromtimestamp(fpath.stat().st_mtime)
            age_days = (now - mtime).days
            if age_days > self.retention_days:
                fpath.unlink()
                removed += 1

        print(f'已清理 {removed} 个过期日志文件')

九、实战案例

理论结合实践才能真正掌握日志自动化处理的技能。本章通过三个完整的实战案例，演示如何将前面介绍的技术综合运用，解决实际工作中的日志处理需求。每个案例都包含完整的代码实现和运行结果，读者可以根据自己的实际场景进行修改和扩展。这三个案例涵盖了日志处理最常见的三类应用场景：Web服务器访问日志分析、应用程序错误日志监控告警、以及系统资源日志报表生成。

Web服务器访问日志分析是最经典的日志处理场景。通过采集Nginx或Apache的访问日志，解析出每个请求的详细信息，然后进行多维度统计分析。我们可以生成每小时请求量趋势图、状态码分布饼图、热门页面排行榜、访问来源IP地理分布等一系列可视化报告。这些报告对于了解网站流量特征、发现性能瓶颈、识别安全攻击都有着重要作用。在实际部署中，建议将分析脚本配置为crontab定时任务，每天凌晨自动处理前一天的日志并生成日报发送给相关人员。

应用程序错误日志监控是保障系统稳定性的关键防线。通过实时采集中间件和业务应用产生的错误日志，结合告警规则引擎，可以在系统出现异常的第一时间通知运维人员。系统资源日志则可以从操作系统层面反映服务器的运行状态，包括CPU使用率、内存占用、磁盘IO、网络流量等指标。将这些指标汇总为可视化报表，可以帮助运维团队了解系统资源的使用趋势，提前进行容量规划和性能优化。三个案例结合起来，构成了一套完整的日志监控分析体系，在本章末尾，还提供了将这些组件集成为统一命令行工具的框架代码。

# 案例1: Web服务器访问日志完整分析管道
import pandas as pd
from collections import Counter

def web_log_analysis_pipeline(log_file_path):
    """Web服务器日志完整分析管道"""
    print('[1/5] 读取日志文件...')
    with open(log_file_path, 'r', encoding='utf-8') as f:
        raw_lines = f.readlines()
    print(f'  共读取 {len(raw_lines)} 行日志')

    print('[2/5] 解析日志...')
    parsed = []
    for line in raw_lines:
        record = parse_apache_log(line.strip())
        if record:
            parsed.append(record)
    print(f'  成功解析 {len(parsed)} 条记录')

    print('[3/5] 转换为DataFrame...')
    df = pd.DataFrame(parsed)

    print('[4/5] 执行统计分析...')
    stats = {
        'total_requests': len(df),
        'unique_ips': df['ip'].nunique(),
        'unique_paths': df['path'].nunique(),
        'error_rate': (df['status'] >= 400).mean() * 100,
        'total_bytes': df['size'].sum(),
        'peak_qps': 0,  # 需要时间计算
    }
    for k, v in stats.items():
        print(f'  {k}: {v}')

    print('[5/5] 生成报表...')
    generate_html_report(df, stats, 'web_log_report.html')
    generate_excel_report(df, 'web_log_report.xlsx')

    print('\n分析完成！')
    return df, stats

# 运行
# df, stats = web_log_analysis_pipeline('/var/log/nginx/access.log')

# 案例2: 应用错误日志实时监控
class AppErrorMonitor:
    """应用错误日志实时监控系统"""

    def __init__(self, log_file, keywords=None):
        self.tailer = LogTailer(log_file)
        self.keywords = keywords or ['ERROR', 'FATAL', 'Exception', 'Traceback']
        self.error_count = 0
        self.last_minute_errors = []
        self.alert_engine = AlertEngine()

        # 注册默认告警规则：连续5个错误/分钟触发警告
        self.alert_engine.add_rule(AlertRule(
            name='高频错误',
            condition_func=lambda s: s.get('errors_per_minute', 0) >= 5,
            level='WARNING',
            channels=['email'],
        ))

    def start_monitoring(self):
        print('开始监控应用错误日志...')
        for batch in self.tailer.follow():
            for line in batch:
                # 检查是否包含错误关键词
                for kw in self.keywords:
                    if kw in line:
                        self.error_count += 1
                        now = datetime.now()
                        self.last_minute_errors.append(now)

                        # 打印错误到控制台
                        print(f'[错误] {now.strftime("%H:%M:%S")} {line[:120]}')

                        # 评估告警规则
                        self.last_minute_errors = [
                            t for t in self.last_minute_errors
                            if (now - t).total_seconds() < 60
                        ]
                        stats = {'errors_per_minute': len(self.last_minute_errors)}
                        self.alert_engine.evaluate(stats)
                        break

# 案例3: 系统资源日志采集与报表
import psutil
import csv
import time

class SystemResourceLogger:
    """系统资源日志采集器，定时采集CPU/内存/磁盘/网络指标"""

    def __init__(self, output_file='system_metrics.csv', interval=60):
        self.output_file = output_file
        self.interval = interval
        self.fields = [
            'timestamp', 'cpu_percent', 'memory_percent',
            'disk_usage_percent', 'net_sent_bytes', 'net_recv_bytes',
            'load_avg_1m', 'load_avg_5m', 'load_avg_15m',
        ]

    def collect_once(self):
        """采集一次系统资源指标"""
        net_io = psutil.net_io_counters()
        return {
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_usage_percent': psutil.disk_usage('/').percent,
            'net_sent_bytes': net_io.bytes_sent,
            'net_recv_bytes': net_io.bytes_recv,
            'load_avg_1m': psutil.getloadavg()[0] if hasattr(psutil, 'getloadavg') else 0,
            'load_avg_5m': psutil.getloadavg()[1] if hasattr(psutil, 'getloadavg') else 0,
            'load_avg_15m': psutil.getloadavg()[2] if hasattr(psutil, 'getloadavg') else 0,
        }

    def start_collection(self, duration_minutes=1440):
        """启动定时采集（默认采集24小时）"""
        file_exists = Path(self.output_file).exists()

        with open(self.output_file, 'a', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=self.fields)
            if not file_exists:
                writer.writeheader()

            end_time = time.time() + duration_minutes * 60
            while time.time() < end_time:
                record = self.collect_once()
                writer.writerow(record)
                f.flush()
                print(f'已采集: {record["timestamp"]} CPU={record["cpu_percent"]}% 内存={record["memory_percent"]}%')
                time.sleep(self.interval)

    def generate_report(self):
        """采集完成后生成可视化报表"""
        df = pd.read_csv(self.output_file)
        fig, axes = plt.subplots(2, 2, figsize=(14, 10))

        # CPU趋势
        axes[0, 0].plot(df['timestamp'], df['cpu_percent'], color='#2196F3')
        axes[0, 0].set_title('CPU使用率')
        axes[0, 0].tick_params(axis='x', rotation=45)

        # 内存趋势
        axes[0, 1].plot(df['timestamp'], df['memory_percent'], color='#4CAF50')
        axes[0, 1].set_title('内存使用率')
        axes[0, 1].tick_params(axis='x', rotation=45)

        # 磁盘使用率
        axes[1, 0].plot(df['timestamp'], df['disk_usage_percent'], color='#FF9800')
        axes[1, 0].set_title('磁盘使用率')
        axes[1, 0].tick_params(axis='x', rotation=45)

        # 负载均值
        axes[1, 1].plot(df['timestamp'], df['load_avg_1m'], label='1分钟', color='#F44336')
        axes[1, 1].plot(df['timestamp'], df['load_avg_5m'], label='5分钟', color='#9C27B0')
        axes[1, 1].set_title('系统负载')
        axes[1, 1].legend()
        axes[1, 1].tick_params(axis='x', rotation=45)

        plt.tight_layout()
        plt.savefig('system_metrics_report.png', dpi=150)
        plt.close()
        print('系统资源报表已生成: system_metrics_report.png')

# 统一入口：集成所有功能的命令行工具框架
import argparse

def main():
    parser = argparse.ArgumentParser(description='日志自动采集、分析与报表系统')
    parser.add_argument('action', choices=['analyze', 'monitor', 'collect', 'report', 'archive'],
                        help='操作类型')
    parser.add_argument('--input', '-i', help='输入日志文件或目录')
    parser.add_argument('--output', '-o', default='./output', help='输出目录')
    parser.add_argument('--config', '-c', help='配置文件路径')
    args = parser.parse_args()

    if args.action == 'analyze':
        if not args.input:
            parser.error('analyze 操作需要 --input 参数')
        web_log_analysis_pipeline(args.input)

    elif args.action == 'monitor':
        if not args.input:
            parser.error('monitor 操作需要 --input 参数')
        monitor = AppErrorMonitor(args.input)
        monitor.start_monitoring()

    elif args.action == 'collect':
        collector = SystemResourceLogger(
            output_file=f'{args.output}/system_metrics.csv',
            interval=60
        )
        collector.start_collection(duration_minutes=1440)

    elif args.action == 'report':
        collector = SystemResourceLogger()
        collector.generate_report()

    elif args.action == 'archive':
        manager = LogArchiveManager(
            hot_dir='./logs/hot',
            cold_dir='./logs/cold',
            retention_days=90
        )
        manager.archive_old_logs(days_threshold=7)
        manager.cleanup_expired()

if __name__ == '__main__':
    main()