Python 多线程下载工具：支持 HTTP、FTP 和 磁力链接

import os
import argparse
import requests
import logging
import re
import transmissionrpc
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
from filelock import FileLock


def check_dir(path):
    if not os.path.exists(path):
        os.makedirs(path)
    elif not os.path.isdir(path):
        raise argparse.ArgumentTypeError(f'{path} is not a directory')
    return path

def check_url(url):
    pattern = r'^https?://|^ftp://|^magnet:'
    if not re.match(pattern, url):
        raise argparse.ArgumentTypeError(f'{url} is not a valid url')
    return url

def download_file(url, filename, num_threads=4):
    response = requests.head(url)
    size = int(response.headers.get('Content-Length'))
    chunk_size = int(size / num_threads) + 1

    with open(filename, 'wb') as f:
        with tqdm(total=size, unit='B', unit_scale=True, desc=os.path.basename(filename), ncols=80, position=0) as bar:
            futures = []
            for i in range(num_threads):
                start = i * chunk_size
                end = min(start + chunk_size, size)
                futures.append((url, start, end, f, bar))

            with ThreadPoolExecutor(max_workers=num_threads) as executor:
                for future in as_completed(executor.submit(download_range, *future) for future in futures):
                    try:
                        future.result()
                    except Exception as e:
                        logging.error(f'Download failed: {e}')

    if os.path.getsize(filename) != size:
        os.remove(filename)
        raise Exception(f'Download incomplete: {filename}')
    logging.info(f'Download completed: {filename}')

def download_range(url, start, end, fileobj, bar):
    headers = {'Range': f'bytes={start}-{end-1}'}
    with requests.get(url, headers=headers, stream=True) as response:
        with open(fileobj.name, 'rb+') as f:
            with FileLock(f'{fileobj.name}.lock'):
                f.seek(start)
                for chunk in response.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)
                        bar.update(len(chunk))

def download_ftp(url, filename, num_threads=4):
    with requests.get(url, stream=True) as response:
        total_size = response.headers.get('Content-Length')
        if total_size:
            total_size = int(total_size.strip())
            with tqdm(total=total_size, unit='B', unit_scale=True, desc=os.path.basename(filename), ncols=80, position=0) as bar:
                with open(filename, 'wb') as f:
                    chunk_size = int(total_size / num_threads) + 1
                    futures = []
                    for i in range(num_threads):
                        start = i * chunk_size
                        end = min(start + chunk_size, total_size)
                        futures.append((url, start, end, f, bar))

                    with ThreadPoolExecutor(max_workers=num_threads) as executor:
                        for future in as_completed(executor.submit(download_range_ftp, *future) for future in futures):
                            try:
                                future.result()
                            except Exception as e:
                                logging.error(f'Download failed: {e}')
            if os.path.getsize(filename) != total_size:
                os.remove(filename)
                raise Exception(f'Download incomplete: {filename}')
            logging.info(f'Download completed: {filename}')
        else:
            with open(filename, 'wb') as f:
                f.write(response.content)
            logging.info(f'Download completed: {filename}')

def download_range_ftp(url, start, end, fileobj, bar):
    headers = {'Range': f'bytes={start}-{end-1}'}
    with requests.get(url, headers=headers, stream=True) as response:
        with open(fileobj.name, 'rb+') as f:
            with FileLock(f'{fileobj.name}.lock'):
                f.seek(start)
                for chunk in response.iter_content(chunk_size=1024):
                    if chunk:
                        f.write(chunk)
                        bar.update(len(chunk))

def download_torrent(magnet, filename):
    tc = transmissionrpc.Client('localhost', port=9091)
    tc.add_torrent(magnet, download_dir=os.path.dirname(filename))
    logging.info(f'Download started: {filename}')

def main():
    parser = argparse.ArgumentParser(description='A command-line tool for downloading files.')
    parser.add_argument('-u', '--url', metavar='<URL>', type=check_url, required=True, help='download url')
    parser.add_argument('-o', '--output', metavar='<FILENAME>', type=str, required=True, help='output filename')
    parser.add_argument('-t', '--threads', metavar='<NUM_THREADS>', type=int, default=4, help='number of threads for downloading')
    parser.add_argument('-d', '--dir', metavar='<DIRECTORY>', type=check_dir, default='.', help='output directory')
    args = parser.parse_args()
    url, filename, num_threads, output_dir = args.url, args.output, args.threads, args.dir

    filename = os.path.join(output_dir, filename)
    if os.path.isfile(filename):
        i = 1
        while True:
            new_filename = f'{os.path.splitext(filename)[0]}_{i}{os.path.splitext(filename)[1]}'
            if not os.path.isfile(new_filename):
                filename = new_filename
                break
            i += 1

    if url.startswith('http'):
        download_file(url, filename, num_threads)
    elif url.startswith('ftp'):
        download_ftp(url, filename, num_threads)
    elif url.startswith('magnet'):
        download_torrent(url, filename)
    else:
        logging.error(f'The url {url} does not start with http, ftp or magnet.')

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
    main()

问题分析

代码中使用 transmissionrpc 客户端连接本地 Transmission Daemon 时，出现 ConnectionRefusedError 错误，表明连接失败。这可能是以下原因造成的：

1. Transmission Daemon 未运行： 确保 Transmission Daemon 已经启动并正常运行。
2. 端口号错误： 确认 Transmission Daemon 监听的端口号是否为 9091。默认情况下，Transmission Daemon 监听 9091 端口，但你可能修改了配置文件中的端口号。

改进建议

1. 检查 Transmission Daemon 运行状态: 在运行代码前，先确保 Transmission Daemon 已经启动并正常运行。可以通过查看 Transmission Daemon 的进程或访问其 web 界面来确认其运行状态。
2. 检查端口号: 确认 Transmission Daemon 监听的端口号是否为 9091。如果端口号不同，需要修改代码中的端口号。
3. 使用其他 BitTorrent 客户端： 如果无法解决上述问题，可以尝试使用其他 BitTorrent 客户端，例如 qBittorrent 或 Deluge。
4. 使用 Magnet 链接下载工具： 可以使用 Magnet 链接下载工具直接下载种子文件，例如迅雷或 aria2。
5. 添加断点续传功能： 代码中没有处理下载中断的情况，可以添加断点续传功能来提高下载的可靠性。具体实现方法可以参考网络上的相关教程。

改进后的代码

# ... (其他代码相同)

def download_torrent(magnet, filename):
    try:
        tc = transmissionrpc.Client('localhost', port=9091)
        tc.add_torrent(magnet, download_dir=os.path.dirname(filename))
        logging.info(f'Download started: {filename}')
    except transmissionrpc.error.TransmissionError as e:
        logging.error(f'Download failed: {e}')
        # 可以尝试使用其他 BitTorrent 客户端或 Magnet 链接下载工具

# ... (其他代码相同)

总结

通过检查 Transmission Daemon 的运行状态和端口号，可以解决代码中出现的连接错误。如果仍然无法解决问题，可以尝试使用其他 BitTorrent 客户端或 Magnet 链接下载工具。另外，添加断点续传功能可以提高下载的可靠性。