python连接远程服务器有很多实现方法，但是在我使用的时候却要要求做成服务的形式，连接信息用后端返回的形式。实现方式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import paramiko
import time
from apscheduler.schedulers.background import BackgroundScheduler


class ssh_tunnel(object):

    def survival_detection(self):
        interval_time = int(time.time()) - self.start_time
        if interval_time > 10800:  # No operation for more than three hours
            self.close_connect()

    def __init__(self, host_ip: str, port: int, user: str, pwd: str):
        self.host_ip = host_ip
        self.port = port
        self.user = user
        self.pwd = pwd
        self.connect_obj = None
        self.start_time = 9000000000  # The default startup time is very large

        self.sched = BackgroundScheduler()
        self.sched.add_job(self.survival_detection,
                           trigger="interval", hours=3)
        self.sched.start()

    def creat_connect(self) -> dict:
        '''建立连接'''
        if self.connect_obj != None:
            return {"status": 1, "info": ">>>"}
        try:
            trans = paramiko.Transport((self.host_ip, self.port))
            trans.start_client()
            trans.auth_password(self.user, self.pwd)
            channel = trans.open_session(timeout=1200)  # 
            channel.get_pty()  # Acquire terminal
            channel.invoke_shell()  # Activate terminal
            while channel.recv_ready() == True:
                time.sleep(0.01)
                break
            time.sleep(1)
            result = channel.recv(10240)
            create_str = result.decode("utf-8")
            self.connect_obj = channel
            self.start_time = int(time.time())
            return {"status": 1, "info": create_str}
        except:
            return {"status": -1, "info": "Remote connection failed！"}

    def execute_cmd(self, commd_str: str = "") -> dict:
        '''执行命令'''
        if self.connect_obj == None:
            return {"status": -1, "info": "Remote connection failed！"}
        try:
            self.connect_obj.send(commd_str + "\n")  # type:ignore /Execute command
            while self.connect_obj.recv_ready() == True:  # type:ignore /Result acquisition
                time.sleep(0.01)
                break
            time.sleep(0.5)
            result = self.connect_obj.recv(10240)  # type:ignore
            result = result.decode("utf-8")
            _result = result[::-1] # These three lines of code are dealing with some redundant characters. Comments do not affect the overall operation, but only the aesthetics
            _str = result[len(result)-_result.index('\r')-1::]
            result = result.replace(_str*2, _str)
            self.start_time = int(time.time())  # Lifetime of each execution refresh
            return {"status": 1, "info": result.replace(commd_str.strip()+'\r\n', '')}
        except:
            return {"status": -1, "info": "Remote connection failed！"}

    def close_connect(self) -> dict:
        if self.connect_obj == None:
            return {"status": -1, "info": "Remote connection failed！"}
        try:
            self.connect_obj.close()
            return {"status": 1, "info": 'Connection closed!'}
        except:
            return {"status": -1, "info": "Connection closing failed！"}


if __name__ == "__main__":
    test = ssh_tunnel(host_ip='*.*.*.*',
                      port=22, user='root', pwd='admin')
    res = test.creat_connect()
    print(res.get('info', ''), end='')

    commd_str = input("")
    while commd_str != '-1':
        res = test.execute_cmd(commd_str + "\n")
        print(res.get('info', ''), end='')
        commd_str = input("")
    test.close_connect()