基于lux的YouTube视频解析，下载

飞熊 • 2023年8月8日下午8:41 • Python • 阅读 231

必须能连外网环境下操作，挂梯子网络不稳定，容易解析失败
Pyton ，lux ，Motrix 百度下，都能找到官网下载即可

使用lux解析视频链接（会解析出很多不同分辨率的视频链接，使用下边的脚本过滤出能下载同时包含音视频的最大分辨率链接）

# !/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@author  : v_jiaohaicheng@baidu.com
@des     :

"""
import re
import json
import subprocess
from cup.util import ThreadPool



def deal_stdout(stdout_list):
    """

    :return:
    """
    # print("stdout_list",stdout_list)
    stdout_str = ""
    for line in stdout_list:
        line = line.decode(encoding="utf-8", errors="ignore")
        if line.startswith("libpng warning"):
            continue
        stdout_str += line
    try:
        tmp_dic = json.loads(stdout_str)
        status_code = tmp_dic["status_code"]
        html_body = tmp_dic["html_body"]
    except BaseException:
        html_body = stdout_str
        status_code = stdout_str.split('"status_code" :')[-1].split("}")[0].strip()
    return status_code, html_body


def _stream_watcher(stream):
    output = []
    while True:
        line = stream.readline()
        if line:
            output.append(line)
        else:
            break
    if not stream.closed:
        stream.close()
        return output


def get_html(url):
    cmd = 'lux --json "{}"'.format(url)
    # print("cmd = ",cmd)
    phandle = subprocess.Popen(
        cmd,
        shell=True,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    stream = phandle.stdout
    status_code, html_body = deal_stdout(_stream_watcher(stream))
    # print("status_code = ", status_code)
    return html_body


def yield_url():
    with open("./url.txt","r",encoding="utf-8")as fp:
        for i in fp:
            yield i.strip()

def main(url):
    with open("./res_url3.txt", "a", encoding="utf-8")as fpout:
        data = get_html(url)
        data_json = json.loads(data)
        # print(json.dumps(data_json).replace("'",'"'))
        title = data_json[0]["title"]
        print(title)
        out_list = []
        for id, value in data_json[0]["streams"].items():
            try:
                quality = int(re.findall("\d+p video/mp4", value["quality"])[0].replace("p video/mp4", ""))

                for args in value["parts"]:
                    url = args["url"]
                    size = args["size"]
                    type = args["ext"]
                    if size == 0 and type == "mp4":
                        if len(out_list) > 0:
                            if quality > out_list[0]:
                                out_lis = [quality, url]
                        else:
                            out_lis = [quality, url]
            except:
                pass
        out_lis = [str(i) for i in out_lis]
        print(out_lis[0])
        fpout.write("{}\n".format(out_lis[1]))
        fpout.flush()




if __name__ == '__main__':
    # for url in yield_url():
    #     main(url)
    #     break

    t = ThreadPool(maxthreads=10,daemon_threads=False)
    t.start()
    for url in yield_url():
        # print(url)
        t.add_1job(main,
            url
        )
    t.stop()