🏷️ refactor: add types

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():
 

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"],

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