本文目录导读：
我来帮你写一个增量文件下载脚本,以下是两种常用实现方式：
基于文件大小和修改时间的简单增量下载
import os
import hashlib
import json
import requests
from pathlib import Path
from urllib.parse import urlparse
class IncrementalDownloader:
def __init__(self, download_dir="downloads", state_file=".download_state.json"):
self.download_dir = Path(download_dir)
self.state_file = Path(state_file)
self.state = self._load_state()
self.download_dir.mkdir(exist_ok=True)
def _load_state(self):
"""加载下载状态"""
if self.state_file.exists():
with open(self.state_file, 'r') as f:
return json.load(f)
return {}
def _save_state(self):
"""保存下载状态"""
with open(self.state_file, 'w') as f:
json.dump(self.state, f, indent=2)
def _get_remote_file_info(self, url):
"""获取远程文件信息"""
try:
response = requests.head(url, allow_redirects=True)
response.raise_for_status()
info = {
'size': int(response.headers.get('content-length', 0)),
'last_modified': response.headers.get('last-modified', ''),
'etag': response.headers.get('etag', '')
}
return info
except Exception as e:
print(f"获取远程文件信息失败: {e}")
return None
def _get_local_file_info(self, filepath):
"""获取本地文件信息"""
if not filepath.exists():
return None
return {
'size': filepath.stat().st_size,
'last_modified': filepath.stat().st_mtime
}
def _calculate_file_hash(self, filepath):
"""计算文件哈希值"""
sha256_hash = hashlib.sha256()
with open(filepath, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
def download_file(self, url, filename=None, verify_hash=False):
"""下载单个文件"""
if not filename:
filename = url.split('/')[-1]
local_filepath = self.download_dir / filename
remote_info = self._get_remote_file_info(url)
if not remote_info:
return False
# 检查是否需要增量下载
state_key = url
previous_state = self.state.get(state_key, {})
needs_download = True
if local_filepath.exists():
local_info = self._get_local_file_info(local_filepath)
# 比较文件大小
if local_info and remote_info['size'] == local_info['size']:
# 比较修改时间或ETag
if remote_info.get('etag') and previous_state.get('etag') == remote_info['etag']:
needs_download = False
elif remote_info.get('last_modified') and previous_state.get('last_modified') == remote_info['last_modified']:
needs_download = False
if needs_download:
print(f"正在下载: {filename}")
try:
response = requests.get(url, stream=True)
response.raise_for_status()
# 支持断点续传
if local_filepath.exists() and previous_state.get('partial'):
# 获取已下载的大小
existing_size = local_filepath.stat().st_size
headers = {'Range': f'bytes={existing_size}-'}
response = requests.get(url, headers=headers, stream=True)
mode = 'ab'
print(f"继续下载: {filename} (从 {existing_size} 字节继续)")
else:
mode = 'wb'
with open(local_filepath, mode) as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
# 可选：验证文件完整性
if verify_hash:
file_hash = self._calculate_file_hash(local_filepath)
print(f"文件哈希值: {file_hash}")
# 更新状态
self.state[state_key] = {
'etag': remote_info['etag'],
'last_modified': remote_info['last_modified'],
'size': remote_info['size'],
'partial': False,
'timestamp': str(Path(local_filepath).stat().st_mtime)
}
self._save_state()
print(f"下载完成: {filename}")
return True
except Exception as e:
print(f"下载失败 {filename}: {e}")
# 保存部分下载状态
if local_filepath.exists():
self.state[state_key] = {'partial': True, 'size': local_filepath.stat().st_size}
self._save_state()
return False
else:
print(f"文件已是最新: {filename}")
return True
def download_multiple_files(self, file_list, verify_hash=False):
"""下载多个文件"""
results = []
for item in file_list:
if isinstance(item, dict):
url = item.get('url')
filename = item.get('filename')
else:
url = item
filename = None
if url:
success = self.download_file(url, filename, verify_hash)
results.append(success)
return all(results)
基于 Rsync 的增量同步（适合大文件）
import subprocess
import os
from pathlib import Path
class RsyncIncrementalDownloader:
def __init__(self, remote_host=None, remote_port=22, ssh_key=None):
self.remote_host = remote_host
self.remote_port = remote_port
self.ssh_key = ssh_key
def download_using_rsync(self, remote_path, local_path, options=None):
"""使用rsync进行增量下载"""
local_path = Path(local_path)
local_path.mkdir(exist_ok=True, parents=True)
# 构建rsync命令
rsync_cmd = ['rsync', '-avz', '--progress']
# 添加增量更新选项
increment_options = [
'--update', # 跳过已经更新的文件
'--partial', # 保留部分下载的文件
'--inplace', # 原地更新文件（节省空间）
]
rsync_cmd.extend(increment_options)
# 添加自定义选项
if options:
rsync_cmd.extend(options)
# 添加SSH选项
if self.remote_host:
ssh_options = f'ssh -p {self.remote_port}'
if self.ssh_key:
ssh_options += f' -i {self.ssh_key}'
rsync_cmd.extend(['-e', ssh_options])
# 构建远程路径
remote_url = f'{self.remote_host}:{remote_path}'
else:
remote_url = remote_path
rsync_cmd.extend([remote_url, str(local_path)])
print(f"执行命令: {' '.join(rsync_cmd)}")
try:
result = subprocess.run(rsync_cmd, capture_output=True, text=True)
print(result.stdout)
if result.returncode == 0:
print("同步完成!")
return True
else:
print(f"同步失败: {result.stderr}")
return False
except Exception as e:
print(f"执行rsync失败: {e}")
return False
使用示例
# 示例1：使用HTTP增量下载
downloader = IncrementalDownloader(download_dir="my_files")
# 单个文件下载
downloader.download_file("https://example.com/file.zip")
# 多个文件下载
files_to_download = [
{"url": "https://example.com/file1.zip", "filename": "file1.zip"},
{"url": "https://example.com/file2.jpg", "filename": "photo.jpg"},
"https://example.com/file3.pdf" # 自动使用文件名
]
downloader.download_multiple_files(files_to_download)
# 示例2：使用Rsync增量同步
rsync_downloader = RsyncIncrementalDownloader(
remote_host="user@example.com",
remote_port=22,
ssh_key="/path/to/ssh/key"
)
# 同步远程目录到本地
rsync_downloader.download_using_rsync(
"/remote/path/to/files/",
"/local/download/directory/"
)
关键特性
- 增量检测：通过比较文件大小、修改时间、ETag 判断是否需要下载
- 断点续传：支持部分下载文件的续传
- 文件验证：可选的文件哈希验证
- 状态持久化：保存下载状态到JSON文件
- Rsync支持：适合大量文件或大文件的增量同步
选择哪种方案取决于你的具体需求：HTTP下载适合网络文件，Rsync适合目录同步和本地/远程服务器之间的传输。
