Skip to content

Commit

Permalink
feat: 将下载视频时选择流的部分参数提取为配置 (#47)
Browse files Browse the repository at this point in the history
  • Loading branch information
amtoaer authored Feb 24, 2024
1 parent bde142a commit c21da25
Show file tree
Hide file tree
Showing 2 changed files with 30 additions and 11 deletions.
2 changes: 1 addition & 1 deletion processor.py
Original file line number Diff line number Diff line change
Expand Up @@ -358,7 +358,7 @@ async def get_video(v: video.Video, page_id: int, tmp_video_path: Path, tmp_audi
await amakedirs(video_path.parent, exist_ok=True)
# 分析对应分p的视频流
detector = video.VideoDownloadURLDataDetecter(await v.get_download_url(page_index=page_id))
streams = detector.detect_best_streams()
streams = detector.detect_best_streams(**settings.stream.model_dump())
if detector.check_flv_stream():
# 对于 flv,直接下载
await download_content(streams[0].url, tmp_video_path)
Expand Down
39 changes: 29 additions & 10 deletions settings.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from pathlib import Path

from bilibili_api.video import VideoCodecs
from pydantic import BaseModel, Field, field_validator
from bilibili_api.video import AudioQuality, VideoCodecs, VideoQuality
from pydantic import BaseModel, Field, field_validator, root_validator
from pydantic_core import PydanticCustomError
from typing_extensions import Annotated

Expand All @@ -17,6 +17,26 @@ class SubtitleConfig(BaseModel):
static_time: float = 10 # 静态弹幕持续时间


class StreamConfig(BaseModel):
video_max_quality: VideoQuality = VideoQuality._8K
audio_max_quality: AudioQuality = AudioQuality._192K
video_min_quality: VideoQuality = VideoQuality._360P
audio_min_quality: AudioQuality = AudioQuality._64K
codecs: list[VideoCodecs] = Field(
default_factory=lambda: [VideoCodecs.AV1, VideoCodecs.AVC, VideoCodecs.HEV], min_length=1
)
no_dolby_video: bool = False
no_dolby_audio: bool = False
no_hdr: bool = False
no_hires: bool = False

@field_validator("codecs", mode="after")
def codec_validator(cls, codecs: list[VideoCodecs]) -> list[VideoCodecs]:
if len(codecs) != len(set(codecs)):
raise PydanticCustomError("unique_list", "List must be unique")
return codecs


class Config(BaseModel):
sessdata: Annotated[str, Field(min_length=1)] = ""
bili_jct: Annotated[str, Field(min_length=1)] = ""
Expand All @@ -26,16 +46,15 @@ class Config(BaseModel):
interval: int = 20
path_mapper: dict[int, str] = Field(default_factory=dict)
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig)
codec: list[VideoCodecs] = Field(
default_factory=lambda: [VideoCodecs.AV1, VideoCodecs.AVC, VideoCodecs.HEV], min_length=1
)
stream: StreamConfig = Field(default_factory=StreamConfig)
paginated_video: bool = False

@field_validator("codec", mode="after")
def codec_validator(cls, codecs: list[VideoCodecs]) -> list[VideoCodecs]:
if len(codecs) != len(set(codecs)):
raise PydanticCustomError("unique_list", "List must be unique")
return codecs
@root_validator(pre=True)
def migrate(cls, values: dict) -> dict:
# 把旧版本的 codec 迁移为 stream 中的 codecs
if "codec" in values and "stream" not in values:
values["stream"] = {"codecs": values.pop("codec")}
return values

@staticmethod
def load(path: Path | None = None) -> "Config":
Expand Down

0 comments on commit c21da25

Please sign in to comment.