modules/bili.py

import requests
import gradio as gr
from utils import (
    download_file,
    extract_fst_url,
    HEADER,
    TIMEOUT,
    API_BILI,
    API_BILI_2,
    API_BILI_1,
    EN_US,
    TMP_DIR,
)

ZH2EN = {
    "状态栏": "Status",
    "请输入B站视频链接": "Please input BiliBili video link",
    "分P": "Part",
    "视频标题": "Video title",
    "视频封面": "Video cover",
    "视频下载": "Download video",
    "视频简介": "Video introduction",
    "视频时长(s)": "Video duration(s)",
    "UP主头像": "Uploader avatar",
    "UP主昵称": "Uploader nickname",
    "B站视频解析": "BiliBili video parser",
    "通道": "Channel",
    "接口1": "Interface 1",
    "接口2": "Interface 2",
}


def _L(zh_txt: str):
    return ZH2EN[zh_txt] if EN_US else zh_txt


def get_fst_url(text):
    fst_url = extract_fst_url(text)
    if fst_url:
        out_url = fst_url.split("?")[0]
        if out_url[-1] == "/":
            out_url = out_url[:-1]

        return out_url

    else:
        return f"https://www.bilibili.com/video/{text}"


def get_real_url(short_url):
    return requests.get(
        short_url,
        headers=HEADER,
        allow_redirects=True,
        timeout=TIMEOUT,
    ).url.split("/?")[0]


def get_video(video_url: str, p: int, cache: str):
    bvid = video_url.split("bilibili.com/video/")[-1]
    response = requests.get(
        API_BILI_2,
        params={
            "bv": bvid,
            "p": p,
            "otype": "json",
        },
        timeout=TIMEOUT,
    )
    response_json = response.json()
    retcode = response_json["code"]
    if retcode == 0:
        qualities: list = response_json["accept_quality"]
        quality = min(qualities[0], 80)
        url = requests.get(
            API_BILI_2,
            params={
                "bv": bvid,
                "p": p,
                "q": quality,
                "otype": "url",
            },
            timeout=TIMEOUT,
        ).text
        return download_file(url, bvid, cache)

    else:
        raise ConnectionError(f"Failed to get video: {retcode}")


def channel_2(video_url: str, p: int, cache: str):
    response = requests.get(
        API_BILI_1,
        params={
            "url": video_url,
            "type": "json",
        },
        timeout=TIMEOUT,
    )
    response_json = response.json()
    retcode = response_json["code"]
    if retcode == 200:
        title = response_json["title"]
        cover = response_json["imgurl"]
        desc = response_json["desc"]

        response_data = response_json["data"][int(p) - 1]
        dur = response_data["duration"]

        author_data = response_json["user"]
        author = author_data["name"]
        avatar = author_data["user_img"]

        video = get_video(video_url, p, cache)

        return title, cover, video, desc, dur, avatar, author

    else:
        raise ConnectionError(f"接口调用失败, 错误码: {retcode}")


def channel_1(video_url: str, p: int, cache: str):
    response = requests.get(API_BILI, params={"url": video_url}, timeout=TIMEOUT)
    response_json = response.json()
    retcode = response_json["code"]
    if retcode == 1:
        title = response_json["title"]
        cover = response_json["imgurl"]
        desc = response_json["desc"]

        author_data = response_json["user"]
        author = author_data["name"]
        avatar = author_data["user_img"]

        response_data = response_json["data"][int(p) - 1]
        dur = response_data["duration"]
        video_id = video_url.split("/")[-1]
        video = download_file(response_data["video_url"], video_id, cache)

        return title, cover, video, desc, dur, avatar, author

    else:
        raise ConnectionError(f"Failed to call API, error code: {retcode}")


# outer func
def infer(ch: str, video_url: str, p: int, cache=f"{TMP_DIR}/bili"):
    status = "Success"
    title = cover = desc = dur = video = author = avatar = None
    if not ch:
        ch == _L("接口1")

    try:
        if not video_url:
            raise ValueError("视频链接为空!")

        video_url = get_fst_url(video_url)
        if "b23.tv" in video_url:
            video_url = get_real_url(video_url)

        title, cover, video, desc, dur, avatar, author = (
            channel_1(video_url, p, cache)
            if ch == _L("接口1")
            else channel_2(video_url, p, cache)
        )

    except Exception as e:
        status = f"视频解析失败: {e}"

    return status, title, cover, video, desc, dur, avatar, author


def bili_parser():
    return gr.Interface(
        fn=infer,
        inputs=[
            gr.Dropdown(
                label=_L("通道"),
                choices=[_L("接口1"), _L("接口2")],
                value=_L("接口1"),
            ),
            gr.Textbox(
                label=_L("请输入B站视频链接"),
                placeholder="https://www.bilibili.com/video/*",
            ),
            gr.Slider(label=_L("分P"), minimum=1, maximum=1000, step=1, value=1),
        ],
        outputs=[
            gr.Textbox(label=_L("状态栏"), show_copy_button=True),
            gr.Textbox(label=_L("视频标题"), show_copy_button=True),
            gr.Image(label=_L("视频封面"), show_share_button=False),
            gr.Video(
                label=_L("视频下载"),
                show_download_button=True,
                show_share_button=False,
                format="mp4",
            ),
            gr.TextArea(label=_L("视频简介"), show_copy_button=True),
            gr.Textbox(label=_L("视频时长(s)"), show_copy_button=True),
            gr.Image(label=_L("UP主头像"), show_share_button=False),
            gr.Textbox(label=_L("UP主昵称"), show_copy_button=True),
        ],
        title=_L("B站视频解析"),
        flagging_mode="never",
        examples=[
            [_L("接口1"), "BV1Dt4y1o7bU", 1],
            [_L("接口1"), "https://b23.tv/LSoJzpW", 1],
            [_L("接口2"), "https://www.bilibili.com/video/BV1G8iRYBE4f", 1],
            [
                _L("接口2"),
                "【『新年おめでとう | 2024』你就如同烟花一般，在我心中绽放！-哔哩哔哩】 https://b23.tv/LuTAbzj",
                1,
            ],
        ],
        cache_examples=False,
    )