脚本怎样统一调用各类脚本

本文目录导读：

1. 主调度脚本方案
2. Shell 主调度方案
3. Node.js 统一调度器
4. 配置文件方案
5. 使用建议

我来详细讲解如何统一调用各类脚本,提供几种实用方案：

主调度脚本方案

Python 主调度器

#!/usr/bin/env python3
# main_dispatcher.py - 统一调度各类脚本
import subprocess
import os
import sys
import json
from pathlib import Path
class ScriptDispatcher:
    def __init__(self, scripts_dir="./scripts"):
        self.scripts_dir = Path(scripts_dir)
        self.script_config = self.load_config()
    def load_config(self):
        """加载脚本配置"""
        config_file = self.scripts_dir / "scripts_config.json"
        if config_file.exists():
            with open(config_file) as f:
                return json.load(f)
        return {}
    def run_script(self, script_path, script_type="auto", **kwargs):
        """统一运行脚本"""
        script_path = Path(script_path)
        # 自动检测脚本类型
        if script_type == "auto":
            script_type = self.detect_script_type(script_path)
        # 根据类型执行
        if script_type == "python":
            return self.run_python(script_path, **kwargs)
        elif script_type == "shell":
            return self.run_shell(script_path, **kwargs)
        elif script_type == "node":
            return self.run_node(script_path, **kwargs)
        elif script_type == "perl":
            return self.run_perl(script_path, **kwargs)
        else:
            raise ValueError(f"不支持的脚本类型: {script_type}")
    def detect_script_type(self, script_path):
        """自动检测脚本类型"""
        extension_map = {
            '.py': 'python',
            '.sh': 'shell', 
            '.bash': 'shell',
            '.js': 'node',
            '.pl': 'perl',
            '.rb': 'ruby'
        }
        ext = script_path.suffix.lower()
        if ext in extension_map:
            return extension_map[ext]
        # 检查 shebang
        with open(script_path) as f:
            first_line = f.readline().strip()
            if 'python' in first_line:
                return 'python'
            elif 'bash' in first_line or 'sh' in first_line:
                return 'shell'
            elif 'node' in first_line:
                return 'node'
        return 'shell'  # 默认
    def run_python(self, script_path, **kwargs):
        """运行 Python 脚本"""
        cmd = [sys.executable, str(script_path)]
        return self.execute_command(cmd, **kwargs)
    def run_shell(self, script_path, **kwargs):
        """运行 Shell 脚本"""
        cmd = ['bash', str(script_path)]
        return self.execute_command(cmd, **kwargs)
    def run_node(self, script_path, **kwargs):
        """运行 Node.js 脚本"""
        cmd = ['node', str(script_path)]
        return self.execute_command(cmd, **kwargs)
    def run_perl(self, script_path, **kwargs):
        """运行 Perl 脚本"""
        cmd = ['perl', str(script_path)]
        return self.execute_command(cmd, **kwargs)
    def execute_command(self, cmd, args=None, env=None, timeout=None):
        """执行命令"""
        if args:
            cmd.extend(args)
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                env={**os.environ, **(env or {})}
            )
            return {
                'success': result.returncode == 0,
                'stdout': result.stdout,
                'stderr': result.stderr,
                'return_code': result.returncode
            }
        except subprocess.TimeoutExpired:
            return {
                'success': False,
                'error': '脚本执行超时',
                'stdout': '',
                'stderr': 'Timeout'
            }
    def run_multiple(self, scripts, parallel=False):
        """运行多个脚本"""
        results = {}
        if parallel:
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as executor:
                futures = {}
                for name, script in scripts.items():
                    future = executor.submit(self.run_script, script['path'])
                    futures[future] = name
                for future in concurrent.futures.as_completed(futures):
                    name = futures[future]
                    results[name] = future.result()
        else:
            for name, script in scripts.items():
                results[name] = self.run_script(script['path'])
        return results
# 使用示例
if __name__ == "__main__":
    dispatcher = ScriptDispatcher()
    # 单脚本执行
    result = dispatcher.run_script("./scripts/data_processor.py", args=["--input", "data.csv"])
    print(result['stdout'])
    # 多脚本顺序执行
    scripts = {
        'clean': {'path': './scripts/clean_data.sh'},
        'process': {'path': './scripts/process.py'},
        'report': {'path': './scripts/generate_report.js'}
    }
    results = dispatcher.run_multiple(scripts)
    # 检查执行结果
    for name, result in results.items():
        if result['success']:
            print(f"{name}: 成功")
        else:
            print(f"{name}: 失败 - {result['stderr']}")

Shell 主调度方案

#!/bin/bash
# unified_runner.sh - Shell 统一调度脚本
# 配置
SCRIPTS_DIR="./scripts"
LOG_DIR="./logs"
CURRENT_DIR=$(pwd)
# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 日志函数
log_info() {
    echo -e "${GREEN}[INFO]${NC} $(date '+%Y-%m-%d %H:%M:%S') - $1"
    echo "[INFO] $(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_DIR/runner.log"
}
log_warn() {
    echo -e "${YELLOW}[WARN]${NC} $(date '+%Y-%m-%d %H:%M:%S') - $1"
    echo "[WARN] $(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_DIR/runner.log"
}
log_error() {
    echo -e "${RED}[ERROR]${NC} $(date '+%Y-%m-%d %H:%M:%S') - $1"
    echo "[ERROR] $(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_DIR/runner.log"
}
# 执行 Python 脚本
run_python() {
    local script=$1
    shift
    log_info "执行 Python 脚本: $script"
    python3 "$SCRIPTS_DIR/$script" "$@"
    return $?
}
# 执行 Shell 脚本
run_shell() {
    local script=$1
    shift
    log_info "执行 Shell 脚本: $script"
    bash "$SCRIPTS_DIR/$script" "$@"
    return $?
}
# 执行 Node.js 脚本
run_node() {
    local script=$1
    shift
    log_info "执行 Node 脚本: $script"
    node "$SCRIPTS_DIR/$script" "$@"
    return $?
}
# 自动检测并执行
run_script() {
    local script=$1
    shift
    if [ ! -f "$SCRIPTS_DIR/$script" ]; then
        log_error "脚本不存在: $script"
        return 1
    fi
    # 根据扩展名判断
    case "${script##*.}" in
        py)
            run_python "$script" "$@"
            ;;
        sh|bash)
            run_shell "$script" "$@"
            ;;
        js)
            run_node "$script" "$@"
            ;;
        *)
            log_warn "未知类型，尝试 shebang 检测"
            run_by_shebang "$script" "$@"
            ;;
    esac
    return $?
}
# 根据 shebang 执行
run_by_shebang() {
    local script=$1
    shift
    local shebang=$(head -1 "$SCRIPTS_DIR/$script")
    case "$shebang" in
        *python*)
            run_python "$script" "$@"
            ;;
        *bash*|*sh*)
            run_shell "$script" "$@"
            ;;
        *node*)
            run_node "$script" "$@"
            ;;
        *)
            log_error "无法识别脚本类型: $script"
            return 1
            ;;
    esac
}
# 批量执行脚本
run_scripts_sequential() {
    log_info "开始顺序执行脚本..."
    for script in "$@"; do
        if ! run_script "$script"; then
            log_error "脚本执行失败: $script"
            exit 1
        fi
    done
    log_info "所有脚本执行完成"
}
# 并行执行脚本
run_scripts_parallel() {
    log_info "开始并行执行脚本..."
    local pids=()
    local scripts=("$@")
    for script in "${scripts[@]}"; do
        run_script "$script" &
        pids+=($!)
    done
    # 等待所有完成
    for pid in "${pids[@]}"; do
        wait $pid
    done
    log_info "所有脚本并行执行完成"
}
# 主函数
main() {
    # 创建日志目录
    mkdir -p "$LOG_DIR"
    # 解析参数
    local mode="sequential"
    local scripts=()
    while [[ $# -gt 0 ]]; do
        case $1 in
            -p|--parallel)
                mode="parallel"
                shift
                ;;
            -s|--sequential)
                mode="sequential"
                shift
                ;;
            -h|--help)
                echo "使用方法: $0 [选项] 脚本1 脚本2 ..."
                echo "选项:"
                echo "  -p, --parallel    并行执行"
                echo "  -s, --sequential  顺序执行 (默认)"
                echo "  -h, --help        显示帮助"
                exit 0
                ;;
            -*)
                log_error "未知选项: $1"
                exit 1
                ;;
            *)
                scripts+=("$1")
                shift
                ;;
        esac
    done
    if [ ${#scripts[@]} -eq 0 ]; then
        log_error "请指定要执行的脚本"
        exit 1
    fi
    # 执行脚本
    if [ "$mode" = "parallel" ]; then
        run_scripts_parallel "${scripts[@]}"
    else
        run_scripts_sequential "${scripts[@]}"
    fi
}
# 运行主函数
main "$@"

Node.js 统一调度器

// unified_runner.js
const { execSync, exec } = require('child_process');
const path = require('path');
const fs = require('fs');
class ScriptRunner {
    constructor(options = {}) {
        this.scriptsDir = options.scriptsDir || './scripts';
        this.logDir = options.logDir || './logs';
        this.timeout = options.timeout || 300000; // 5 minutes
    }
    // 运行脚本
    runScript(scriptName, args = []) {
        const scriptPath = path.join(this.scriptsDir, scriptName);
        if (!fs.existsSync(scriptPath)) {
            throw new Error(`Script not found: ${scriptPath}`);
        }
        const ext = path.extname(scriptName).toLowerCase();
        switch (ext) {
            case '.py':
                return this.runPython(scriptPath, args);
            case '.sh':
            case '.bash':
                return this.runShell(scriptPath, args);
            case '.js':
                return this.runNode(scriptPath, args);
            case '.rb':
                return this.runRuby(scriptPath, args);
            case '.pl':
                return this.runPerl(scriptPath, args);
            default:
                return this.runByShebang(scriptPath, args);
        }
    }
    runPython(scriptPath, args) {
        return this.execute('python3', [scriptPath, ...args]);
    }
    runShell(scriptPath, args) {
        return this.execute('bash', [scriptPath, ...args]);
    }
    runNode(scriptPath, args) {
        return this.execute('node', [scriptPath, ...args]);
    }
    runRuby(scriptPath, args) {
        return this.execute('ruby', [scriptPath, ...args]);
    }
    runPerl(scriptPath, args) {
        return this.execute('perl', [scriptPath, ...args]);
    }
    runByShebang(scriptPath, args) {
        const content = fs.readFileSync(scriptPath, 'utf8');
        const firstLine = content.split('\n')[0];
        if (firstLine.includes('python')) {
            return this.runPython(scriptPath, args);
        } else if (firstLine.includes('bash') || firstLine.includes('sh')) {
            return this.runShell(scriptPath, args);
        } else if (firstLine.includes('node')) {
            return this.runNode(scriptPath, args);
        } else {
            throw new Error(`Cannot determine script type: ${scriptPath}`);
        }
    }
    execute(command, args) {
        return new Promise((resolve, reject) => {
            const cmd = `${command} ${args.join(' ')}`;
            exec(cmd, {
                timeout: this.timeout,
                cwd: this.scriptsDir
            }, (error, stdout, stderr) => {
                if (error) {
                    reject({ error, stdout, stderr });
                } else {
                    resolve({ stdout, stderr });
                }
            });
        });
    }
    // 批量运行
    async runMultiple(scripts, options = {}) {
        const results = {};
        const parallel = options.parallel || false;
        if (parallel) {
            const promises = scripts.map(async (script) => {
                try {
                    const result = await this.runScript(script.name, script.args);
                    return { name: script.name, success: true, ...result };
                } catch (error) {
                    return { name: script.name, success: false, error };
                }
            });
            const results_arr = await Promise.all(promises);
            results_arr.forEach(r => {
                results[r.name] = r;
            });
        } else {
            for (const script of scripts) {
                try {
                    const result = await this.runScript(script.name, script.args);
                    results[script.name] = { success: true, ...result };
                } catch (error) {
                    results[script.name] = { success: false, error };
                }
            }
        }
        return results;
    }
}
// 使用示例
async function main() {
    const runner = new ScriptRunner();
    // 单脚本运行
    try {
        await runner.runScript('data_processor.py', ['--input', 'data.csv']);
        console.log('Script completed successfully');
    } catch (error) {
        console.error('Script failed:', error);
    }
    // 多脚本顺序运行
    const scripts = [
        { name: 'clean_data.sh', args: [] },
        { name: 'process.py', args: ['--mode', 'fast'] },
        { name: 'generate_report.js', args: ['--format', 'html'] }
    ];
    const results = await runner.runMultiple(scripts, { parallel: false });
    // 检查结果
    for (const [name, result] of Object.entries(results)) {
        if (result.success) {
            console.log(`${name}: Success`);
        } else {
            console.error(`${name}: Failed - ${result.error}`);
        }
    }
}
main();

配置文件方案

// scripts_config.json
{
    "scripts": {
        "data_clean": {
            "path": "clean_data.sh",
            "type": "shell",
            "description": "数据清洗",
            "dependencies": [],
            "timeout": 60
        },
        "data_process": {
            "path": "process.py",
            "type": "python",
            "description": "数据处理",
            "dependencies": ["data_clean"],
            "timeout": 300
        },
        "report_generate": {
            "path": "generate_report.js",
            "type": "node",
            "description": "生成报告",
            "dependencies": ["data_process"],
            "timeout": 120
        }
    },
    "workflows": {
        "full_pipeline": {
            "description": "完整数据处理流程",
            "steps": [
                "data_clean",
                "data_process", 
                "report_generate"
            ],
            "parallel_steps": {
                "group1": ["data_clean"],
                "group2": ["data_process", "report_generate"]
            }
        }
    }
}

使用建议

选择方案

• Python方案: 功能强大，适合复杂流程
• Shell方案: 轻量级，系统自带
• Node.js方案: 适合前端/全栈团队
• 配置文件方案: 灵活配置，便于管理

最佳实践

1. 统一错误处理: 所有脚本返回标准格式
2. 日志记录: 集中管理执行日志
3. 环境隔离: 确保脚本独立运行
4. 超时控制: 防止脚本卡死
5. 依赖管理: 自动处理脚本间的依赖

你更倾向于使用哪种语言方案？我可以帮你进一步优化具体的实现。