Python中使用 you-get 批量爬取视频(支持b站,优酷

1.环境依赖

• python3
• 插件 you-get 官网: https://you-get.org/

2. you-get 安装

pip3 install you-get

image.png

3. cmd下测试you-get

cmd 输入 you-get https://www.bilibili.com/video/BV1nU4y1472g

image.png

批量下载, python代码如下

#!/usr/bin/python3
# -*- coding: UTF-8 -*-

import os
import queue
import threading
import time
from you_get import common

exitFlag = 0


class myThread(threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q

    def run(self):
        print("开启线程：{}".format(self.name))
        process_data(self.name, self.q)
        print("退出线程：{}".format(self.name))


def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print("{} processing {}".format(threadName, data['vname']))
            # common.any_download(url=data['url'], stream_id='flv', info_only=False, output_dir=r'E:\BiliBili', merge=True)
            os.system('you-get -o {path} -O {vname} {url}'.format(path=r'E:\BiliBili',
                                                                  vname=data['vname'] + '-' + data['code'],
                                                                  url=data['url']))  # 使用os操作you-get
        else:
            queueLock.release()
        time.sleep(1)


threadList = ["Thread-1", "Thread-2"]
queueLock = threading.Lock()
workQueue = queue.Queue(24)
threads = []
threadID = 1
videoList = [
    {'vname': '视频名称1', 'code': 'ABC01', 'url': 'https://www.bilibili.com/video/BV14t41157xp/?spm_id_from=333.788.videocard.10'},
    {'vname': '视频名称2', 'code': 'ABC02', 'url': 'https://www.bilibili.com/video/BV1nU4y1472g'},
]

# 创建新线程
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1

# 填充队列
queueLock.acquire()
for video in videoList:
    workQueue.put(video)
queueLock.release()

# 等待队列清空
while not workQueue.empty():
    pass

# 通知线程是时候退出
exitFlag = 1

# 等待所有线程完成
for t in threads:
    t.join()
print("退出主线程")