164 lines
4.6 KiB
Python
164 lines
4.6 KiB
Python
# License: GNU AGPL v3: http://www.gnu.org/licenses/
|
|
# This file is part of `delarte` (https://git.afpy.org/fcode/delarte.git)
|
|
|
|
"""delarte - ArteTV downloader."""
|
|
|
|
__version__ = "0.1"
|
|
|
|
from .error import *
|
|
from .model import *
|
|
|
|
|
|
def fetch_sources(http_session, url):
|
|
"""Fetch sources at a given ArteTV page URL."""
|
|
from .api import fetch_program_info
|
|
from .hls import fetch_program_tracks
|
|
from .www import parse_url
|
|
|
|
site, program_id, slug = parse_url(url)
|
|
|
|
variants = dict()
|
|
renditions = dict()
|
|
|
|
p_meta, program_index_urls = fetch_program_info(http_session, site, program_id)
|
|
|
|
program = Program(program_id, slug, p_meta)
|
|
|
|
for program_index_url in program_index_urls:
|
|
v_tracks, a_track, s_track = fetch_program_tracks(
|
|
http_session, program_index_url
|
|
)
|
|
for v_meta, v_url in v_tracks:
|
|
if v_meta not in variants:
|
|
variants[v_meta] = v_url
|
|
elif variants[v_meta] != v_url:
|
|
raise ValueError
|
|
|
|
a_meta, a_url = a_track
|
|
s_meta, s_url = s_track or (None, None)
|
|
|
|
if (a_meta, s_meta) not in renditions:
|
|
renditions[(a_meta, s_meta)] = (a_url, s_url)
|
|
elif renditions[(a_meta, s_meta)] != (a_url, s_url):
|
|
raise ValueError
|
|
|
|
return Sources(
|
|
program,
|
|
[Variant(key, source) for key, source in variants.items()],
|
|
[Rendition(key, source) for key, source in renditions.items()],
|
|
)
|
|
|
|
|
|
def iter_renditions(sources):
|
|
"""Iterate over renditions (code, key) of the given sources."""
|
|
keys = [r.key for r in sources.renditions]
|
|
|
|
keys.sort(
|
|
key=lambda k: (
|
|
not k[0].is_original,
|
|
k[0].language,
|
|
k[0].is_descriptive,
|
|
k[1].language if k[1] else "",
|
|
k[1].is_descriptive if k[1] else False,
|
|
)
|
|
)
|
|
|
|
for (a_meta, s_meta) in keys:
|
|
code = a_meta.language
|
|
|
|
if a_meta.is_descriptive:
|
|
code += "[AD]"
|
|
|
|
if s_meta:
|
|
if s_meta.is_descriptive:
|
|
code += f"-{s_meta.language}[CC]"
|
|
elif s_meta.language != a_meta.language:
|
|
code += f"-{s_meta.language}"
|
|
|
|
yield code, (a_meta, s_meta)
|
|
|
|
|
|
def select_rendition(sources, key):
|
|
"""Reject all other renditions from the given sources."""
|
|
renditions = [r for r in sources.renditions if r.key == key]
|
|
match len(renditions):
|
|
case 0:
|
|
raise ValueError("rendition not found")
|
|
case 1:
|
|
pass
|
|
case _:
|
|
raise ValueError("non unique rendition")
|
|
|
|
sources.renditions[:] = renditions
|
|
|
|
|
|
def iter_variants(sources):
|
|
"""Iterate over variants (code, key) of the given sources."""
|
|
import itertools
|
|
|
|
keys = [v.key for v in sources.variants]
|
|
|
|
keys.sort(key=lambda k: (k.height, k.frame_rate), reverse=True)
|
|
|
|
for height, group in itertools.groupby(keys, lambda m: m.height):
|
|
group = list(group)
|
|
if len(group) == 1:
|
|
yield f"{height}p", group[0]
|
|
else:
|
|
for m in group:
|
|
yield f"{height}p@{m.frame_rate}", m
|
|
|
|
|
|
def select_variant(sources, key):
|
|
"""Reject all other variants from the given sources."""
|
|
variants = [v for v in sources.variants if v.key == key]
|
|
match len(variants):
|
|
case 0:
|
|
raise ValueError("variant not found")
|
|
case 1:
|
|
pass
|
|
case _:
|
|
raise ValueError("non unique variant")
|
|
|
|
sources.variants[:] = variants
|
|
|
|
|
|
def compile_sources(sources, **naming_options):
|
|
"""Return target from the given sources."""
|
|
from .naming import file_name_builder
|
|
|
|
match len(sources.variants):
|
|
case 0:
|
|
raise ValueError("no variants")
|
|
case 1:
|
|
v_meta, v_url = sources.variants[0]
|
|
case _:
|
|
raise ValueError("multiple variants")
|
|
|
|
match len(sources.renditions):
|
|
case 0:
|
|
raise ValueError("no renditions")
|
|
case 1:
|
|
(a_meta, s_meta), (a_url, s_url) = sources.renditions[0]
|
|
case _:
|
|
raise ValueError("multiple renditions")
|
|
|
|
build_file_name = file_name_builder(v_meta, a_meta, s_meta, **naming_options)
|
|
|
|
return Target(
|
|
sources.program.meta,
|
|
VideoTrack(v_meta, v_url),
|
|
AudioTrack(a_meta, a_url),
|
|
SubtitlesTrack(s_meta, s_url) if s_meta else None,
|
|
build_file_name(sources.program),
|
|
)
|
|
|
|
|
|
def download_target(http_session, target, progress):
|
|
"""Download the given target."""
|
|
from .hls import download_target_tracks
|
|
from .muxing import mux_target
|
|
|
|
with download_target_tracks(http_session, target, progress) as local_target:
|
|
mux_target(local_target, progress)
|