Spaces:
Sleeping
Sleeping
File size: 2,064 Bytes
0aee47a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
"""
ivitools.download
下载互动视频
"""
import os
import tqdm
import requests
from colorama import Fore
from bilibili_api import sync, interactive_video
async def download_file(url: str, out: str):
resp = requests.get(
url,
headers={"User-Agent": "Mozilla/5.0", "Referer": "https://www.bilibili.com"},
stream=True,
)
headers = resp.headers
CHUNK_SIZE = 1024
parts = int(headers["Content-Length"]) // CHUNK_SIZE
if int(headers["Content-Length"]) % CHUNK_SIZE != 0:
parts += 1
fp = open(out, "wb")
bar = tqdm.tqdm(range(parts))
bar.set_description("DOWNLOADING...")
bar.display()
for chunk in resp.iter_content(CHUNK_SIZE):
fp.write(chunk)
bar.update(1)
def download_interactive_video(bvid: str, out: str):
ivideo = interactive_video.InteractiveVideo(bvid)
downloader = interactive_video.InteractiveVideoDownloader(
ivideo, out, download_file
)
@downloader.on("START")
async def on_start(data):
print("Start downloading " + bvid + "...")
@downloader.on("GET")
async def on_get(data):
print(f'Get node {data["title"]} (node_id: {data["node_id"]}). ')
@downloader.on("PREPARE_DOWNLOAD")
async def on_prepare_download(data):
print(
f'Start download the video for cid {data["cid"]}'
)
@downloader.on("PACKAGING")
async def on_packaing(data):
print("Almost!!! It's packaging your interactive video now! ")
@downloader.on("SUCCESS")
async def on_success(data):
print(
Fore.GREEN
+ "Congratulations! Your IVI file is ready. Check it at "
+ os.path.abspath(out)
+ ". "
+ Fore.RESET
)
try:
sync(downloader.start())
except KeyboardInterrupt:
sync(downloader.abort())
print(Fore.YELLOW + "[WRN]: Aborted by user. " + Fore.RESET)
except Exception as e:
raise e
|