stream-fetch-dl 0.1.4

一个用于从各种来源获取流数据的Python库。

stream_fetch_dl

轻量、高效、稳定的 Python 流式下载库，专注普通文件与 M3U8 视频下载。

简介

stream_fetch_dl 是一个极简实用的文件下载工具包，提供两种核心下载能力：

• 通用文件流式下载（支持所有直链格式）
• M3U8/HLS 视频分片下载、并发限流与 AES-128 解密

API 简洁、资源占用低，可安全嵌入 GUI 或后台程序，不卡顿、易控制。

特性

• 单文件流式下载
• M3U8 多 TS 并发下载与合并
• AES-128 解密支持
• 并发限流（Semaphore）
• 任务可停止、可控制
• 轻量无依赖，结构干净

---

安装

pip install stream_fetch_dl

快速上手

请确保已安装 stream_fetch_dl 库 以下示例代码基于 Python 3.x 如需在 Qt 环境中使用，请参考 Qt 线程池下载示例 如需在命令行环境中使用，请参考命令行下载示例

同步调用

# coding: utf-8
from time import sleep

from stream_fetch_dl.task import StreamTask, TaskResult


def callback(result: TaskResult):
    print(result.message, end="\r\n", flush=True)


params = {
    # 测试URL，随时可能失效
    "url": 'https://v16.toutiao50.com/e8fb833110916841a0dd460146f80fc4/69e863a1/video/tos/alisg/tos-alisg-v-90231e-sg/oEXeKB2bEEQfq7QU3B1isqq48ED1gwiKpbBRkv/',
    "headers": {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
    },
    "save_path": "./downloads",
    "timeout": 10,
    "callback": callback
}

task = StreamTask(params)
task.run()

sleep(10)
task.stop()

单线程调用

# coding: utf-8
from time import sleep

from stream_fetch_dl.task import TaskResult
from stream_fetch_dl.threads import StreamThread


def callback(result: TaskResult):
    print(result.message, end="\r\n", flush=True)


params = {
    # 测试URL，随时可能失效
    "url": 'https://v16.toutiao50.com/ca050714cd76ceb56f5718d376d99b27/69e98e29/video/tos/alisg/tos-alisg-v-90231e-sg/oQOBfGsfbUkpFBE4IwCI173DgEpUQzm7Zz1JIr/',
    "headers": {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
    },
    "save_path": "./downloads",
    "timeout": 10,
    "callback": callback
}

task = StreamThread(params)
task.start()

sleep(10)
task.stop()

线程池下载

# coding: utf-8
from time import sleep

from stream_fetch_dl.threads import ThreadPoolManager
from stream_fetch_dl.task import TaskResult


def callback(result: TaskResult):
    print(result.message, end="\r\n", flush=True)


pool = ThreadPoolManager(5)

m3u8_params = {
    "id": "52091_553fc92e",
    "url": "https://vip.ffzy-plays.com/20260406/52091_553fc92e/index.m3u8",
    "name": "茉莉花酱的好感度正在崩坏-第1集",
    "suffix": ".mp4",
    "save_path": "./downloads",
    "headers": {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"
    },
    "callback": callback
}
pool.add_m3u8(m3u8_params, True)
sleep(10)
worker = pool.task(m3u8_params["id"])
if worker:
    worker.stop()

Qt线程池下载

# coding: utf-8
from PySide6.QtWidgets import QApplication
from PySide6.QtCore import QThreadPool, QTimer
import sys
from stream_fetch_dl.qt import QThreadPoolManager
from stream_fetch_dl.task import TaskResult


def callback(result: TaskResult):
    print(result.message, end="\r\n", flush=True)


app = QApplication(sys.argv)
pool = QThreadPoolManager()

m3u8_params = {
    "id": "52091_553fc92e",
    "url": "https://vip.ffzy-plays.com/20260406/52091_553fc92e/index.m3u8",
    "name": "茉莉花酱的好感度正在崩坏-第1集",
    "suffix": ".mp4",
    "save_path": "./downloads",
    "headers": {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"
    },
    "callback": callback
}
pool.add_m3u8(m3u8_params, True)


def stop_worker():
    worker = pool.task(m3u8_params["id"])
    worker.stop()


QTimer.singleShot(10000, stop_worker)
sys.exit(app.exec())