Youtube 工具集

youtube_videos.py

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

__author__  = "ChenyangGao <https://chenyanggao.github.io>"
__all__ = ["video_iter"]
__version__ = (0, 0, 1)

if __name__ == "__main__":
    from argparse import ArgumentParser

    parser = ArgumentParser(description="YouTube 用户的视频罗列")
    parser.add_argument("user", help="用户名或者包含用户名的链接")
    parser.add_argument("-d", "--detail", action="store_true", help="输出完整信息，json 格式")
    parser.add_argument("-t", "--title", action="store_true", help="输出标题")
    args = parser.parse_args()

from json import load, loads, dumps
from urllib.request import urlopen, Request


def extact_data(text):
    data = {}
    start = text.index(b"ytcfg.set({") + 10
    stop = text.index(b"});", start) + 1
    data["context"] = loads(text[start:stop])
    start = text.index(b"ytInitialData = ") + 16
    stop = text.index(b"};", start) + 1
    data["initialData"] = loads(text[start:stop])
    return data


def video_iter(user):
    if "@" in user:
        start = user.index("@") + 1
        stop = user.find("/", start)
        if stop == -1:
            user = user[start:]
        else:
            user = user[start:stop]
    url = f"https://www.youtube.com/@{user}/videos"
    data = extact_data(urlopen(url).read())
    context = data["context"]["INNERTUBE_CONTEXT"]
    browser = data["initialData"]["contents"]["twoColumnBrowseResultsRenderer"]["tabs"][1]["tabRenderer"]["endpoint"]["browseEndpoint"]

    api = "https://www.youtube.com/youtubei/v1/browse?prettyPrint=false"
    data = {"context": context, **browser}
    resp = load(urlopen(Request(api, data=dumps(data).encode("ascii"), method="POST", headers={"Content-Type": "application/json"})))
    contents = resp["contents"]["twoColumnBrowseResultsRenderer"]["tabs"][1]["tabRenderer"]["content"]["richGridRenderer"]["contents"]
    yield from contents[:30]

    while len(contents) > 30:
        continuation = contents[-1]["continuationItemRenderer"]["continuationEndpoint"]["continuationCommand"]["token"]
        data = {"context": context, "continuation": continuation}
        resp = load(urlopen(Request(api, data=dumps(data).encode("ascii"), method="POST", headers={"Content-Type": "application/json"})))
        contents = resp["onResponseReceivedActions"][0]["appendContinuationItemsAction"]["continuationItems"]
        yield from contents[:30]


if __name__ == "__main__":
    show_detail = args.detail
    show_title = args.title
    try:
        for video in video_iter(args.user):
            info = video["richItemRenderer"]["content"]["videoRenderer"]
            if show_title:
                print("#", info["title"]["runs"][0]["text"])
            if show_detail:
                print(dumps(video, ensure_ascii=False), flush=True)
            else:
                print("https://www.youtube.com/watch?v=", info["videoId"], sep="", flush=True)
    except BrokenPipeError:
        from sys import stderr
        stderr.close()
    except KeyboardInterrupt:
        pass

yt1s.py

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

__author__  = "ChenyangGao <https://chenyanggao.github.io>"
__all__ = ["query", "convert"]
__version__ = (0, 0, 3)

if __name__ == "__main__":
    from argparse import ArgumentParser

    parser = ArgumentParser(description="YouTube 视频下载链接获取 by https://www.yt1s.com")
    parser.add_argument("qs", nargs="*", help="视频的 链接 或者 id")
    parser.add_argument("-d", "--detail", action="store_true", help="输出完整信息，json 格式")
    parser.add_argument("-t", "--title", action="store_true", help="输出标题")
    parser.add_argument("-s", "--select", help="提供一个表达式（会注入一个变量 links），选择一个视频，默认选择第 1 个")
    args = parser.parse_args()

from gzip import GzipFile
from json import load
from urllib.parse import urlencode
from urllib.request import urlopen, Request


def post(url, data=None, headers={"Accept-Encoding": "gzip", "User-Agent": "Mozilla/5.0"}):
    if data and not isinstance(data, bytes):
        data = urlencode(data).encode()
    with urlopen(Request(url, data=data, method="POST", headers=headers)) as resp:
        js = load(GzipFile(fileobj=resp))
    if js["status"] != "ok":
        raise ValueError(js["mess"])
    return js


def query(q):
    "查询视频信息"
    if q.startswith("http"):
        url = q
    elif q.startswith("/watch?"):
        url = "https://www.youtube.com" + q
    else:
        url = "https://www.youtube.com/watch?v=" + q
    api = "https://www.yt1s.com/api/ajaxSearch/index"
    return post(api, data={"q": url, "vt": "home"})


def convert(vid, k):
    "获取下载链接"
    url = "https://www.yt1s.com/api/ajaxConvert/convert"
    return post(url, data={"vid": vid, "k": k})


if __name__ == "__main__":
    from json import dumps
    from sys import stderr, stdin

    qs = args.qs
    if not qs:
        qs = map(str.strip, stdin)
    show_detail = args.detail
    show_title = args.title
    select = args.select
    if select is None:
        select = lambda links: next(v for d in links.values() for v in d.values())
    else:
        select = eval("lambda links:" + select)
    try:
        for q in qs:
            js = query(q)
            video = select(js["links"])
            if video is None:
                continue
            k = video["k"]
            video_info = convert(js["vid"], k)
            if show_title:
                print("#", video_info["title"])
            if show_detail:
                print(dumps(video_info, ensure_ascii=False), flush=True)
            else:
                print(video_info["dlink"], flush=True)
    except BrokenPipeError:
        stderr.close()
    except EOFError:
        raise
    except KeyboardInterrupt:
        pass