#适用场景#

很多用户需要下载大量的图片或是视频，目前影刀内部指令只支持异步下载图片，如果有其他压缩包或者视频的需求那么就可以用到这窜代码去并发下载其他类型的文件啦。

#首先附上使用截图一张#（咋用 ）

其实你只需要准备：
URL 列表（要下的链接） + 下载文件夹路径即可

#参数说明#

1. url_list:用于传入下载链接的列表
2. download_folder:传入文件下载的文件夹地址（若文件夹不存在会自动创建）
3. mode:下载模式分别有smart,fast,ultra,async四种模式
4. filename_list:可不填，用于给下载下来的文件列表命名，比如：0.mp4，1.jpg等等，默认为None则自动以0,1,2,3,4命名,文件后缀会自动识别
5. max_workers:最大线程数，选填，默认为None，不同模式会自动选择不同线程数，后续会细讲。
6. 使用异步模式时有max_concurrent，是最大并发数。

#接下来具体讲讲不同模式#

mode中分为以下四种模式：

• smart：通过计算自动评估文件数量选择最优线程（已更新计算逻辑，会自动获取电脑进程数选择合理最大进程数限制）
• fast：固定最大并发20个线程限制
• ultra：固定最大高并发60线程限制（CPU线程数要求较高，普通电脑不建议用，适合服务器，目前主流最顶级CPU线程数也达不到这个限制）
• async：真异步下载（已更新会自动计算最大并发数也可以手动指定，非服务器或者工作站也可以使用，默认限制是100）

#只想并发无脑拉smart即可#

#最后当然是附上源码啦#

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv
import os
import time
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import asyncio
import aiohttp
import aiofiles
from aiohttp import TCPConnector
import multiprocessing

# -------------------------------------------------------------------
# 线程数计算
# -------------------------------------------------------------------
def calculate_optimal_workers(file_count, max_workers=32):
    """ 根据文件数量和 CPU 核心数自动计算最优线程数 """
    cpu_threads = multiprocessing.cpu_count()
    max_allowed = min(cpu_threads * 2, max_workers)

    if file_count <= 5:
        return min(file_count, max_allowed)
    elif file_count <= 20:
        return min(8, max_allowed)
    elif file_count <= 50:
        return min(16, max_allowed)
    elif file_count <= 100:
        return min(24, max_allowed)
    else:
        return max_allowed

# -------------------------------------------------------------------
# 单文件下载函数（同步）
# -------------------------------------------------------------------
def download_file(url, filename, download_folder, session, retries=3):
    """ 单个文件下载，支持失败重试 """
    for attempt in range(retries):
        try:
            response = session.get(url, stream=True, timeout=30)
            response.raise_for_status()

            os.makedirs(download_folder, exist_ok=True)
            file_path = os.path.join(download_folder, filename)

            with open(file_path, "wb") as f:
                for data in response.iter_content(chunk_size=65536):
                    if data:
                        f.write(data)

            return True, None

        except Exception as e:
            error_msg = str(e)
            if attempt < retries - 1:
                time.sleep(2 ** attempt)
            else:
                print(f"下载失败：{filename} - {error_msg}")
                return False, error_msg

# -------------------------------------------------------------------
# 多线程下载
# -------------------------------------------------------------------
def multithread_download(url_list, filename_list, download_folder, max_workers=None, retries=3):
    result = {
        "success_count": 0,
        "fail_count": 0,
        "failed_items": []
    }

    if not url_list:
        print("没有可处理的 URL。")
        return result

    file_count = len(url_list)
    if max_workers is None:
        max_workers = calculate_optimal_workers(file_count)

    print(f"开始多线程下载")
    print(f"CPU 核心数: {multiprocessing.cpu_count()}，线程数: {max_workers}，文件数量: {file_count}")

    with requests.Session() as session:
        session.headers.update({"User-Agent": "Mozilla/5.0"})

        with ThreadPoolExecutor(max_workers=max_workers) as ex:
            tasks = {
                ex.submit(download_file, url_list[i], filename_list[i], download_folder, session, retries): i
                for i in range(file_count)
            }

            for future in as_completed(tasks):
                idx = tasks[future]
                ok, reason = future.result()

                if ok:
                    result["success_count"] += 1
                else:
                    result["fail_count"] += 1
                    result["failed_items"].append({
                        "url": url_list[idx],
                        "filename": filename_list[idx],
                        "reason": reason
                    })

    print(f"完成。成功 {result['success_count']} 个，失败 {result['fail_count']} 个。")
    return result

# -------------------------------------------------------------------
# 异步下载（全异步 aiohttp + aiofiles）
# -------------------------------------------------------------------
async def async_download_one(session, url, filename, download_folder, semaphore):
    async with semaphore:
        try:
            async with session.get(url) as resp:
                resp.raise_for_status()
                os.makedirs(download_folder, exist_ok=True)
                path = os.path.join(download_folder, filename)

                # 异步写文件
                async with aiofiles.open(path, "wb") as f:
                    async for chunk in resp.content.iter_chunked(65536):
                        await f.write(chunk)

            print(f"完成：{filename}")
            return True, None

        except Exception as e:
            error_msg = str(e)
            print(f"失败：{filename} - {error_msg}")
            return False, error_msg

async def async_download_batch(url_list, filename_list, download_folder, max_concurrent=None):
    result = {
        "success_count": 0,
        "fail_count": 0,
        "failed_items": []
    }

    if max_concurrent is None:
        max_concurrent = calculate_optimal_workers(len(url_list), max_workers=100)

    print(f"开始异步下载")
    print(f"CPU 核心数: {multiprocessing.cpu_count()}，最大并发数: {max_concurrent}，文件数量: {len(url_list)}")

    semaphore = asyncio.Semaphore(max_concurrent)
    connector = TCPConnector(limit=max_concurrent)
    timeout = aiohttp.ClientTimeout(total=300)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [
            async_download_one(session, url_list[i], filename_list[i], download_folder, semaphore)
            for i in range(len(url_list))
        ]
        results = await asyncio.gather(*tasks)

    for i, (ok, reason) in enumerate(results):
        if ok:
            result["success_count"] += 1
        else:
            result["fail_count"] += 1
            result["failed_items"].append({
                "url": url_list[i],
                "filename": filename_list[i],
                "reason": reason
            })

    print(f"完成。成功 {result['success_count']} 个，失败 {result['fail_count']} 个。")
    return result

def run_async(url_list, filename_list, download_folder, max_concurrent=None):
    return asyncio.run(async_download_batch(url_list, filename_list, download_folder, max_concurrent))

# -------------------------------------------------------------------
# 智能模式
# -------------------------------------------------------------------
def smart_download(url_list, filename_list, download_folder):
    file_count = len(url_list)
    workers = calculate_optimal_workers(file_count)

    print("智能模式")
    print(f"CPU 核心数: {multiprocessing.cpu_count()}，推荐线程数: {workers}，文件数量: {file_count}")

    return multithread_download(url_list, filename_list, download_folder, max_workers=workers)

# -------------------------------------------------------------------
# 主调度函数
# -------------------------------------------------------------------
def main(url_list, download_folder, mode="smart", filename_list=None, max_workers=None):
    if filename_list is None:
        filename_list = []
        for i, url in enumerate(url_list):
            ext = os.path.splitext(url)[1] or ""
            filename_list.append(f"{i}{ext}")

    print("任务启动...")

    file_count = len(url_list)

    if mode == "smart":
        # 自动选择线程数
        result = smart_download(url_list, filename_list, download_folder)

    elif mode == "fast":
        # 固定 20 线程
        max_workers = 20
        result = multithread_download(url_list, filename_list, download_folder, max_workers=max_workers)

    elif mode == "ultra":
        # 固定 60 线程，高并发模式
        max_workers = 60
        result = multithread_download(url_list, filename_list, download_folder, max_workers=max_workers)

    elif mode == "async":
        # 异步下载模式，可自动计算最大并发数，也可手动传入 max_workers
        if max_workers is None:
            max_workers = calculate_optimal_workers(file_count, max_workers=100)
        result = run_async(url_list, filename_list, download_folder, max_concurrent=max_workers)

    else:
        print("未知模式，将使用智能模式。")
        result = smart_download(url_list, filename_list, download_folder)

    if result["fail_count"] == 0:
        print("全部文件已下载成功！")

    return result

可直接导入影刀编码版中

记得先装库！！！！

-------------------------------------------------------------
pip install aiohttp
pip install requests
pip install aiofiles
------------------------------------------------------------------

当然最后其实还是会取决于网速~~~