【Python】多线程批量下载视频

#!/usr/bin/env python3
# coding: utf-8

import os
import logging
import pathlib
import requests
import threading
from urllib import parse

def to_logger(file_name):
    logger = logging.getLogger(file_name)
    logger.setLevel(logging.INFO)
    fmt = logging.Formatter("%(asctime)s - %(levelname)s: %(message)s")
    fh = logging.FileHandler(file_name, encoding="utf-8")
    fh.setFormatter(fmt)
    if not logger.handlers:
        logger.addHandler(fh)
    return logger

def down2video(video_url, timeout=3, retry=5, error_log="error.log"):
    r = None
    ct = 0
    # 请求头部
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.1.4753.73 Safari/527.56 Edg/102.1.4753.73",
        "Accept-Encoding": "identity"
    }
    
    # 请求URL，例如http://www.example.com/video/to1.mp4
    url = video_url
    # 请求超时时间，默认3秒
    timeout = timeout
    # 失败重试次数。默认5次
    retry = retry
    # 错误日志文件，默认error.log
    error_log = error_log
    
    to_string = parse.urlparse(url)
    # 存放目录，使用绝对路径，例如/root/video
    video_dir = os.getcwd() + os.path.dirname(to_string.path)
    # 创建目录，目录不存在则创建，即使目录已存在也不会抛出异常
    pathlib.Path(video_dir).mkdir(parents=True, exist_ok=True)
    # 文件名，使用绝对路径，例如/root/video/to1.mp4
    video_name = video_dir + "/" + os.path.basename(to_string.path)
    # 错误日志
    logger = to_logger(error_log)
    
    while ct < retry:
        try:
            r = requests.get(url=url, headers=headers, timeout=timeout)
            r.raise_for_status()
            if r.status_code == 200:
                # 将body写入文件，适合小文件，大文件可以用chunk方式
                with open(video_name, "wb") as fp:
                    fp.write(r.content)
                return r
        except Exception as e:
            logger.info(e)
        ct += 1
    logger.info(url)
    logger.info(r)
    return r

def main(urls):
    urls = urls
    th_list = []
    th_num = 4
    for url in urls:
        t = threading.Thread(target=down2video, args=[url])
        th_list.append(t)
        t.start()
    while len(th_list) > th_num:
        th_list = [th for th in th_list if th.is_alive()]

if __name__ == "__main__":
    import time
    start_time = time.time()

    import sys
    url_file = sys.argv[1]
    urls = []
    with open(url_file, "r") as fp:
        for url in fp.readlines():
            urls.append(url.strip("\n"))
    main(urls)

    end_time = time.time()
    print("Total time:{}(s)".format(end_time-start_time))

#    import sys, m3u8
#    m3u8_url = sys.argv[1]
#    play_list = m3u8.load(m3u8_url)
#    urls = []
#    for line in play_list.files:
#        urls.append(play_list.base_uri + line)
#    main(urls)

#    url = ["http://192.168.3.120/video/video10/0000.ts"]
#    # 根据URI自动创建指定的目录，并将视频存放在此目录下
#    main(url)

root@ubuntu:~/py# cat url_list.txt 
http://192.168.3.120/video/video10/0000.ts
http://192.168.3.120/video/video10/0001.ts
http://192.168.3.120/video/video10/0002.ts
http://192.168.3.120/video/video10/0003.ts
http://192.168.3.120/video/video10/0004.ts
http://192.168.3.120/video/video10/0005.ts
root@ubuntu:~/py#  
root@ubuntu:~/py# ./down2video.py url_list.txt 
Total time:0.03328990936279297(s)
root@ubuntu:~/py# 
root@ubuntu:~/py# tree
.
├── down2video.py
├── error.log
├── url_list.txt
└── video
    └── video10
        ├── 0000.ts
        ├── 0001.ts
        ├── 0002.ts
        ├── 0003.ts
        ├── 0004.ts
        └── 0005.ts

2 directories, 9 files
root@ubuntu:~/py# 

本人并不会写代码，基本上都是网上抄来改的。
目前也一直在用，还是可以的。