最近要下一些AI大模型文件，发现用浏览器自带的下载工具下载很慢，想找一个多线程下载器。但是 IDM 的破解资源现在不是很好找，要么就算找到了版本也是很久之前的了。刚好最近在捣鼓 AI ，就尝试用 AI 工具自己写一个，简单调试后，还挺好用的。这里做个记录吧。

源代码如下：
PS：用 pyinstall 打包成 windows 支持的 exe 程序。我自己简单打了个包，可用但没详细调试，如有需要下载使用即可。

import os
import re
import time
import math
import requests
import threading
from queue import Queue
import argparse
import urllib3


# 禁用SSL证书验证警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class Downloader:
    def __init__(self, url, num_threads=8, output_path=None, verify_ssl=False, max_retries=3, initial_timeout=10):
        self.url = url
        self.num_threads = num_threads
        self.output_path = self.auto_detect_filename(
            url) if output_path is None else output_path
        self.file_size = 0
        self.downloaded = 0
        self.lock = threading.Lock()
        self.supports_range = False
        self.verify_ssl = verify_ssl
        self.session = requests.Session()
        self.max_retries = max_retries
        self.initial_timeout = initial_timeout
        self.size_unit = 'B'
        self.size_divisor = 1
        self.start_time = None  # 下载开始时间
        self.last_update_time = None  # 最后进度更新时间
        self.session.mount(
            'http://', requests.adapters.HTTPAdapter(max_retries=0))
        self.session.mount(
            'https://', requests.adapters.HTTPAdapter(max_retries=0))

    def format_size(self, bytes):
        """自动格式化文件大小显示单位"""
        if bytes < 1024:
            return bytes, 'B'
        elif bytes < 1024 * 1024:
            return bytes / 1024, 'KB'
        elif bytes < 1024 * 1024 * 1024:
            return bytes / (1024 * 1024), 'MB'
        else:
            return bytes / (1024 * 1024 * 1024), 'GB'

    def format_time(self, seconds):
        """格式化时间显示"""
        if seconds < 60:
            return f"{seconds:.1f}秒"
        elif seconds < 3600:
            return f"{seconds // 60:.0f}分{seconds % 60:.0f}秒"
        else:
            return f"{seconds // 3600:.0f}时{(seconds % 3600) // 60:.0f}分{seconds % 60:.0f}秒"

    def calculate_speed(self):
        """计算当前下载速度"""
        if not self.start_time or not self.last_update_time:
            return 0, 'B/s'

        elapsed = time.time() - self.start_time
        if elapsed <= 0:
            return 0, 'B/s'

        speed = self.downloaded / elapsed
        if speed < 1024:
            return speed, 'B/s'
        elif speed < 1024 * 1024:
            return speed / 1024, 'KB/s'
        else:
            return speed / (1024 * 1024), 'MB/s'

    def request_with_retry(self, method, **kwargs):
        """带重试机制的请求方法"""
        timeout = self.initial_timeout
        last_exception = None

        for attempt in range(1, self.max_retries + 1):
            try:
                current_timeout = timeout * (1 + 0.5 * (attempt - 1))
                kwargs['timeout'] = current_timeout

                if method.lower() == 'head':
                    resp = self.session.head(**kwargs)
                else:
                    resp = self.session.get(**kwargs)

                resp.raise_for_status()
                return resp

            except requests.exceptions.RequestException as e:
                last_exception = e
                print(
                    f"请求失败 (尝试 {attempt}/{self.max_retries}), 超时时间: {current_timeout:.1f}s - {str(e)}")
                if attempt < self.max_retries:
                    time.sleep(1)

        raise last_exception if last_exception else Exception("未知请求错误")

    # ...（其他方法保持不变，仅修改show_progress方法）
    def check_range_support(self):
        """检测服务器是否支持Range请求"""
        try:
            resp = self.request_with_retry(
                'head',
                url=self.url,
                allow_redirects=True,
                verify=self.verify_ssl
            )
            if resp.headers.get('Accept-Ranges', '').lower() == 'bytes':
                self.supports_range = True
                return True

            headers = {'Range': 'bytes=0-100'}
            resp = self.request_with_retry(
                'get',
                url=self.url,
                headers=headers,
                stream=True,
                verify=self.verify_ssl
            )
            if resp.status_code == 206 and 'Content-Range' in resp.headers:
                self.supports_range = True
                return True

        except Exception as e:
            print(f"Range检测异常: {e}")
        return False

    def auto_detect_filename(self, url):
        """从URL或HTTP头信息自动提取文件名"""
        filename = os.path.basename(url.split('?')[0])
        if not filename or '.' not in filename:
            try:
                resp = self.request_with_retry(
                    'head',
                    url=url,
                    allow_redirects=True,
                    verify=self.verify_ssl
                )
                if 'content-disposition' in resp.headers:
                    match = re.search(r'filename="?(.+?)"?(;|$)',
                                      resp.headers['content-disposition'])
                    if match:
                        filename = match.group(1)
            except:
                pass
        return filename or "downloaded_file"

    def get_file_size(self):
        """获取文件大小并设置显示单位"""
        try:
            resp = self.request_with_retry(
                'head',
                url=self.url,
                allow_redirects=True,
                verify=self.verify_ssl
            )
            self.file_size = int(resp.headers.get('content-length', 0))

            # 设置显示单位
            _, self.size_unit = self.format_size(self.file_size)
            self.size_divisor = 1024 ** {'B': 0,
                                         'KB': 1, 'MB': 2, 'GB': 3}[self.size_unit]

            return True
        except Exception as e:
            print(f"获取文件大小失败: {e}")
            return False

    def download_chunk(self, start, end, chunk_queue):
        """下载文件分块（带重试机制）"""
        headers = {'Range': f'bytes={start}-{end}'} if self.supports_range else {}
        last_exception = None

        for attempt in range(1, self.max_retries + 1):
            try:
                current_timeout = self.initial_timeout * \
                    (1 + 0.5 * (attempt - 1))

                resp = self.session.get(
                    self.url,
                    headers=headers,
                    stream=True,
                    verify=self.verify_ssl,
                    timeout=current_timeout
                )
                resp.raise_for_status()

                chunk = resp.content
                with self.lock:
                    chunk_queue.put((start, chunk))
                return

            except Exception as e:
                last_exception = e
                print(
                    f"分块 {start}-{end} 下载失败 (尝试 {attempt}/{self.max_retries}), 超时: {current_timeout:.1f}s - {str(e)}")
                if attempt < self.max_retries:
                    time.sleep(1)

        if last_exception:
            print(f"分块 {start}-{end} 下载最终失败: {str(last_exception)}")

    def single_thread_download(self):
        """单线程下载实现（带重试机制）"""
        print("服务器不支持Range请求，转为单线程下载...")
        last_exception = None

        for attempt in range(1, self.max_retries + 1):
            try:
                current_timeout = self.initial_timeout * \
                    (1 + 0.5 * (attempt - 1))

                with self.session.get(
                    self.url,
                    stream=True,
                    verify=self.verify_ssl,
                    timeout=current_timeout
                ) as response:
                    response.raise_for_status()

                    with open(self.output_path, 'wb') as f:
                        for chunk in response.iter_content(chunk_size=8192):
                            if chunk:
                                f.write(chunk)
                                self.downloaded += len(chunk)
                                self.show_progress()
                return True

            except Exception as e:
                last_exception = e
                print(
                    f"单线程下载失败 (尝试 {attempt}/{self.max_retries}), 超时: {current_timeout:.1f}s - {str(e)}")
                if attempt < self.max_retries:
                    time.sleep(1)
                else:
                    if os.path.exists(self.output_path):
                        os.remove(self.output_path)

        print(f"单线程下载最终失败: {str(last_exception)}")
        return False

    def save_chunks(self):
        """保存下载的分块数据"""
        with open(self.output_path, 'wb') as f:
            while True:
                chunk = self.chunk_queue.get()
                if chunk is None:
                    break
                start, data = chunk
                f.seek(start)
                f.write(data)
                with self.lock:
                    self.downloaded += len(data)
                    self.show_progress()
                self.chunk_queue.task_done()

    def show_progress(self):
        """显示下载进度（含用时和速度）"""
        if self.file_size <= 0:
            return

        # 初始化计时器
        if self.start_time is None:
            self.start_time = time.time()
            self.last_update_time = self.start_time

        current_time = time.time()
        elapsed = current_time - self.start_time
        self.last_update_time = current_time

        # 计算进度和速度
        percent = (self.downloaded / self.file_size) * 100
        downloaded_size = self.downloaded / self.size_divisor
        total_size = self.file_size / self.size_divisor
        speed, speed_unit = self.calculate_speed()

        # 预估剩余时间
        remaining_time = 0
        if speed > 0:
            remaining_bytes = self.file_size - self.downloaded
            remaining_time = remaining_bytes / \
                (speed *
                 (1024 ** {'B/s': 0, 'KB/s': 1, 'MB/s': 2}[speed_unit]))

        print(
            f"\r下载进度: {percent:.1f}% "
            f"({downloaded_size:.1f}/{total_size:.1f} {self.size_unit}) "
            f"速度: {speed:.1f} {speed_unit} "
            f"用时: {self.format_time(elapsed)} "
            f"剩余: {self.format_time(remaining_time)}",
            end=''
        )

        if self.downloaded >= self.file_size:
            print(f"\n下载完成! 总用时: {self.format_time(elapsed)}")

    def run(self):
        """执行下载流程"""
        print(f"目标文件: {self.output_path}")

        # 检测Range支持情况
        self.supports_range = self.check_range_support()

        # 获取文件大小
        if not self.get_file_size():
            return False

        # 显示启动信息
        total_size, unit = self.format_size(self.file_size)
        print(f"文件大小: {total_size:.1f} {unit}")
        print(f"最大重试次数: {self.max_retries}")
        print(f"初始超时时间: {self.initial_timeout}s")
        print(f"服务器Range支持: {'是' if self.supports_range else '否'}")
        print(f"SSL证书验证: {'已启用' if self.verify_ssl else '已禁用'}")
        print("开始下载...")

        if not self.supports_range:
            return self.single_thread_download()

        print(f"启动多线程下载（线程数: {self.num_threads}）...")
        self.chunk_queue = Queue()
        threads = []

        # 启动写入线程
        writer_thread = threading.Thread(target=self.save_chunks)
        writer_thread.start()

        # 计算分块范围
        chunk_size = self.file_size // self.num_threads
        ranges = [(i * chunk_size, (i + 1) * chunk_size - 1)
                  for i in range(self.num_threads - 1)]
        ranges.append((ranges[-1][1] + 1, self.file_size - 1))

        # 启动下载线程
        for start, end in ranges:
            thread = threading.Thread(
                target=self.download_chunk, args=(start, end, self.chunk_queue))
            thread.start()
            threads.append(thread)

        # 等待下载线程完成
        for thread in threads:
            thread.join()

        # 通知写入线程结束
        self.chunk_queue.put(None)
        writer_thread.join()

        return True


def show_full_help():
    """显示完整的使用帮助信息"""
    print("""
智能多线程下载器 - 完整使用指南

基本语法:
  python downloader.py [选项] URL

必需参数:
  URL                   要下载的文件网址 (必须以http://或https://开头)

常用选项:
  -o PATH, --output PATH   指定输出文件名或路径
  -t N, --threads N        设置下载线程数 (1-32, 默认16)

高级选项:
  --ssl-verify           启用SSL证书验证 (默认不验证)
  --max-retries N        设置最大重试次数 (默认3)
  --initial-timeout N    设置初始超时秒数 (默认10.0)

示例命令:
  1. 基本下载:
     python downloader.py https://example.com/file.zip

  2. 指定输出路径:
     python downloader.py https://example.com/file.zip -o ~/Downloads/myfile.zip

  3. 使用16个线程下载大文件:
     python downloader.py https://example.com/large.iso -t 16

  4. 下载需要SSL验证的文件:
     python downloader.py https://secure.com/data --ssl-verify

  5. 对不稳定连接增加重试次数:
     python downloader.py https://unstable.com/file --max-retries 5 --initial-timeout 30

注意事项:
  - 按Ctrl+C可随时中断下载
  - 中断后会自动清理未完成的临时文件
  - 线程数不是越多越好，建议根据网络状况调整
""")


def main():
    class CustomArgumentParser(argparse.ArgumentParser):
        def error(self, message):
            """自定义错误处理，显示完整帮助"""
            print(f"\n错误: {message}\n")
            show_full_help()
            self.exit(2)

    parser = CustomArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description='''智能多线程下载器（版本: 2.1）

核心功能:
  ✔ 多线程加速下载 (1-32线程)
  ✔ 智能断点续传
  ✔ 实时速度/进度显示
  ✔ 自动重试机制
  ✔ 下载用时统计''',
        epilog='''输入 --help 查看完整帮助信息''',
        add_help=False  # 禁用默认的help，使用自定义的
    )

    # 添加帮助选项
    parser.add_argument(
        '-h', '--help',
        action='store_true',
        help='显示完整帮助信息'
    )

    # 必需参数
    parser.add_argument(
        'url',
        nargs='?',  # 设为可选以便处理--help情况
        help='要下载的文件URL (http/https)'
    )

    # 可选参数
    optional = parser.add_argument_group('常用选项')
    optional.add_argument(
        '-o', '--output',
        metavar='PATH',
        help='自定义输出文件路径'
    )
    optional.add_argument(
        '-t', '--threads',
        type=int,
        default=16,
        help='下载线程数 (1-32, 默认: %(default)s)'
    )

    # 高级选项
    advanced = parser.add_argument_group('高级选项')
    advanced.add_argument(
        '--ssl-verify',
        action='store_true',
        help='启用SSL证书验证'
    )
    advanced.add_argument(
        '--max-retries',
        type=int,
        default=3,
        help='最大重试次数 (默认: %(default)s)'
    )
    advanced.add_argument(
        '--initial-timeout',
        type=float,
        default=10.0,
        help='初始超时秒数 (默认: %(default)s)'
    )

    # 版本信息
    parser.add_argument(
        '-v', '--version',
        action='store_true',
        help='显示版本信息'
    )

    # 解析参数
    args = parser.parse_args()

    # 处理帮助和版本信息
    if args.help:
        show_full_help()
        return

    if args.version:
        print("智能多线程下载器 版本 2.1 (2024-05-15)")
        return

    # 验证必需参数
    if not args.url:
        parser.error("必须指定要下载的URL")

    if not args.url.startswith(('http://', 'https://')):
        parser.error("URL必须以http://或https://开头")

    # 验证参数范围
    if args.threads < 1 or args.threads > 32:
        parser.error("线程数必须在1-32之间")

    if args.max_retries < 0:
        parser.error("重试次数不能为负数")

    if args.initial_timeout <= 0:
        parser.error("超时时间必须大于0")

    # 执行下载
    print(f"\n{' 智能多线程下载器 ':=^60}")
    print(f"正在准备下载: {args.url}")

    downloader = Downloader(
        args.url,
        args.threads,
        args.output,
        verify_ssl=args.ssl_verify,
        max_retries=args.max_retries,
        initial_timeout=args.initial_timeout
    )

    try:
        if downloader.run():
            print(f"\n{' 下载成功 ':=^60}")
            print(f"文件位置: {os.path.abspath(downloader.output_path)}")
            if downloader.file_size > 0:
                size, unit = downloader.format_size(downloader.file_size)
                print(f"文件大小: {size:.2f} {unit}")
                speed, speed_unit = downloader.calculate_speed()
                print(f"平均速度: {speed:.1f} {speed_unit}")
        else:
            print(f"\n{' 下载失败 ':=^60}")
    except KeyboardInterrupt:
        print(f"\n{' 下载已取消 ':=^60}")
        if os.path.exists(downloader.output_path):
            os.remove(downloader.output_path)
            print("已清理未完成下载的文件")
    except Exception as e:
        print(f"\n{' 下载错误 ':=^60}")
        print(f"错误类型: {type(e).__name__}")
        print(f"错误详情: {str(e)}")
        if os.path.exists(downloader.output_path):
            os.remove(downloader.output_path)
            print("已清理未完成下载的文件")
        print("\n建议解决方案:")
        print("- 检查URL是否正确可用")
        print("- 尝试增加 --max-retries 和 --initial-timeout 参数")
        print("- 使用 --ssl-verify 如果服务器需要证书验证")
        print("- 使用 -t 减少线程数（网络不稳定时）")


if __name__ == '__main__':
    main()