如何编写插件状态检测脚本

wen 实用脚本 2026-06-27 225

本文目录导读：

如何编写插件状态检测脚本

Nagios/Icinga 插件检测脚本
WordPress 插件状态检测
Python - 通用插件检测脚本
监控系统整合脚本 (Prometheus格式)
Shell 脚本 - 系统插件检测
使用建议

我来为您介绍几种编写插件状态检测脚本的方法,针对不同场景和系统：

Nagios/Icinga 插件检测脚本

#!/bin/bash
# check_plugin_status.sh - Nagios兼容的插件状态检测
# 定义插件目录
PLUGIN_DIR="/usr/lib/nagios/plugins"
TIMEOUT=30
# 检查函数
check_plugin() {
    local plugin=$1
    local args=$2
    if [ ! -f "$PLUGIN_DIR/$plugin" ]; then
        echo "CRITICAL - Plugin $plugin not found"
        exit 2
    fi
    if [ ! -x "$PLUGIN_DIR/$plugin" ]; then
        echo "CRITICAL - Plugin $plugin is not executable"
        exit 2
    fi
    # 执行插件并检测超时
    timeout $TIMEOUT $PLUGIN_DIR/$plugin $args
    local exit_code=$?
    case $exit_code in
        0) echo "OK - $plugin ran successfully" ;;
        1) echo "WARNING - $plugin returned warning" ;;
        2) echo "CRITICAL - $plugin returned critical" ;;
        3) echo "UNKNOWN - $plugin returned unknown" ;;
        *) echo "CRITICAL - $plugin timed out or errored" ;;
    esac
    return $exit_code
}
# 使用示例
check_plugin "check_ping" "-H 8.8.8.8 -w 100.0,20% -c 200.0,50%"

WordPress 插件状态检测

<?php
// wp_plugin_status.php - WordPress插件状态检测
class PluginStatusChecker {
    private $required_plugins = [
        'akismet/akismet.php',
        'wordfence/wordfence.php',
        'yoast-seo/wp-seo.php'
    ];
    public function check_all_plugins() {
        if (!function_exists('is_plugin_active')) {
            require_once(ABSPATH . 'wp-admin/includes/plugin.php');
        }
        $results = [];
        foreach ($this->required_plugins as $plugin) {
            $status = $this->check_single_plugin($plugin);
            $results[$plugin] = $status;
        }
        return $results;
    }
    private function check_single_plugin($plugin) {
        $result = [
            'name' => $plugin,
            'installed' => false,
            'active' => false,
            'version' => null,
            'status' => 'missing'
        ];
        // 检查插件是否安装
        $plugin_data = get_plugin_data(WP_PLUGIN_DIR . '/' . $plugin);
        if (!empty($plugin_data)) {
            $result['installed'] = true;
            $result['version'] = $plugin_data['Version'];
            // 检查插件是否激活
            if (is_plugin_active($plugin)) {
                $result['active'] = true;
                $result['status'] = 'ok';
            } else {
                $result['status'] = 'inactive';
            }
        }
        return $result;
    }
    public function generate_report() {
        $plugins = $this->check_all_plugins();
        $html = "<table border='1'>";
        $html .= "<tr><th>Plugin</th><th>Status</th><th>Version</th></tr>";
        foreach ($plugins as $plugin => $status) {
            $color = ($status['status'] == 'ok') ? 'green' : 'red';
            $html .= "<tr style='color: $color'>";
            $html .= "<td>" . basename($plugin) . "</td>";
            $html .= "<td>" . $status['status'] . "</td>";
            $html .= "<td>" . ($status['version'] ?? 'N/A') . "</td>";
            $html .= "</tr>";
        }
        $html .= "</table>";
        return $html;
    }
}
// 使用示例
$checker = new PluginStatusChecker();
echo $checker->generate_report();

Python - 通用插件检测脚本

#!/usr/bin/env python3
# plugin_detector.py - 多平台插件检测
import os
import importlib
import pkg_resources
import subprocess
from datetime import datetime
class PluginDetector:
    def __init__(self):
        self.plugins = {}
        self.results = {}
    def check_python_package(self, package_name, min_version=None):
        """检测Python包"""
        try:
            package = pkg_resources.get_distribution(package_name)
            version = package.version
            if min_version:
                from pkg_resources import parse_version
                if parse_version(version) < parse_version(min_version):
                    status = 'outdated'
                else:
                    status = 'ok'
            else:
                status = 'ok'
            return {
                'name': package_name,
                'version': version,
                'status': status,
                'installed': True
            }
        except pkg_resources.DistributionNotFound:
            return {
                'name': package_name,
                'version': None,
                'status': 'missing',
                'installed': False
            }
    def check_browser_extension(self, extension_id, browser='chrome'):
        """检测浏览器扩展"""
        if browser == 'chrome':
            extension_path = os.path.expanduser(
                f"~/.config/google-chrome/Default/Extensions/{extension_id}"
            )
        elif browser == 'firefox':
            extension_path = os.path.expanduser(
                "~/.mozilla/firefox/*.default/extensions/"
            )
        else:
            return {'error': f'Unsupported browser: {browser}'}
        if os.path.exists(extension_path) or os.path.isdir(extension_path):
            return {'installed': True, 'status': 'ok'}
        return {'installed': False, 'status': 'missing'}
    def check_vscode_extension(self, extension_name):
        """检测VS Code扩展"""
        try:
            result = subprocess.run(
                ['code', '--list-extensions'],
                capture_output=True,
                text=True,
                timeout=10
            )
            installed = extension_name in result.stdout
            return {
                'name': extension_name,
                'installed': installed,
                'status': 'ok' if installed else 'missing'
            }
        except subprocess.TimeoutExpired:
            return {'error': 'Timeout checking VSCode extensions'}
        except FileNotFoundError:
            return {'error': 'VSCode command not found'}
    def generate_json_report(self):
        """生成JSON格式报告"""
        import json
        report = {
            'timestamp': datetime.now().isoformat(),
            'plugins': self.results,
            'summary': {
                'total': len(self.results),
                'ok': sum(1 for r in self.results.values() 
                         if r.get('status') == 'ok'),
                'missing': sum(1 for r in self.results.values() 
                             if r.get('status') == 'missing'),
                'errors': sum(1 for r in self.results.values() 
                            if 'error' in r)
            }
        }
        return json.dumps(report, indent=2)
# 使用示例
if __name__ == "__main__":
    detector = PluginDetector()
    # 检查Python包
    detector.results['requests'] = detector.check_python_package('requests', '2.25.0')
    detector.results['flask'] = detector.check_python_package('flask', '2.0.0')
    # 检查浏览器扩展 (需要替换为实际的扩展ID)
    detector.results['adblock'] = detector.check_browser_extension(
        'gighmmpiobklfepjocnamgkkbiglidom', 'chrome'
    )
    # 检查VS Code扩展
    detector.results['python-extension'] = detector.check_vscode_extension(
        'ms-python.python'
    )
    # 输出报告
    print(detector.generate_json_report())

监控系统整合脚本 (Prometheus格式)

#!/usr/bin/env python3
# prometheus_plugin_exporter.py - Prometheus格式输出
import time
import importlib
import sys
class PrometheusPluginExporter:
    def __init__(self):
        self.metrics = []
    def add_metric(self, name, value, labels=None):
        """添加Prometheus指标"""
        labels_str = ""
        if labels:
            label_parts = [f'{k}="{v}"' for k, v in labels.items()]
            labels_str = "{" + ",".join(label_parts) + "}"
        self.metrics.append(f'{name}{labels_str} {value}')
    def check_plugin_health(self, plugin_name, module_name):
        """检查插件健康状态"""
        try:
            module = importlib.import_module(module_name)
            has_init = hasattr(module, '__init__')
            version = getattr(module, '__version__', 'unknown')
            self.add_metric(
                'plugin_status',
                1,
                {'name': plugin_name, 'version': str(version), 'status': 'ok'}
            )
            return True
        except ImportError:
            self.add_metric(
                'plugin_status',
                0,
                {'name': plugin_name, 'version': '0', 'status': 'missing'}
            )
            return False
        except Exception as e:
            self.add_metric(
                'plugin_status',
                0,
                {'name': plugin_name, 'version': '0', 'status': 'error'}
            )
            return False
    def export_metrics(self):
        """导出Prometheus格式指标"""
        print("# HELP plugin_status Plugin installation status")
        print("# TYPE plugin_status gauge")
        print(f"# Timestamp: {int(time.time())}")
        for metric in self.metrics:
            print(metric)
# 使用示例
if __name__ == "__main__":
    exporter = PrometheusPluginExporter()
    plugins_to_check = [
        ('numpy', 'numpy'),
        ('pandas', 'pandas'),
        ('matplotlib', 'matplotlib'),
        ('scikit-learn', 'sklearn'),
        ('tensorflow', 'tensorflow')
    ]
    for name, module in plugins_to_check:
        exporter.check_plugin_health(name, module)
    exporter.export_metrics()

Shell 脚本 - 系统插件检测

#!/bin/bash
# system_plugin_check.sh - 系统级插件检测
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 检测函数
check_plugin() {
    local plugin_name=$1
    local check_command=$2
    echo -n "Checking $plugin_name... "
    if eval "$check_command" &>/dev/null; then
        echo -e "${GREEN}[OK]${NC}"
        return 0
    else
        echo -e "${RED}[FAIL]${NC}"
        return 1
    fi
}
# 检查进程是否运行
check_process_running() {
    local process_name=$1
    pgrep -x "$process_name" > /dev/null 2>&1
}
# 检查端口是否监听
check_port_listening() {
    local port=$1
    ss -tlnp | grep -q ":$port "
}
# 主检测流程
main() {
    echo "=== Plugin Status Check ==="
    echo "Date: $(date)"
    echo "=============================="
    # 检查系统服务插件
    check_plugin "Apache" "systemctl is-active --quiet apache2 || systemctl is-active --quiet httpd"
    check_plugin "MySQL" "systemctl is-active --quiet mysql || systemctl is-active --quiet mariadb"
    check_plugin "PHP-FPM" "systemctl is-active --quiet php*-fpm"
    # 检查进程
    check_plugin "Nginx Process" "check_process_running nginx"
    check_plugin "Redis Process" "check_process_running redis-server"
    # 检查端口
    check_plugin "Port 80" "check_port_listening 80"
    check_plugin "Port 443" "check_port_listening 443"
    check_plugin "Port 3306" "check_port_listening 3306"
    # 检查自定义插件目录
    PLUGIN_DIRS=(
        "/usr/local/plugins"
        "/etc/plugins"
        "/opt/plugins"
    )
    for dir in "${PLUGIN_DIRS[@]}"; do
        if [ -d "$dir" ]; then
            echo -e "${GREEN}[OK]${NC} Plugin directory exists: $dir"
            # 检查目录中的可执行文件
            for file in "$dir"/*; do
                if [ -x "$file" ] && [ -f "$file" ]; then
                    echo "  -> Executable plugin: $(basename $file)"
                fi
            done
        else
            echo -e "${YELLOW}[INFO]${NC} Plugin directory not found: $dir"
        fi
    done
    echo "=============================="
    echo "Check completed."
}
# 运行主函数
main "$@"