工作中，在做一些提效的指令碼開發時，總會連線一些常用的工具，包括ssl、redis、zookeeper、dubbo、jenkins等
下邊我整理了自己的常用的三方庫

ssl連線伺服器，進行檔案互動

import paramiko

#方式一：命令列互動
ssh = paramiko.SSHClient()      #建立連線物件
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect("xx.xx.xx.xx", 22, "xxx", "xxx")      #ip、埠號、使用者、密碼；建立連結
stdin, stdout, stderr = ssh.exec_command("pwd")   #命令列執行命令
result = stdout.read().decode('utf-8')            #返回輸出結果
print(result)
ssh.close()      #關閉連結

#方式二：檔案互動
transport = paramiko.Transport(('ip', 22))      #ip/port
transport.connect(username='xxx', password='xxx') #user/pwd 建立連結
ssh = paramiko.SSHClient()
ssh._transport = transport
sftp = paramiko.SFTPClient.from_transport(transport)      #sftp連線
sftp.put(localpath, remotepath)      #localpath:本地路徑 remotepath：遠端路徑      上傳檔案
sftp.get(remotepath, localpath)      #localpath:本地路徑 remotepath：遠端路徑      下載檔案
stdin, stdout, stderr = ssh.exec_command()
result = stdout.read().decode('utf-8')
print(result)
transport.close()
 

連zk，獲取dubbo服務列表

# -*- coding: utf-8 -*-
__author__ = 'felix'
from kazoo.client import KazooClient
import urllib, sys

host = {
    "evn1": "10.6.64.96:2181",
    "evn2": "172.17.12.253:2181",
    "evn3": "172.17.253.253:2181"
}

def get_dubbo_full(evn, serviceName):            #返回某服務資訊，字典型別{'ip:port': ['serviceName']}
    Host = host[evn]      #獲取不同環境zk地址
    zk = KazooClient(hosts="{}".format(Host))      #建立例項
    zk.start()            #連線
    urls = []
    list = zk.get_children("dubbo")      #獲取子節點

    for i in list:
        if serviceName in i:  # 關鍵字過濾      #檢視某個服務
            try:
                gg = zk.get_children("/dubbo/{}/providers".format(i))
                if gg:
                    for j in gg:
                        url = urllib.parse.unquote(j)
                        if url.startswith('dubbo:'):
                            urls.append(url.split('?')[0].split('dubbo://')[1])
            except Exception as e:
                print(e)
                print(i)
    services = {}
    for i in urls:
        try:
            path, service = i.split('/')
            if not services.get(path):
                services.update({path: []})
            services[path].append(service)
        except Exception as e:
            print(e)
    # print(json.dumps(services, indent=4))
    return services


def get_service(evn, servicename):            #返回某服務資訊，字串型別'ip:port'
    services = get_dubbo_full(evn, servicename)
    for k, v in services.items():
        if servicename in v:
            return k
    return False

def get_service_ip_port(evn, servicename):      #返回某服務所在的ip與埠，陣列型別['ip', 'port']
    ip_port = get_service(evn, servicename)
    if ip_port:
        return str(ip_port).split(":")
    else:
        return "no service"
 

呼叫dubbo服務方法（需要自己學習telnet呼叫dubbo服務）

# -*- coding: utf-8 -*-
__author__ = 'felix'

import socket
import telnetlib
import os

class Dubbo(telnetlib.Telnet):
    prompt = 'dubbo>'
    coding = 'utf-8'
    serv = ''
    def __init__(self, ip, port, service, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
        self.serv = service
        super().__init__(ip, port)
        self.write(b'\n')

    def command(self, flag, str_=""):
        data = self.read_until(flag.encode())
        self.write(str_.encode() + b"\n")
        return data

    def invoke(self, method_name, arg):      
        command_str = "invoke {0}.{1}({2})".format(self.serv, method_name, arg)
        self.command(Dubbo.prompt, command_str)
        data = self.command(Dubbo.prompt, "")
        data = data.decode(Dubbo.coding,errors='ignore').split('\n')[0]
        return data

    def do(self, arg):
        command_str = arg
        self.command(Dubbo.prompt, command_str)
        data = self.command(Dubbo.prompt, command_str)
        data = data.decode(Dubbo.coding,errors='ignore').split('\n')[0]
        # for i in data:
        #     print (i.strip())
        return data


    def ls_l_all(self):
        command_str = "ls -l {0}".format(self.serv)
        self.command(Dubbo.prompt, command_str)
        data = self.command(Dubbo.prompt, "")
        data = data.decode(Dubbo.coding,errors='ignore').split('\n')
        for i in data:
            print(i)

    def ls_l_func(self, func_name):
        command_str = "ls -l {0}".format(self.serv)
        self.command(Dubbo.prompt, command_str)
        data = self.command(Dubbo.prompt, "")
        data = data.decode(Dubbo.coding,errors='ignore').split('\n')
        for i in data:
            if func_name in i:
                print(i)
                break
 

連線redis（redis相關命令語法、具體操作需要自行學習）

#方式一：直連
import redis
conn = redis.Redis(host=host, port=port)
result = conn.get(key) #獲取key資訊

#方式二：叢集
from rediscluster import StrictRedisCluster
startup_nodes = [            #叢集節點
    {'host': 'xxx', 'port': 6379},
    {'host': 'xxx', 'port': 6379},
    {'host': 'xxx', 'port': 6379},
    {'host': 'xxx', 'port': 6379},
    {'host': 'xxx', 'port': 6379},
    {'host': 'xxx', 'port': 6379},
]
rc = StrictRedisCluster(startup_nodes=startup_nodes, decode_responses=True)
result = rc.get(key)