import paramiko

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('host', username='user', password='pass')
stdin, stdout, stderr = ssh.exec_command('ls -l')
print(stdout.read().decode())
ssh.close()

command = "/path/to/venv/bin/python /path/to/your_script.py"
stdin, stdout, stderr = ssh.exec_command(command)

from fabric import Connection

c = Connection('user@host', connect_kwargs={"password": "pass"})
result = c.run('df -h', hide=True)
print(result.stdout)

def run_in_venv(c):
    return c.run("/path/to/venv/bin/python /path/to/script.py", hide=True)

result = run_in_venv(c)
print(result.stdout)

ssh.connect('host', username='user', key_filename='/path/to/key')

command_template = "echo {var} > /tmp/output"
formatted_cmd = command_template.format(var="value")
stdin, stdout, stderr = ssh.exec_command(formatted_cmd)

sftp = ssh.open_sftp()
sftp.put('local.txt', '/remote/path.txt')
sftp.close()

import socket

socket.setdefaulttimeout(10)  # 全局超時(shí)
# 或在exec_command中設(shè)置channel超時(shí)

import logging
logging.basicConfig(filename='ssh.log', level=logging.INFO)
logging.info(f"Executed: {command} on {host}")

from ansible_runner import run
r = run(private_data_dir='/path/to/playbook', playbook='site.yml')
print(r.stats)

import asyncssh

async def run_client():
    async with asyncssh.connect('host', username='user') as conn:
        result = await conn.run('uptime')
        print(result.stdout)

asyncio.get_event_loop().run_until_complete(run_client())

import paramiko

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('host', username='user', password='pass')

# 激活虛擬環(huán)境并執(zhí)行命令（Linux示例）
command = "source /path/to/venv/bin/activate && python your_script.py"
stdin, stdout, stderr = ssh.exec_command(command)

print("STDOUT:", stdout.read().decode())
print("STDERR:", stderr.read().decode())
ssh.close()

command = "/path/to/venv/bin/python /path/to/your_script.py"
stdin, stdout, stderr = ssh.exec_command(command)

from fabric import Connection

c = Connection('user@host', connect_kwargs={"password": "pass"})

# 定義任務(wù)：激活虛擬環(huán)境并執(zhí)行命令
def run_in_venv(c):
    return c.run("/path/to/venv/bin/python /path/to/script.py", hide=True)

result = run_in_venv(c)
print(result.stdout)

import os

venv_path = os.getenv("REMOTE_VENV_PATH", "/default/path/to/venv")
command = f"{venv_path}/bin/python /path/to/script.py"

import paramiko
from concurrent.futures import ThreadPoolExecutor

def execute_in_venv(host, venv_path, script_path):
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(host, username='user', key_filename='/path/to/key')
        
        # 激活虛擬環(huán)境并執(zhí)行（推薦直接調(diào)用Python解釋器）
        command = f"{venv_path}/bin/python {script_path} --arg1 value1"
        stdin, stdout, stderr = ssh.exec_command(command, timeout=30)
        
        print(f"[{host}] STDOUT:\n{stdout.read().decode()}")
        print(f"[{host}] STDERR:\n{stderr.read().decode()}")
    finally:
        ssh.close()

# 批量執(zhí)行
hosts = ["host1", "host2"]
venv_path = "/opt/venvs/myenv"
script_path = "/home/user/scripts/process_data.py"

with ThreadPoolExecutor(2) as pool:
    pool.map(
        lambda h: execute_in_venv(h, venv_path, script_path),
        hosts
    )