怎么用脚本监控文本输出

wen 实用脚本 2026-08-01 2

本文目录导读：

怎么用脚本监控文本输出

监控实时日志文件（最常用）
Python 实时监控
Node.js 监控脚本
Go 脚本监控
带过滤和告警的完整脚本
Bash 一行命令版本
使用建议

我来介绍几种监控文本输出的脚本方案,从简单到复杂：

监控实时日志文件（最常用）

Bash 脚本

#!/bin/bash
# 监控日志文件的新增内容
LOG_FILE="/var/log/application.log"
# 方法1：使用 tail -f 持续监控
tail -f "$LOG_FILE" | while read line; do
    echo "检测到新输出: $line"
    # 检测特定关键词
    if echo "$line" | grep -q "ERROR"; then
        echo "发现错误！"
        # 发送告警等
    fi
done
# 方法2：监控文件变化并显示差异
previous_hash=""
while true; do
    current_hash=$(md5sum "$LOG_FILE" | awk '{print $1}')
    if [ "$current_hash" != "$previous_hash" ]; then
        # 显示新增内容
        diff <(echo "$previous_content") <(cat "$LOG_FILE")
        previous_hash="$current_hash"
        previous_content=$(cat "$LOG_FILE")
    fi
    sleep 1
done

Python 实时监控

实时读取文件尾部

#!/usr/bin/env python3
import time
import os
import subprocess
from datetime import datetime
class LogMonitor:
    def __init__(self, file_path, keywords=None):
        self.file_path = file_path
        self.keywords = keywords or ['ERROR', 'WARNING', 'SUCCESS']
        self.file_pos = 0
    def monitor(self):
        """持续监控文件输出"""
        try:
            with open(self.file_path, 'r') as f:
                # 跳转到文件末尾开始监控
                f.seek(0, 2)  # 移动到文件末尾
                while True:
                    line = f.readline()
                    if not line:
                        time.sleep(0.1)  # 等待新数据
                        continue
                    self.process_line(line.strip())
        except FileNotFoundError:
            print(f"文件不存在: {self.file_path}")
    def process_line(self, line):
        """处理每一行日志"""
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        print(f"[{timestamp}] {line}")
        # 关键词检测
        for keyword in self.keywords:
            if keyword.lower() in line.lower():
                self.handle_alert(keyword, line)
    def handle_alert(self, keyword, line):
        """处理告警"""
        print(f"⚠️ 告警: 发现关键词 [{keyword}]")
        # 可以在这里添加邮件、短信等通知
# 使用示例
if __name__ == "__main__":
    monitor = LogMonitor("/var/log/nginx/access.log")
    monitor.monitor()

监控子进程输出

#!/usr/bin/env python3
import subprocess
import sys
def monitor_command_output(command):
    """监控命令的实时输出"""
    process = subprocess.Popen(
        command,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1
    )
    try:
        for line in iter(process.stdout.readline, ''):
            if line:
                print(f"输出: {line.strip()}")
                # 自定义处理逻辑
                if "ERROR" in line:
                    print("检测到错误！")
                elif "INFO" in line:
                    print("正常信息")
    except KeyboardInterrupt:
        print("\n监控停止")
        process.terminate()
# 监控一个命令的输出
monitor_command_output("ping google.com")

Node.js 监控脚本

#!/usr/bin/env node
const { spawn, exec } = require('child_process');
const fs = require('fs');
const readline = require('readline');
// 监控文件输出
function monitorFile(filePath) {
    let lastSize = 0;
    setInterval(() => {
        try {
            const stats = fs.statSync(filePath);
            if (stats.size > lastSize) {
                const buffer = Buffer.alloc(stats.size - lastSize);
                const fd = fs.openSync(filePath, 'r');
                fs.readSync(fd, buffer, 0, buffer.length, lastSize);
                fs.closeSync(fd);
                const output = buffer.toString('utf8');
                console.log('新输出:', output.trim());
                lastSize = stats.size;
            }
        } catch (err) {
            console.error('读取文件出错:', err);
        }
    }, 1000);
}
// 监控命令输出
function monitorCommand(command) {
    const process = spawn(command, { shell: true });
    process.stdout.on('data', (data) => {
        const lines = data.toString().split('\n');
        for (const line of lines) {
            if (line.trim()) {
                console.log(`[${new Date().toLocaleTimeString()}] ${line}`);
                // 检测关键词
                if (line.includes('ERROR')) {
                    console.log('⚠️ Error detected!');
                }
            }
        }
    });
    process.stderr.on('data', (data) => {
        console.error('stderr:', data.toString());
    });
    process.on('close', (code) => {
        console.log(`进程结束，退出码: ${code}`);
    });
}
// 使用示例
monitorCommand('ls -la; sleep 2; echo "完成"');

Go 脚本监控

package main
import (
    "bufio"
    "fmt"
    "log"
    "os"
    "strings"
    "time"
)
// 监控文件
func monitorFile(filename string) {
    file, err := os.Open(filename)
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()
    // 跳到文件末尾
    file.Seek(0, 2)
    reader := bufio.NewReader(file)
    for {
        line, err := reader.ReadString('\n')
        if err == nil {
            line = strings.TrimSpace(line)
            fmt.Printf("[%s] %s\n", time.Now().Format("15:04:05"), line)
            // 检测关键词
            if strings.Contains(line, "ERROR") {
                fmt.Println("⚠️ ERROR detected!")
            }
        }
        time.Sleep(100 * time.Millisecond)
    }
}
func main() {
    // 监控日志文件
    monitorFile("/var/log/application.log")
}

带过滤和告警的完整脚本

#!/usr/bin/env python3
"""
功能强大的日志监控脚本
支持：文件监控、命令监控、关键词过滤、告警通知
"""
import argparse
import os
import re
import subprocess
import sys
import time
from datetime import datetime
class AdvancedLogMonitor:
    def __init__(self, source, filters=None, alerts=None):
        self.source = source
        self.filters = filters or ['ERROR', 'CRITICAL']
        self.alerts = alerts or []
    def tail_file(self, filename):
        """tail -f 实现"""
        with open(filename, 'r') as f:
            f.seek(0, 2)  # 到末尾
            while True:
                line = f.readline()
                if not line:
                    time.sleep(0.1)
                    continue
                yield line.strip()
    def filter_line(self, line):
        """过滤日志行"""
        for pattern in self.filters:
            if re.search(pattern, line, re.IGNORECASE):
                return True
        return False
    def send_alert(self, message):
        """发送告警"""
        for alert_method in self.alerts:
            # 可以扩展为邮件、Slack、Webhook等
            print(f"🚨 告警: {message}")
    def monitor_file(self, filepath):
        """监控文件"""
        print(f"开始监控文件: {filepath}")
        for line in self.tail_file(filepath):
            if self.filter_line(line):
                timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                self.send_alert(f"[{timestamp}] {line}")
    def monitor_process(self, command):
        """监控进程输出"""
        print(f"监控命令: {command}")
        process = subprocess.Popen(
            command.split(),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT
        )
        for line in iter(process.stdout.readline, b''):
            text = line.decode('utf-8').strip()
            if text and self.filter_line(text):
                self.send_alert(text)
# 命令行使用
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='日志监控工具')
    parser.add_argument('-f', '--file', help='监控的文件路径')
    parser.add_argument('-c', '--command', help='监控的命令')
    parser.add_argument('-k', '--keywords', help='关键词列表，用逗号分隔')
    args = parser.parse_args()
    monitor = AdvancedLogMonitor(
        source=args.file or args.command,
        filters=args.keywords.split(',') if args.keywords else None
    )
    if args.file:
        monitor.monitor_file(args.file)
    elif args.command:
        monitor.monitor_process(args.command)
    else:
        parser.print_help()

Bash 一行命令版本

# 监控文件并grep特定内容
tail -f app.log | grep --line-buffered "ERROR"
# 监控命令输出
while true; do command; echo "---$(date)---"; sleep 1; done
# 简单文件监控
watch -n 1 'cat /var/log/syslog | tail -20'
# 带时间戳的监控
tail -f app.log | while read line; do echo "$(date '+%H:%M:%S') $line"; done

使用建议

选择合适的方法：
- 文件监控：适合日志文件
- 命令监控：适合临时命令输出
- 进程监控：适合长时间运行的程序
性能考虑：
- 避免频繁轮询文件
- 使用增量读取而非全量读取
- 适当使用缓冲
告警机制：
- 可以接入邮件通知
- 支持 Webhook（Slack、企业微信等）
- 记录告警历史
健壮性：
- 添加错误处理
- 支持文件轮转
- 断线重连

这些脚本可以根据你的具体需求调整,比如添加邮件发送、数据库记录、Webhook通知等功能。