需求

项目需要定时移植多个客户服务器的文件到公司服务器上，确保文件定时同步和生成监控日志

机制原理

1.客户和公司服务器同时安装vpn，绕过复杂的网关，linux下使用的OpenVPN

2.服务器定时运行Python移植脚本

3.使用Python的原因，支持多线程和完善ftp类库

代码实现（Python）

Python脚本

# coding=utf-8

import os

import ftplib

import threading

import time

import pymysql

# ftp操作类

class myFtp:

    # 初始化类，连接生产服务器，链接mysql

    def __init__(self, host, port, user, passwd, dbCursor):

        self.ftp = ftplib.FTP()

        self.msgHost = host

        self.dbCursor = dbCursor

        self.user = user

        self.passwd = passwd

        self.filepath = '/'

        try:

            self.ftp.connect(host, port)

            msgInsert(self.user, "info", "ftp连接成功", self.dbCursor)

            print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + self.msgHost + " ftp连接成功")

        except:

            msgInsert(self.user, "error", "ftp连接失败", self.dbCursor)

            print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + self.msgHost + " ftp连接失败")

        # ftp错误级别

        self.ftp.set_debuglevel(0)

        # self.ftp.set_pasv(0)

        try:

            self.ftp.login(user, passwd)

            self.loginSuccess = 1

            msgInsert(self.user, "info", "ftp用户登录成功", self.dbCursor)

            print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + self.msgHost + " ftp用户登录成功")

        except:

            self.loginSuccess = 0

            msgInsert(self.user, "error", "ftp用户登录失败", self.dbCursor)

            print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + self.msgHost + " ftp用户登录失败")

    # 检查ftp是否登录成功

    def isLogin(self):

        # return 1

        return self.loginSuccess

    # 下载单个文件

    def downLoadFile(self, LocalFile, RemoteFile):

        curTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

        # 如果本地有文件删除

        if os.path.exists(LocalFile):

            os.remove(LocalFile)

        # 如果本地写入

        try:

            buffer_size = 1024 * 1024

            file_handler = open(LocalFile, 'wb')

            self.ftp.retrbinary('RETR ' + RemoteFile, file_handler.write, buffer_size)

            file_handler.close()

            msgInsert(self.user, "info", LocalFile + "写入本地成功", self.dbCursor)

            print(curTime + " " + self.msgHost + " " + LocalFile + "写入本地成功")

            self.isFileSuccess = 1

        except:

            msgInsert(self.user, "error", LocalFile + "写入本地失败", self.dbCursor)

            print(curTime + " " + self.msgHost + " " + LocalFile + "写入本地失败")

            self.isFileSuccess = 0

        # 删除ftp文件

        if self.isFileSuccess == 1:

            try:

                self.ftp.delete(RemoteFile)

                msgInsert(self.user, "info", self.filepath + RemoteFile + "删除ftp文件成功", self.dbCursor)

                print(curTime + " " + self.msgHost + " " + self.filepath + RemoteFile + "删除ftp文件成功")

            except:

                msgInsert(self.user, "error", self.filepath + RemoteFile + "删除ftp文件失败", self.dbCursor)

                print(curTime + " " + self.msgHost + " " + self.filepath + RemoteFile + "删除ftp文件失败")

        return True

    # 下载整个目录下的文件

    def downLoadFileTree(self, LocalDir, RemoteDir):

        if not os.path.exists(LocalDir):

            curTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

            try:

                os.makedirs(LocalDir)

                msgInsert(self.user, "info", LocalDir + "本地目录创建成功", self.dbCursor)

            except:

                msgInsert(self.user, "error", LocalDir + "本地目录创建失败", self.dbCursor)

                print(curTime + " " + self.msgHost + " " + LocalDir + "本地目录创建失败")

        try:

            curTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

            self.ftp.cwd(RemoteDir)

            RemoteNames = self.ftp.nlst()

            msgInsert(self.user, "info", self.filepath + "切换目录和读取列表成功", self.dbCursor)

            print(curTime + " " + self.msgHost + " " + self.filepath + "切换目录和读取列表成功")

        except:

            msgInsert(self.user, "error", self.filepath + "切换目录和读取列表失败", self.dbCursor)

            print(curTime + " " + self.msgHost + " " + self.filepath + "切换目录和读取列表失败")

        for file in RemoteNames:

            Local = os.path.join(LocalDir, file)

            # print(file)

            if file.find(".") == -1:

                self.filepath += file + '/'

                if not os.path.exists(Local):

                    try:

                        os.makedirs(Local)

                        msgInsert(self.user, "info", Local + "本地目录创建成功", self.dbCursor)

                        print(curTime + " " + self.msgHost + " " + Local + "本地目录创建成功")

                    except:

                        msgInsert(self.user, "error", Local + "本地目录创建失败", self.dbCursor)

                        print(curTime + " " + self.msgHost + " " + Local + "本地目录创建失败")

                self.downLoadFileTree(Local, file)

                # 删除目录

                self.ftp.rmd(file)

            elif file.find(".") == 0:

                pass

            else:

                self.downLoadFile(Local, file)

        self.ftp.cwd("..")

        # print('start')

        # print(self.filepath)

        self.filepath = self.filepath[0:self.filepath.rstrip("/").rfind("/")+1]

        # print(self.filepath)

        # print('end')

        return

    def close(self):

        self.ftp.quit()

# 移植函数

def getSiteFtpFile(host, port, username, passwd, localDir, dbconn):

    # dbconn = sqlite3.connect('getFile.sqlite3', check_same_thread=False)

    try:

        dbconn = pymysql.connect("127.0.0.1", "root", "root", "ftpwork")

        dbCursor = dbconn.cursor()

        print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + host + " 连接mysql成功")

    except:

        print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + host + " 连接mysql失败")

    ftp = myFtp(host, port, username, passwd, dbCursor)

    isLogin = ftp.isLogin()

    if isLogin == 1:

        try:

            ftp.downLoadFileTree(localDir, '/')

        except:

            msgInsert(username, "error", "ftp.downLoadFileTree方法执行中断", dbCursor)

            print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + " " + host + " ftp.downLoadFileTree方法执行中断")

    ftp.close()

    print(time.strftime("########### " + "%Y-%m-%d %H:%M:%S", time.localtime()) + " " + host + " 执行完成 ###########")

    # dbCursor.close()

    dbconn.commit()

    dbconn.close()

# 自定义线程

class myThread(threading.Thread):

    def __init__(self, threadID, name, host, port, username, passwd, localDir, dbConn):

        threading.Thread.__init__(self)

        self.threadID = threadID

        self.name = name

        self.host = host

        self.port = port

        self.username = username

        self.passwd = passwd

        self.localDir = localDir

        self.dbConn = dbConn

    def run(self):

        print("开始线程：" + self.name)

        getSiteFtpFile(self.host, self.port, self.username, self.passwd, self.localDir, self.dbConn)

        print("退出线程：" + self.name)

# 监控消息写入mysql

def msgInsert(msgHost, msgType, msgContent, dbCur):

    datetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

    dbCur.execute("INSERT INTO ftp_getfile (title,content,w_time,msgtype) \

          VALUES ('" + msgHost + "', '" + msgContent.replace("\\","/") + "','" + datetime + "' ,'" + msgType + "')")

if __name__ == "__main__":

    conn = ''

    # 创建两个线程

    thread1 = myThread(1, "Thread-1", "ftp_ip", 21, 'ftp_user', 'ftp_passwd', 'Z:/', conn)

    # thread2 = myThread(2, "Thread-2", "ftp_ip", 21, 'ftp_user', 'ftp_passwd', "Z:/", conn)

    # thread3 = myThread(3, "Thread-3", "ftp_ip", 21, 'ftp_user', 'ftp_passwd', 'D:/root-python/test/03', conn)

    # thread4 = myThread(4, "Thread-4", "ftp_ip", 82, 'ftp_user', 'ftp_passwd', 'D:/root-python/test/04', conn)

    # thread5 = myThread(5, "Thread-5", "ftp_ip", 21, 'ftp_user', 'ftp_passwd', 'D:/root-python/test/05', conn)

    # thread6 = myThread(6, "Thread-6", "ftp_ip", 82, 'ftp_user', 'ftp_passwd', 'D:/root-python/test/06', conn)

    try:

        # 开启新线程

        thread1.start()

        # thread2.start()

        # thread3.start()

        # thread4.start()

        # thread5.start()

        # thread6.start()

        thread1.join()

        # thread2.join()

        # thread3.join()

        # thread4.join()

        # thread5.join()

        # thread6.join()

    except:

        print("Error: unable to start thread")

    # while 1:

    #     pass