mirror of https://github.com/yutto-dev/yutto
♻️ refactor: refactor code about metadata
This commit is contained in:
SigureMo
parent
51b5983abd
commit
270c2b9f97
|
|
@ -126,6 +126,7 @@ dmypy.json
|
|||
*.pb
|
||||
*.ass
|
||||
*.srt
|
||||
*.nfo
|
||||
|
||||
# test files
|
||||
*.test.py
|
||||
|
|
|
|||
12
README.md
12
README.md
|
|
@ -292,14 +292,6 @@ yutto <url> -c "d8bc7493%2C2843925707%2C08c3e*81"
|
|||
- 参数 `--with-metadata`
|
||||
- 默认值 `False`
|
||||
|
||||
#### 指定媒体元数据文件格式
|
||||
|
||||
- 参数 `--metadata-type`
|
||||
- 可选值 `nfo`
|
||||
- 默认值 `nfo`
|
||||
|
||||
只有在`--with-metadata`参数开启时，格式参数才会生效。**目前只支持一种格式**
|
||||
|
||||
</details>
|
||||
|
||||
### 批量参数
|
||||
|
|
@ -378,7 +370,7 @@ yutto <url> -b -p ^~3,10,12~14,16,-4~$
|
|||
- 播放列表生成
|
||||
- 源格式修改功能（不再支持 flv 源视频下载，如果仍有视频不支持 dash 源，请继续使用 bilili）
|
||||
- 对 Python3.8 的支持，最低支持 Python3.9
|
||||
- 下载询问
|
||||
- 下载前询问
|
||||
|
||||
### 默认行为的修改
|
||||
|
||||
|
|
@ -398,6 +390,7 @@ yutto <url> -b -p ^~3,10,12~14,16,-4~$
|
|||
- 更多的批下载支持（现已支持 UP 主全部视频下载，扩展其它批下载支持也很简单）
|
||||
- 更加完善的 warning 与 error 提示
|
||||
- 支持仅输入 id 即可下载（aid、bvid、episode_id 等）
|
||||
- 支持描述文件生成（当前仅番剧）
|
||||
|
||||
## 小技巧
|
||||
|
||||
|
|
@ -492,6 +485,7 @@ yutto 现在也还不是非常稳定，需要稳定的体验的话请继续使
|
|||
|
||||
### future
|
||||
|
||||
- [ ] feat: 投稿视频描述文件支持
|
||||
- [ ] refactor: 尽可能使下载器成为独立模块（也许？）
|
||||
- [ ] feat: 字幕、弹幕嵌入视频支持（也许？）
|
||||
- [ ] refactor: 以插件形式支持更多音视频处理方面的功能，比如 autosub（也许？）
|
||||
|
|
|
|||
1
justfile
1
justfile
|
|
@ -34,6 +34,7 @@ clean:
|
|||
find . -name "*.srt" -print0 | xargs -0 rm -f
|
||||
find . -name "*.xml" -print0 | xargs -0 rm -f
|
||||
find . -name "*.ass" -print0 | xargs -0 rm -f
|
||||
find . -name "*.nfo" -print0 | xargs -0 rm -f
|
||||
find . -name "*.pb" -print0 | xargs -0 rm -f
|
||||
rm -rf .pytest_cache/
|
||||
rm -rf .mypy_cache/
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import copy
|
|||
from yutto.__version__ import VERSION as yutto_version
|
||||
from yutto.cli import batch_get, checker, get
|
||||
from yutto.media.quality import audio_quality_priority_default, video_quality_priority_default
|
||||
from yutto.typing import metadata_type
|
||||
|
||||
from yutto.processor.urlparser import alias_parser, file_scheme_parser, bare_name_parser
|
||||
from yutto.utils.console.logger import Logger, Badge
|
||||
|
||||
|
|
@ -54,12 +54,12 @@ def main():
|
|||
)
|
||||
group_common.add_argument("--no-danmaku", action="store_true", help="不生成弹幕文件")
|
||||
group_common.add_argument("--no-subtitle", action="store_true", help="不生成字幕文件")
|
||||
group_common.add_argument("--with-metadata", action="store_true", help="生成元数据文件")
|
||||
group_common.add_argument("--metadata-format", default="nfo", choices=["nfo"], help="（待实现）元数据文件类型，目前仅支持 nfo")
|
||||
group_common.add_argument("--embed-danmaku", action="store_true", help="（待实现）将弹幕文件嵌入到视频中")
|
||||
group_common.add_argument("--embed-subtitle", default=None, help="（待实现）将字幕文件嵌入到视频中（需输入语言代码）")
|
||||
group_common.add_argument("--no-color", action="store_true", help="不使用颜色")
|
||||
group_common.add_argument("--debug", action="store_true", help="启用 debug 模式")
|
||||
group_common.add_argument("--with-metadata", action="store_true", help="生成元数据文件")
|
||||
group_common.add_argument("--metadata-type", default="nfo", choices=metadata_type, help="元数据文件类型，目前仅支持nfo")
|
||||
|
||||
# 仅批量下载使用
|
||||
group_batch = parser.add_argument_group("batch", "批量下载参数")
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import re
|
||||
from typing import Literal, TypedDict
|
||||
from typing import Any, Literal, TypedDict
|
||||
|
||||
from aiohttp import ClientSession
|
||||
|
||||
|
|
@ -15,10 +15,10 @@ from yutto.typing import (
|
|||
MultiLangSubtitle,
|
||||
SeasonId,
|
||||
VideoUrlMeta,
|
||||
MetadataInfo,
|
||||
)
|
||||
from yutto.utils.console.logger import Logger
|
||||
from yutto.utils.fetcher import Fetcher
|
||||
from yutto.utils.metadata import MetaData
|
||||
from yutto.utils.time import get_time_str_by_now, get_time_str_by_stamp
|
||||
|
||||
|
||||
|
|
@ -29,7 +29,7 @@ class BangumiListItem(TypedDict):
|
|||
episode_id: EpisodeId
|
||||
avid: AvId
|
||||
is_section: bool # 是否属于专区
|
||||
metadata: MetadataInfo
|
||||
metadata: MetaData
|
||||
|
||||
|
||||
async def get_season_id_by_media_id(session: ClientSession, media_id: MediaId) -> SeasonId:
|
||||
|
|
@ -77,18 +77,6 @@ async def get_bangumi_title_from_html(session: ClientSession, season_id: SeasonI
|
|||
return title
|
||||
|
||||
|
||||
def parse_episode_data(item) -> MetadataInfo:
|
||||
|
||||
return MetadataInfo(
|
||||
title=item["long_title"],
|
||||
show_title=item["share_copy"],
|
||||
plot=item["share_copy"],
|
||||
thumb=item["cover"],
|
||||
premiered=get_time_str_by_stamp(item["pub_time"]),
|
||||
dataadded=get_time_str_by_now(),
|
||||
)
|
||||
|
||||
|
||||
async def get_bangumi_list(session: ClientSession, season_id: SeasonId) -> list[BangumiListItem]:
|
||||
list_api = "http://api.bilibili.com/pgc/view/web/season?season_id={season_id}"
|
||||
resp_json = await Fetcher.fetch_json(session, list_api.format(season_id=season_id))
|
||||
|
|
@ -109,7 +97,7 @@ async def get_bangumi_list(session: ClientSession, season_id: SeasonId) -> list[
|
|||
"episode_id": EpisodeId(str(item["id"])),
|
||||
"avid": BvId(item["bvid"]),
|
||||
"is_section": i >= len(result["episodes"]),
|
||||
"metadata": parse_episode_data(item),
|
||||
"metadata": _parse_bangumi_metadata(item),
|
||||
}
|
||||
for i, item in enumerate(result["episodes"] + section_episodes)
|
||||
]
|
||||
|
|
@ -170,3 +158,17 @@ async def get_bangumi_subtitles(session: ClientSession, avid: AvId, cid: CId) ->
|
|||
}
|
||||
for sub_info in subtitles_info["subtitles"]
|
||||
]
|
||||
|
||||
|
||||
def _parse_bangumi_metadata(item: dict[str, Any]) -> MetaData:
|
||||
|
||||
return MetaData(
|
||||
title=item["long_title"],
|
||||
show_title=item["share_copy"],
|
||||
plot=item["share_copy"],
|
||||
thumb=item["cover"],
|
||||
premiered=get_time_str_by_stamp(item["pub_time"]),
|
||||
dataadded=get_time_str_by_now(),
|
||||
source="", # TODO
|
||||
original_filename="", # TODO
|
||||
)
|
||||
|
|
|
|||
|
|
@ -35,7 +35,6 @@ from yutto.typing import AId, BvId, EpisodeData, EpisodeId, FId, MediaId, MId, S
|
|||
from yutto.utils.console.logger import Badge, Logger
|
||||
from yutto.utils.fetcher import Fetcher
|
||||
from yutto.utils.functiontools.sync import sync
|
||||
from yutto.utils.metadata import save_episode_metadata_file
|
||||
|
||||
|
||||
@sync
|
||||
|
|
@ -202,15 +201,6 @@ async def run(args: argparse.Namespace):
|
|||
Logger.error("url 不正确～")
|
||||
sys.exit(ErrorCode.WRONG_URL_ERROR.value)
|
||||
|
||||
# 生成metadata file
|
||||
if args.with_metadata:
|
||||
type = args.metadata_type
|
||||
for i, episode_data in enumerate(download_list):
|
||||
Logger.info(
|
||||
"[{0}/{1}][{2}]正在生成【{3}】媒体描述文件".format(i, len(download_list), type, episode_data["filename"])
|
||||
)
|
||||
save_episode_metadata_file(episode_data, type)
|
||||
|
||||
for i, episode_data in enumerate(download_list):
|
||||
Logger.custom(
|
||||
f"{episode_data['filename']}", Badge(f"[{i+1}/{len(download_list)}]", fore="black", back="cyan")
|
||||
|
|
|
|||
|
|
@ -30,7 +30,6 @@ from yutto.utils.console.logger import Badge, Logger
|
|||
from yutto.utils.danmaku import EmptyDanmakuData
|
||||
from yutto.utils.fetcher import Fetcher
|
||||
from yutto.utils.functiontools.sync import sync
|
||||
from yutto.utils.metadata import save_episode_metadata_file
|
||||
|
||||
|
||||
async def fetch_bangumi_data(
|
||||
|
|
@ -59,6 +58,7 @@ async def fetch_bangumi_data(
|
|||
videos, audios = await get_bangumi_playurl(session, avid, episode_id, cid)
|
||||
subtitles = await get_bangumi_subtitles(session, avid, cid) if not args.no_subtitle else []
|
||||
danmaku = await get_danmaku(session, cid, args.danmaku_format) if not args.no_danmaku else EmptyDanmakuData
|
||||
metadata = bangumi_info["metadata"] if args.with_metadata else None
|
||||
subpath_variables_base: PathTemplateVariableDict = {
|
||||
"id": id,
|
||||
"name": name,
|
||||
|
|
@ -73,10 +73,10 @@ async def fetch_bangumi_data(
|
|||
audios=audios,
|
||||
subtitles=subtitles,
|
||||
danmaku=danmaku,
|
||||
metadata=metadata,
|
||||
output_dir=output_dir,
|
||||
tmp_dir=args.tmp_dir or output_dir,
|
||||
filename=filename,
|
||||
metadata=bangumi_info["metadata"],
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -98,6 +98,10 @@ async def fetch_acg_video_data(
|
|||
videos, audios = await get_acg_video_playurl(session, avid, cid)
|
||||
subtitles = await get_acg_video_subtitles(session, avid, cid) if not args.no_subtitle else []
|
||||
danmaku = await get_danmaku(session, cid, args.danmaku_format) if not args.no_danmaku else EmptyDanmakuData
|
||||
# TODO: 支持投稿视频的 metadata 文件生成
|
||||
if args.with_metadata:
|
||||
Logger.warning("目前仅支持番剧 metadata 生成")
|
||||
metadata = None
|
||||
subpath_variables_base: PathTemplateVariableDict = {
|
||||
"id": id,
|
||||
"name": name,
|
||||
|
|
@ -112,6 +116,7 @@ async def fetch_acg_video_data(
|
|||
audios=audios,
|
||||
subtitles=subtitles,
|
||||
danmaku=danmaku,
|
||||
metadata=metadata,
|
||||
output_dir=output_dir,
|
||||
tmp_dir=args.tmp_dir or output_dir,
|
||||
filename=filename,
|
||||
|
|
@ -170,11 +175,6 @@ async def run(args: argparse.Namespace):
|
|||
Logger.error("url 不正确，也许该 url 仅支持批量下载，如果是这样，请使用参数 -b～")
|
||||
sys.exit(ErrorCode.WRONG_URL_ERROR.value)
|
||||
|
||||
if args.with_metadata:
|
||||
Logger.info("[{0}]正在生成【{1}】媒体描述文件".format(type, episode_data["filename"]))
|
||||
metadata_type = args.metadata_type
|
||||
save_episode_metadata_file(episode_data, metadata_type)
|
||||
|
||||
await process_video_download(
|
||||
session,
|
||||
episode_data,
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ import aiohttp
|
|||
from yutto.media.quality import audio_quality_map, video_quality_map
|
||||
from yutto.processor.filter import filter_none_value, select_audio, select_video
|
||||
from yutto.processor.progressbar import show_progress
|
||||
from yutto.typing import AudioUrlMeta, VideoUrlMeta, EpisodeData, DownloaderOptions
|
||||
from yutto.typing import AudioUrlMeta, DownloaderOptions, EpisodeData, VideoUrlMeta
|
||||
from yutto.utils.asynclib import CoroutineTask, parallel_with_limit
|
||||
from yutto.utils.console.colorful import colored_string
|
||||
from yutto.utils.console.logger import Badge, Logger
|
||||
|
|
@ -16,6 +16,7 @@ from yutto.utils.danmaku import write_danmaku
|
|||
from yutto.utils.fetcher import Fetcher
|
||||
from yutto.utils.ffmpeg import FFmpeg
|
||||
from yutto.utils.file_buffer import AsyncFileBuffer
|
||||
from yutto.utils.metadata import write_metadata
|
||||
from yutto.utils.subtitle import write_subtitle
|
||||
|
||||
|
||||
|
|
@ -198,6 +199,7 @@ async def process_video_download(
|
|||
audios = episode_data["audios"]
|
||||
subtitles = episode_data["subtitles"]
|
||||
danmaku = episode_data["danmaku"]
|
||||
metadata = episode_data["metadata"]
|
||||
output_dir = episode_data["output_dir"]
|
||||
tmp_dir = episode_data["tmp_dir"]
|
||||
filename = episode_data["filename"]
|
||||
|
|
@ -252,6 +254,11 @@ async def process_video_download(
|
|||
)
|
||||
Logger.custom("{} 弹幕已生成".format(danmaku["save_type"]).upper(), badge=Badge("弹幕", fore="black", back="cyan"))
|
||||
|
||||
# 保存媒体描述文件
|
||||
if metadata is not None:
|
||||
write_metadata(metadata, output_path)
|
||||
Logger.custom("NFO 媒体描述文件已生成", badge=Badge("描述文件", fore="black", back="cyan"))
|
||||
|
||||
# 下载视频 / 音频
|
||||
await download_video_and_audio(session, video, video_path, audio, audio_path, options)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
from typing import NamedTuple, TypedDict
|
||||
from typing import NamedTuple, TypedDict, Optional
|
||||
|
||||
from yutto.media.codec import AudioCodec, VideoCodec
|
||||
from yutto.media.quality import AudioQuality, VideoQuality
|
||||
from yutto.utils.danmaku import DanmakuData
|
||||
from yutto.utils.subtitle import SubtitleData
|
||||
from yutto.utils.metadata import MetaData
|
||||
|
||||
|
||||
class BilibiliId(NamedTuple):
|
||||
|
|
@ -108,26 +109,15 @@ class MultiLangSubtitle(TypedDict):
|
|||
lines: SubtitleData
|
||||
|
||||
|
||||
class MetadataInfo(TypedDict):
|
||||
title: str
|
||||
show_title: str
|
||||
plot: str
|
||||
thumb: str
|
||||
premiered: str
|
||||
dataadded: str
|
||||
source: str
|
||||
original_filename: str
|
||||
|
||||
|
||||
class EpisodeData(TypedDict):
|
||||
videos: list[VideoUrlMeta]
|
||||
audios: list[AudioUrlMeta]
|
||||
subtitles: list[MultiLangSubtitle]
|
||||
metadata: Optional[MetaData]
|
||||
danmaku: DanmakuData
|
||||
output_dir: str
|
||||
tmp_dir: str
|
||||
filename: str
|
||||
metadata: MetadataInfo
|
||||
|
||||
|
||||
class DownloaderOptions(TypedDict):
|
||||
|
|
@ -149,9 +139,6 @@ class FavouriteMetaData(TypedDict):
|
|||
title: str
|
||||
|
||||
|
||||
metadata_type: list[str] = ["nfo"]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
aid = AId("add")
|
||||
cid = CId("xxx")
|
||||
|
|
|
|||
|
|
@ -1,36 +1,28 @@
|
|||
import os.path
|
||||
|
||||
import dicttoxml
|
||||
import os
|
||||
from typing import TypedDict
|
||||
from xml.dom.minidom import parseString
|
||||
|
||||
from yutto.typing import EpisodeData
|
||||
import dicttoxml
|
||||
|
||||
|
||||
def save_season_metadata_file(episode_data: EpisodeData, ext: str = "nfo"):
|
||||
output_dir = episode_data["output_dir"]
|
||||
filename = episode_data["filename"] + "." + ext
|
||||
metadata = episode_data["metadata"]
|
||||
|
||||
save(output_dir, filename, metadata, custom_root="tvshow")
|
||||
class MetaData(TypedDict):
|
||||
title: str
|
||||
show_title: str
|
||||
plot: str
|
||||
thumb: str
|
||||
premiered: str
|
||||
dataadded: str
|
||||
source: str
|
||||
original_filename: str
|
||||
|
||||
|
||||
def save_episode_metadata_file(episode_data: EpisodeData, ext: str = "nfo"):
|
||||
output_dir = episode_data["output_dir"]
|
||||
filename = episode_data["filename"] + "." + ext
|
||||
metadata = episode_data["metadata"]
|
||||
def write_metadata(metadata: MetaData, video_path: str):
|
||||
video_path_no_ext = os.path.splitext(video_path)[0]
|
||||
metadata_path = video_path_no_ext + ".nfo"
|
||||
custom_root = "episodedetails"
|
||||
|
||||
save(output_dir, filename, metadata)
|
||||
|
||||
|
||||
def save(dir: str, filename: str, data, encoding="utf8", custom_root="episodedetails"):
|
||||
path = os.path.join(dir, filename)
|
||||
|
||||
if not os.path.exists(dir):
|
||||
os.makedirs(dir)
|
||||
|
||||
xml = dicttoxml.dicttoxml(data, custom_root=custom_root, attr_type=False)
|
||||
dom = parseString(xml)
|
||||
xml_content = dicttoxml.dicttoxml(metadata, custom_root=custom_root, attr_type=False)
|
||||
dom = parseString(xml_content)
|
||||
pretty_content = dom.toprettyxml()
|
||||
f = open(path, "w", encoding=encoding)
|
||||
f.write(pretty_content)
|
||||
f.close()
|
||||
with open(metadata_path, "w", encoding="utf-8") as f:
|
||||
f.write(pretty_content)
|
||||
|
|
|
|||
|
|
@ -1,14 +1,13 @@
|
|||
import datetime
|
||||
import time
|
||||
|
||||
TIME_FMT = "%Y-%m-%d %H:%M:%S"
|
||||
|
||||
|
||||
def get_time_str_by_now():
|
||||
now = datetime.datetime.now()
|
||||
return now.strftime(TIME_FMT)
|
||||
time_stamp_now = time.time()
|
||||
return get_time_str_by_stamp(time_stamp_now)
|
||||
|
||||
|
||||
def get_time_str_by_stamp(stamp):
|
||||
def get_time_str_by_stamp(stamp: int):
|
||||
local_time = time.localtime(stamp)
|
||||
return time.strftime(TIME_FMT, local_time)
|
||||
|
|
|
|||
Loading…
Reference in New Issue