怎么用脚本监控文本文件增长

wen 实用脚本 2026-08-02 245

本文目录导读：

怎么用脚本监控文本文件增长

Python 脚本监控
Bash Shell 脚本
Node.js 脚本
Go 语言版本
实用技巧
使用建议

Python 脚本监控

基础版（实时监控新增内容）

#!/usr/bin/env python3
import time
import os
def monitor_file(filepath, callback=None):
    """监控文件新增内容"""
    # 获取文件初始大小
    file_size = os.path.getsize(filepath)
    print(f"开始监控 {filepath}，当前大小: {file_size} 字节")
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            # 跳到文件末尾
            f.seek(file_size)
            while True:
                # 读取新增内容
                line = f.readline()
                if line:
                    if callback:
                        callback(line)
                    else:
                        print(line, end='')
                else:
                    # 检查文件是否被截断（变小）
                    current_size = os.path.getsize(filepath)
                    if current_size < file_size:
                        print(f"\n文件被截断，重置位置...")
                        f.seek(0)
                        file_size = current_size
                    time.sleep(0.1)
    except FileNotFoundError:
        print(f"文件 {filepath} 不存在")
    except KeyboardInterrupt:
        print("\n监控停止")
# 自定义回调函数
def process_line(line):
    if 'ERROR' in line:
        print(f"[错误] {time.ctime()}: {line}", end='')
if __name__ == "__main__":
    monitor_file("/path/to/your/file.log", process_line)

高级版（带统计功能）

#!/usr/bin/env python3
import time
import os
import sys
from collections import Counter
from datetime import datetime
class FileMonitor:
    def __init__(self, filepath):
        self.filepath = filepath
        self.initial_size = os.path.getsize(filepath)
        self.total_lines = 0
        self.keyword_counts = Counter()
        self.start_time = time.time()
    def monitor(self, keywords=None):
        """监控文件增长并统计"""
        with open(self.filepath, 'r', encoding='utf-8') as f:
            f.seek(self.initial_size)
            while True:
                line = f.readline()
                if line:
                    self.total_lines += 1
                    self.process_line(line, keywords)
                else:
                    self.display_stats()
                    time.sleep(1)
    def process_line(self, line, keywords):
        """处理每一行"""
        if keywords:
            for keyword in keywords:
                if keyword in line:
                    self.keyword_counts[keyword] += 1
    def display_stats(self):
        """显示统计信息"""
        os.system('clear')  # 或 'cls' 在Windows
        elapsed = time.time() - self.start_time
        file_size = os.path.getsize(self.filepath)
        print(f"= 文件监控报告 =")
        print(f"文件: {self.filepath}")
        print(f"运行时间: {elapsed:.1f} 秒")
        print(f"当前文件大小: {file_size} 字节")
        print(f"新增内容: {file_size - self.initial_size} 字节")
        print(f"新增行数: {self.total_lines}")
        print(f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        if self.keyword_counts:
            print("\n关键词统计:")
            for keyword, count in self.keyword_counts.most_common():
                print(f"  {keyword}: {count}次")
# 使用示例
if __name__ == "__main__":
    monitor = FileMonitor("/var/log/syslog")
    monitor.monitor(keywords=["ERROR", "WARNING", "INFO"])

Bash Shell 脚本

简单版（tail 命令）

#!/bin/bash
# 监控文件增长（类似 tail -f）
tail -f /var/log/syslog

增强版 Bash 脚本

#!/bin/bash
LOG_FILE="/path/to/your/file.log"
PATTERN="ERROR"
echo "开始监控 $LOG_FILE"
echo "搜索模式: $PATTERN"
# 使用 tail -f 并过滤
tail -f "$LOG_FILE" | while read line; do
    # 时间戳
    timestamp=$(date "+%Y-%m-%d %H:%M:%S")
    # 检查是否匹配模式
    if echo "$line" | grep -qi "$PATTERN"; then
        echo "[$timestamp] 匹配: $line"
    else
        echo "[$timestamp] $line"
    fi
    # 可选：执行其他命令
    # echo "$line" | grep "$PATTERN" >> matched_lines.log
done

带轮转处理

#!/bin/bash
LOG_FILE="/var/log/application.log"
LAST_SIZE=0
CHECK_INTERVAL=2  # 秒
# 获取初始大小
if [ -f "$LOG_FILE" ]; then
    LAST_SIZE=$(stat -c%s "$LOG_FILE")
else
    echo "文件不存在: $LOG_FILE"
    exit 1
fi
echo "开始监控，初始大小: $LAST_SIZE 字节"
while true; do
    sleep $CHECK_INTERVAL
    # 检查文件是否存在
    if [ ! -f "$LOG_FILE" ]; then
        echo "文件被删除或轮转!"
        # 等待文件重建
        while [ ! -f "$LOG_FILE" ]; do
            sleep 1
        done
        LAST_SIZE=0
        echo "文件重新创建，继续监控..."
        continue
    fi
    CURRENT_SIZE=$(stat -c%s "$LOG_FILE")
    # 检查文件是否被截断（大小变小）
    if [ $CURRENT_SIZE -lt $LAST_SIZE ]; then
        echo "检测到日志轮转，重置位置"
        LAST_SIZE=$CURRENT_SIZE
        continue
    fi
    # 如果有新内容
    if [ $CURRENT_SIZE -gt $LAST_SIZE ]; then
        # 读取新增部分
        tail -c $((CURRENT_SIZE - LAST_SIZE)) "$LOG_FILE"
        LAST_SIZE=$CURRENT_SIZE
    fi
done

Node.js 脚本

const fs = require('fs');
const readline = require('readline');
class LogMonitor {
    constructor(filepath) {
        this.filepath = filepath;
        this.currentPosition = fs.existsSync(filepath) 
            ? fs.statSync(filepath).size 
            : 0;
        this.keywords = ['ERROR', 'WARNING'];
    }
    start() {
        console.log(`开始监控 ${this.filepath}`);
        // 每100ms检查一次
        setInterval(() => this.checkForUpdates(), 100);
    }
    checkForUpdates() {
        if (!fs.existsSync(this.filepath)) {
            console.log('文件不存在');
            return;
        }
        const stats = fs.statSync(this.filepath);
        // 检测文件截断
        if (stats.size < this.currentPosition) {
            console.log('文件被截断，重置位置');
            this.currentPosition = 0;
        }
        // 读取新增内容
        if (stats.size > this.currentPosition) {
            const stream = fs.createReadStream(this.filepath, {
                start: this.currentPosition,
                end: stats.size - 1
            });
            const rl = readline.createInterface({ input: stream });
            rl.on('line', (line) => {
                this.processLine(line);
            });
            rl.on('close', () => {
                this.currentPosition = stats.size;
            });
        }
    }
    processLine(line) {
        const timestamp = new Date().toISOString();
        // 检查关键词
        for (const keyword of this.keywords) {
            if (line.includes(keyword)) {
                console.log(`[${timestamp}] [${keyword}] ${line}`);
                return;
            }
        }
        // 或直接输出
        console.log(`[${timestamp}] ${line}`);
    }
}
// 使用
const monitor = new LogMonitor('/var/log/app.log');
monitor.start();

Go 语言版本

package main
import (
    "fmt"
    "os"
    "time"
)
func main() {
    filepath := "/var/log/app.log"
    // 获取初始大小
    fileInfo, err := os.Stat(filepath)
    if err != nil {
        fmt.Println("文件不存在")
        return
    }
    lastSize := fileInfo.Size()
    fmt.Printf("开始监控 %s, 初始大小: %d\n", filepath, lastSize)
    for {
        // 每2秒检查一次
        time.Sleep(2 * time.Second)
        fileInfo, err = os.Stat(filepath)
        if err != nil {
            fmt.Println("文件被删除")
            continue
        }
        currentSize := fileInfo.Size()
        // 检测轮转
        if currentSize < lastSize {
            fmt.Println("检测到日志轮转")
            lastSize = 0
        }
        // 读取新增内容
        if currentSize > lastSize {
            file, _ := os.Open(filepath)
            defer file.Close()
            buffer := make([]byte, currentSize-lastSize)
            file.Seek(lastSize, 0)
            file.Read(buffer)
            fmt.Printf("[%s] %s", time.Now().Format("2006-01-02 15:04:05"), buffer)
            lastSize = currentSize
        }
    }
}

实用技巧

实时显示在Web界面

# 简单的Web监控
from flask import Flask, Response
import time
app = Flask(__name__)
@app.route('/monitor')
def stream():
    def generate():
        with open("/var/log/app.log", "r") as f:
            f.seek(0, 2)  # 跳到末尾
            while True:
                line = f.readline()
                if line:
                    yield f"data: {line}\n\n"
                else:
                    time.sleep(1)
    return Response(generate(), mimetype="text/event-stream")
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)