Python资料 - xingzhihe/hello-world GitHub Wiki

[ssh scp]

yum -y install gcc
yum -y install python-devel

wget https://bootstrap.pypa.io/get-pip.py
python get-pip.py
pip -V
pip install ssh

# coding=utf-8

import os
import sys
import subprocess
import ssh
import logging

ch_host = "10.1.23.XX"
ch_user = "XXX"
ch_pwd = "XXX"

dn_host = "10.2.43.XX"
dn_user = "XXX"
dn_pwd = "XXX"

local_file = "/home/XXXX/clickhouse/gjzs/gjzs1203.csv"
remote_file = "/home/XXX/gjzs1203.csv"

logger = logging.getLogger("mylog")
logger.setLevel(level=logging.DEBUG)

handler = logging.FileHandler("log.txt")
handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)

logger.addHandler(handler)


def subprocess_popen(statement):
    p = subprocess.Popen(statement, shell=True, stdout=subprocess.PIPE)
    while p.poll() is None:
        if p.wait() is not 0:
            return [False,"请检查设备连接状态"]
        else:
            re = p.stdout.readlines()
            result = []
            for i in range(len(re)):
                res = re[i].decode('utf-8').strip('\r\n')
                result.append(res)
            return [True,result]

def exec_cmd(cmd):
    flag,msg = subprocess_popen(cmd)
    if flag:
        logger.info("命令[" + cmd + "]执行成功，返回结果如下:")
        print("命令[" + cmd + "]执行成功，返回结果如下:")
        for i in range(len(msg)):
            print(msg[i])
            logger.info(msg[i])
    else:
        print("命令[" + cmd + "]执行失败:" + msg)
        logger.info("命令[" + cmd + "]执行失败:" + msg)

def get_remote_client(host, port, user, pwd):
    client = ssh.SSHClient()
    client.set_missing_host_key_policy(ssh.AutoAddPolicy())
    client.connect(host, port, username=user, password=pwd)
    return client

def remote_exec_cmd(host, user, pwd, cmd):
    client = get_remote_client(host, 22, user, pwd)
    print(cmd)
    logger.info(cmd)
    stdin, stdout, stderr = client.exec_command(cmd)
    r = stderr.read()
    o = stdout.read()
    client.close()
    if(len(r)== 0):
        return [True, o]
    else:
        return [False, r]

def scp_upload(host, user, pwd, localFile, remoteFile):
    client = get_remote_client(host, 22, user, pwd)
    sftp = client.open_sftp()
    sftp.put(localFile, remoteFile)
    sftp.close()
    client.close()

def scp_download(host, user, pwd, localFile, remoteFile):
    client = get_remote_client(host, 22, user, pwd)
    sftp = client.open_sftp()
    sftp.get(remoteFile, localFile)
    sftp.close()
    client.close()

def export_csv():
    cmd = 'clickhouse-client -h ' + ch_host + ' -u ' + ch_user + ' --password ' + ch_pwd + ' --query="SELECT product, un, uuid, userid, orgid, username, orgname, nsrsbh, act, actname, time, day from foresee_user_behavior.gjzs_behavior"  > ' + local_file
    exec_cmd(cmd)

def scp():
    try:
        scp_upload(dn_host, dn_user, dn_pwd, local_file, remote_file)
        print("文件[" + local_file + "]上传成功")
        logger.info("文件[" + local_file + "]上传成功")
    except Exception as e:
        print("文件[" + local_file + "]上传失败：" + e.args)
        print(repr(e))
        logger.error("文件[" + local_file + "]上传失败：" + e.args)
        logger.error(e)

def load2hive():
    cmd = '/usr/bin/hive -e "load data local inpath \'' + remote_file + '\' overwrite into table yypt.gjzs_behavior;"'
    flag, msg = remote_exec_cmd(dn_host, dn_user, dn_pwd, cmd)
    print("[" + str(flag) + "]" + msg)
    logger.info("[" + str(flag) + "]" + msg)


if __name__ == '__main__':
    if len(sys.argv) == 2 :
        cmd = sys.argv[1]
        #exec_cmd(cmd)
        #flag, msg = remote_exec_cmd(dn_host, dn_user, dn_pwd, cmd)
        #print("[" + str(flag) + "]" + msg)
        export_csv()
        scp()
        load2hive()
    else:
        print("缺少参数，必须传递名cmd参数")
        logger.warn("缺少参数，必须传递名cmd参数")