如何用Python从air.inc API下载资产并维护目录结构、避免重复下载？

前期准备 #

• 确认air.inc的数据获取方式：先明确是通过API、FTP/SFTP还是预生成文件列表下载，是否需要身份验证（如API密钥、账号密码），以及是否有单文件大小限制、并发请求限制等规则。
• 本地存储规划：预留至少15GB磁盘空间（10GB目标数据+日志/元数据缓存），优先选择IO性能较好的存储路径。

核心实现：去重与下载逻辑 #

1. 重复下载规避机制 #

推荐优先利用air.inc提供的文件元数据（如文件ID、大小、最后修改时间）做快速校验，哈希校验作为兜底方案。

方案1：基于文件元数据校验（高效）

本地维护一个JSON格式的下载记录文件，存储已下载文件的关键信息，每次下载前对比校验：

import json
import os
from datetime import datetime

# 本地下载记录存储路径
DOWNLOAD_RECORD_PATH = "./downloaded_records.json"

def load_downloaded_records():
    """加载已下载文件的记录"""
    if os.path.exists(DOWNLOAD_RECORD_PATH):
        with open(DOWNLOAD_RECORD_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def save_download_record(file_id, local_path, file_size, last_modified):
    """保存下载成功的文件记录"""
    records = load_downloaded_records()
    records[file_id] = {
        "local_path": local_path,
        "file_size": file_size,
        "last_modified": last_modified,
        "download_time": datetime.now().isoformat()
    }
    with open(DOWNLOAD_RECORD_PATH, "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2)

def is_file_downloaded(file_metadata):
    """校验文件是否已下载且未变更"""
    records = load_downloaded_records()
    file_id = file_metadata["file_id"]
    
    if file_id not in records:
        return False
    
    local_record = records[file_id]
    # 校验文件大小和修改时间是否匹配
    if not os.path.exists(local_record["local_path"]):
        return False
    local_file_size = os.path.getsize(local_record["local_path"])
    
    return (local_file_size == file_metadata["file_size"] 
            and local_record["last_modified"] == file_metadata["last_modified"])

方案2：基于哈希校验（兜底）

如果air.inc不提供可靠的文件元数据，可通过计算文件哈希值确保文件完整性与唯一性：

import hashlib

def calculate_file_hash(file_path, chunk_size=4096):
    """计算文件的SHA256哈希值"""
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        while chunk := f.read(chunk_size):
            sha256.update(chunk)
    return sha256.hexdigest()

# 使用示例：下载完成后保存哈希值到记录
# save_download_record(file_id, local_path, file_size, last_modified, calculate_file_hash(local_path))

2. 批量下载实现 #

根据air.inc的下载方式选择对应逻辑，以下是两种常见场景的实现：

场景1：HTTP/API下载（分块+断点续传）

用requests库分块下载，避免内存溢出，同时支持断点续传应对网络中断：

import requests

def download_via_api(file_url, save_path, chunk_size=1024*1024):  # 1MB分块
    # 检查是否已存在部分下载的文件，支持断点续传
    downloaded_size = os.path.getsize(save_path) if os.path.exists(save_path) else 0
    headers = {"Range": f"bytes={downloaded_size}-"} if downloaded_size > 0 else {}
    
    try:
        with requests.get(file_url, stream=True, headers=headers) as resp:
            resp.raise_for_status()
            total_size = downloaded_size + int(resp.headers.get("content-length", 0))
            
            with open(save_path, "ab") as f:
                for chunk in resp.iter_content(chunk_size=chunk_size):
                    f.write(chunk)
                    # 可选：打印下载进度
                    current_size = downloaded_size + f.tell()
                    print(f"Downloading: {current_size}/{total_size} bytes", end="\r")
        return True
    except Exception as e:
        print(f"Download failed: {str(e)}")
        return False

场景2：FTP/SFTP下载

如果是FTP协议，用ftplib实现；SFTP则用paramiko：

# FTP下载示例
from ftplib import FTP

def download_via_ftp(ftp_host, ftp_user, ftp_pass, remote_file_path, local_save_path):
    try:
        with FTP(ftp_host) as ftp:
            ftp.login(ftp_user, ftp_pass)
            # 检查远程文件大小（可选，用于去重）
            remote_size = ftp.size(remote_file_path)
            if os.path.exists(local_save_path) and os.path.getsize(local_save_path) == remote_size:
                print(f"File already exists: {local_save_path}")
                return True
            
            with open(local_save_path, "wb") as f:
                ftp.retrbinary(f"RETR {remote_file_path}", f.write)
        return True
    except Exception as e:
        print(f"FTP download failed: {str(e)}")
        return False

3. 任务调度与日志 #

• 多线程批量下载：利用concurrent.futures开启多线程，提高下载效率（注意不要超过air.inc的并发限制）：

from concurrent.futures import ThreadPoolExecutor

def batch_download_files(file_metadata_list, max_workers=4):
    """批量下载文件，自动跳过已下载的文件"""
    records = load_downloaded_records()
    
    def _download_task(file_meta):
        if is_file_downloaded(file_meta):
            print(f"Skipping already downloaded file: {file_meta['file_id']}")
            return
        # 根据下载方式调用对应的下载函数，这里以API为例
        success = download_via_api(file_meta["url"], file_meta["local_path"])
        if success:
            save_download_record(file_meta["file_id"], file_meta["local_path"], 
                                file_meta["file_size"], file_meta["last_modified"])
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        executor.map(_download_task, file_metadata_list)

• 日志记录：用Python内置logging模块记录下载状态，方便排查问题：

import logging

logging.basicConfig(
    filename="./download_logs.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

# 在下载成功/失败时添加日志
# logging.info(f"Successfully downloaded file: {file_meta['file_id']}")
# logging.error(f"Failed to download file {file_meta['file_id']}: {str(e)}")

4. 性能优化建议 #

• 控制并发数：根据air.inc的服务限制调整max_workers，避免因请求过多被限流。
• 批量更新记录：将已下载记录加载到内存，批量完成下载后再一次性写入文件，减少磁盘IO操作。
• 优先下载小文件：避免大文件长时间占用线程，可按文件大小排序后下载。

内容的提问来源于stack exchange，提问作者Calm Rain