Beispiele für die Verwendung von Python FTP und 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()Nach dem Login kopieren
Das obige ist der detaillierte Inhalt vonBeispiele für die Verwendung von Python FTP und SFTP. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
Erklärung dieser Website
Der Inhalt dieses Artikels wird freiwillig von Internetnutzern beigesteuert und das Urheberrecht liegt beim ursprünglichen Autor. Diese Website übernimmt keine entsprechende rechtliche Verantwortung. Wenn Sie Inhalte finden, bei denen der Verdacht eines Plagiats oder einer Rechtsverletzung besteht, wenden Sie sich bitte an admin@php.cn
Heiße KI -Werkzeuge
Undresser.AI Undress
KI-gestützte App zum Erstellen realistischer Aktfotos
AI Clothes Remover
Online-KI-Tool zum Entfernen von Kleidung aus Fotos.
Undress AI Tool
Ausziehbilder kostenlos
Clothoff.io
KI-Kleiderentferner
AI Hentai Generator
Erstellen Sie kostenlos Ai Hentai.
Heißer Artikel
R.E.P.O. Energiekristalle erklärten und was sie tun (gelber Kristall)
3 Wochen vor
By 尊渡假赌尊渡假赌尊渡假赌
R.E.P.O. Beste grafische Einstellungen
3 Wochen vor
By 尊渡假赌尊渡假赌尊渡假赌
Assassin's Creed Shadows: Seashell Riddle -Lösung
2 Wochen vor
By DDD
R.E.P.O. So reparieren Sie Audio, wenn Sie niemanden hören können
3 Wochen vor
By 尊渡假赌尊渡假赌尊渡假赌
WWE 2K25: Wie man alles in Myrise freischaltet
3 Wochen vor
By 尊渡假赌尊渡假赌尊渡假赌
Heiße Werkzeuge
Notepad++7.3.1
Einfach zu bedienender und kostenloser Code-Editor
SublimeText3 chinesische Version
Chinesische Version, sehr einfach zu bedienen
Senden Sie Studio 13.0.1
Leistungsstarke integrierte PHP-Entwicklungsumgebung
Dreamweaver CS6
Visuelle Webentwicklungstools
SublimeText3 Mac-Version
Codebearbeitungssoftware auf Gottesniveau (SublimeText3)
Heiße Themen
CakePHP-Tutorial
1376
52
1376
52
Was ist der Grund, warum PS immer wieder Laden zeigt?
Apr 06, 2025 pm 06:39 PM
PS "Laden" Probleme werden durch Probleme mit Ressourcenzugriff oder Verarbeitungsproblemen verursacht: Die Lesegeschwindigkeit von Festplatten ist langsam oder schlecht: Verwenden Sie Crystaldiskinfo, um die Gesundheit der Festplatte zu überprüfen und die problematische Festplatte zu ersetzen. Unzureichender Speicher: Upgrade-Speicher, um die Anforderungen von PS nach hochauflösenden Bildern und komplexen Schichtverarbeitung zu erfüllen. Grafikkartentreiber sind veraltet oder beschädigt: Aktualisieren Sie die Treiber, um die Kommunikation zwischen PS und der Grafikkarte zu optimieren. Dateipfade sind zu lang oder Dateinamen haben Sonderzeichen: Verwenden Sie kurze Pfade und vermeiden Sie Sonderzeichen. Das eigene Problem von PS: Installieren oder reparieren Sie das PS -Installateur neu.
Wie löst ich das Problem des Ladens beim Starten von PS?
Apr 06, 2025 pm 06:36 PM
Ein PS, der beim Booten auf "Laden" steckt, kann durch verschiedene Gründe verursacht werden: Deaktivieren Sie korrupte oder widersprüchliche Plugins. Eine beschädigte Konfigurationsdatei löschen oder umbenennen. Schließen Sie unnötige Programme oder aktualisieren Sie den Speicher, um einen unzureichenden Speicher zu vermeiden. Upgrade auf ein Solid-State-Laufwerk, um die Festplatte zu beschleunigen. PS neu installieren, um beschädigte Systemdateien oder ein Installationspaketprobleme zu reparieren. Fehlerinformationen während des Startprozesses der Fehlerprotokollanalyse anzeigen.
Wie löste ich das Problem des Ladens, wenn die PS die Datei öffnet?
Apr 06, 2025 pm 06:33 PM
Das Laden von Stottern tritt beim Öffnen einer Datei auf PS auf. Zu den Gründen gehören: zu große oder beschädigte Datei, unzureichender Speicher, langsame Festplattengeschwindigkeit, Probleme mit dem Grafikkarten-Treiber, PS-Version oder Plug-in-Konflikte. Die Lösungen sind: Überprüfen Sie die Dateigröße und -integrität, erhöhen Sie den Speicher, aktualisieren Sie die Festplatte, aktualisieren Sie den Grafikkartentreiber, deinstallieren oder deaktivieren Sie verdächtige Plug-Ins und installieren Sie PS. Dieses Problem kann effektiv gelöst werden, indem die PS -Leistungseinstellungen allmählich überprüft und genutzt wird und gute Dateimanagementgewohnheiten entwickelt werden.
Wie kontrolliert PS -Federn die Weichheit des Übergangs?
Apr 06, 2025 pm 07:33 PM
Der Schlüssel zur Federkontrolle liegt darin, seine allmähliche Natur zu verstehen. PS selbst bietet nicht die Möglichkeit, die Gradientenkurve direkt zu steuern, aber Sie können den Radius und die Gradientenweichheit flexius durch mehrere Federn, Matching -Masken und feine Selektionen anpassen, um einen natürlichen Übergangseffekt zu erzielen.
So verwenden Sie MySQL nach der Installation
Apr 08, 2025 am 11:48 AM
Der Artikel führt den Betrieb der MySQL -Datenbank vor. Zunächst müssen Sie einen MySQL -Client wie MySQLworkBench oder Befehlszeilen -Client installieren. 1. Verwenden Sie den Befehl mySQL-uroot-P, um eine Verbindung zum Server herzustellen und sich mit dem Stammkonto-Passwort anzumelden. 2. Verwenden Sie die Erstellung von Createdatabase, um eine Datenbank zu erstellen, und verwenden Sie eine Datenbank aus. 3.. Verwenden Sie CreateTable, um eine Tabelle zu erstellen, Felder und Datentypen zu definieren. 4. Verwenden Sie InsertInto, um Daten einzulegen, Daten abzufragen, Daten nach Aktualisierung zu aktualisieren und Daten nach Löschen zu löschen. Nur indem Sie diese Schritte beherrschen, lernen, mit gemeinsamen Problemen umzugehen und die Datenbankleistung zu optimieren, können Sie MySQL effizient verwenden.
Wie richte ich PS -Federn ein?
Apr 06, 2025 pm 07:36 PM
PS Federn ist ein Bildkantenschwärcheneffekt, der durch den gewichteten Durchschnitt der Pixel im Randbereich erreicht wird. Das Einstellen des Federradius kann den Grad der Unschärfe steuern und je größer der Wert ist, desto unscharfer ist er. Eine flexible Einstellung des Radius kann den Effekt entsprechend den Bildern und Bedürfnissen optimieren. Verwenden Sie beispielsweise einen kleineren Radius, um Details bei der Verarbeitung von Charakterfotos zu erhalten und einen größeren Radius zu verwenden, um ein dunstiges Gefühl bei der Verarbeitung von Kunst zu erzeugen. Es ist jedoch zu beachten, dass zu groß der Radius leicht an Kantendetails verlieren kann, und zu klein ist der Effekt nicht offensichtlich. Der Federneffekt wird von der Bildauflösung beeinflusst und muss anhand des Bildverständnisses und des Griffs von Effekten angepasst werden.
So optimieren Sie die Datenbankleistung nach der MySQL -Installation
Apr 08, 2025 am 11:36 AM
Die MySQL -Leistungsoptimierung muss von drei Aspekten beginnen: Installationskonfiguration, Indexierung und Abfrageoptimierung, Überwachung und Abstimmung. 1. Nach der Installation müssen Sie die my.cnf -Datei entsprechend der Serverkonfiguration anpassen, z. 2. Erstellen Sie einen geeigneten Index, um übermäßige Indizes zu vermeiden und Abfrageanweisungen zu optimieren, z. B. den Befehl Erklärung zur Analyse des Ausführungsplans; 3. Verwenden Sie das eigene Überwachungstool von MySQL (ShowProcessList, Showstatus), um die Datenbankgesundheit zu überwachen und die Datenbank regelmäßig zu sichern und zu organisieren. Nur durch kontinuierliche Optimierung dieser Schritte kann die Leistung der MySQL -Datenbank verbessert werden.
Muss MySQL bezahlen?
Apr 08, 2025 pm 05:36 PM
MySQL hat eine kostenlose Community -Version und eine kostenpflichtige Enterprise -Version. Die Community -Version kann kostenlos verwendet und geändert werden, die Unterstützung ist jedoch begrenzt und für Anwendungen mit geringen Stabilitätsanforderungen und starken technischen Funktionen geeignet. Die Enterprise Edition bietet umfassende kommerzielle Unterstützung für Anwendungen, die eine stabile, zuverlässige Hochleistungsdatenbank erfordern und bereit sind, Unterstützung zu bezahlen. Zu den Faktoren, die bei der Auswahl einer Version berücksichtigt werden, gehören Kritikalität, Budgetierung und technische Fähigkeiten von Anwendungen. Es gibt keine perfekte Option, nur die am besten geeignete Option, und Sie müssen die spezifische Situation sorgfältig auswählen.
