yt_dlp/extractor/adn.py

   1 import base64
   2 import binascii
   3 import json
   4 import os
   5 import random
   6 import time
   7
   8 from .common import InfoExtractor
   9 from ..aes import aes_cbc_decrypt_bytes, unpad_pkcs7
  10 from ..networking.exceptions import HTTPError
  11 from ..utils import (
  12     ExtractorError,
  13     ass_subtitles_timecode,
  14     bytes_to_long,
  15     float_or_none,
  16     int_or_none,
  17     join_nonempty,
  18     long_to_bytes,
  19     parse_iso8601,
  20     pkcs1pad,
  21     str_or_none,
  22     strip_or_none,
  23     try_get,
  24     unified_strdate,
  25     urlencode_postdata,
  26 )
  27 from ..utils.traversal import traverse_obj
  28
  29
  30 class ADNBaseIE(InfoExtractor):
  31     IE_DESC = 'Animation Digital Network'
  32     _NETRC_MACHINE = 'animationdigitalnetwork'
  33     _BASE = 'animationdigitalnetwork.fr'
  34     _API_BASE_URL = f'https://gw.api.{_BASE}/'
  35     _PLAYER_BASE_URL = f'{_API_BASE_URL}player/'
  36     _HEADERS = {}
  37     _LOGIN_ERR_MESSAGE = 'Unable to log in'
  38     _RSA_KEY = (0x9B42B08905199A5CCE2026274399CA560ECB209EE9878A708B1C0812E1BB8CB5D1FB7441861147C1A1F2F3A0476DD63A9CAC20D3E983613346850AA6CB38F16DC7D720FD7D86FC6E5B3D5BBC72E14CD0BF9E869F2CEA2CCAD648F1DCE38F1FF916CEFB2D339B64AA0264372344BC775E265E8A852F88144AB0BD9AA06C1A4ABB, 65537)
  39     _POS_ALIGN_MAP = {
  40         'start': 1,
  41         'end': 3,
  42     }
  43     _LINE_ALIGN_MAP = {
  44         'middle': 8,
  45         'end': 4,
  46     }
  47
  48
  49 class ADNIE(ADNBaseIE):
  50     _VALID_URL = r'https?://(?:www\.)?animationdigitalnetwork\.com/(?:(?P<lang>de)/)?video/[^/?#]+/(?P<id>\d+)'
  51     _TESTS = [{
  52         'url': 'https://animationdigitalnetwork.com/video/558-fruits-basket/9841-episode-1-a-ce-soir',
  53         'md5': '1c9ef066ceb302c86f80c2b371615261',
  54         'info_dict': {
  55             'id': '9841',
  56             'ext': 'mp4',
  57             'title': 'Fruits Basket - Episode 1',
  58             'description': 'md5:14be2f72c3c96809b0ca424b0097d336',
  59             'series': 'Fruits Basket',
  60             'duration': 1437,
  61             'release_date': '20190405',
  62             'comment_count': int,
  63             'average_rating': float,
  64             'season_number': 1,
  65             'episode': 'À ce soir !',
  66             'episode_number': 1,
  67             'thumbnail': str,
  68             'season': 'Season 1',
  69         },
  70         'skip': 'Only available in French and German speaking Europe',
  71     }, {
  72         'url': 'https://animationdigitalnetwork.com/de/video/973-the-eminence-in-shadow/23550-folge-1',
  73         'md5': '5c5651bf5791fa6fcd7906012b9d94e8',
  74         'info_dict': {
  75             'id': '23550',
  76             'ext': 'mp4',
  77             'episode_number': 1,
  78             'duration': 1417,
  79             'release_date': '20231004',
  80             'series': 'The Eminence in Shadow',
  81             'season_number': 2,
  82             'episode': str,
  83             'title': str,
  84             'thumbnail': str,
  85             'season': 'Season 2',
  86             'comment_count': int,
  87             'average_rating': float,
  88             'description': str,
  89         },
  90         # 'skip': 'Only available in French and German speaking Europe',
  91     }]
  92
  93     def _get_subtitles(self, sub_url, video_id):
  94         if not sub_url:
  95             return None
  96
  97         enc_subtitles = self._download_webpage(
  98             sub_url, video_id, 'Downloading subtitles location', fatal=False) or '{}'
  99         subtitle_location = (self._parse_json(enc_subtitles, video_id, fatal=False) or {}).get('location')
 100         if subtitle_location:
 101             enc_subtitles = self._download_webpage(
 102                 subtitle_location, video_id, 'Downloading subtitles data',
 103                 fatal=False, headers={'Origin': 'https://' + self._BASE})
 104         if not enc_subtitles:
 105             return None
 106
 107         # http://animationdigitalnetwork.fr/components/com_vodvideo/videojs/adn-vjs.min.js
 108         dec_subtitles = unpad_pkcs7(aes_cbc_decrypt_bytes(
 109             base64.b64decode(enc_subtitles[24:]),
 110             binascii.unhexlify(self._K + '7fac1178830cfe0c'),
 111             base64.b64decode(enc_subtitles[:24])))
 112         subtitles_json = self._parse_json(dec_subtitles.decode(), None, fatal=False)
 113         if not subtitles_json:
 114             return None
 115
 116         subtitles = {}
 117         for sub_lang, sub in subtitles_json.items():
 118             ssa = '''[Script Info]
 119 ScriptType:V4.00
 120 [V4 Styles]
 121 Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,TertiaryColour,BackColour,Bold,Italic,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,AlphaLevel,Encoding
 122 Style: Default,Arial,18,16777215,16777215,16777215,0,-1,0,1,1,0,2,20,20,20,0,0
 123 [Events]
 124 Format: Marked,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text'''
 125             for current in sub:
 126                 start, end, text, line_align, position_align = (
 127                     float_or_none(current.get('startTime')),
 128                     float_or_none(current.get('endTime')),
 129                     current.get('text'), current.get('lineAlign'),
 130                     current.get('positionAlign'))
 131                 if start is None or end is None or text is None:
 132                     continue
 133                 alignment = self._POS_ALIGN_MAP.get(position_align, 2) + self._LINE_ALIGN_MAP.get(line_align, 0)
 134                 ssa += os.linesep + 'Dialogue: Marked=0,{},{},Default,,0,0,0,,{}{}'.format(
 135                     ass_subtitles_timecode(start),
 136                     ass_subtitles_timecode(end),
 137                     '{\\a%d}' % alignment if alignment != 2 else '',
 138                     text.replace('\n', '\\N').replace('<i>', '{\\i1}').replace('</i>', '{\\i0}'))
 139
 140             if sub_lang == 'vostf':
 141                 sub_lang = 'fr'
 142             elif sub_lang == 'vostde':
 143                 sub_lang = 'de'
 144             subtitles.setdefault(sub_lang, []).extend([{
 145                 'ext': 'json',
 146                 'data': json.dumps(sub),
 147             }, {
 148                 'ext': 'ssa',
 149                 'data': ssa,
 150             }])
 151         return subtitles
 152
 153     def _perform_login(self, username, password):
 154         try:
 155             access_token = (self._download_json(
 156                 self._API_BASE_URL + 'authentication/login', None,
 157                 'Logging in', self._LOGIN_ERR_MESSAGE, fatal=False,
 158                 data=urlencode_postdata({
 159                     'password': password,
 160                     'rememberMe': False,
 161                     'source': 'Web',
 162                     'username': username,
 163                 })) or {}).get('accessToken')
 164             if access_token:
 165                 self._HEADERS['Authorization'] = f'Bearer {access_token}'
 166         except ExtractorError as e:
 167             message = None
 168             if isinstance(e.cause, HTTPError) and e.cause.status == 401:
 169                 resp = self._parse_json(
 170                     e.cause.response.read().decode(), None, fatal=False) or {}
 171                 message = resp.get('message') or resp.get('code')
 172             self.report_warning(message or self._LOGIN_ERR_MESSAGE)
 173
 174     def _real_extract(self, url):
 175         lang, video_id = self._match_valid_url(url).group('lang', 'id')
 176         self._HEADERS['X-Target-Distribution'] = lang or 'fr'
 177         video_base_url = self._PLAYER_BASE_URL + f'video/{video_id}/'
 178         player = self._download_json(
 179             video_base_url + 'configuration', video_id,
 180             'Downloading player config JSON metadata',
 181             headers=self._HEADERS)['player']
 182         options = player['options']
 183
 184         user = options['user']
 185         if not user.get('hasAccess'):
 186             start_date = traverse_obj(options, ('video', 'startDate', {str}))
 187             if (parse_iso8601(start_date) or 0) > time.time():
 188                 raise ExtractorError(f'This video is not available yet. Release date: {start_date}', expected=True)
 189             self.raise_login_required('This video requires a subscription', method='password')
 190
 191         token = self._download_json(
 192             user.get('refreshTokenUrl') or (self._PLAYER_BASE_URL + 'refresh/token'),
 193             video_id, 'Downloading access token', headers={
 194                 'X-Player-Refresh-Token': user['refreshToken'],
 195             }, data=b'')['token']
 196
 197         links_url = try_get(options, lambda x: x['video']['url']) or (video_base_url + 'link')
 198         self._K = ''.join(random.choices('0123456789abcdef', k=16))
 199         message = list(json.dumps({
 200             'k': self._K,
 201             't': token,
 202         }).encode())
 203
 204         # Sometimes authentication fails for no good reason, retry with
 205         # a different random padding
 206         links_data = None
 207         for _ in range(3):
 208             padded_message = bytes(pkcs1pad(message, 128))
 209             n, e = self._RSA_KEY
 210             encrypted_message = long_to_bytes(pow(bytes_to_long(padded_message), e, n))
 211             authorization = base64.b64encode(encrypted_message).decode()
 212
 213             try:
 214                 links_data = self._download_json(
 215                     links_url, video_id, 'Downloading links JSON metadata', headers={
 216                         'X-Player-Token': authorization,
 217                         **self._HEADERS,
 218                     }, query={
 219                         'freeWithAds': 'true',
 220                         'adaptive': 'false',
 221                         'withMetadata': 'true',
 222                         'source': 'Web',
 223                     })
 224                 break
 225             except ExtractorError as e:
 226                 if not isinstance(e.cause, HTTPError):
 227                     raise e
 228
 229                 if e.cause.status == 401:
 230                     # This usually goes away with a different random pkcs1pad, so retry
 231                     continue
 232
 233                 error = self._parse_json(e.cause.response.read(), video_id)
 234                 message = error.get('message')
 235                 if e.cause.code == 403 and error.get('code') == 'player-bad-geolocation-country':
 236                     self.raise_geo_restricted(msg=message)
 237                 raise ExtractorError(message)
 238         else:
 239             raise ExtractorError('Giving up retrying')
 240
 241         links = links_data.get('links') or {}
 242         metas = links_data.get('metadata') or {}
 243         sub_url = (links.get('subtitles') or {}).get('all')
 244         video_info = links_data.get('video') or {}
 245         title = metas['title']
 246
 247         formats = []
 248         for format_id, qualities in (links.get('streaming') or {}).items():
 249             if not isinstance(qualities, dict):
 250                 continue
 251             for quality, load_balancer_url in qualities.items():
 252                 load_balancer_data = self._download_json(
 253                     load_balancer_url, video_id,
 254                     f'Downloading {format_id} {quality} JSON metadata',
 255                     headers=self._HEADERS,
 256                     fatal=False) or {}
 257                 m3u8_url = load_balancer_data.get('location')
 258                 if not m3u8_url:
 259                     continue
 260                 m3u8_formats = self._extract_m3u8_formats(
 261                     m3u8_url, video_id, 'mp4', 'm3u8_native',
 262                     m3u8_id=format_id, fatal=False)
 263                 if format_id == 'vf':
 264                     for f in m3u8_formats:
 265                         f['language'] = 'fr'
 266                 elif format_id == 'vde':
 267                     for f in m3u8_formats:
 268                         f['language'] = 'de'
 269                 formats.extend(m3u8_formats)
 270
 271         if not formats:
 272             self.raise_login_required('This video requires a subscription', method='password')
 273
 274         video = (self._download_json(
 275             self._API_BASE_URL + f'video/{video_id}', video_id,
 276             'Downloading additional video metadata', fatal=False, headers=self._HEADERS) or {}).get('video') or {}
 277         show = video.get('show') or {}
 278
 279         return {
 280             'id': video_id,
 281             'title': title,
 282             'description': strip_or_none(metas.get('summary') or video.get('summary')),
 283             'thumbnail': video_info.get('image') or player.get('image'),
 284             'formats': formats,
 285             'subtitles': self.extract_subtitles(sub_url, video_id),
 286             'episode': metas.get('subtitle') or video.get('name'),
 287             'episode_number': int_or_none(video.get('shortNumber')),
 288             'series': show.get('title'),
 289             'season_number': int_or_none(video.get('season')),
 290             'duration': int_or_none(video_info.get('duration') or video.get('duration')),
 291             'release_date': unified_strdate(video.get('releaseDate')),
 292             'average_rating': float_or_none(video.get('rating') or metas.get('rating')),
 293             'comment_count': int_or_none(video.get('commentsCount')),
 294         }
 295
 296
 297 class ADNSeasonIE(ADNBaseIE):
 298     _VALID_URL = r'https?://(?:www\.)?animationdigitalnetwork\.com/(?:(?P<lang>de)/)?video/(?P<id>\d+)[^/?#]*/?(?:$|[#?])'
 299     _TESTS = [{
 300         'url': 'https://animationdigitalnetwork.com/video/911-tokyo-mew-mew-new',
 301         'playlist_count': 12,
 302         'info_dict': {
 303             'id': '911',
 304             'title': 'Tokyo Mew Mew New',
 305         },
 306         # 'skip': 'Only available in French end German speaking Europe',
 307     }]
 308
 309     def _real_extract(self, url):
 310         lang, video_show_slug = self._match_valid_url(url).group('lang', 'id')
 311         self._HEADERS['X-Target-Distribution'] = lang or 'fr'
 312         show = self._download_json(
 313             f'{self._API_BASE_URL}show/{video_show_slug}/', video_show_slug,
 314             'Downloading show JSON metadata', headers=self._HEADERS)['show']
 315         show_id = str(show['id'])
 316         episodes = self._download_json(
 317             f'{self._API_BASE_URL}video/show/{show_id}', video_show_slug,
 318             'Downloading episode list', headers=self._HEADERS, query={
 319                 'order': 'asc',
 320                 'limit': '-1',
 321             })
 322
 323         def entries():
 324             for episode_id in traverse_obj(episodes, ('videos', ..., 'id', {str_or_none})):
 325                 yield self.url_result(join_nonempty(
 326                     'https://animationdigitalnetwork.com', lang, 'video',
 327                     video_show_slug, episode_id, delim='/'), ADNIE, episode_id)
 328
 329         return self.playlist_result(entries(), show_id, show.get('title'))