怎样实现增量文件下载脚本

本文目录导读：

1. 方案一：基于文件大小和修改时间的简单增量下载
2. 方案二：基于 Rsync 的增量同步（适合大文件）
3. 使用示例
4. 关键特性

我来帮你写一个增量文件下载脚本,以下是两种常用实现方式：

基于文件大小和修改时间的简单增量下载

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/"
)

关键特性

1. 增量检测：通过比较文件大小、修改时间、ETag 判断是否需要下载
2. 断点续传：支持部分下载文件的续传
3. 文件验证：可选的文件哈希验证
4. 状态持久化：保存下载状态到JSON文件
5. Rsync支持：适合大量文件或大文件的增量同步

选择哪种方案取决于你的具体需求：HTTP下载适合网络文件，Rsync适合目录同步和本地/远程服务器之间的传输。