77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
from biliup_next.core.errors import ModuleError
|
|
|
|
|
|
class YtDlpAdapter:
|
|
def probe(self, *, yt_dlp_cmd: str, source_url: str) -> dict[str, Any]:
|
|
cmd = [
|
|
yt_dlp_cmd,
|
|
"--dump-single-json",
|
|
"--no-warnings",
|
|
"--no-playlist",
|
|
source_url,
|
|
]
|
|
result = self._run(cmd)
|
|
if result.returncode != 0:
|
|
raise ModuleError(
|
|
code="YT_DLP_PROBE_FAILED",
|
|
message="yt-dlp 获取视频信息失败",
|
|
retryable=True,
|
|
details={"stdout": result.stdout[-2000:], "stderr": result.stderr[-2000:]},
|
|
)
|
|
try:
|
|
payload = json.loads(result.stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise ModuleError(
|
|
code="YT_DLP_PROBE_INVALID_JSON",
|
|
message="yt-dlp 返回了非法 JSON",
|
|
retryable=True,
|
|
details={"stdout": result.stdout[-2000:]},
|
|
) from exc
|
|
if not isinstance(payload, dict):
|
|
raise ModuleError(
|
|
code="YT_DLP_PROBE_INVALID_JSON",
|
|
message="yt-dlp 返回结果不是对象",
|
|
retryable=True,
|
|
)
|
|
return payload
|
|
|
|
def download(
|
|
self,
|
|
*,
|
|
yt_dlp_cmd: str,
|
|
source_url: str,
|
|
output_template: str,
|
|
format_selector: str | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
cmd = [
|
|
yt_dlp_cmd,
|
|
"--no-warnings",
|
|
"--no-playlist",
|
|
]
|
|
if format_selector:
|
|
cmd.extend(["-f", format_selector])
|
|
cmd.extend([
|
|
"--merge-output-format",
|
|
"mp4",
|
|
"-o",
|
|
output_template,
|
|
source_url,
|
|
])
|
|
return self._run(cmd)
|
|
|
|
def _run(self, cmd: list[str]) -> subprocess.CompletedProcess[str]:
|
|
try:
|
|
return subprocess.run(cmd, capture_output=True, text=True, check=False)
|
|
except FileNotFoundError as exc:
|
|
raise ModuleError(
|
|
code="YT_DLP_NOT_FOUND",
|
|
message=f"找不到 yt-dlp 命令: {cmd[0]}",
|
|
retryable=False,
|
|
) from exc
|