openapi/source/bilibili/bilibili.py

import asyncio
import sys
import base64
import os
import re
import tempfile
from pathlib import Path

import httpx
from aiohttp import web

COOKIE_FILE = str(Path(__file__).parent / "cookies.txt")
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"


def _prepare_cookie_file():
    """Convert simple name=value cookies.txt to Netscape format for yt-dlp."""
    try:
        with open(COOKIE_FILE) as f:
            cookies = [line.strip() for line in f if line.strip() and not line.startswith("#")]
    except FileNotFoundError:
        return None
    if not cookies:
        return None
    tmp = tempfile.NamedTemporaryFile(
        mode="w", suffix=".txt", prefix="bilibili_cookie_", delete=False
    )
    tmp.write("# Netscape HTTP Cookie File\n")
    for c in cookies:
        if "=" not in c:
            continue
        name, value = c.split("=", 1)
        # Netscape format: domain flag path secure expires name value
        tmp.write(f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n")
    tmp.close()
    return tmp.name


async def resolve_bvid(link):
    # 直接是 BV 号
    m = re.search(r"(BV[a-zA-Z0-9]{10})", link)
    if m:
        return m.group(1)

    # b23.tv 短链接，跟随跳转
    if "b23.tv" in link:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(
                link.strip(),
                follow_redirects=False,
                headers={"User-Agent": UA},
            )
        location = resp.headers.get("Location", "")
        if m := re.search(r"(BV[a-zA-Z0-9]{10})", location):
            return m.group(1)
        if m := re.search(r"(BV[a-zA-Z0-9]{10})", resp.text):
            return m.group(1)

    # 普通 bilibili.com URL
    if m := re.search(r"(BV[a-zA-Z0-9]{10})", link):
        return m.group(1)

    raise ValueError(f"unable to resolve BV from: {link}")


def decode_url(url_b64):
    try:
        return base64.urlsafe_b64decode(url_b64.encode()).decode()
    except Exception:
        return base64.b64decode(url_b64.encode()).decode()


async def get_video(url_b64=None, bv=None):
    if bv:
        bvid = bv
    else:
        try:
            url = decode_url(url_b64)
        except Exception:
            return web.json_response({"error": "invalid base64 encoding"}, status=400)

        if not url:
            return web.json_response({"error": "empty url"}, status=400)

        if "bilibili.com" not in url and "b23.tv" not in url and not url.startswith("BV"):
            return web.json_response({"error": "not a bilibili url"}, status=400)

        try:
            bvid = await resolve_bvid(url)
        except ValueError as e:
            return web.json_response({"error": str(e)}, status=400)

    cookie_tmp = _prepare_cookie_file()
    if not cookie_tmp:
        return web.json_response({"error": "missing bilibili cookie"}, status=500)

    tmpdir = tempfile.mkdtemp(prefix="bilibili_")
    output_template = os.path.join(tmpdir, "video.%(ext)s")
    cmd = [
        os.path.join(os.path.dirname(sys.executable), "yt-dlp"),
        "--cookies", cookie_tmp,
        "--add-header", f"User-Agent:{UA}",
        "--add-header", "Referer:https://www.bilibili.com/",
        "-f", "bestvideo*+bestaudio/best",
        "--merge-output-format", "mp4",
        "-o", output_template,
        f"https://www.bilibili.com/video/{bvid}",
    ]
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=300
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            return web.json_response({"error": "download timeout"}, status=504)
        finally:
            if cookie_tmp and os.path.exists(cookie_tmp):
                os.unlink(cookie_tmp)

        video_files = [p for p in Path(tmpdir).glob("*") if p.is_file()]
        if proc.returncode != 0 and not video_files:
            err_msg = stderr.decode()[-1000:] if stderr else "unknown error"
            return web.json_response(
                {"error": "download failed", "detail": err_msg},
                status=502,
            )

        if not video_files:
            return web.json_response({"error": "no output file"}, status=502)

        video_path = max(video_files, key=lambda p: p.stat().st_size)

        ext = video_path.suffix.lower()
        content_types = {
            ".mp4": "video/mp4",
            ".flv": "video/x-flv",
            ".mkv": "video/x-matroska",
            ".webm": "video/webm",
        }
        content_type = content_types.get(ext, "application/octet-stream")

        return web.FileResponse(
            path=video_path,
            headers={
                "Content-Type": content_type,
                "Content-Disposition": f'attachment; filename="{video_path.name}"',
            },
        )
    except Exception as e:
        print(f"[bilibili] Error: {e}")
        return web.json_response({"error": str(e)}, status=500)


async def handle_bilibili(request):
    bv = request.query.get("bv", "")
    if bv:
        return await get_video(bv=bv)
    url_b64 = request.query.get("url", "")
    if not url_b64:
        return web.json_response({"error": "missing url or bv parameter"}, status=400)
    return await get_video(url_b64=url_b64)


app = web.Application()
app.router.add_get("/bilibili/get_video", handle_bilibili)

if __name__ == "__main__":
    web.run_app(app, host="127.0.0.1", port=19990)