使用python ftp和sftp的实例介绍
python
这篇文章详解使用python ftp和sftp的实例介绍
python ftp 上传、下载文件
#获取昨天日期
TODAY = datetime.date.today()
YESTERDAY = TODAY - datetime.timedelta(days=1)CURRENTDAY=YESTERDAY.strftime('%Y%m%d')
---------------------------------------------------------------------------------------
#!/usr/bin/env
python
# -*- coding: cp936 -*-
#导入ftplib扩展库
import ftplib
#创建ftp对象实例
ftp = ftplib.FTP()
#指定IP地址和端口，连接到FTP服务，上面显示的是FTP服务器的Welcome信息
FTPIP= "218.108.***.***"
FTPPORT= 21
USERNAME= "ybmftp"
USERPWD= "ybm***"
ftp.connect(FTPIP,FTPPORT)
#通过账号和密码登录FTP服务器
ftp.login(USERNAME,USERPWD)
#如果参数 pasv 为真，打开被动模式传输 (PASV MODE) ，
#否则，如果参数 pasv 为假则关闭被动传输模式。
#在被动模式打开的情况下，数据的传送由客户机启动，而不是由服务器开始。
#这里要根据不同的服务器配置
ftp.set_pasv(0)
#在FTP连接中切换当前目录
CURRTPATH= "/home1/ftproot/ybmftp/testupg/payment"
ftp.cwd(CURRTPATH)
#为准备下载到本地的文件，创建文件对象
DownLocalFilename="YBM_20110629_9001_CHK"
f = open(DownLocalFilename,'wb')
#从FTP服务器下载文件到前一步创建的文件对象，其中写对象为f.write，1024是缓冲区大小
DownRoteFilename="YBM_20110629_9001_CHK"
ftp.retrbinary('RETR ' + DownRoteFilename , f.write ,1024)
#关闭下载到本地的文件
#提醒：虽然Python可以自动关闭文件，但实践证明，如果想下载完后立即读该文件，最好关闭后重新打开一次
f.close()
#关闭FTP客户端连接
ftp.close()
###上传文件
#! /usr/bin/env python
from ftplib import FTP
import sys, getpass, os.path
host="218.108.***.***"
username="ybmftp"
password="ybm!***"
localfile="/home/gws/xym/script/duizhang.txt"
remotepath="~/testpayment"
f=FTP(host)
f.login(username, password)
f.cwd(remotepath)
fd=open(localfile,'rb')print os.path.basename(localfile)
#否则，如果参数
pasv 为假则关闭被动传输模式。
#在被动模式打开的情况下，数据的传送由客户机启动，而不是由服务器开始。
#这里要根据不同的服务器配置
ftp.set_pasv(0)
f.storbinary('STOR %s ' % os.path.basename(localfile),fd)
fd.close()
f.quit
Python中的ftplib模块
Python中默认安装的ftplib模块定义了FTP类，其中函数有限，可用来实现简单的ftp客户端，用于上传或下载文件
FTP的工作流程及基本操作可参考协议RFC959
ftp登陆连接
from ftplib import FTP #加载ftp模块
ftp=FTP() #设置变量ftp.set_debuglevel(2) #打开调试级别2，显示详细信息
ftp.connect("IP","port") #连接的ftp sever和端口
ftp.login("user","password")#连接的用户名，密码
print ftp.getwelcome() #打印出欢迎信息
ftp.cmd("xxx/xxx") #更改远程目录
bufsize=1024 #设置的缓冲区大小
filename="filename.txt" #需要下载的文件
file_handle=open(filename,"wb").write #以写模式在本地打开文件
ftp.retrbinaly("RETR filename.txt",file_handle,bufsize) #接收服务器上文件并写入本地文件
ftp.set_debuglevel(0) #关闭调试模式ftp.quit #退出ftp
ftp相关命令操作
ftp.cwd(pathname) #设置FTP当前操作的路径
ftp.dir() #显示目录下文件信息
ftp.nlst() #获取目录下的文件
ftp.mkd(pathname) #新建远程目录
ftp.pwd() #返回当前所在位置
ftp.rmd(dirname) #删除远程目录
ftp.delete(filename) #删除远程文件ftp.rename(fromname, toname)#将fromname修改名称为toname。
ftp.storbinaly("STOR filename.txt",file_handel,bufsize) #上传目标文件
ftp.retrbinary("RETR filename.txt",file_handel,bufsize)#下载FTP文件
from ftplib import FTP
ftp = FTP()
timeout = 30
port = 21
ftp.connect('192.168.1.188',port,timeout) # 连接FTP服务器
ftp.login('UserName','888888') # 登录
print ftp.getwelcome() # 获得欢迎信息
ftp.cwd('file/test') # 设置FTP路径 list = ftp.nlst() # 获得目录列表
for name in list:
print(name) # 打印文件名字
path = 'd:/data/' + name # 文件保存路径
f = open(path,'wb') # 打开要保存文件
filename = 'RETR ' + name # 保存FTP文件
ftp.retrbinary(filename,f.write) # 保存FTP上的文件
ftp.delete(name) # 删除FTP文件
ftp.storbinary('STOR '+filename, open(path, 'rb')) # 上传FTP文件
ftp.quit() # 退出FTP服务器
import ftplib
import os
import socket
HOST = 'ftp.mozilla.org'
DIRN = 'pub/mozilla.org/webtools'
FILE = 'bugzilla-3.6.7.tar.gz'
def main():
try:
f = ftplib.FTP(HOST)
except (socket.error, socket.gaierror):
print 'ERROR:cannot reach " %s"' % HOST
return
print '***Connected to host "%s"' % HOST
try:
f.login()
except ftplib.error_perm:
print 'ERROR: cannot login anonymously'
f.quit()
return
print '*** Logged in as "anonymously"'
try:
f.cwd(DIRN)
except ftplib.error_perm:
print 'ERRORL cannot CD to "%s"' % DIRN
f.quit()
return
print '*** Changed to "%s" folder' % DIRN
try:
#传一个回调函数给retrbinary() 它在每接收一个二进制数据时都会被调用
f.retrbinary('RETR %s' % FILE, open(FILE, 'wb').write)
except ftplib.error_perm:
print 'ERROR: cannot read file "%s"' % FILE
os.unlink(FILE)
else:
print '*** Downloaded "%s" to CWD' % FILE
f.quit()
return
if name == 'main':
main()
os.listdir(dirname)：列出dirname下的目录和文件
os.getcwd()：获得当前工作目录
os.curdir:返回当前目录（'.')
os.chdir(dirname):改变工作目录到dirname
os.path.isdir(name):判断name是不是一个目录，name不是目录就返回false
os.path.isfile(name):判断name是不是一个文件，不存在name也返回false
os.path.exists(name):判断是否存在文件或目录name
os.path.getsize(name):获得文件大小，如果name是目录返回0L
os.path.abspath(name):获得绝对路径
os.path.normpath(path):规范path字符串形式
os.path.split(name):分割文件名与目录（事实上，如果你完全使用目录，它也会将最后一个目录作为文件名而分离，同时它不会判断文件或目录是否存在）
os.path.splitext():分离文件名与扩展名
os.path.join(path,name):连接目录与文件名或目录
os.path.basename(path):返回文件名
os.path.dirname(path):返回文件路径
os.remove(dir) #dir为要删除的文件夹或者文件路径
os.rmdir(path) #path要删除的目录的路径。需要说明的是，使用os.rmdir删除的目录必须为空目录，否则函数出错。
os.path.getmtime(name) ＃获取文件的修改时间
os.stat(path).st_mtime＃获取文件的修改时间
os.stat(path).st_ctime #获取文件修改时间
os.path.getctime(name)#获取文件的创建时间
python中对文件、文件夹的操作需要涉及到os模块和shutil模块。
创建文件：
1) os.mknod("test.txt") 创建空文件
2) open("test.txt",w) 直接打开一个文件，如果文件不存在则创建文件
创建目录：
os.mkdir("file") 创建目录
复制文件：
shutil.copyfile("oldfile","newfile") oldfile和newfile都只能是文件
shutil.copy("oldfile","newfile") oldfile只能是文件夹，newfile可以是文件，也可以是目标目录
复制文件夹：
shutil.copytree("olddir","newdir") olddir和newdir都只能是目录，且newdir必须不存在
重命名文件（目录）
os.rename("oldname","newname") 文件或目录都是使用这条命令
移动文件（目录）
shutil.move("oldpos","newpos")
删除文件
os.remove("file")
删除目录
os.rmdir("dir") 只能删除空目录
shutil.rmtree("dir") 空目录、有内容的目录都可以删
转换目录
os.chdir("path") 换路径
判断目标
os.path.exists("goal") 判断目标是否存在
os.path.isdir("goal") 判断目标是否目录
os.path.isfile("goal") 判断目标是否文件
Python 实现文件复制、删除
import os
import shutil
filelist=[]
rootdir="/home/zoer/aaa"
filelist=os.listdir(rootdir)
for f in filelist:
filepath = os.path.join( rootdir, f )
if os.path.isfile(filepath):
os.remove(filepath)
print filepath+" removed!"
elif os.path.isdir(filepath):
shutil.rmtree(filepath,True)
print "dir "+filepath+" removed!"
用python实现了一个小型的自动发版本的工具。这个“自动发版本”有点虚， 只是简单地把debug 目录下的配置文件复制到指定目录，把Release下的生成文件复制到同一指定，过滤掉不需要的文件夹(.svn)，然后再往这个指定目录添加几个特定的 文件。
这个是我的第一个python小程序。
下面就来看其代码的实现。
首先插入必要的库：
import os
import os.path
import shutil
import time, datetime
然后就是一大堆功能函数。第一个就是把某一目录下的所有文件复制到指定目录中：
def copyFiles(sourceDir, targetDir):
if sourceDir.find(".svn") > 0:
return
for file in os.listdir(sourceDir):
sourceFile = os.path.join(sourceDir, file)
targetFile = os.path.join(targetDir, file)
if os.path.isfile(sourceFile):
if not os.path.exists(targetDir):
os.makedirs(targetDir)
if not os.path.exists(targetFile) or(os.path.exists(targetFile) and (os.path.getsize(targetFile) != os.path.getsize(sourceFile))):
open(targetFile, "wb").write(open(sourceFile, "rb").read())
if os.path.isdir(sourceFile):
First_Directory = False
copyFiles(sourceFile, targetFile)
删除一级目录下的所有文件：
def removeFileInFirstDir(targetDir):
for file in os.listdir(targetDir):
targetFile = os.path.join(targetDir, file)
if os.path.isfile(targetFile):
os.remove(targetFile)
复制一级目录下的所有文件到指定目录：
def coverFiles(sourceDir, targetDir):
for file in os.listdir(sourceDir):
sourceFile = os.path.join(sourceDir, file)
targetFile = os.path.join(targetDir, file)
#cover the files
if os.path.isfile(sourceFile):
open(targetFile, "wb").write(open(sourceFile, "rb").read())
复制指定文件到目录：
def moveFileto(sourceDir, targetDir):
shutil.copy(sourceDir, targetDir)
往指定目录写文本文件：
def writeVersionInfo(targetDir):
open(targetDir, "wb").write("Revison:")
返回当前的日期，以便在创建指定目录的时候用：
def getCurTime():
nowTime = time.localtime()
year = str(nowTime.tm_year)
month = str(nowTime.tm_mon)
if len(month) < 2:
month = '0' + month
day = str(nowTime.tm_yday)
if len(day) < 2:
day = '0' + day
return (year + '-' + month + '-' + day)
然后就是主函数的实现了：
if name =="main":
print "Start(S) or Quilt(Q) \n"
flag = True
while (flag):
answer = raw_input()
if 'Q' == answer:
flag = False
elif 'S'== answer :
formatTime = getCurTime()
targetFoldername = "Build " + formatTime + "-01"
Target_File_Path += targetFoldername
copyFiles(Debug_File_Path, Target_File_Path)
removeFileInFirstDir(Target_File_Path)
coverFiles(Release_File_Path, Target_File_Path)
moveFileto(Firebird_File_Path, Target_File_Path)
moveFileto(AssistantGui_File_Path, Target_File_Path)
writeVersionInfo(Target_File_Path+"\\ReadMe.txt")
print "all sucess"
else:
print "not the correct command"linux下python脚本判断目录和文件是否存在
if os.path.isdir('E:test'):
pass
else:
os.mkdir('E:test')
##os.mkdir() 只会创建一个目录，不可以级联创建
eg2：
if not os.path.exists('E:test'): ###判断文件是否存在，返回布尔值
os.makedirs('E:test')
##os.makedirs() 这个连同中间的目录都会创建，类似于参数mkdir -p
eg3：
try:
fp = open("file_path")
catch exception: except 和catch的区别？
os.mkdir('file_path') ##os.mkdir() 只会创建一个目录，不可级联创建，但要有异常处理的意识
fp = open("file_path"
eg4：实测
#!/<a href="http://so.21ops.com/cse/search?s=9181936462520079739&entry=1&q=usr" class="bdcs-inlinelink" target="_blank">usr</a>/bin/env python
import os
FILE_PATH='/home/wuxy/aaa111/222/333/444.txt' ###444.txt 不会当做文件，而是当做目录
if os.path.exists('FILE_PATH'): ##目录存在，返回为真
print 'dir not exists'
os.makedirs(FILE_PATH) ###FILE_PATH不用加引号。否则会报错
else:
print 'dir exists'
python实现ftp上传下载文件
#!/usr/bin/env python
# encoding: utf-8
author = "pwy"
'''
上传：上传文件并备份到其他目录
下载：下载文件，并删除远端文件
'''
from ftplib import FTP
from time import sleep
import os,datetime,logging
from shutil import move
HOST = "192.168.1.221"
USER = "sxit"
PASSWORD = "1qaz!QAZ"
#PORT = ""
#Upload the file, change the directory
remotedir = "/home/test/"
localdir = "/home/sxit/object/"
bakdir = "/home/sxit/bak"
#Download the file, change the directory
Remoredir = "/home/sxit/object1/"
Localdir = "/root/ftp-logging"
LOGFILE = datetime.datetime.now().strftime('%Y-%m-%d')+'.log'
logging.basicConfig(level=logging.INFO,
format='%(asctime)s %(filename)s %(levelname)s %(message)s',
# datefmt='%a, %d %b %Y %H:%M:%S',
filename= LOGFILE,
filemode='a')
logging.FileHandler(LOGFILE)
class CLASS_FTP:
def init(self,HOST,USER,PASSWORD,PORT='21'):
self.HOST = HOST
self.USER = USER
self.PASSWORD = PASSWORD
self.PORT = PORT
self.ftp=FTP()
self.flag=0 # 0:no connected, 1: connting
def Connect(self):
try:
if self.flag == 1:
logging.info("ftp Has been connected")
else:
self.ftp.connect(self.HOST,self.PORT)
self.ftp.login(self.USER,self.PASSWORD)
# self.ftp.set_pasv(False)
self.ftp.set_debuglevel(0)
self.flag=1
except Exception:
logging.info("FTP login failed")
def Up_load(self,remotedir,localdir,bakdir):
try:
self.ftp.cwd(remotedir)
for i in os.listdir(localdir):
if i.endswith('.txt'):
file_handler = open(i,'rb')
self.ftp.storbinary('STOR %s' % i,file_handler)
logging.info("%s already upload ."%i)
try:
if os.path.isdir(bakdir):
move(i,bakdir)
logging.info("%s move to %s ." % (i,bakdir))
else:
print "Move the file FAILED"
logging.info("Move the %s to %s FAILED!"%(i,bakdir))
except Exception:
logging.info("ftp delete file faild !!!!!")
file_handler.close()
# self.ftp.quit()
except Exception:
logging.info("Up_load failed")
def Down_load(self,Remoredir,Localdir):
try:
self.ftp.cwd(Remoredir)
for i in self.ftp.nlst():
if i.endswith('.NET'): #match file
file_handler = open(i,'wb')
self.ftp.retrbinary('RETR %s' % i,file_handler.write)
logging.info("%s already down ."%i)
try:
self.ftp.delete(i)
logging.info("%s already deleted!"%i)
except Exception:
logging.info("ftp delete file faild !!!!!")
file_handler.close()
#self.ftp.quit()
except Exception:
logging.info("Down_load failed")
if name == 'main':
ftp = CLASS_FTP(HOST,USER,PASSWORD)
while True:
ftp.Connect()
# ftp.Down_load(Remoredir,Localdir)
ftp.Up_load(remotedir,localdir,bakdir)
sleep(30)常用函数用手册查看，以下只是简略，因为没用用到，[待整理]：
login(user='',passwd='', acct='') 登录到FTP 服务器，所有的参数都是可选的
pwd() 当前工作目录
cwd(path) 把当前工作目录设置为path
dir([path[,...[,cb]]) 显示path 目录里的内容，可选的参数cb 是一个回调函数，会被传给retrlines()方法
nlst([path[,...]) 与dir()类似，但返回一个文件名的列表，而不是显示这些文件名
retrlines(cmd [, cb]) 给定FTP 命令（如“RETR filename”），用于下载文本文件。可选的回调函数cb 用于处理文件的每一行
retrbinary(cmd, cb[,bs=8192[, ra]]) 与retrlines()类似，只是这个指令处理二进制文件。回调函数cb 用于处理每一块（块大小默认为8K）下载的数据。
storlines(cmd, f) 给定FTP 命令（如“STOR filename”），以上传文本文件。要给定一个文件对象f
storbinary(cmd, f[,bs=8192]) 与storlines()类似，只是这个指令处理二进制文件。要给定一个文件对象f，上传块大小bs 默认为8Kbs=8192])
rename(old, new) 把远程文件old 改名为new
delete(path) 删除位于path 的远程文件
mkd(directory) 创建远程目录
ftp
'''第一个例子'''
def get_C(self,target_dir=None):
C = []
print "PWD:", self.ftp.pwd()
if target_dir is not None:
self.ftp.cwd(target_dir)# change working directory to target_dir
server_file_list = []
fuck_callback = lambda x: (server_file_list.append(x))
self.ftp.retrlines('LIST', fuck_callback)
# print server_file_list
server_file_items = self.filter_dir_list(server_file_list)
for item in server_file_items:
if item.is_dir:
print 'name = ', item.name
sub_C = self.get_C(item.name)
# sub_C = ['/'+item.name+'/'+cc.name for cc in sub_C]
for cc in sub_C:
cc.name = '/' + item.name + cc.name
print 'name --- ',cc.name
C.extend(sub_C)
else:
item.name = '/' + item.name
C.append(item)
self.ftp.cwd('..')
return C
def runtest(self,next_dir):
C = ftp.get_C(next_dir)
next_dir2=next_dir[2:]
C = [cc.pack for cc in C]
for i in C:
print i
next_dir1=i
pos=next_dir1.rindex('/')
next_dir3= next_dir1[0:pos]
all_path=next_dir2 + next_dir3
print all_path
next_dir_local = all_path.decode('utf8').encode('gbk')
try:
print next_dir_local
#os.makedirs(next_dir_local)
except OSError:
pass
#os.chdir(next_dir_local)
localfile=next_dir1[pos+1:]
print localfile
allall_path=all_path + "/" + localfile
self.ftp.cwd('/')
print self.ftp.pwd()
#file_handler = open(localfile, 'wb')
#self.ftp.retrbinary('RETR %s' % (allall_path), file_handler.write)
#file_handler.close()
'''第一个例子获取成/home/user/test.txt这样的列表'''
第二个例子
def download_files(self, localdir='./', remotedir='./'):
try:
self.ftp.cwd(remotedir)
except:
debug_print('目录%s不存在，继续...' % remotedir)
return
if not os.path.isdir(localdir):
pass
#os.makedirs(localdir)
debug_print('切换至目录 %s' % self.ftp.pwd())
self.file_list = []
self.ftp.dir(self.get_file_list)
remotenames = self.file_list
print(remotenames)
# return
for item in remotenames:
filetype = item[0]
filename = item[1]
print "filename:",filename
local = os.path.join(localdir, filename)
if filetype == 'd':
self.download_files(local, filename)
elif filetype == '-':
self.download_file(local, filename)
self.ftp.cwd('..')
debug_print('返回上层目录 %s' % self.ftp.pwd())
f.download_files(rootdir_local, rootdir_remote)
'''第二个例子'''
sftp
s_file = path.join(path_name,name).replace('\\','/')
def process_sftp_dir(path_name):
"""
此函数递归处理sftp server端的目录和文件,并在client端创建所有不存在的目录,然后针对每个文件在两端的全路径执行get操作.
path_name第一次的引用值应该是source_path的值
"""
d_path = path_name.replace(source_path,destination_path,1)
if not path.exists(d_path): # 若目标目录不存在则创建
print('%s----Create Local Dir: %s' % (' '*8,d_path))
try:
makedirs(d_path) # 递归创建不存在的目录
except Exception as err:
print('%s----Create %s Failed' % (' '*8,d_path))
print('{}----{}'.format(' '*8,err))
exit(10)
for name in (i for i in sftp.listdir(path=path_name) if not i.startswith('.')):
"""去掉以.开头的文件或目录"""
s_file = path.join(path_name,name).replace('\\','/') # 在win环境下组合路径所用的'\\'换成'/'
d_file = s_file.replace(source_path,destination_path,1) # 目标端全路径
chk_r_path_result = check_remote_path(s_file)
if chk_r_path_result == 'file': # 文件
sftp_get(s_file,d_file,12)
elif chk_r_path_result == 'directory': # 目录
process_sftp_dir(s_file) # 递归调用本身
process_sftp_dir(source_path)
区别很大
ftp:
ftp.retrlines('LIST', fuck_callback)
完全是循环，目录的进行循环操作，而文件下载。最底层目录的文件下载完，回归上级目录。继续循环。
self.ftp.pwd()
self.ftp.dir(self.get_file_list)
get_file_list(self, line)
self.ftp.cwd('..')
self.ftp.cwd(remotedir)
self.download_file(local, filename)
建立好本地目录，然后cd到远程目录，下载
sftp：
sftp.listdir
s_file = path.join(path_name,name).replace('\\','/')
指定源全路径下载
代码格式乱了，详细例子
ftp 第一个例子
# !/usr/bin/env python
# -*-coding:utf-8-*-
from ftplib import FTP
from time import sleep
import os, datetime,logging,time
import string,re
d1 = datetime.datetime.now()
'''months=['Jan','Feb','March','Apr','May','Jun','Jul','Aug','Sep']
patternm = r'2017.*|201611.*|201612.*|201610.*'
patternxml = r'.*2016'
patternx = r'xx.*'''''
HOST = "192.168.1.100"
USER = "ftpuser3"
PASSWORD = "test1passwd"
class Myfile(object):
def init(self, name, size, mtime):
self.name = name # 文件名字
self.mtime = mtime # 文件创建时间
self.is_dir = False # 是否为文件夹，默认为不是文件夹
#self.size = float(size) / (1024 * 1024) # 文件大小
size = float(size)
if size > 1024*1024:
self.size = str('%.2f'%(size / (1024*1024))) + 'MB'
elif size > 1024:
self.size = str('%.2f'%(size / 1024)) + 'KB'
else:
self.size = str(size) + 'Bytes'
@property
def is_file(self):
return not self.is_dir
@property
def dir_property(self):
if self.is_dir==True:
return 'dir'
return 'file'
def show(self):
print '[%s], [%s], [%s], [%s]' % (self.name, self.size, self.mtime, self.dir_property)
@property
def pack(self):
"""
将myfile对象封装为一个字符串
:return:
"""
#return '[%s][%s][%s]'%(self.name, self.size, self.mtime)
#return '[%s][%s]'%(self.name, self.size)
return '%s' %(self.name)
class CLASS_FTP:
def init(self, HOST, USER, PASSWORD, PORT='21'):
self.HOST = HOST
self.USER = USER
self.PASSWORD = PASSWORD
self.PORT = PORT
self.ftp = FTP()
self.flag = 0 # 0:no connected, 1: connting
def Connect(self):
try:
if self.flag == 1:
logging.info("ftp Has been connected")
else:
self.ftp.connect(self.HOST, self.PORT)
self.ftp.login(self.USER, self.PASSWORD)
# self.ftp.set_pasv(False)
self.ftp.set_debuglevel(0)
self.flag = 1
except Exception:
logging.info("FTP login failed")
def str_codec_std(self,mystr):
return mystr.decode('utf8').encode('gbk')
def dirmakedirs(self,next_dir_local,local_dir):
# next_dir_local2= next_dir_local.split('/')[1:]
next_dir_local2 = next_dir_local[1:].replace('/', '\\')
# next_dir_localw = next_dir_local2.decode('utf8').encode('gbk') # windows用这个
s_file = os.path.join(local_dir, next_dir_local2)
print "s_file", s_file
if not os.path.exists(s_file):
try:
os.makedirs(s_file)
except OSError:
pass
os.chdir(s_file)
def filter_dir_list(self,mystr_list):
res = []
for mystr in mystr_list:
#mystr = self.str_codec_std(mystr)
# print "mystr is :%s" % mystr
file_info = string.split(mystr, maxsplit=8)
name = file_info[8]
print 'name = ', name
if name == '.' or name == '..':
continue
size = file_info[4]
mtime = '%s-%s-%s' % (file_info[5], file_info[6], file_info[7])
myfile = Myfile(name=name, size=size, mtime=mtime)
dir_info = file_info[0]
if dir_info[0] == 'd':
myfile.is_dir = True
res.append(myfile)
return res
def get_C(self,target_dir=None,local_dir=None):
C = []
if target_dir is not None:
self.ftp.cwd(target_dir)# change working directory to target_dir
server_file_list = []
fuck_callback = lambda x: (server_file_list.append(x))
self.ftp.retrlines('LIST', fuck_callback)
next_dir_local = self.ftp.pwd()
self.dirmakedirs(next_dir_local, local_dir)
server_file_items = self.filter_dir_list(server_file_list)
for item in server_file_items:
if item.is_dir:
sub_C = self.get_C(item.name,local_dir)
for cc in sub_C:
cc.name = '/' + item.name + cc.name
C.extend(sub_C)
else:
item.name = '/' + item.name
C.append(item)
self.ftp.cwd('..')
return C
def runtest(self,local_dir,next_dir):
os.chdir(local_dir)
C = ftp.get_C(next_dir,local_dir)
next_dir2=next_dir[2:]
C = [cc.pack for cc in C]
print "C:",C
for i in C:
next_dir1=i
pos=next_dir1.rindex('/')
next_dir3= next_dir1[0:pos]
all_path=next_dir2 + next_dir3
self.dirmakedirs(all_path, local_dir)
next_dir_localz = all_path[1:].replace('/', '\\')
'''# next_dir_local = next_dir_localz
# next_dir_local = next_dir_localz.decode('utf8').encode('gbk') #windows用这个'''
# s_file = os.path.join(local_dir, next_dir_localz)
# try:
# os.makedirs(s_file)
# except OSError:
# pass
# os.chdir(s_file)
localfile=next_dir1[pos+1:]
print localfile
allall_path=all_path + "/" + localfile
file_handler = open(localfile, 'wb')
self.ftp.retrbinary('RETR %s' % (allall_path), file_handler.write)
file_handler.close()
if name == 'main':
ftp = CLASS_FTP(HOST, USER, PASSWORD)
ftp.Connect()
ftp.runtest('D:\\ftp','./')
d2 = datetime.datetime.now()
print d2 - d1
'''参数乱七八糟'''
ftp 第二个例子 别人2010写好的
# !/usr/bin/env python
# coding:utf-8
from ftplib import FTP
import os, sys, string, datetime, time
import socket
class MYFTP:
def init(self, hostaddr, username, password, remotedir, port=21):
self.hostaddr = hostaddr
self.username = username
self.password = password
self.remotedir = remotedir
self.port = port
self.ftp = FTP()
self.file_list = []
# self.ftp.set_debuglevel(2)
def del(self):
self.ftp.close()
# self.ftp.set_debuglevel(0)
def login(self):
ftp = self.ftp
try:
timeout = 60
socket.setdefaulttimeout(timeout)
ftp.set_pasv(True)
print '开始连接到 %s' % (self.hostaddr)
ftp.connect(self.hostaddr, self.port)
print '成功连接到 %s' % (self.hostaddr)
print '开始登录到 %s' % (self.hostaddr)
ftp.login(self.username, self.password)
print '成功登录到 %s' % (self.hostaddr)
debug_print(ftp.getwelcome())
except Exception:
deal_error("连接或登录失败")
try:
print "now:",self.ftp.pwd()
self.ftp.cwd(self.remotedir)
except(Exception):
deal_error('切换目录失败')
def is_same_size(self, localfile, remotefile):
try:
remotefile_size = self.ftp.size(remotefile)
except:
remotefile_size = -1
try:
localfile_size = os.path.getsize(localfile)
except:
localfile_size = -1
debug_print('lo:%d re:%d' % (localfile_size, remotefile_size), )
if remotefile_size == localfile_size:
return 1
else:
return 0
def download_file(self, localfile, remotefile):
if self.is_same_size(localfile, remotefile):
debug_print('%s 文件大小相同，无需下载' % localfile)
return
else:
print "remotefile:",remotefile
debug_print('>>>>>>>>>>>>下载文件 %s ... ...' % localfile)
# return
file_handler = open(localfile, 'wb')
self.ftp.retrbinary('RETR %s' % (remotefile), file_handler.write)
file_handler.close()
def download_files(self, localdir='./', remotedir='./'):
try:
print "remotedir:",remotedir
self.ftp.cwd(remotedir)
except:
debug_print('目录%s不存在，继续...' % remotedir)
return
if not os.path.isdir(localdir):
# pass
os.makedirs(localdir)
debug_print('切换至目录 %s' % self.ftp.pwd())
self.file_list = []
print(self.ftp.dir())
self.ftp.dir(self.get_file_list)
remotenames = self.file_list
# print(remotenames)
# return
for item in remotenames:
filetype = item[0]
filename = item[1]
print "filename:",filename
local = os.path.join(localdir, filename).replace('\\', '/')
if filetype == 'd':
self.download_files(local, filename)
elif filetype == '-':
self.download_file(local, filename)
self.ftp.cwd('..')
debug_print('返回上层目录 %s' % self.ftp.pwd())
def upload_file(self, localfile, remotefile):
if not os.path.isfile(localfile):
return
if self.is_same_size(localfile, remotefile):
debug_print('跳过[相等]: %s' % localfile)
return
file_handler = open(localfile, 'rb')
self.ftp.storbinary('STOR %s' % remotefile, file_handler)
file_handler.close()
debug_print('已传送: %s' % localfile)
def upload_files(self, localdir='./', remotedir='./'):
if not os.path.isdir(localdir):
return
localnames = os.listdir(localdir)
self.ftp.cwd(remotedir)
for item in localnames:
src = os.path.join(localdir, item)
if os.path.isdir(src):
try:
self.ftp.mkd(item)
except:
debug_print('目录已存在 %s' % item)
self.upload_files(src, item)
else:
self.upload_file(src, item)
self.ftp.cwd('..')
def get_file_list(self, line):
print "line1:", line
ret_arr = []
file_arr = self.get_filename(line)
print "file_arr:",file_arr
if file_arr[1] not in ['.', '..']:
self.file_list.append(file_arr)
def get_filename(self, line):
print "line2:",line
print type(line)
pos = line.rfind(':')
while (line[pos] != ' '):
pos += 1
while (line[pos] == ' '):
pos += 1
print pos
file_arr = [line[0], line[pos:]]
return file_arr
def debug_print(s):
print (s)
def deal_error(e):
timenow = time.localtime()
datenow = time.strftime('%Y-%m-%d', timenow)
logstr = '%s 发生错误: %s' % (datenow, e)
debug_print(logstr)
file.write(logstr)
sys.exit()
if name == 'main':
file = open("log.txt", "a")
timenow = time.localtime()
datenow = time.strftime('%Y-%m-%d', timenow)
logstr = datenow
# 配置如下变量
hostaddr = '192.168.1.100' # ftp地址
username = 'ftpuser3' # 用户名
password = 'test1passwd' # 密码
port = 21 # 端口号
#rootdir_local = '.' + os.sep + 'bak/' # 本地目录
rootdir_local = 'D:/ftp/'
rootdir_remote = './' # 远程目录
f = MYFTP(hostaddr, username, password, rootdir_remote, port)
f.login()
f.download_files(rootdir_local, rootdir_remote)
timenow = time.localtime()
datenow = time.strftime('%Y-%m-%d', timenow)
logstr += " - %s 成功执行了备份\n" % datenow
debug_print(logstr)
file.write(logstr)
file.close()登录后复制
以上是使用python ftp和sftp的实例介绍的详细内容。更多信息请关注PHP中文网其他相关文章！
本站声明
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn
热AI工具
Undresser.AI Undress
人工智能驱动的应用程序，用于创建逼真的裸体照片
AI Clothes Remover
用于从照片中去除衣服的在线人工智能工具。
Undress AI Tool
免费脱衣服图片
Clothoff.io
AI脱衣机
Video Face Swap
使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸！
热门文章
刺客信条阴影：贝壳谜语解决方案
4 周前
By DDD
Windows 11 KB5054979中的新功能以及如何解决更新问题
3 周前
By DDD
在哪里可以找到原子中的起重机控制钥匙卡
4 周前
By DDD
<🎜>：死铁路 - 如何完成所有挑战
1 个月前
By DDD
Atomfall指南：项目位置，任务指南和技巧
1 个月前
By DDD
热工具
记事本++7.3.1
好用且免费的代码编辑器
SublimeText3汉化版
中文版，非常好用
禅工作室 13.0.1
功能强大的PHP集成开发环境
Dreamweaver CS6
视觉化网页开发工具
SublimeText3 Mac版
神级代码编辑软件(SublimeText3)
PHP和Python：解释了不同的范例
Apr 18, 2025 am 12:26 AM
PHP主要是过程式编程，但也支持面向对象编程（OOP）；Python支持多种范式，包括OOP、函数式和过程式编程。PHP适合web开发，Python适用于多种应用，如数据分析和机器学习。
在PHP和Python之间进行选择：指南
Apr 18, 2025 am 12:24 AM
PHP适合网页开发和快速原型开发，Python适用于数据科学和机器学习。1.PHP用于动态网页开发，语法简单，适合快速开发。2.Python语法简洁，适用于多领域，库生态系统强大。
Python vs. JavaScript：学习曲线和易用性
Apr 16, 2025 am 12:12 AM
Python更适合初学者，学习曲线平缓，语法简洁；JavaScript适合前端开发，学习曲线较陡，语法灵活。1.Python语法直观，适用于数据科学和后端开发。2.JavaScript灵活，广泛用于前端和服务器端编程。
vscode怎么在终端运行程序
Apr 15, 2025 pm 06:42 PM
在 VS Code 中，可以通过以下步骤在终端运行程序：准备代码和打开集成终端确保代码目录与终端工作目录一致根据编程语言选择运行命令（如 Python 的 python your_file_name.py）检查是否成功运行并解决错误利用调试器提升调试效率
vs code 可以在 Windows 8 中运行吗
Apr 15, 2025 pm 07:24 PM
VS Code可以在Windows 8上运行，但体验可能不佳。首先确保系统已更新到最新补丁，然后下载与系统架构匹配的VS Code安装包，按照提示安装。安装后，注意某些扩展程序可能与Windows 8不兼容，需要寻找替代扩展或在虚拟机中使用更新的Windows系统。安装必要的扩展，检查是否正常工作。尽管VS Code在Windows 8上可行，但建议升级到更新的Windows系统以获得更好的开发体验和安全保障。
visual studio code 可以用于 python 吗
Apr 15, 2025 pm 08:18 PM
VS Code 可用于编写 Python，并提供许多功能，使其成为开发 Python 应用程序的理想工具。它允许用户：安装 Python 扩展，以获得代码补全、语法高亮和调试等功能。使用调试器逐步跟踪代码，查找和修复错误。集成 Git，进行版本控制。使用代码格式化工具，保持代码一致性。使用 Linting 工具，提前发现潜在问题。
PHP和Python：深入了解他们的历史
Apr 18, 2025 am 12:25 AM
PHP起源于1994年，由RasmusLerdorf开发，最初用于跟踪网站访问者，逐渐演变为服务器端脚本语言，广泛应用于网页开发。Python由GuidovanRossum于1980年代末开发，1991年首次发布，强调代码可读性和简洁性，适用于科学计算、数据分析等领域。
vscode 扩展是否是恶意的
Apr 15, 2025 pm 07:57 PM
VS Code 扩展存在恶意风险，例如隐藏恶意代码、利用漏洞、伪装成合法扩展。识别恶意扩展的方法包括：检查发布者、阅读评论、检查代码、谨慎安装。安全措施还包括：安全意识、良好习惯、定期更新和杀毒软件。
