def open_telnet_connection(ip_address, port=3000):
socket.setdefaulttimeout(3)
try:
session = telnetlib.Telnet(ip_address, int(port))
except (ConnectionRefusedError,
socket.timeout):
pass
else:
return session
def telnet_command(session, command, UntilContent=‘\n‘, timeout=10):
_write(session,command)
return read_buff(session,UntilContent, timeout)
def _write(session, command):
if session:
session.write(command.encode() + "\n".encode())
def read_buff(session,UntilContent, timeout=30):
if session:
response = session.read_until((UntilContent).encode(), timeout)
response = response.decode()
return response
def telnet_close(session):
session.close()
def get_alive_topyang(alive_host_ip, port=1):
topyang_alive_ip = []
for host in alive_host_ip:
try:
session = open_telnet_connection(host, 3000)
if session is None: raise ConnectionRefusedError
response = telnet_command(session, "READ:%s" % port).split(":")[2].strip("\r\n")
telnet_close(session)
except (ConnectionRefusedError, # 端口没开
):
print(host,‘: ‘,‘telnet failed !‘)
continue
else:
topyang_alive_ip.append((host,response))
return topyang_alive_ip
python【telnet】使用