记录维维的每一篇Hello World

previewFix.jpg

需要用到的库

  1. import json
  2. import tqdm
  3. import typing
  4. import os
  5. import requests
  6. import retry
  7. import signal
  8. import multitasking

完整源码

  1. # import time
  2. import json
  3. from tqdm import tqdm
  4. from typing import List, Tuple
  5. import os
  6. import requests
  7. from retry import retry
  8. import signal
  9. import multitasking
  11. signal.signal(signal.SIGINT, multitasking.killall)
  14. class Download:
  15.     def __init__(self):
  16.         """
  17.         多线程下载文件
  18.         """
  19.         self.__file_size = None     # 文件大小/单位b
  20.         self.__file_url = None  # 文件路径
  21.         self.__MB = pow(1024, 2)  # 1MB == 1024^2B
  22.         self.savePath = "D:\\AppData\\Download"  # 保存路径
  23.         self.__headers = {
  24.             'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
  25.                           'Chrome/99.0.4844.74 Safari/537.36 Edg/99.0.1150.46'
  26.         }
  27.         self.__split_max = 20  # 分块数
  28.         self.__file_min_mb = 10  # 每块最小内存
  29.         self.__retry_tries = 3  # 每次连接失败重试次数
  31.     @property
  32.     def split_max(self):
  33.         """
  34.         最大线程数
  35.         :return:
  36.         """
  37.         return self.__split_max
  39.     @split_max.setter
  40.     def split_max(self, value):
  41.         """
  42.         设置最大线程数
  43.         :param value:
  44.         :return:
  45.         """
  46.         if isinstance(value, int) or value.isdigit():
  47.             self.__split_max = int(value)
  48.         else:
  49.             raise ValueError("split_max value not is int")
  51.     def download(self, file_url: str, file_name: str):
  52.         """
  53.         下载文件
  54.         :param file_url: 文件链接
  55.         :param file_name: 文件保存名
  56.         :return:
  57.         """
  58.         # 如果保存目录不存在，创建
  59.         if not os.path.exists(self.savePath):
  60.             os.makedirs(self.savePath)
  61.         self.__file_url = file_url
  62.         # 获取文件大小/单位b
  63.         self.__file_size = self.__get_file_size()
  64.         self.__download(file_name=file_name)
  66.     def __download(self, file_name):
  67.         @retry(tries=self.__retry_tries)
  68.         @multitasking.task
  69.         def split_download(start, ending):
  70.             """
  71.             分块下载
  72.             :param start: 开头
  73.             :param ending: 结尾
  74.             :return:
  75.             """
  76.             # 拷贝，避开线程之间的影响
  77.             part_headers = self.__headers.copy()
  78.             # 分段下载的核心/设置下载的文件部分
  79.             part_headers['Range'] = 'bytes={start}-{ending}'.format(start=start, ending=ending)
  80.             # 发起请求并获取响应（流式）
  81.             response = session.get(self.__file_url, headers=part_headers, stream=True)
  82.             # 每次读取的流式响应大小/128b
  83.             data_size = 128
  84.             # 暂存已获取的数据
  85.             datas = []
  86.             for data in response.iter_content(chunk_size=data_size):
  87.                 # 暂存获取的响应
  88.                 datas.append(data)
  89.                 # 更新进度条
  90.                 download_tqdm.update(data_size)
  91.             file.seek(start)
  92.             for data in datas:
  93.                 file.write(data)
  94.             # 释放已写入的资源
  95.             del datas
  97.         # session保持浏览器状态
  98.         session = requests.Session()
  99.         # 打开文件
  100.         file = open(os.path.join(self.savePath, file_name), "wb")
  101.         # 创建进度条
  102.         download_tqdm = tqdm(total=self.__file_size, desc=f'下载文件[{file_name}]')
  103.         # 开始分线下载
  104.         for part in self.__split_list():
  105.             split_download(part[0], part[1])
  106.         # 等待全部线程结束
  107.         multitasking.wait_for_tasks()
  108.         file.close()
  109.         download_tqdm.close()
  111.     def __split_list(self) -> List[Tuple[int, int]]:
  112.         """
  113.         获取分块列表
  114.         :return:
  115.         """
  116.         # 获取文件大小/单位mb
  117.         file_size = self.__file_size / self.__MB
  118.         # 判断文件大小是否小于最小分块内存
  119.         if file_size < self.__file_min_mb:
  120.             return [(0, self.__file_size)]
  121.         # 获取实际线程数
  122.         split_sum = round(file_size / self.__file_min_mb)
  123.         split_sum = self.__split_max if split_sum > self.__split_max else split_sum
  124.         # 获取每块下载的内存大小
  125.         split_size = self.__file_size // split_sum
  127.         # print("file_size:", self.__file_size)
  128.         # print("file_MB:", file_size)
  129.         # print("split_sum:", split_sum)
  130.         # print("split_size:", split_size)
  131.         # print("split_size_mb:", split_size / self.__MB)
  133.         # 分块列表
  134.         split_list = []
  135.         # 获取分块列表
  136.         for start in range(0, self.__file_size, split_size):
  137.             if start + 2 * split_size > self.__file_size:
  138.                 ending = self.__file_size
  139.                 split_list.append((start, ending))
  140.                 break
  141.             else:
  142.                 ending = start + split_size
  143.                 split_list.append((start, ending))
  144.         return split_list
  146.     def __get_file_size(self):
  147.         """
  148.         获取文件大小/单位b
  149.         :return:
  150.         """
  151.         response = requests.head(self.__file_url)
  152.         file_size = response.headers.get('Content-Length')
  153.         # print(json.dumps(dict(response.headers), indent=4, ensure_ascii=False))
  154.         if file_size is None:
  155.             raise ValueError('该文件不支持多线程分段下载！')
  156.         return eval(file_size)

示例代码

  1. if __name__ == '__main__':
  2.     # Anaconda3
  3.     # 文件下载路径
  4.     url = "https://repo.anaconda.com/archive/Anaconda3-2021.11-Windows-x86_64.exe"
  5.     # 文件保存名字
  6.     file_name = "Anaconda3-2021.11-Windows-x86_64.exe"
  7.     # 实例化
  8.     d = Download()
  9.     # 设置最大线程数
  10.     # d.split_max = 36
  11.     d.download(file_url=url, file_name=file_name)

注意事项

1、使用时更改文件下载路径和文件保存的名字

2、文件下载后默认保存在”D:\AppData\Download”下（可以自行更改）

3、下载速度和网速、线程数相关，小于10M的文件单线程下载（可以自行更改）

4、当网速不佳时，开启过多的线程反而会比正常下载更慢

python环境

电脑：Windows10

python：3.10.4