Skip to content

Instantly share code, notes, and snippets.

@acgnhiki
Created February 19, 2023 07:27
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save acgnhiki/21a8da348f3e6495d71baa466f06a913 to your computer and use it in GitHub Desktop.
Save acgnhiki/21a8da348f3e6495d71baa466f06a913 to your computer and use it in GitHub Desktop.
a simple bilibili live api reverse proxy
"""
PREREQUISITES:
pip install httpx fastapi uvicorn
USAGE:
uvicorn bili_api_reverse_proxy:app --no-server-header --no-date-header
uvicorn bili_api_reverse_proxy:app --host 0.0.0.0 --port 52233 --no-server-header --no-date-header
TEST:
curl 'http://localhost:52233/xlive/web-room/v2/index/getRoomPlayInfo?room_id=23058&protocol=0,1&format=0,1,2&codec=0,1&qn=0&platform=web&ptype=8&dolby=5&panorama=1' \
-H 'referer: https://live.bilibili.com/3' \
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36' \
--compressed -v | jq .
curl 'http://localhost:52233/xlive/web-room/v2/index/getRoomPlayInfo?room_id=23058&protocol=0,1&format=0,1,2&codec=0,1&qn=0&platform=web&ptype=8&dolby=5&panorama=1' \
-H 'referer: https://live.bilibili.com/3' \
-H 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36' \
--compressed -v | python -m json.tool
"""
from typing import Dict, Mapping
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import Response
PROXY = 'http://localhost:10809'
BASE_URL = 'https://api.live.bilibili.com'
app = FastAPI()
client = httpx.AsyncClient(proxies=PROXY, timeout=10)
@app.get('/{path:path}')
async def proxy_get(req: Request, path: str):
res = await client.get(
f'{BASE_URL}/{path}',
params=req.query_params,
headers=transform_request_headers(req.headers),
)
return Response(
status_code=res.status_code,
headers=transform_response_headers(res.headers),
content=res.content,
)
def transform_request_headers(headers: Mapping[str, str]) -> Dict[str, str]:
return {
name: value for name, value in headers.items() if name.lower() not in ('host',)
}
def transform_response_headers(headers: Mapping[str, str]) -> Dict[str, str]:
return {
name: value
for name, value in headers.items()
if name.lower()
not in ('content-encoding', 'content-length', 'transfer-encoding', 'connection')
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment