diff --git a/yutto/__main__.py b/yutto/__main__.py index deefdb3..e34bf9c 100644 --- a/yutto/__main__.py +++ b/yutto/__main__.py @@ -1,14 +1,22 @@ import asyncio import json -from typing import Any +from typing import Any, Optional import aiofiles import aiohttp -from yutto.api.acg_video import get_acg_video_list, get_acg_video_playurl, get_acg_video_subtitile, get_video_info +from yutto.api.acg_video import ( + AudioUrlMeta, + VideoUrlMeta, + get_acg_video_list, + get_acg_video_playurl, + get_acg_video_subtitile, + get_video_info, +) from yutto.api.types import AId, BvId, CId -from yutto.media.quality import AudioQuality, VideoQuality, gen_video_quality_priority, gen_audio_quality_priority -from yutto.media.codec import VideoCodec, AudioCodec, gen_acodec_priority, gen_vcodec_priority +from yutto.filter import select_audio, select_video +from yutto.media.codec import AudioCodec, VideoCodec, gen_acodec_priority, gen_vcodec_priority +from yutto.media.quality import AudioQuality, VideoQuality, gen_audio_quality_priority, gen_video_quality_priority from yutto.utils.asynclib import LimitParallelsPool, run_with_n_workers from yutto.utils.fetcher import Fetcher from yutto.utils.file_buffer import AsyncFileBuffer, BufferChunk @@ -17,52 +25,10 @@ from yutto.utils.logger import logger def gen_headers(): return { - "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.135 Safari/537.36", + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.85 Safari/537.36", "Referer": "https://www.bilibili.com", } -def select_video( - videos: list[dict[str, Any]], - video_quality: VideoQuality = 125, - video_codec: VideoCodec = "hevc" -) -> dict[str, Any]: - video_quality_priority = gen_video_quality_priority(video_quality) - video_codec_priority = gen_vcodec_priority(video_codec) - - # fmt: off - video_combined_priority = [ - (vqn, vcodec) - for vqn in video_quality_priority - for vcodec in video_codec_priority - ] - - for vqn, vcodec in video_combined_priority: - for video in videos: - if video["quality"] == vqn and video["codec"] == vcodec: - return video - return {} - -def select_audio( - audios: list[dict[str, Any]], - audio_quality: AudioQuality = 30280, - audio_codec: AudioCodec = "mp4a", -) -> dict[str, Any]: - audio_quality_priority = gen_audio_quality_priority(audio_quality) - audio_codec_priority = gen_acodec_priority(audio_codec) - - # fmt: off - audio_combined_priority = [ - (aqn, acodec) - for aqn in audio_quality_priority - for acodec in audio_codec_priority - ] - - for aqn, acodec in audio_combined_priority: - for audio in audios: - if audio["quality"] == aqn and audio["codec"] == acodec: - return audio - return {} - async def main(): diff --git a/yutto/api/acg_video.py b/yutto/api/acg_video.py index 320a9dd..5ac386d 100644 --- a/yutto/api/acg_video.py +++ b/yutto/api/acg_video.py @@ -1,12 +1,14 @@ import json import re -from typing import Any, TypedDict +from typing import Any, TypedDict, Literal from aiohttp import ClientSession from yutto.api.types import AId, AvId, BvId, CId, EpisodeId from yutto.urlparser import regexp_bangumi_ep from yutto.utils.fetcher import Fetcher +from yutto.media.codec import VideoCodec, AudioCodec +from yutto.media.quality import VideoQuality, AudioQuality class HttpStatusError(Exception): @@ -32,6 +34,30 @@ class VideoInfo(TypedDict): title: str +class AcgVideoListItem(TypedDict): + id: int + name: str + cid: CId + + +class VideoUrlMeta(TypedDict): + url: str + mirrors: list[str] + codec: VideoCodec + width: int + height: int + quality: VideoQuality + + +class AudioUrlMeta(TypedDict): + url: str + mirrors: list[str] + codec: AudioCodec + width: int + height: int + quality: AudioQuality + + async def get_video_info(session: ClientSession, avid: AvId) -> VideoInfo: info_api = "http://api.bilibili.com/x/web-interface/view?aid={aid}&bvid={bvid}" res_json = await Fetcher.fetch_json(session, info_api.format(**avid.to_dict())) @@ -56,7 +82,7 @@ async def get_acg_video_title(session: ClientSession, avid: AvId) -> str: return (await get_video_info(session, avid))["title"] -async def get_acg_video_list(session: ClientSession, avid: AvId) -> list[dict[str, Any]]: +async def get_acg_video_list(session: ClientSession, avid: AvId) -> list[AcgVideoListItem]: list_api = "https://api.bilibili.com/x/player/pagelist?aid={aid}&bvid={bvid}&jsonp=jsonp" res_json = await Fetcher.fetch_json(session, list_api.format(**avid.to_dict())) return [ @@ -64,7 +90,7 @@ async def get_acg_video_list(session: ClientSession, avid: AvId) -> list[dict[st { "id": i + 1, "name": item["part"], - "cid": str(item["cid"]) + "cid": CId(str(item["cid"])) } for i, item in enumerate(res_json["data"]) ] @@ -72,8 +98,9 @@ async def get_acg_video_list(session: ClientSession, avid: AvId) -> list[dict[st async def get_acg_video_playurl( session: ClientSession, avid: AvId, cid: CId -) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: +) -> tuple[list[VideoUrlMeta], list[AudioUrlMeta]]: play_api = "https://api.bilibili.com/x/player/playurl?avid={aid}&bvid={bvid}&cid={cid}&qn=125&type=&otype=json&fnver=0&fnval=80&fourk=1" + codecid_map: dict[Literal[7, 12], VideoCodec] = {7: "avc", 12: "hevc"} async with session.get(play_api.format(**avid.to_dict(), cid=cid)) as resp: if not resp.ok: @@ -86,7 +113,7 @@ async def get_acg_video_playurl( { "url": video["base_url"], "mirrors": video["backup_url"], - "codec": {7: "avc", 12: "hevc"}[video["codecid"]], + "codec": codecid_map[video["codecid"]], "width": video["width"], "height": video["height"], "quality": video["id"], diff --git a/yutto/filter.py b/yutto/filter.py new file mode 100644 index 0000000..43c7544 --- /dev/null +++ b/yutto/filter.py @@ -0,0 +1,51 @@ +from typing import Optional + +from yutto.api.acg_video import AudioUrlMeta, VideoUrlMeta +from yutto.media.codec import (AudioCodec, VideoCodec, gen_acodec_priority, + gen_vcodec_priority) +from yutto.media.quality import (AudioQuality, VideoQuality, + gen_audio_quality_priority, + gen_video_quality_priority) + + +def select_video( + videos: list[VideoUrlMeta], + video_quality: VideoQuality = 125, + video_codec: VideoCodec = "hevc" +) -> Optional[VideoUrlMeta]: + video_quality_priority = gen_video_quality_priority(video_quality) + video_codec_priority = gen_vcodec_priority(video_codec) + + # fmt: off + video_combined_priority = [ + (vqn, vcodec) + for vqn in video_quality_priority + for vcodec in video_codec_priority + ] + + for vqn, vcodec in video_combined_priority: + for video in videos: + if video["quality"] == vqn and video["codec"] == vcodec: + return video + return None + +def select_audio( + audios: list[AudioUrlMeta], + audio_quality: AudioQuality = 30280, + audio_codec: AudioCodec = "mp4a", +) -> Optional[AudioUrlMeta]: + audio_quality_priority = gen_audio_quality_priority(audio_quality) + audio_codec_priority = gen_acodec_priority(audio_codec) + + # fmt: off + audio_combined_priority = [ + (aqn, acodec) + for aqn in audio_quality_priority + for acodec in audio_codec_priority + ] + + for aqn, acodec in audio_combined_priority: + for audio in audios: + if audio["quality"] == aqn and audio["codec"] == acodec: + return audio + return None