diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md
index 654555be..0c85c4c6 100644
--- a/docs/guide/downloading.md
+++ b/docs/guide/downloading.md
@@ -843,6 +843,10 @@ The [mediaexport](https://github.com/unshackle-dl/mediaexport) package specifies
service. If a service never sets it, importing that export fails with a message naming
the service. Neither end guesses a language for you, so the fix belongs in the service.
+ An HLS master playlist gives no language for a video track. When the export gives no
+ title language, the video track takes the language of the default audio track. When
+ the master playlist gives no audio language either, the import stops with a message.
+
## Metadata and tagging
unshackle looks up metadata automatically, but you can override the identifiers used for
diff --git a/tests/core/test_import_foreign_hls.py b/tests/core/test_import_foreign_hls.py
new file mode 100644
index 00000000..d07e2156
--- /dev/null
+++ b/tests/core/test_import_foreign_hls.py
@@ -0,0 +1,144 @@
+"""An HLS export from another tool: only a master playlist URL, title keys, and no track rows.
+
+The importer parses the master playlist for the tracks. Such a file can have no title
+language, and its master playlist can name no key, so each track's own media playlist is
+the only place that names its KIDs.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import struct
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, Optional, cast
+from unittest.mock import patch
+from uuid import UUID
+
+import click
+import pytest
+
+from unshackle.core.import_service import ImportService
+from unshackle.core.manifests import HLS
+from unshackle.core.tracks import Video
+
+MASTER_URL = "https://example.invalid/hls/playlist.m3u8?t=1"
+VIDEO_KID = UUID("00000000-499d-c8c7-6332-202020202020")
+AUDIO_KID = UUID("00000000-499d-c8c7-6336-202020202020")
+OTHER_KID = UUID("11111111-2222-3333-4444-555555555555")
+KEYS = {VIDEO_KID: "0000000000000000000000000000000a", AUDIO_KID: "0000000000000000000000000000000b"}
+KEYS[OTHER_KID] = "0000000000000000000000000000000c"
+
+
+def playready_object(kid: UUID) -> str:
+ """A base64 PlayReady object whose one WRM header names ``kid``."""
+ value = base64.b64encode(kid.bytes_le).decode()
+ xml = (
+ ''
+ f''
+ ).encode("utf-16-le")
+ record = struct.pack(" str:
+ key = ""
+ if kid:
+ key = (
+ '#EXT-X-KEY:METHOD=SAMPLE-AES,KEYFORMAT="com.microsoft.playready",KEYFORMATVERSIONS="1",'
+ f'URI="data:text/plain;charset=UTF-16;base64,{playready_object(kid)}"\n'
+ )
+ return (
+ "#EXTM3U\n#EXT-X-TARGETDURATION:6\n#EXT-X-PLAYLIST-TYPE:VOD\n"
+ f'#EXT-X-MAP:URI="init.mp4"\n{key}#EXTINF:6.0,\nseg0.m4s\n#EXT-X-ENDLIST\n'
+ )
+
+
+def master(audio_language: str = "en") -> str:
+ lang = f'LANGUAGE="{audio_language}",' if audio_language else ""
+ return (
+ "#EXTM3U\n"
+ f'#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",{lang}NAME="Main",DEFAULT=YES,URI="audio.m3u8"\n'
+ '#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="sub",LANGUAGE="en",NAME="English",URI="sub.m3u8"\n'
+ '#EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080,CODECS="avc1.640028,ec-3",'
+ 'AUDIO="aud",SUBTITLES="sub"\nvideo.m3u8\n'
+ )
+
+
+def write_export(tmp_path: Path, language: str = "", manifest_url: str = MASTER_URL) -> Path:
+ """An export in the other tool's own format, as that tool writes it."""
+ title: dict[str, Any] = {"id": "ep-1", "kind": "episode", "name": "Example Show", "season": 1, "episode": 1}
+ if language:
+ title["language"] = language
+ doc = {
+ "kind": "unidl-export",
+ "version": 1,
+ "service": "example",
+ "titles": [
+ {
+ "save_name": "Example.Show.S01E01",
+ "title": title,
+ "keys": [f"{kid.hex}:{key}" for kid, key in KEYS.items()],
+ "manifest_url": manifest_url,
+ "headers": {"User-Agent": "Example/1.0"},
+ "drm": {"system": "playready", "wrm_header": ""},
+ }
+ ],
+ }
+ path = tmp_path / "export.json"
+ path.write_text(json.dumps(doc), encoding="utf8")
+ return path
+
+
+def import_title(export: Path, master_text: str, playlists: dict[str, Optional[UUID]]) -> tuple[ImportService, Any]:
+ ctx = cast(click.Context, SimpleNamespace(parent=None, params={}))
+ svc = ImportService(ctx, "EXAMPLE", "ep-1", str(export))
+
+ def get(url: str, **_: Any) -> SimpleNamespace:
+ name = url.split("?")[0].rsplit("/", 1)[-1]
+ return SimpleNamespace(text=media_playlist(playlists.get(name)), raise_for_status=lambda: None)
+
+ setattr(svc.session, "get", get)
+ episode = next(iter(svc.get_titles()))
+ with patch.object(HLS, "from_url", side_effect=lambda url, session=None, **kw: HLS.from_text(master_text, url)):
+ episode.tracks = svc.get_tracks(episode)
+ return svc, episode
+
+
+def test_no_title_language_takes_the_default_audio_language(tmp_path: Path) -> None:
+ _, episode = import_title(write_export(tmp_path), master(), {})
+
+ assert [str(t.language) for t in episode.tracks.videos] == ["en"]
+
+
+def test_no_language_anywhere_names_the_missing_language(tmp_path: Path) -> None:
+ with pytest.raises(click.ClickException, match="No language"):
+ import_title(write_export(tmp_path), master(audio_language=""), {})
+
+
+def test_each_hls_track_gets_the_keys_its_media_playlist_names(tmp_path: Path) -> None:
+ """The master playlist names no key, so only the media playlist tells the tracks apart."""
+ svc, episode = import_title(
+ write_export(tmp_path, "en"), master(), {"video.m3u8": VIDEO_KID, "audio.m3u8": AUDIO_KID}
+ )
+ svc.resolve_server_keys(episode)
+
+ video, audio = episode.tracks.videos[0], episode.tracks.audio[0]
+ assert video.drm and video.drm[0].content_keys == {VIDEO_KID: KEYS[VIDEO_KID]}
+ assert audio.drm and audio.drm[0].content_keys == {AUDIO_KID: KEYS[AUDIO_KID]}
+ assert all(not t.drm for t in episode.tracks.subtitles)
+
+
+def test_an_hls_track_whose_playlist_names_no_key_stays_clear(tmp_path: Path) -> None:
+ svc, episode = import_title(write_export(tmp_path, "en"), master(), {"video.m3u8": VIDEO_KID})
+ svc.resolve_server_keys(episode)
+
+ assert not episode.tracks.audio[0].drm
+ assert isinstance(episode.tracks.videos[0], Video) and episode.tracks.videos[0].drm
+
+
+def test_a_manifest_that_unshackle_cannot_read_stops_the_import(tmp_path: Path) -> None:
+ """A manifest held only in the other tool's own form must not import as a title with no tracks."""
+ with pytest.raises(click.ClickException, match="cannot read"):
+ import_title(write_export(tmp_path, "en", manifest_url="x-unidl:json_manifest"), master(), {})
diff --git a/unshackle/core/import_service.py b/unshackle/core/import_service.py
index 5f8537c3..efba9b18 100644
--- a/unshackle/core/import_service.py
+++ b/unshackle/core/import_service.py
@@ -8,6 +8,7 @@ from typing import Any, Callable, Optional, TypeVar, Union
from uuid import UUID
import click
+import m3u8
import mediaexport
import requests
from langcodes import tag_is_valid
@@ -42,6 +43,43 @@ def fallback_language(language: Any) -> Optional[Any]:
return language
+def hls_audio_language(manifest: HLS) -> Optional[str]:
+ """The language of the default audio rendition, else of the first audio rendition, else None.
+
+ An HLS master playlist gives no language for a video variant, so the parser needs a
+ fallback. An export from another tool can give no title language.
+ """
+ audio = [m for m in manifest.manifest.media if m.type == "AUDIO" and fallback_language(m.language)]
+ audio.sort(key=lambda m: m.default != "YES")
+ return str(audio[0].language) if audio else None
+
+
+def hls_playlist_kids(track: Any, session: Optional[requests.Session] = None) -> Optional[set[UUID]]:
+ """Return the KIDs that the Widevine and PlayReady keys of the track's media playlist name.
+
+ Return None when the playlist has no key or only AES-128 keys: the download gets an AES-128
+ key from its URI, so such a track takes no exported content key. An empty set means that
+ the track is encrypted but names no KID that unshackle can read.
+ """
+ try:
+ text = (session or requests.Session()).get(str(track.url)).text
+ keys = [k for k in m3u8.loads(text, uri=str(track.url)).keys if k is not None and k.method != "NONE"]
+ except Exception as e:
+ log.warning(f"Cannot read the media playlist of {track.id} for its KIDs: {e!r}")
+ return None
+ kids: set[UUID] = set()
+ for key in keys:
+ if key.method == "AES-128":
+ continue
+ try:
+ kids.update(real_kids(HLS.get_drm(key)))
+ except Exception as e:
+ log.debug(f"Skipping HLS key {key.keyformat or key.method} of {track.id}: {e!r}")
+ if kids or any(k.method != "AES-128" for k in keys):
+ return kids
+ return None
+
+
def resolve_import_manifest_data(
tracks: Tracks,
manifest_type: Optional[str],
@@ -316,8 +354,16 @@ class ImportService:
f"Failed to fetch the {manifest_type} manifest for '{title}'. "
f"The manifest URL may have expired since export. ({e})"
)
+ language = fallback_language(title.language)
+ if language is None and isinstance(manifest, HLS):
+ language = hls_audio_language(manifest)
+ if language is None:
+ raise click.ClickException(
+ f"No language for '{title}': the export gives no title language and the HLS "
+ "manifest gives no audio language."
+ )
try:
- parsed = manifest.to_tracks(language=fallback_language(title.language))
+ parsed = manifest.to_tracks(language=language)
except ValueError as e:
if "Language information could not be derived" in str(e):
raise click.ClickException(
@@ -365,6 +411,11 @@ class ImportService:
track.drm = drm
tracks.add(track, warn_only=True)
+ if manifest_url and parser is None and not [*tracks.videos, *tracks.audio, *tracks.subtitles]:
+ raise click.ClickException(
+ f"Cannot import '{title}': unshackle cannot read its manifest {str(manifest_url)[:80]!r}."
+ )
+
resolve_import_manifest_data(
tracks,
manifest_type,
@@ -608,7 +659,8 @@ class ImportService:
reads only ``track.drm``. A rung that misses content key injection here downloads
encrypted and muxes without error. A DASH manifest with no ContentProtection can still
describe an encrypted track, with the PSSH only in the init segment, so that case
- probes the init segment with `session`.
+ probes the init segment with `session`. An HLS track with no DRM gets its KIDs from the
+ EXT-X-KEY tags of its media playlist, because a master playlist can name no KID.
"""
if track.drm:
return {kid for drm in track.drm for kid in real_kids(drm)}
@@ -642,6 +694,8 @@ class ImportService:
manifest = ism.get("manifest")
if manifest is not None and manifest.findall(".//ProtectionHeader"):
return set()
+ if track.descriptor == Track.Descriptor.HLS and track.url:
+ return hls_playlist_kids(track, session)
return None
def exported_drm_system(self) -> str: