如何用Python实现开机自动下载FTP文件
在很多场景下,我们需要让计算机在开机后自动从FTP服务器下载特定文件,比如定时同步数据、自动更新程序资源等。本文将详细介绍如何使用Python实现这一功能。
一、准备工作
1. 安装必要的库
我们将使用ftplib库来处理FTP连接和文件传输,这是Python标准库的一部分,无需额外安装。但如果需要更复杂的任务调度,可能需要安装schedule库:
pip install schedule
2. 准备FTP服务器信息
确保你有FTP服务器的以下信息:
FTP服务器地址
端口号(默认是21)
用户名和密码
要下载的文件路径
本地保存路径
二、编写Python脚本
1. 基本FTP下载功能
首先创建一个基本的Python脚本来连接FTP服务器并下载文件:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FTP文件下载脚本
"""
from ftplib import FTP
import os
import sys
def download_file_from_ftp(ftp_host, ftp_port, ftp_user, ftp_pass, remote_path, local_path):
"""
从FTP服务器下载文件
参数:
ftp_host: FTP服务器地址
ftp_port: FTP服务器端口
ftp_user: FTP用户名
ftp_pass: FTP密码
remote_path: 远程文件路径
local_path: 本地保存路径
"""
try:
# 连接到FTP服务器
print(f"正在连接到FTP服务器 {ftp_host}:{ftp_port}...")
ftp = FTP()
ftp.connect(ftp_host, ftp_port)
ftp.login(ftp_user, ftp_pass)
print("登录成功")
# 获取文件名
filename = os.path.basename(remote_path)
# 创建本地目录(如果不存在)
local_dir = os.path.dirname(local_path)
if not os.path.exists(local_dir):
os.makedirs(local_dir)
print(f"创建本地目录: {local_dir}")
# 下载文件
print(f"开始下载文件: {filename}")
with open(local_path, 'wb') as local_file:
ftp.retrbinary(f'RETR {remote_path}', local_file.write)
print(f"文件下载成功,保存至: {local_path}")
# 关闭FTP连接
ftp.quit()
return True
except Exception as e:
print(f"下载失败: {str(e)}")
return False
if __name__ == "__main__":
# FTP服务器配置
FTP_HOST = "ftp.example.com" # 替换为你的FTP服务器地址
FTP_PORT = 21 # FTP默认端口
FTP_USER = "your_username" # 替换为你的FTP用户名
FTP_PASS = "your_password" # 替换为你的FTP密码
# 文件路径配置
REMOTE_PATH = "/path/to/remote/file.txt" # 替换为远程文件路径
LOCAL_PATH = "./downloads/file.txt" # 本地保存路径
# 执行下载
success = download_file_from_ftp(FTP_HOST, FTP_PORT, FTP_USER, FTP_PASS, REMOTE_PATH, LOCAL_PATH)
# 根据下载结果设置退出码
sys.exit(0 if success else 1)2. 添加错误处理和日志记录
为了让脚本更健壮,我们可以添加更详细的错误处理和日志记录:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
增强版FTP文件下载脚本,包含错误处理和日志记录
"""
from ftplib import FTP, error_perm, error_temp
import os
import sys
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('ftp_download.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
def download_file_with_retry(ftp_host, ftp_port, ftp_user, ftp_pass, remote_path, local_path, max_retries=3):
"""
带重试机制的FTP文件下载
参数:
ftp_host: FTP服务器地址
ftp_port: FTP服务器端口
ftp_user: FTP用户名
ftp_pass: FTP密码
remote_path: 远程文件路径
local_path: 本地保存路径
max_retries: 最大重试次数
"""
retries = 0
last_error = None
while retries < max_retries:
try:
logger.info(f"尝试下载文件 (第 {retries + 1} 次)")
# 连接到FTP服务器
ftp = FTP()
ftp.connect(ftp_host, ftp_port, timeout=30)
ftp.login(ftp_user, ftp_pass)
# 获取文件名
filename = os.path.basename(remote_path)
# 创建本地目录(如果不存在)
local_dir = os.path.dirname(local_path)
if not os.path.exists(local_dir):
os.makedirs(local_dir)
logger.info(f"创建本地目录: {local_dir}")
# 检查远程文件是否存在
try:
file_size = ftp.size(remote_path)
logger.info(f"远程文件大小: {file_size} bytes")
except error_perm as e:
if "550" in str(e): # 550表示文件不存在
raise FileNotFoundError(f"远程文件不存在: {remote_path}")
raise
# 下载文件
logger.info(f"开始下载文件: {filename}")
start_time = datetime.now()
with open(local_path, 'wb') as local_file:
ftp.retrbinary(f'RETR {remote_path}', local_file.write)
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
# 验证文件是否下载完整
local_size = os.path.getsize(local_path)
if local_size == file_size:
logger.info(f"文件下载成功,大小: {local_size} bytes,耗时: {duration:.2f}秒")
ftp.quit()
return True
else:
logger.warning(f"文件大小不匹配,期望: {file_size},实际: {local_size}")
os.remove(local_path) # 删除不完整的文件
raise IOError("文件下载不完整")
except (error_perm, error_temp) as e:
last_error = e
logger.error(f"FTP错误: {str(e)}")
retries += 1
if retries < max_retries:
logger.info(f"等待5秒后重试...")
import time
time.sleep(5)
except Exception as e:
last_error = e
logger.error(f"下载失败: {str(e)}")
retries += 1
if retries < max_retries:
logger.info(f"等待5秒后重试...")
import time
time.sleep(5)
finally:
try:
if 'ftp' in locals():
ftp.quit()
except:
pass
logger.error(f"达到最大重试次数 {max_retries},下载失败")
if last_error:
logger.error(f"最后错误: {str(last_error)}")
return False
if __name__ == "__main__":
# FTP服务器配置
FTP_HOST = "ftp.example.com"
FTP_PORT = 21
FTP_USER = "your_username"
FTP_PASS = "your_password"
# 文件路径配置
REMOTE_PATH = "/path/to/remote/file.txt"
LOCAL_PATH = "./downloads/file.txt"
# 执行下载
success = download_file_with_retry(FTP_HOST, FTP_PORT, FTP_USER, FTP_PASS, REMOTE_PATH, LOCAL_PATH)
sys.exit(0 if success else 1)三、设置开机自启动
Windows系统
在Windows系统中,有多种方法可以实现开机自启动:
方法1:使用启动文件夹
按下Win + R键,输入shell:startup,回车打开启动文件夹
将你的Python脚本或快捷方式放入此文件夹
方法2:使用任务计划程序
按下Win + R键,输入taskschd.msc,回车打开任务计划程序
创建新任务,设置触发器为"计算机启动时"
操作设置为启动你的Python脚本
方法3:修改注册表
按下Win + R键,输入regedit,回车打开注册表编辑器
导航到HKEY_CURRENT_USER\Software\Microsoft\Windows\CurrentVersion\Run
新建字符串值,名称为你的脚本名称,值为Python解释器路径和脚本路径
Linux系统
在Linux系统中,可以通过以下方式实现开机自启动:
方法1:使用crontab
打开终端,输入crontab -e
添加一行:@reboot /usr/bin/python3 /path/to/your/script.py
方法2:使用systemd服务
创建一个新的systemd服务文件:sudo nano /etc/systemd/system/ftp-downloader.service
添加以下内容:
[Unit] Description=FTP Downloader Service After=network.target [Service] ExecStart=/usr/bin/python3 /path/to/your/script.py Restart=always User=your_username [Install] WantedBy=multi-user.target
启用并启动服务:
sudo systemctl daemon-reload sudo systemctl enable ftp-downloader.service sudo systemctl start ftp-downloader.service
macOS系统
在macOS系统中,可以通过以下方式实现开机自启动:
方法1:使用LaunchAgents
创建plist文件:~/Library/LaunchAgents/com.user.ftpdownloader.plist
添加以下内容:
Label com.user.ftpdownloader ProgramArguments /usr/bin/python3 /path/to/your/script.py RunAtLoad KeepAlive
加载plist文件:launchctl load ~/Library/LaunchAgents/com.user.ftpdownloader.plist
四、进阶功能
1. 定时下载
如果需要定时下载而不是仅在开机时下载,可以使用schedule库:
import schedule
import time
def job():
print("执行定时下载任务...")
# 调用下载函数
# 每天凌晨2点执行
schedule.every().day.at("02:00").do(job)
# 每隔1小时执行
schedule.every().hour.do(job)
while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次2. 下载多个文件
修改脚本以支持下载多个文件:
def download_multiple_files(ftp_config, file_list): """ 下载多个文件 参数: ftp_config: FTP配置字典 file_list: 文件路径列表 [(remote_path, local_path), ...] """ results = [] for remote_path, local_path in file_list: success = download_file_from_ftp( ftp_config['host'], ftp_config['port'], ftp_config['user'], ftp_config['pass'], remote_path, local_path ) results.append((remote_path, success)) return results
3. 断点续传
对于大文件,可以实现断点续传功能:
def resume_download(ftp, remote_path, local_path):
"""
断点续传下载
"""
# 检查本地文件是否存在
if os.path.exists(local_path):
local_size = os.path.getsize(local_path)
try:
# 获取远程文件大小
remote_size = ftp.size(remote_path)
# 如果本地文件小于远程文件,继续下载
if local_size < remote_size:
print(f"继续下载,已下载 {local_size}/{remote_size} bytes")
with open(local_path, 'ab') as f:
ftp.retrbinary(f'RETR {remote_path}', f.write, rest=local_size)
return True
elif local_size == remote_size:
print("文件已完整下载")
return True
else:
print("本地文件大于远程文件,重新下载")
os.remove(local_path)
except Exception as e:
print(f"断点续传失败: {str(e)}")
os.remove(local_path)
# 正常下载
with open(local_path, 'wb') as f:
ftp.retrbinary(f'RETR {remote_path}', f.write)
return True五、注意事项
确保FTP服务器的凭据安全,不要在脚本中硬编码敏感信息,考虑使用环境变量或配置文件
处理网络异常和超时情况,确保脚本的稳定性
考虑防火墙和网络策略对FTP连接的影响
对于大文件下载,注意磁盘空间的使用
在生产环境中,建议添加监控和告警机制
通过以上步骤,你可以实现一个可靠的Python脚本,在开机时自动从FTP服务器下载文件。根据你的具体需求,可以进一步扩展和优化这个脚本。