hugo
e27cc22cfb
Derived from pengchujin/jzsub at 222a90265d2a8797ca258eb1a980cee0863a8311; preserve the upstream MIT license and attribution.
528 lines
21 KiB
Python
Executable File
528 lines
21 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Ingest one browser-confirmed, authorized embedded HLS video.
|
|
|
|
The browser-facing agent writes a private resource map after it has verified
|
|
that playback is authorized. This script never receives signed URLs on the
|
|
command line, never prints them, and propagates a playlist's authorization
|
|
query only to same-origin HLS resources. It rejects master playlists and DRM
|
|
methods so the ordinary yt-dlp route remains the preferred path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import tempfile
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any, Sequence
|
|
|
|
import fetch_video as fetch
|
|
|
|
|
|
RESOURCE_MAP_SCHEMA = 1
|
|
ALLOWED_SUBTITLE_KINDS = {"manual", "automatic"}
|
|
_URI_ATTRIBUTE = re.compile(r'URI="([^"]+)"')
|
|
_EXTENSION = re.compile(r"^[A-Za-z0-9]{1,8}$")
|
|
|
|
|
|
class IngestError(fetch.FetchError):
|
|
"""An embedded-resource ingest failed safely."""
|
|
|
|
|
|
def _read_private_resource_map(path: Path) -> dict[str, Any]:
|
|
path = path.expanduser().resolve()
|
|
if not path.is_file():
|
|
raise IngestError(f"Resource map does not exist or is not a file: {path}")
|
|
if path.stat().st_mode & 0o077:
|
|
raise IngestError("Resource map must be private (chmod 600) because it may contain signed URLs")
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise IngestError("Could not read the private resource map") from exc
|
|
if not isinstance(value, dict) or value.get("schema_version") != RESOURCE_MAP_SCHEMA:
|
|
raise IngestError(f"Resource map must be a schema_version={RESOURCE_MAP_SCHEMA} object")
|
|
return value
|
|
|
|
|
|
def _prepare_output_dir(
|
|
output_dir: Path,
|
|
resource_map_path: Path,
|
|
*,
|
|
resume: bool,
|
|
cleanup_resource_map: bool,
|
|
) -> Path:
|
|
if output_dir.exists() and not output_dir.is_dir():
|
|
raise IngestError(f"Output path is not a directory: {output_dir}")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
if resume:
|
|
return output_dir
|
|
entries = list(output_dir.iterdir())
|
|
allowed_map = (
|
|
cleanup_resource_map
|
|
and resource_map_path.parent == output_dir
|
|
and resource_map_path in entries
|
|
)
|
|
unexpected = [path for path in entries if not (allowed_map and path == resource_map_path)]
|
|
if unexpected or (entries and not allowed_map):
|
|
raise IngestError(
|
|
f"Output directory is not empty: {output_dir}. Choose a new directory or pass --resume explicitly."
|
|
)
|
|
return output_dir
|
|
|
|
|
|
def _required_text(value: Any, field: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise IngestError(f"Resource map field {field!r} must be a non-empty string")
|
|
return value.strip()
|
|
|
|
|
|
def _optional_number(value: Any, field: str) -> float | int | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)) or value <= 0:
|
|
raise IngestError(f"Resource map field {field!r} must be a positive number")
|
|
return value
|
|
|
|
|
|
def _validated_url(value: Any, field: str) -> str:
|
|
try:
|
|
return fetch.validate_url(_required_text(value, field))
|
|
except fetch.FetchError as exc:
|
|
raise IngestError(f"Invalid resource map URL field {field!r}") from exc
|
|
|
|
|
|
def _same_origin(left: str, right: str) -> bool:
|
|
a = urllib.parse.urlsplit(left)
|
|
b = urllib.parse.urlsplit(right)
|
|
return (a.scheme.lower(), a.hostname, a.port) == (b.scheme.lower(), b.hostname, b.port)
|
|
|
|
|
|
def _authorized_resource_url(resource_url: str, playlist_url: str) -> str:
|
|
"""Add missing root-playlist query fields to same-origin resources only."""
|
|
absolute = urllib.parse.urljoin(playlist_url, resource_url)
|
|
if not _same_origin(absolute, playlist_url):
|
|
return absolute
|
|
root = urllib.parse.urlsplit(playlist_url)
|
|
target = urllib.parse.urlsplit(absolute)
|
|
root_query = urllib.parse.parse_qsl(root.query, keep_blank_values=True)
|
|
target_query = urllib.parse.parse_qsl(target.query, keep_blank_values=True)
|
|
existing = {key for key, _ in target_query}
|
|
merged = target_query + [(key, value) for key, value in root_query if key not in existing]
|
|
return urllib.parse.urlunsplit(
|
|
(target.scheme, target.netloc, target.path, urllib.parse.urlencode(merged), "")
|
|
)
|
|
|
|
|
|
def _validate_hls_protection(playlist: str) -> None:
|
|
for line in playlist.splitlines():
|
|
if not line.startswith("#EXT-X-KEY:"):
|
|
continue
|
|
attributes = line.split(":", 1)[1]
|
|
method = re.search(r"(?:^|,)METHOD=([^,]+)", attributes)
|
|
method_value = method.group(1).strip().upper() if method else ""
|
|
keyformat = re.search(r'(?:^|,)KEYFORMAT="?([^,"]+)', attributes)
|
|
keyformat_value = keyformat.group(1).strip().lower() if keyformat else "identity"
|
|
if method_value not in {"NONE", "AES-128"} or keyformat_value != "identity":
|
|
raise IngestError(
|
|
"Embedded HLS uses an unsupported protection method; MaterialSub will not bypass DRM"
|
|
)
|
|
|
|
|
|
def rewrite_media_playlist(playlist: str, playlist_url: str) -> str:
|
|
"""Return a signed, absolute media playlist without logging its secrets."""
|
|
if not playlist.lstrip().startswith("#EXTM3U"):
|
|
raise IngestError("The selected resource is not an HLS playlist")
|
|
if "#EXT-X-STREAM-INF" in playlist or "#EXT-X-MEDIA:" in playlist:
|
|
raise IngestError(
|
|
"The resource is an HLS master playlist; use the browser-observed media playlist selected during playback"
|
|
)
|
|
_validate_hls_protection(playlist)
|
|
rewritten: list[str] = []
|
|
for line in playlist.splitlines():
|
|
if line.startswith("#"):
|
|
line = _URI_ATTRIBUTE.sub(
|
|
lambda match: f'URI="{_authorized_resource_url(match.group(1), playlist_url)}"',
|
|
line,
|
|
)
|
|
elif line.strip():
|
|
line = _authorized_resource_url(line.strip(), playlist_url)
|
|
rewritten.append(line)
|
|
return "\n".join(rewritten) + "\n"
|
|
|
|
|
|
def _request_bytes(url: str, *, playlist_url: str | None = None) -> bytes:
|
|
resolved = _authorized_resource_url(url, playlist_url) if playlist_url else url
|
|
request = urllib.request.Request(resolved, headers={"User-Agent": "Mozilla/5.0 MaterialSub/1"})
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=45) as response:
|
|
return response.read()
|
|
except Exception as exc:
|
|
raise IngestError(f"Failed to download an authorized resource from {fetch.display_url(url)}") from exc
|
|
|
|
|
|
def _atomic_download(url: str, destination: Path, *, playlist_url: str | None = None) -> Path:
|
|
if destination.exists():
|
|
raise IngestError(f"Refusing to replace existing artifact without --resume: {destination}")
|
|
payload = _request_bytes(url, playlist_url=playlist_url)
|
|
if not payload:
|
|
raise IngestError(f"Downloaded resource is empty: {fetch.display_url(url)}")
|
|
fd, temporary_name = tempfile.mkstemp(
|
|
prefix=f".{destination.stem}.", suffix=destination.suffix, dir=destination.parent
|
|
)
|
|
try:
|
|
with os.fdopen(fd, "wb") as handle:
|
|
handle.write(payload)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary_name, destination)
|
|
finally:
|
|
Path(temporary_name).unlink(missing_ok=True)
|
|
return destination
|
|
|
|
|
|
def _download_cover(
|
|
url: str | None,
|
|
destination: Path,
|
|
*,
|
|
playlist_url: str,
|
|
ffmpeg: str,
|
|
resume: bool,
|
|
) -> Path | None:
|
|
if not url:
|
|
return None
|
|
if destination.exists() and resume:
|
|
return destination
|
|
with tempfile.TemporaryDirectory(prefix="materialsub-cover-") as directory:
|
|
raw = Path(directory) / "cover.input"
|
|
raw.write_bytes(_request_bytes(url, playlist_url=playlist_url))
|
|
if not raw.stat().st_size:
|
|
raise IngestError("Downloaded cover is empty")
|
|
temporary = Path(directory) / "cover.jpg"
|
|
result = fetch._run(
|
|
[
|
|
ffmpeg,
|
|
"-hide_banner",
|
|
"-loglevel",
|
|
"error",
|
|
"-nostdin",
|
|
"-i",
|
|
str(raw),
|
|
"-frames:v",
|
|
"1",
|
|
str(temporary),
|
|
],
|
|
"converting the embedded cover to JPEG",
|
|
secrets=(url, str(raw), str(temporary)),
|
|
check=False,
|
|
)
|
|
if result.returncode or not temporary.is_file() or not temporary.stat().st_size:
|
|
raise IngestError("Failed to convert the embedded cover to JPEG")
|
|
os.replace(temporary, destination)
|
|
return destination
|
|
|
|
|
|
def _subtitle_choice(resource_map: dict[str, Any], target_language: str) -> tuple[fetch.SubtitleChoice | None, dict[str, Any] | None]:
|
|
subtitle = resource_map.get("subtitle")
|
|
if subtitle is None:
|
|
return None, None
|
|
if not isinstance(subtitle, dict):
|
|
raise IngestError("Resource map subtitle must be an object or null")
|
|
url = _validated_url(subtitle.get("url"), "subtitle.url")
|
|
language = _required_text(subtitle.get("language"), "subtitle.language")
|
|
kind = _required_text(subtitle.get("kind"), "subtitle.kind").lower()
|
|
if kind not in ALLOWED_SUBTITLE_KINDS:
|
|
raise IngestError("subtitle.kind must be 'manual' or 'automatic'")
|
|
original_format = str(subtitle.get("format") or Path(urllib.parse.urlsplit(url).path).suffix.lstrip(".") or "vtt").lower()
|
|
if not _EXTENSION.fullmatch(original_format):
|
|
raise IngestError("subtitle.format must be a short filename extension")
|
|
if fetch._excluded_language(language, fetch._target_language_bases(target_language)):
|
|
raise IngestError("Selected subtitle is not a foreign-language dialogue track for the requested target language")
|
|
return fetch.SubtitleChoice(language, kind, original_format, (original_format,)), subtitle
|
|
|
|
|
|
def _download_hls(
|
|
playlist_url: str,
|
|
output_dir: Path,
|
|
base: str,
|
|
*,
|
|
yt_dlp: str,
|
|
concurrent_fragments: int,
|
|
) -> tuple[Path, Any]:
|
|
source = _request_bytes(playlist_url).decode("utf-8-sig")
|
|
rewritten = rewrite_media_playlist(source, playlist_url)
|
|
with tempfile.TemporaryDirectory(prefix="materialsub-authorized-hls-") as directory:
|
|
playlist_path = Path(directory) / "media.m3u8"
|
|
playlist_path.write_text(rewritten, encoding="utf-8")
|
|
os.chmod(playlist_path, 0o600)
|
|
result = fetch._run(
|
|
[
|
|
yt_dlp,
|
|
"--ignore-config",
|
|
"--no-playlist",
|
|
"--no-write-playlist-metafiles",
|
|
"--no-progress",
|
|
"--enable-file-urls",
|
|
"--concurrent-fragments",
|
|
str(concurrent_fragments),
|
|
"-P",
|
|
str(output_dir),
|
|
"-f",
|
|
fetch.FORMAT_SELECTOR,
|
|
"--merge-output-format",
|
|
"mkv",
|
|
"--remux-video",
|
|
"mkv",
|
|
"--no-overwrites",
|
|
"--no-post-overwrites",
|
|
"-o",
|
|
f"{base}.intermediate.%(ext)s",
|
|
playlist_path.as_uri(),
|
|
],
|
|
"downloading the authorized embedded HLS media playlist",
|
|
secrets=(playlist_url, urllib.parse.urlsplit(playlist_url).query, rewritten),
|
|
)
|
|
intermediate = fetch._artifact(output_dir, f"{base}.intermediate.")
|
|
if intermediate is None:
|
|
raise IngestError("yt-dlp completed but no embedded HLS intermediate was written")
|
|
return intermediate, result
|
|
|
|
|
|
def _build_manifest(
|
|
resource_map: dict[str, Any],
|
|
output_dir: Path,
|
|
*,
|
|
target_language: str,
|
|
deliverable: str,
|
|
choice: fetch.SubtitleChoice | None,
|
|
) -> dict[str, Any]:
|
|
title = _required_text(resource_map.get("title"), "title")
|
|
video_id = _required_text(resource_map.get("id"), "id")
|
|
page_url = _validated_url(resource_map.get("page_url"), "page_url")
|
|
info = {
|
|
"title": title,
|
|
"id": video_id,
|
|
"webpage_url": page_url,
|
|
"extractor_key": "AuthorizedEmbeddedHLS",
|
|
"duration": _optional_number(resource_map.get("duration_seconds"), "duration_seconds"),
|
|
"width": _optional_number(resource_map.get("width"), "width"),
|
|
"height": _optional_number(resource_map.get("height"), "height"),
|
|
"language": choice.language if choice else resource_map.get("declared_language"),
|
|
}
|
|
manifest = fetch._manifest_base(
|
|
info=info,
|
|
url=page_url,
|
|
output_dir=output_dir,
|
|
browser_cookies="browser-session",
|
|
allow_remote_ejs=False,
|
|
choice=choice,
|
|
deliverable=deliverable,
|
|
target_language=target_language,
|
|
)
|
|
manifest["authentication"]["mode"] = "browser-confirmed-resource-map"
|
|
manifest["execution"] = {
|
|
"resume": False,
|
|
"embedded_hls": True,
|
|
"authorization_query_logged": False,
|
|
"temporary_playlist_cleaned": True,
|
|
}
|
|
return manifest
|
|
|
|
|
|
def execute(args: argparse.Namespace) -> int:
|
|
resource_map_path = args.resource_map.expanduser().resolve()
|
|
resource_map = _read_private_resource_map(resource_map_path)
|
|
target_language = fetch._validated_target_language(args.target_lang)
|
|
playlist_url = _validated_url(resource_map.get("playlist_url"), "playlist_url")
|
|
choice, subtitle_config = _subtitle_choice(resource_map, target_language)
|
|
if args.deliver in fetch.SUBTITLE_ONLY_DELIVERABLES and choice is None:
|
|
raise IngestError("A subtitle-only delivery requires one selected source subtitle")
|
|
|
|
yt_dlp = fetch._require_executable("yt-dlp", "Install the current official yt-dlp release.")
|
|
ffmpeg = fetch._require_executable("ffmpeg", "Install FFmpeg.")
|
|
ffprobe = fetch._require_executable("ffprobe", "Install FFmpeg with ffprobe.")
|
|
output_dir = _prepare_output_dir(
|
|
args.output_dir.expanduser().resolve(),
|
|
resource_map_path,
|
|
resume=args.resume,
|
|
cleanup_resource_map=args.cleanup_resource_map,
|
|
)
|
|
manifest = _build_manifest(
|
|
resource_map,
|
|
output_dir,
|
|
target_language=target_language,
|
|
deliverable=args.deliver,
|
|
choice=choice,
|
|
)
|
|
manifest["execution"]["resume"] = bool(args.resume)
|
|
title = manifest["source"]["title"]
|
|
video_id = manifest["source"]["id"]
|
|
base = fetch.safe_stem(title, video_id)
|
|
names = manifest["delivery_names"]
|
|
subtitles_only = args.deliver in fetch.SUBTITLE_ONLY_DELIVERABLES
|
|
warnings: list[str] = []
|
|
completed: list[Any] = []
|
|
|
|
intermediate: Path | None = None
|
|
media_probe: dict[str, Any] = {}
|
|
master: Path | None = None
|
|
fallback: Path | None = None
|
|
fallback_error: str | None = None
|
|
if not subtitles_only:
|
|
print("Downloading authorized embedded HLS at maximum observed quality…", file=os.sys.stderr)
|
|
intermediate, completed_result = _download_hls(
|
|
playlist_url,
|
|
output_dir,
|
|
base,
|
|
yt_dlp=yt_dlp,
|
|
concurrent_fragments=args.concurrent_fragments,
|
|
)
|
|
completed.append(completed_result)
|
|
media_probe = fetch._ffprobe(intermediate, ffprobe)
|
|
if not any(
|
|
isinstance(stream, dict) and stream.get("codec_type") == "video"
|
|
for stream in media_probe.get("streams", [])
|
|
):
|
|
raise IngestError("Embedded HLS intermediate has no video stream")
|
|
master, remux_error = fetch._try_lossless_mp4(
|
|
intermediate,
|
|
output_dir / f"{base}.master.mp4",
|
|
ffmpeg,
|
|
replace_existing=args.resume,
|
|
)
|
|
if remux_error:
|
|
warnings.append(f"Lossless MP4 remux unavailable: {remux_error}")
|
|
if master is None and args.mp4_fallback:
|
|
fallback, fallback_error = fetch._create_fallback_mp4(
|
|
intermediate,
|
|
output_dir / f"{base}.fallback.mp4",
|
|
ffmpeg,
|
|
replace_existing=args.resume,
|
|
)
|
|
if fallback_error:
|
|
raise IngestError(f"Requested MP4 fallback failed: {fallback_error}")
|
|
|
|
cover = _download_cover(
|
|
resource_map.get("cover_url"),
|
|
output_dir / names["cover"],
|
|
playlist_url=playlist_url,
|
|
ffmpeg=ffmpeg,
|
|
resume=args.resume,
|
|
)
|
|
if not subtitles_only and cover is None:
|
|
warnings.append("The embedded player resource map did not declare a cover")
|
|
|
|
original_subtitle: Path | None = None
|
|
source_srt: Path | None = None
|
|
conversion_method: str | None = None
|
|
parent_hash: str | None = None
|
|
if choice and subtitle_config:
|
|
language_label = fetch._subtitle_language_label(choice.language)
|
|
original_subtitle = output_dir / (
|
|
f"{base}.source-original.{language_label}.{choice.original_format}"
|
|
)
|
|
if not (args.resume and original_subtitle.is_file()):
|
|
_atomic_download(
|
|
_validated_url(subtitle_config.get("url"), "subtitle.url"),
|
|
original_subtitle,
|
|
playlist_url=playlist_url,
|
|
)
|
|
source_srt, conversion_method, parent_hash = fetch._derive_source_srt(
|
|
original_subtitle,
|
|
output_dir / f"{base}.source-srt.{language_label}.srt",
|
|
ffmpeg=ffmpeg,
|
|
replace_existing=args.resume,
|
|
)
|
|
|
|
original_record = fetch._file_record(original_subtitle, output_dir, checksum=True)
|
|
source_srt_record = fetch._file_record(source_srt, output_dir, checksum=True)
|
|
if original_record and source_srt_record:
|
|
if original_record.get("sha256") != parent_hash:
|
|
raise IngestError("Original subtitle changed during SRT derivation")
|
|
original_record["content_role"] = "immutable-parent"
|
|
source_srt_record["conversion_method"] = conversion_method
|
|
source_srt_record["derived_from"] = {
|
|
"path": original_record["path"],
|
|
"sha256": parent_hash,
|
|
}
|
|
|
|
manifest["status"] = "downloaded"
|
|
manifest["artifacts"] = {
|
|
"intermediate": fetch._file_record(intermediate, output_dir),
|
|
"media_streams": media_probe.get("streams", []),
|
|
"lossless_mp4_master": fetch._file_record(master, output_dir),
|
|
"lossy_mp4_fallback": {
|
|
"requested": bool(args.mp4_fallback),
|
|
"created": fetch._file_record(fallback, output_dir),
|
|
"reason_not_created": (
|
|
"lossless_master_available"
|
|
if args.mp4_fallback and master is not None
|
|
else (fallback_error if args.mp4_fallback and fallback is None else None)
|
|
),
|
|
"video_encoding": "libx264 preset=slow crf=18" if fallback else None,
|
|
"audio_encoding": "aac 256k" if fallback else None,
|
|
},
|
|
"cover": fetch._file_record(cover, output_dir, checksum=True),
|
|
"subtitle": {
|
|
"language": choice.language,
|
|
"kind": choice.kind,
|
|
"label": subtitle_config.get("label"),
|
|
"original": original_record,
|
|
"source_srt": source_srt_record,
|
|
"original_is_never_modified_by_this_script": True,
|
|
}
|
|
if choice and subtitle_config
|
|
else None,
|
|
}
|
|
manifest["warnings"].extend(warnings)
|
|
manifest["warnings"].extend(
|
|
warning
|
|
for warning in fetch._warning_lines(
|
|
completed, (playlist_url, urllib.parse.urlsplit(playlist_url).query)
|
|
)
|
|
if warning not in manifest["warnings"]
|
|
)
|
|
destination = fetch._write_manifest(output_dir, manifest)
|
|
exit_code = fetch._advance_bilingual_stage(destination)
|
|
if args.cleanup_resource_map:
|
|
resource_map_path.unlink()
|
|
return exit_code
|
|
|
|
|
|
def _parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Ingest one browser-confirmed authorized embedded HLS media playlist."
|
|
)
|
|
parser.add_argument("--resource-map", type=Path, required=True)
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
|
parser.add_argument("--target-lang", default=fetch.DEFAULT_TARGET_LANGUAGE)
|
|
parser.add_argument("--deliver", choices=fetch.DELIVERABLES, default="full")
|
|
parser.add_argument("--concurrent-fragments", type=int, default=8)
|
|
parser.add_argument("--mp4-fallback", action="store_true")
|
|
parser.add_argument("--resume", action="store_true")
|
|
parser.add_argument(
|
|
"--cleanup-resource-map",
|
|
action="store_true",
|
|
help="delete the exact private resource-map file after a successful manifest write",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = _parser()
|
|
args = parser.parse_args(argv)
|
|
if not 1 <= args.concurrent_fragments <= 32:
|
|
parser.error("--concurrent-fragments must be between 1 and 32")
|
|
try:
|
|
return execute(args)
|
|
except (IngestError, fetch.FetchError) as exc:
|
|
print(f"error: {fetch.sanitize_diagnostic(str(exc))}", file=os.sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|