nyanpasu/m3u8dl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
root
2026-09-26 13:55:43 +08:00
commit 9dfc684869
40 changed files with 5632 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
N_m3u8DL_py/tests/test_api.py
+142
View File
@@ -0,0 +1,142 @@
from __future__ import annotations
from pathlib import Path
import pytest
from n_m3u8dl_py import (
DownloadCancelledError,
DownloadEventKind,
DownloadRequest,
N_m3u8DL,
SelectionError,
TrackSelection,
)
def test_public_api_inspects_dash_with_stable_track_ids() -> None:
root = Path(__file__).parents[2]
manifest = root / "src/N_m3u8DL-RE.Tests/Resources/Dash/Manifest_DuplicateSegments.mpd"
info = N_m3u8DL().inspect(manifest)
assert info.protocol == "DASH"
assert info.tracks[0].track_id == "dash:0"
assert info.tracks[0].segments_count == 3
assert info.tracks[0].duration == 6
def test_api_downloads_hls_with_default_auto_selection_and_events(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"one")
(tmp_path / "two.ts").write_bytes(b"two")
manifest = tmp_path / "movie.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXTINF:1,\ntwo.ts\n#EXT-X-ENDLIST\n")
events = []
client = N_m3u8DL()
info = client.inspect(manifest, on_event=events.append)
result = client.download(
info,
DownloadRequest(
output_dir=tmp_path / "output",
temporary_dir=tmp_path / "temp",
file_name="movie",
binary_merge=True,
),
on_event=events.append,
)
assert result.selected_track_ids == ("hls:0",)
assert result.files[0].path.read_bytes() == b"onetwo"
assert result.files[0].size == 6
assert [event.kind for event in events] == [
DownloadEventKind.INSPECT_STARTED,
DownloadEventKind.INSPECT_COMPLETED,
DownloadEventKind.DOWNLOAD_STARTED,
DownloadEventKind.PLAYLIST_LOADED,
DownloadEventKind.TRACK_STARTED,
DownloadEventKind.SEGMENT_COMPLETED,
DownloadEventKind.SEGMENT_COMPLETED,
DownloadEventKind.TRACK_COMPLETED,
DownloadEventKind.DOWNLOAD_COMPLETED,
]
segment_events = [event for event in events if event.kind is DownloadEventKind.SEGMENT_COMPLETED]
assert [(event.completed_segments, event.total_segments) for event in segment_events] == [(1, 2), (2, 2)]
def test_api_download_url_is_one_call_convenience(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"one")
manifest = tmp_path / "movie.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
result = N_m3u8DL().download_url(
manifest,
DownloadRequest(output_dir=tmp_path / "output", temporary_dir=tmp_path / "temp", file_name="one-call", binary_merge=True),
)
assert result.files[0].path.read_bytes() == b"one"
def test_api_downloads_only_explicit_track_id(tmp_path: Path) -> None:
(tmp_path / "video.ts").write_bytes(b"video")
(tmp_path / "audio.ts").write_bytes(b"audio")
(tmp_path / "video.m3u8").write_text("#EXTM3U\n#EXTINF:1,\nvideo.ts\n#EXT-X-ENDLIST\n")
(tmp_path / "audio.m3u8").write_text("#EXTM3U\n#EXTINF:1,\naudio.ts\n#EXT-X-ENDLIST\n")
manifest = tmp_path / "master.m3u8"
manifest.write_text(
"#EXTM3U\n"
'#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",NAME="English",LANGUAGE="en",URI="audio.m3u8"\n'
'#EXT-X-STREAM-INF:BANDWIDTH=1000,RESOLUTION=640x360,AUDIO="aud"\n'
"video.m3u8\n"
)
client = N_m3u8DL()
info = client.inspect(manifest)
video = next(track for track in info.tracks if track.resolution == "640x360")
result = client.download(
info,
DownloadRequest(
output_dir=tmp_path / "output",
temporary_dir=tmp_path / "temp",
file_name="video-only",
binary_merge=True,
selection=TrackSelection(track_ids=(video.track_id,)),
),
)
assert result.selected_track_ids == (video.track_id,)
assert len(result.files) == 1
assert result.files[0].path.read_bytes() == b"video"
def test_api_rejects_unknown_track_id(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"one")
manifest = tmp_path / "movie.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
client = N_m3u8DL()
info = client.inspect(manifest)
with pytest.raises(SelectionError, match="Unknown track ID"):
client.download(info, DownloadRequest(selection=TrackSelection(track_ids=("hls:404",))))
def test_api_event_callback_can_cancel_download(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"one")
manifest = tmp_path / "movie.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
events = []
def on_event(event):
events.append(event)
return False if event.kind is DownloadEventKind.TRACK_STARTED else None
client = N_m3u8DL()
info = client.inspect(manifest, on_event=on_event)
with pytest.raises(DownloadCancelledError):
client.download(
info,
DownloadRequest(output_dir=tmp_path / "output", temporary_dir=tmp_path / "temp", file_name="cancelled"),
on_event=on_event,
)
assert events[-1].kind is DownloadEventKind.CANCELLED
assert not (tmp_path / "output" / "cancelled.ts").exists()
N_m3u8DL_py/tests/test_cli.py
+18
View File
@@ -0,0 +1,18 @@
from n_m3u8dl_py.cli import build_parser
def test_cli_accepts_compatibility_options() -> None:
args = build_parser().parse_args(
[
"input.m3u8",
"--urlprocessor-args",
"token=value",
"--no-date-info",
"--mp4-real-time-decryption",
"--live-fix-vtt-by-audio",
]
)
assert args.urlprocessor_args == "token=value"
assert args.no_date_info
assert args.mp4_real_time_decryption
assert args.live_fix_vtt_by_audio
N_m3u8DL_py/tests/test_dash.py
+26
View File
@@ -0,0 +1,26 @@
from pathlib import Path
from n_m3u8dl_py.dash import DASHExtractor
ROOT = Path(__file__).parents[2]
def test_dash_resource_parses_expected_stream_shape() -> None:
path = ROOT / "src/N_m3u8DL-RE.Tests/Resources/Dash/Manifest_1080p.mpd"
streams = DASHExtractor(path.resolve().as_uri()).extract(path.read_text())
assert len(streams) == 23
first = streams[0]
assert first.bandwidth == 386437
assert first.extension == "m4s"
assert first.playlist is not None
assert first.playlist.total_duration == 736
assert first.playlist.media_init is not None
assert first.playlist.media_init.url.endswith("/1/init.mp4")
def test_dash_removes_duplicate_segment_urls() -> None:
path = ROOT / "src/N_m3u8DL-RE.Tests/Resources/Dash/Manifest_DuplicateSegments.mpd"
stream = DASHExtractor(path.resolve().as_uri()).extract(path.read_text())[0]
assert stream.playlist is not None
assert [segment.url.rsplit("/", 1)[-1] for segment in stream.playlist.segments] == ["seg-1.m4s", "seg-2.m4s", "seg-3.m4s"]
N_m3u8DL_py/tests/test_hls.py
+127
View File
@@ -0,0 +1,127 @@
from __future__ import annotations
import contextlib
import functools
import http.server
from pathlib import Path
import shutil
import subprocess
import threading
from n_m3u8dl_py.cli import main
from n_m3u8dl_py.hls import HLSExtractor
from n_m3u8dl_py.http import HttpClient
def test_hls_master_and_media_playlist(tmp_path: Path) -> None:
(tmp_path / "video.m3u8").write_text("#EXTM3U\n#EXTINF:2,\na.ts\n#EXT-X-ENDLIST\n")
master = """#EXTM3U
#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",NAME="English",LANGUAGE="en",URI="audio.m3u8"
#EXT-X-STREAM-INF:BANDWIDTH=1000,RESOLUTION=640x360,AUDIO="aud"
video.m3u8
"""
extractor = HLSExtractor((tmp_path / "master.m3u8").as_uri(), HttpClient())
streams = extractor.extract(master)
assert len(streams) == 2
video = next(stream for stream in streams if stream.resolution == "640x360")
extractor.fetch_playlists([video])
assert video.playlist is not None
assert video.playlist.segments[0].url.endswith("/a.ts")
def test_local_hls_download_and_binary_merge(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"first")
(tmp_path / "two.ts").write_bytes(b"second")
manifest = tmp_path / "test.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXTINF:1,\ntwo.ts\n#EXT-X-ENDLIST\n")
output = tmp_path / "out"
assert main([str(manifest), "--auto-select", "--binary-merge", "--save-name", "joined", "--save-dir", str(output), "--tmp-dir", str(tmp_path / "tmp")]) == 0
assert (output / "joined.ts").read_bytes() == b"firstsecond"
def test_local_hls_ffmpeg_merge_remuxes_ts_to_mp4(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"first")
manifest = tmp_path / "test.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
ffmpeg = tmp_path / "fake-ffmpeg"
ffmpeg.write_text("#!/bin/sh\nfor argument do output=$argument; done\nprintf remuxed > \"$output\"\n", encoding="utf-8")
ffmpeg.chmod(0o755)
output = tmp_path / "out"
assert main([
str(manifest),
"--auto-select",
"--ffmpeg-binary-path",
str(ffmpeg),
"--save-name",
"episode",
"--save-dir",
str(output),
"--tmp-dir",
str(tmp_path / "tmp"),
]) == 0
assert (output / "episode.mp4").read_bytes() == b"remuxed"
assert not (output / "episode.ts").exists()
def test_local_hls_save_pattern(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"first")
manifest = tmp_path / "test.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
output = tmp_path / "out"
assert main([str(manifest), "--auto-select", "--binary-merge", "--save-name", "episode", "--save-pattern", "<SaveName>_<MediaType>.<Ext>", "--save-dir", str(output), "--tmp-dir", str(tmp_path / "tmp")]) == 0
assert (output / "episode_video.ts").read_bytes() == b"first"
def test_local_hls_file_uri_decodes_spaces(tmp_path: Path) -> None:
source = tmp_path / "directory with spaces"
source.mkdir()
(source / "one.ts").write_bytes(b"space-safe")
manifest = source / "test.m3u8"
manifest.write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
output = tmp_path / "out"
assert main([str(manifest), "--auto-select", "--binary-merge", "--save-name", "spaced", "--save-dir", str(output), "--tmp-dir", str(tmp_path / "tmp")]) == 0
assert (output / "spaced.ts").read_bytes() == b"space-safe"
def test_http_hls_download(tmp_path: Path) -> None:
(tmp_path / "one.ts").write_bytes(b"served-over-http")
(tmp_path / "test.m3u8").write_text("#EXTM3U\n#EXTINF:1,\none.ts\n#EXT-X-ENDLIST\n")
output = tmp_path / "out"
with _http_server(tmp_path) as base_url:
assert main([f"{base_url}/test.m3u8", "--no-use-system-proxy", "--auto-select", "--binary-merge", "--save-name", "http", "--save-dir", str(output), "--tmp-dir", str(tmp_path / "tmp")]) == 0
assert (output / "http.ts").read_bytes() == b"served-over-http"
def test_local_hls_aes128_download(tmp_path: Path) -> None:
key = bytes(range(16))
iv = bytes(16)
plaintext = b"encrypted payload"
padding = 16 - len(plaintext) % 16
ciphertext = subprocess.run(
[shutil.which("openssl") or "openssl", "enc", "-aes-128-cbc", "-K", key.hex(), "-iv", iv.hex(), "-nopad"],
input=plaintext + bytes([padding]) * padding,
capture_output=True,
check=True,
).stdout
(tmp_path / "key.bin").write_bytes(key)
(tmp_path / "encrypted.ts").write_bytes(ciphertext)
manifest = tmp_path / "encrypted.m3u8"
manifest.write_text("#EXTM3U\n#EXT-X-KEY:METHOD=AES-128,URI=\"key.bin\",IV=0x00000000000000000000000000000000\n#EXTINF:1,\nencrypted.ts\n#EXT-X-ENDLIST\n")
output = tmp_path / "out"
assert main([str(manifest), "--auto-select", "--binary-merge", "--save-name", "plain", "--save-dir", str(output), "--tmp-dir", str(tmp_path / "tmp")]) == 0
assert (output / "plain.ts").read_bytes() == plaintext
@contextlib.contextmanager
def _http_server(directory: Path):
handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=str(directory))
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield f"http://127.0.0.1:{server.server_port}"
finally:
server.shutdown()
thread.join()
N_m3u8DL_py/tests/test_http_socks5.py
+137
View File
@@ -0,0 +1,137 @@
from __future__ import annotations
import contextlib
import functools
import http.server
import socket
import threading
from pathlib import Path
from urllib.request import Request
from n_m3u8dl_py.http import HttpClient, _Socks5HTTPSHandler
def test_http_client_routes_requests_through_no_auth_socks5_proxy(tmp_path: Path) -> None:
(tmp_path / "payload.txt").write_text("proxied", encoding="utf-8")
with _http_server(tmp_path) as target_url, _socks5_proxy() as (proxy_url, address_types):
target_url = target_url.replace("127.0.0.1", "localhost")
data, resolved, _ = HttpClient(proxy=proxy_url, use_system_proxy=False).get_bytes(f"{target_url}/payload.txt")
assert data == b"proxied"
assert resolved.endswith("/payload.txt")
assert address_types == [3]
def test_socks5_https_handler_uses_the_standard_library_context_contract() -> None:
handler = _Socks5HTTPSHandler("127.0.0.1", 1080)
captured: dict[str, object] = {}
def do_open(http_class, request, **kwargs):
captured.update(kwargs)
return "opened"
handler.do_open = do_open # type: ignore[method-assign]
assert handler.https_open(Request("https://example.test/video.m3u8")) == "opened"
assert captured == {"context": handler._context}
@contextlib.contextmanager
def _http_server(directory: Path):
handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=str(directory))
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield f"http://127.0.0.1:{server.server_port}"
finally:
server.shutdown()
server.server_close()
thread.join()
@contextlib.contextmanager
def _socks5_proxy():
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(("127.0.0.1", 0))
listener.listen()
listener.settimeout(0.2)
stop = threading.Event()
workers: list[threading.Thread] = []
address_types: list[int] = []
def serve() -> None:
while not stop.is_set():
try:
client, _ = listener.accept()
except TimeoutError:
continue
except OSError:
if stop.is_set():
return
raise
worker = threading.Thread(target=_handle_client, args=(client, address_types), daemon=True)
workers.append(worker)
worker.start()
thread = threading.Thread(target=serve, daemon=True)
thread.start()
try:
yield f"socks5://127.0.0.1:{listener.getsockname()[1]}", address_types
finally:
stop.set()
listener.close()
thread.join()
for worker in workers:
worker.join()
def _handle_client(client: socket.socket, address_types: list[int]) -> None:
with client:
assert _receive_exact(client, 2) == b"\x05\x01"
assert _receive_exact(client, 1) == b"\x00"
client.sendall(b"\x05\x00")
header = _receive_exact(client, 4)
assert header[:3] == b"\x05\x01\x00"
address_types.append(header[3])
if header[3] == 1:
host = socket.inet_ntoa(_receive_exact(client, 4))
elif header[3] == 3:
host = _receive_exact(client, _receive_exact(client, 1)[0]).decode("idna")
else:
raise AssertionError("Unexpected SOCKS address type")
port = int.from_bytes(_receive_exact(client, 2), "big")
with socket.create_connection((host, port)) as upstream:
client.sendall(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
left = threading.Thread(target=_copy, args=(client, upstream), daemon=True)
right = threading.Thread(target=_copy, args=(upstream, client), daemon=True)
left.start()
right.start()
left.join()
right.join()
def _copy(source: socket.socket, target: socket.socket) -> None:
try:
while data := source.recv(64 * 1024):
target.sendall(data)
except OSError:
pass
finally:
try:
target.shutdown(socket.SHUT_WR)
except OSError:
pass
def _receive_exact(connection: socket.socket, count: int) -> bytes:
chunks: list[bytes] = []
while count:
chunk = connection.recv(count)
if not chunk:
raise OSError("Socket closed")
chunks.append(chunk)
count -= len(chunk)
return b"".join(chunks)
N_m3u8DL_py/tests/test_utils.py
+22
View File
@@ -0,0 +1,22 @@
from n_m3u8dl_py.utils import replace_vars, truncate_filename, valid_filename
def test_replace_vars_supports_dash_number_formats() -> None:
values = {"RepresentationID": "audio", "Number": 28324}
assert replace_vars("$RepresentationID$/$Number%08x$/$Number%05d$/$Number%08X$", values) == "audio/00006ea4/28324/00006EA4"
def test_replace_vars_keeps_unknown_values() -> None:
assert replace_vars("$Number%08x$", {}) == "$Number%08x$"
def test_filename_truncation_is_stable_and_unicode_safe() -> None:
value = "中" * 100
result = truncate_filename(value, 50)
assert len(result.encode("utf-8")) <= 50
assert "\ufffd" not in result
assert valid_filename("a:b*c") == "a_b_c"
def test_replace_vars_supports_upper_and_lower_hex() -> None:
assert replace_vars("$Number%08x$/$Number%08X$", {"Number": 22999}) == "000059d7/000059D7"