1. 连接 ftp 服务器
代码如下
from ftplib import FTP
from log import Logger
class LinkFTP:
"""
连接 FTP 服务器
"""
def __init__(self, host: str, port: int, username: str, password: str):
ftp_ = FTP()
ftp_.connect(host, port)
ftp_.login(username, password)
Logger.info(f"{host} {port} {username} {password} 连接成功")
self.ftp = ftp_
self.buffer_size = 2048
def download_file(self, remote_path: str, local_path: str) -> None:
"""
从 ftp 下载文件
:param remote_path: 远程服务器的目录绝对路径
:param local_path:
:return:
"""
with open(local_path, 'wb') as fp:
self.ftp.retrbinary('RETR ' + remote_path, fp.write, self.buffer_size)
self.ftp.set_debuglevel(0)
def upload_file(self, remote_path: str, local_path: str) -> None:
"""
从本地上传文件到 ftp
:param remote_path: 远程服务器的目录绝对路径
:param local_path:
:return:
"""
with open(local_path, 'rb') as fp:
self.ftp.storbinary('STOR ' + remote_path, fp, self.buffer_size)
self.ftp.set_debuglevel(0)
def path_list(self, path: str) -> list:
"""
获取路径信息
:param path: 路径
:return:
"""
ftp = self.ftp
ftp.cwd(path)
ftp.dir()
dir_list: list = ftp.nlst()
dir_list.sort()
return dir_list
2. 基本命令
ftp = FTP()
ftp.set_debuglevel(2)
ftp.connect("你的IP", "port")
ftp.login("user", "password")
ftp.getwelcome()
ftp.cmd("xxx/xxx")
bufsize = 1024
filename = "filename.txt"
file_handle = open(filename, "wb").write
ftp.retrbinaly("RETR %s"%filename, file_handle, bufsize)
ftp.set_debuglevel(0)
ftp.quit()
ftp.cwd(pathname)
ftp.dir()
ftp.nlst()
ftp.mkd(pathname)
ftp.pwd()
ftp.rmd(dirname)
ftp.delete(filename)
ftp.rename(fromname, toname)
ftp.storbinaly("STOR filename.txt", file_handel, bufsize)
ftp.retrbinary("RETR filename.txt", file_handel, bufsize)
参考连接: [1]: https://blog.51cto.com/u_11949039/2834779
|