前言

没什么用，闲出来的……

此下载只适用于xy.z格式的文件（x是前缀，y是递增的数字，z是文件扩展名）。

代码

下载

import os
import time
import requests
from tqdm import tqdm

# 记录开始时间，用于计算下载过程的耗时
start_time = time.time()


def download_images(url, path, prefix, start, end, save_path, fill, length, file_extension):
    """
    下载指定范围内的文件。

    参数:

    :param url: 文件在服务器上的基础URL，需要“/”结尾。
    :param path: 文件在服务器上的路径。
    :param prefix: 文件名称的前缀。
    :param start: 开始下载的文件序号。
    :param end: 结束下载的文件序号。
    :param save_path: 文件保存的根目录。
    :param fill: 文件序号的填充字符，可为空。
    :param length: 文件名称的长度。
    :param file_extension: 文件的文件扩展名。

    返回:
    无
    """
    # 构建文件在服务器上的基础URL
    base_url = url
    # 构建文件保存的目录
    save_dir = os.path.join(save_path, path, prefix.rstrip('-'))

    # 创建保存目录，如果已存在则不覆盖
    os.makedirs(save_dir, exist_ok=True)

    # 使用tqdm显示下载进度条
    for ordinal in tqdm(range(start, end + 1), desc="下载进度"):
        # 将文件序号格式化为定义位数字符串
        ordinal_formatted = f"{ordinal:{fill}{length}}"
        # 构建文件的完整URL
        url = f"{base_url}{path}/{prefix}{ordinal_formatted}.{file_extension}"
        # 构建文件的保存路径
        file_save_path = os.path.join(save_dir, f"{prefix}{ordinal_formatted}.{file_extension}")

        try:
            # 发起GET请求获取文件数据
            response = requests.get(url, stream=True)
            # 检查请求是否成功
            response.raise_for_status()

            # 逐块写入文件数据到文件
            with open(file_save_path, 'wb') as f:
                for chunk in response.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)
        except requests.exceptions.RequestException as e:
            # 打印下载失败的文件URL和错误信息
            print(f"Failed to download {url}: {e}")

    # 记录结束时间
    end_time = time.time()
    # 计算下载总耗时
    elapsed_time = end_time - start_time
    # 打印下载耗时
    print(f"下载完成，总耗时: {elapsed_time:.2f} 秒")

上传

import os
from aligo import Aligo


def upload_to_aliyunpan(local_directory, aliyun_folder_path):
    """
    将指定目录下的文件上传到阿里云盘的指定文件夹中。

    :param local_directory: 需要上传的本地文件夹路径
    :param aliyun_folder_path: 阿里云盘上的目标文件夹路径
    """
    # 初始化阿里云盘客户端
    ali = Aligo()

    # 初始化当前文件夹ID为根目录
    current_folder_id = 'root'

    # 通过路径分段创建或获取目标文件夹
    for folder in aliyun_folder_path.split('/'):
        if folder:
            # 尝试获取已存在的文件夹
            target_folder = ali.get_folder_by_path(folder, parent_file_id=current_folder_id)
            if target_folder is None:
                # 如果文件夹不存在，则创建新文件夹
                target_folder = ali.create_folder(name=folder, parent_file_id=current_folder_id)
            # 更新当前文件夹ID为新获取或创建的文件夹ID
            current_folder_id = target_folder.file_id

    # 遍历本地目录下的文件
    for file in os.listdir(local_directory):
        file_path = os.path.join(local_directory, file)
        if os.path.isfile(file_path):
            # 上传文件到当前目标文件夹
            ali.upload_file(file_path, parent_file_id=current_folder_id)
            # 输出文件上传成功的提示信息
            print(f"文件”{file_path}”上传至阿里云盘文件夹'{aliyun_folder_path}'")

示例

from pathlib import Path
from 文件上传.阿里云.aliyun import upload_to_aliyunpan
from 文件下载.图片.main import download_images


url = ""
path = ""
prefix = ""
start = 1
end = 16
save_path = r"D:\Pictures\download"  # 自定义存储路径
fill = ''
length = 3
file_extension = "webp"  # 自定义文件扩展名

# 拼接 local_directory
local_directory = Path(save_path) / path / prefix.rstrip('-')

# 获取最后一部分
image_name = local_directory.name

# 替换 aliyun_path 中的 'aliyun_path' 为 image_name
aliyun_path = 'test/aliyun_path'
aliyun_path = aliyun_path.replace('aliyun_path', image_name)

# 下载图像
download_images(url, path, prefix, start, end, save_path, fill, length, file_extension)

# 询问用户是否上传
upload_choice = input("是否上传到阿里云盘？(yes/no): ").strip().lower()

if upload_choice == 'yes':
    # 上传到阿里云盘
    upload_to_aliyunpan(local_directory, aliyun_path)
    print("上传成功！")
else:
    print("取消上传。")