大文件并发分片下载

程序的主要功能是通过多线程的方式下载一个文件，并将下载进度条显示给用户。下面是对其各部分功能的详细总结：

1. 主要功能：

1.1. 并行分段下载文件

使用 requests 库的 Range 请求头，通过多个线程并行下载不同的文件片段，将这些片段组合起来完成整个文件的下载。
下载过程中使用 ThreadPoolExecutor 来管理线程，并行化下载以提高下载效率。

1.2. 显示下载进度

通过 tqdm 库为每个线程显示一个进度条，追踪每个线程下载的文件片段大小，并在主进度条中汇总所有线程的下载进度。

1.3. 断点续传

使用 .fetch 文件来追踪下载进度，如果文件已经部分下载，再次运行程序时可以从断点续传，而不必重新下载整个文件。

2. 关键模块与功能解释：

2.1. 文件的分段请求 (fetch_one)

根据文件的总大小，将下载任务分成多个段，每个段交给不同的线程处理。
使用 Range 头部请求文件的特定字节范围，确保每个线程下载对应的文件片段。

2.2. 多线程下载 (insert_data)

使用 ThreadPoolExecutor 来创建多个线程，进行并发的文件片段下载。
每个线程负责从指定的开始位置下载到结束位置，并将下载的数据写入文件的相应部分。

2.3. 显示下载进度 (tqdm)

对每个线程使用单独的 tqdm 进度条来显示下载进度，并汇总显示整个文件的总进度。下载过程中通过 pbar_thread.update() 来更新进度条的显示，便于用户了解下载的实时进度。

2.4. 断点续传

当下载中断时，通过保存文件进度在 .fetch 文件中，可以在下次运行时从中断的位置继续下载。

2.5. 错误处理

通过 try-except 捕获异常，并使用 loguru 记录错误信息。当下载失败时，程序不会崩溃，可以记录错误以便后续分析和处理。

3. 总体流程：

获取文件大小：首先通过 get_file_length() 获取文件的总大小。
创建文件占位：如果目标文件还未存在，则预先创建一个与下载文件大小一致的文件（填充空字节），以便可以使用 seek 精确写入下载的文件片段。
分段下载：使用 ThreadPoolExecutor 将下载任务分配给多个线程，线程根据 Range 头部请求文件片段，写入相应部分。
显示进度：通过 tqdm 显示下载的进度。
下载完成：下载结束后，删除 .fetch 文件并提示下载成功。

4. 优化点：

错误处理：可以进一步优化每个线程的错误处理，避免某个线程的失败导致整体下载失败。断点续传功能扩展：目前 .fetch 文件的实现较为简单，后续可以将每个片段的下载状态保存到 .fetch 文件中，保证下载中断时能更智能地恢复下载。

This is a tip

一个技巧横幅

或者

This is a tip

一个技巧横幅

 1import requests
 2from concurrent.futures import ThreadPoolExecutor
 3import os
 4from pathlib import Path
 5from loguru import logger
 6import traceback
 7from tqdm import tqdm
 8from threading import Lock
 9from functools import partial
10
11# url = "https://scontent.xx.fbcdn.net/m1/v/t6/An_YmP5OIPXun-vu3hkckAZZ2s4lPYoVkiyvCcWiVY21mu1Ng5_1HeCa2CWiSTsskj8HQ8bN013HxNpYDdSC_7jWQq_svcg.tar?ccb=10-5&oh=00_AfBn8XrMhiHu6w1KuS1X8rkuLzzZJnRs8B9jFMvVRfQnfg&oe=64659C28&_nc_sid=fb0754"
12#
13# tar_path = Path("/Users/chennan/Desktop/sa_000020.tar")
14# fetching_path = Path(f"{tar_path.as_posix()}.fetch")
15lock = Lock()
16downloaded = 0
17url = "https://images.pexels.com/photos/15983035/pexels-photo-15983035.jpeg"
18tar_path = Path("/Users/chennan/Desktop/pexels-photo-15983035.jpeg")
19fetching_path = Path(f"{tar_path.as_posix()}.fetch")
20pbar_threads = []
21
22def insert_data(info):
23    headers, start, end = info
24    pbar_thread = tqdm(total=end - start)
25    pbar_threads.append(pbar_thread)
26    with requests.get(url, stream=True, headers=headers) as response:
27        with tar_path.open('rb+') as f:  # path 文件保存路径
28            # 从文件的start位置开始写入
29            f.seek(start)
30            for chunk in response.iter_content(chunk_size=1024):
31                if chunk:
32                    f.write(chunk)
33                # pbar.update(len(chunk))
34                pbar_thread.update(len(chunk))
35            pbar_thread.close()
36
37
38def get_file_length():
39    # 改为requests.head可能更高效，待测试。
40    req = requests.get(url, stream=True)
41    return int(req.headers['content-Length'])
42
43
44def fetch_one(content_length):
45    all_thread = 64  # 线程数量
46    part = content_length // all_thread  # 每个线程请求的大小
47    for i in range(all_thread):
48        # 每个线程开始， 结束爬取的位置
49        start = part * i
50        if i == all_thread - 1:
51            end = content_length
52        else:
53            end = start + part
54        if i > 0:
55            start += 1
56        # 设置从网站请求开始和结束的位置
57        headers = {
58            'Range': f'bytes={start}-{end}',
59        }
60        yield headers, start, end
61
62
63if __name__ == '__main__':
64    content_length = get_file_length()
65    if not tar_path.exists():
66        with tar_path.open('wb') as f:
67            f.seek(content_length - 1)
68            f.write(b'\0')
69    if not fetching_path.exists():
70        with fetching_path.open("wb") as fs:
71            fs.write(b"\0")
72
73    try:
74        with ThreadPoolExecutor(max_workers=16) as pool:
75            pool.map(insert_data, fetch_one(content_length))
76        with tqdm(total=content_length) as pbar:
77            progress = sum(pbar_thread.n for pbar_thread in pbar_threads)
78            pbar.update(progress)
79        fetching_path.unlink()
80        logger.info("下载完成")
81    except Exception:
82        logger.error(f"{traceback.format_exc()}")