MetaCore-startup/MetaCore/test_open.py

import subprocess
import os
import sys
from pathlib import Path


def run_cmd_command(command, cwd=None, shell=True, timeout=30):
    """
    执行CMD命令的通用方法

    Args:
        command (str or list): 要执行的命令
        cwd (str, optional): 工作目录
        shell (bool): 是否使用shell执行命令
        timeout (int): 超时时间（秒）

    Returns:
        tuple: (return_code, stdout, stderr)
    """
    try:
        # 执行命令
        result = subprocess.run(
            command,
            cwd=cwd,
            shell=shell,
            capture_output=True,
            text=True,
            timeout=timeout,
            encoding='utf-8'
        )

        return result.returncode, result.stdout, result.stderr

    except subprocess.TimeoutExpired:
        return -1, "", "命令执行超时"
    except Exception as e:
        return -2, "", f"执行命令时出错: {str(e)}"


def run_cmd_with_venv(venv_path, command, cwd=None, timeout=30):
    """
    在指定虚拟环境中执行CMD命令

    Args:
        venv_path (str): 虚拟环境路径
        command (str): 要执行的命令
        cwd (str, optional): 工作目录
        timeout (int): 超时时间（秒）

    Returns:
        tuple: (return_code, stdout, stderr)
    """
    try:
        # 激活虚拟环境并执行命令
        if os.name == 'nt':  # Windows
            activate_script = Path(venv_path) / 'Scripts' / 'activate.bat'
            full_command = f'"{activate_script}" && {command}'
        else:  # Unix-like systems
            activate_script = Path(venv_path) / 'bin' / 'activate'
            full_command = f'source "{activate_script}" && {command}'

        return run_cmd_command(full_command, cwd=cwd, timeout=timeout)

    except Exception as e:
        return -2, "", f"在虚拟环境中执行命令时出错: {str(e)}"


def run_python_in_venv(venv_path, script_path, args=None, cwd=None, timeout=30):
    """
    在指定虚拟环境中运行Python脚本

    Args:
        venv_path (str): 虚拟环境路径
        script_path (str): Python脚本路径
        args (list, optional): 脚本参数
        cwd (str, optional): 工作目录
        timeout (int): 超时时间（秒）

    Returns:
        tuple: (return_code, stdout, stderr)
    """
    try:
        # 获取虚拟环境中的Python解释器
        if os.name == 'nt':  # Windows
            python_executable = Path(venv_path) / 'Scripts' / 'python.exe'
        else:  # Unix-like systems
            python_executable = Path(venv_path) / 'bin' / 'python'

        # 构建命令
        cmd = [str(python_executable), str(script_path)]
        if args:
            cmd.extend(args)

        # 设置环境变量
        env = os.environ.copy()
        env['VIRTUAL_ENV'] = str(venv_path)
        if os.name == 'nt':  # Windows
            env['PATH'] = str(Path(venv_path) / 'Scripts') + os.pathsep + env['PATH']
        else:  # Unix-like
            env['PATH'] = str(Path(venv_path) / 'bin') + os.pathsep + env['PATH']

        # 执行命令
        result = subprocess.run(
            cmd,
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=timeout,
            encoding='utf-8',
            env=env
        )

        return result.returncode, result.stdout, result.stderr

    except subprocess.TimeoutExpired:
        return -1, "", "命令执行超时"
    except Exception as e:
        return -2, "", f"在虚拟环境中运行Python脚本时出错: {str(e)}"


def execute_commands_sequentially(commands, cwd=None, timeout=30):
    """
    顺序执行多个命令

    Args:
        commands (list): 命令列表
        cwd (str, optional): 工作目录
        timeout (int): 超时时间（秒）

    Returns:
        list: 每个命令的执行结果列表
    """
    results = []

    for i, command in enumerate(commands):
        print(f"执行命令 {i + 1}/{len(commands)}: {command}")
        return_code, stdout, stderr = run_cmd_command(command, cwd=cwd, timeout=timeout)
        results.append({
            'command': command,
            'return_code': return_code,
            'stdout': stdout,
            'stderr': stderr
        })

        # 如果命令执行失败，停止执行后续命令
        if return_code != 0:
            print(f"命令执行失败，停止执行后续命令: {command}")
            break

    return results


def run_text():
    # 示例1: 执行简单命令
    print("=== 示例1: 执行简单命令 ===")
    return_code, stdout, stderr = run_cmd_command("dir" if os.name == 'nt' else "ls -la")
    print(f"返回码: {return_code}")
    print(f"输出: {stdout}")
    if stderr:
        print(f"错误: {stderr}")

    # 示例2: 在虚拟环境中执行命令
    print("\n=== 示例2: 在虚拟环境中执行命令 ===")
    # return_code, stdout, stderr = run_cmd_with_venv(
    #     r"C:\path\to\your\venv",
    #     "pip list"
    # )
    # print(f"返回码: {return_code}")
    # print(f"输出: {stdout}")

    # 示例3: 在虚拟环境中运行Python脚本
    print("\n=== 示例3: 在虚拟环境中运行Python脚本 ===")
    # return_code, stdout, stderr = run_python_in_venv(
    #     r"C:\path\to\your\venv",
    #     r"C:\path\to\your\script.py",
    #     args=["arg1", "arg2"]
    # )
    # print(f"返回码: {return_code}")
    # print(f"输出: {stdout}")

    # 示例4: 顺序执行多个命令
    print("\n=== 示例4: 顺序执行多个命令 ===")
    commands = [
        "echo Hello",
        "echo World",
        "python --version"
    ]
    results = execute_commands_sequentially(commands)
    for result in results:
        print(f"命令: {result['command']}")
        print(f"返回码: {result['return_code']}")
        print(f"输出: {result['stdout']}")
        if result['stderr']:
            print(f"错误: {result['stderr']}")
        print("-" * 40)


# 使用示例
if __name__ == "__main__":
    venv_path = r"d:\PythonProject\CH_EG\EG\.venv"
    script_path = r"d:\PythonProject\CH_EG\EG\Start_Run.py"

    print(f"虚拟环境路径: {venv_path}")
    print(f"脚本路径: {script_path}")

    # 检查路径是否存在
    if not os.path.exists(venv_path):
        print(f"错误: 虚拟环境路径不存在: {venv_path}")
        sys.exit(1)

    if not os.path.exists(script_path):
        print(f"错误: 脚本文件不存在: {script_path}")
        sys.exit(1)

    # 在虚拟环境中运行Python脚本
    return_code, stdout, stderr = run_python_in_venv(venv_path, script_path)

    print(f"返回码: {return_code}")
    print(f"标准输出: {stdout}")
    print(f"错误输出: {stderr}")

    if return_code == 0:
        print("脚本执行成功!")
    else:
        print("脚本执行失败!")