code:
import base64
import binascii
import codecs
import functools
import hashlib
import hmac
import io
import json
import re
import struct
import time
import urllib.parse
import urllib.request
import urllib.response
import uuid

from .common import InfoExtractor
from ..aes import aes_ecb_decrypt
from ..utils import (
    ExtractorError,
    bytes_to_intlist,
    decode_base_n,
    int_or_none,
    intlist_to_bytes,
    OnDemandPagedList,
    request_to_url,
    time_seconds,
    traverse_obj,
    urljoin,
)

# NOTE: network handler related code is temporary thing until network stack overhaul PRs are merged (#2861/#2862)

def add_opener(ydl, handler):
    ''' Add a handler for opening URLs, like _download_webpage '''
    # https://github.com/python/cpython/blob/main/Lib/urllib/request.py#L426
    # https://github.com/python/cpython/blob/main/Lib/urllib/request.py#L605
    assert isinstance(ydl._opener, urllib.request.OpenerDirector)
    ydl._opener.add_handler(handler)

def remove_opener(ydl, handler):
    '''
    Remove handler(s) for opening URLs
    @param handler Either handler object itself or handler type.
    Specifying handler type will remove all handler which isinstance returns True.
    '''
    # https://github.com/python/cpython/blob/main/Lib/urllib/request.py#L426
    # https://github.com/python/cpython/blob/main/Lib/urllib/request.py#L605
    opener = ydl._opener
    assert isinstance(ydl._opener, urllib.request.OpenerDirector)
    if isinstance(handler, (type, tuple)):
        find_cp = lambda x: isinstance(x, handler)
    else:
        find_cp = lambda x: x is handler

    removed = []
    for meth in dir(handler):
        if meth in ["redirect_request", "do_open", "proxy_open"]:
            # oops, coincidental match
            continue

        i = meth.find("_")
        protocol = meth[:i]
        condition = meth[i + 1:]

        if condition.startswith("error"):
            j = condition.find("_") + i + 1
            kind = meth[j + 1:]
            try:
                kind = int(kind)
            except ValueError:
                pass
            lookup = opener.handle_error.get(protocol, {})
            opener.handle_error[protocol] = lookup
        elif condition == "open":
            kind = protocol
            lookup = opener.handle_open
        elif condition == "response":
            kind = protocol
            lookup = opener.process_response
        elif condition == "request":
            kind = protocol
            lookup = opener.process_request
        else:
            continue

        handlers = lookup.setdefault(kind, [])
        if handlers:
            handlers[:] = [x for x in handlers if not find_cp(x)]

        removed.append(x for x in handlers if find_cp(x))

    if removed:
        for x in opener.handlers:
            if find_cp(x):
                x.add_parent(None)
        opener.handlers[:] = [x for x in opener.handlers if not find_cp(x)]

class AbemaLicenseHandler(urllib.request.BaseHandler):
    handler_order = 499
    STRTABLE = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
    HKEY = b'3AF0298C219469522A313570E8583005A642E73EDD58E3EA2FB7339D3DF1597E'

    def __init__(self, ie: 'AbemaTVIE'):
        # the protocol that this should really handle is 'abematv-license://'
        # abematv_license_open is just a placeholder for development purposes
        # ref. https://github.com/python/cpython/blob/f4c03484da59049eb62a9bf7777b963e2267d187/Lib/urllib/request.py#L510
        setattr(self, 'abematv-license_open', getattr(self, 'abematv_license_open'))
        self.ie = ie

    def _get_videokey_from_ticket(self, ticket):
        to_show = self.ie.get_param('verbose', False)
        media_token = self.ie._get_media_token(to_show=to_show)

        license_response = self.ie._download_json(
            'https://license.abema.io/abematv-hls', None, note='Requesting playback license' if to_show else False,
            query={'t': media_token},
            data=json.dumps({
                'kv': 'a',
                'lt': ticket
            }).encode('utf-8'),
            headers={
                'Content-Type': 'application/json',
            })

        res = decode_base_n(license_response['k'], table=self.STRTABLE)
        encvideokey = bytes_to_intlist(struct.pack('>QQ', res >> 64, res & 0xffffffffffffffff))

        h = hmac.new(
            binascii.unhexlify(self.HKEY),
            (license_response['cid'] + self.ie._DEVICE_ID).encode('utf-8'),
            digestmod=hashlib.sha256)
        enckey = bytes_to_intlist(h.digest())

        return intlist_to_bytes(aes_ecb_decrypt(encvideokey, enckey))

    def abematv_license_open(self, url):
        url = request_to_url(url)
        ticket = urllib.parse.urlparse(url).netloc
        response_data = self._get_videokey_from_ticket(ticket)
        return urllib.response.addinfourl(io.BytesIO(response_data), headers={
            'Content-Length': len(response_data),
        }, url=url, code=200)

    def augment_key_handler(self, handler):
        # For /key/:ticket
        ticket = handler.route_params['ticket']
        response_data = self._get_videokey_from_ticket(ticket)
        handler.send_header('Content-Length', str(len(response_data)))
        handler.end_headers()
        handler.wfile.write(response_data)
        return True

    def augment_hls_handler(self, handler):
        # for /hls/:hex_url
        hex_url = handler.route_params['hex_url']
        url = codecs.decode(hex_url, 'hex').decode('utf8')
        headers = dict(handler.headers)
        # drop some "bad" headers which ffmpeg likes to add or needed to be
        if headers.get('Range') == 'bytes=0-':
            headers.pop('Range', None)
        headers.pop('Host', None)
        content, uoh = self.ie._download_webpage_handle(
            url, False, note='Proxying HLS manifest request',
            fatal=True, data=None if handler.command == 'GET' else handler.rfile.read(),
            headers=headers)

        def convert_lines(line):
            if 'abematv-license://' in line:
                # rewrite abematv-license
                return line.replace('abematv-license://', '/key/')
            if line.startswith('#'):
                # metadata
                return line
            # segment url (which is usually relative)
            return urljoin(url, line)

        content = '\n'.join(map(convert_lines, content.splitlines())).encode('utf-8')
        resp_headers = dict(uoh.info())
        # drop/update some headers
        resp_headers.pop('Content-Encoding', None)
        resp_headers.pop('Keep-Alive', None)
        resp_headers['Connection'] = 'close'
        resp_headers['Content-Length'] = str(len(content))
        for k, v in resp_headers.items():
            handler.send_header(k, v)
        handler.end_headers()
        handler.wfile.write(content)
        return True

class AbemaTVBaseIE(InfoExtractor):
    _USERTOKEN = None
    _DEVICE_ID = None
    _MEDIATOKEN = None

    _SECRETKEY = b'v+Gjs=25Aw5erR!J8ZuvRrCx*rGswhB&qdHd_SYerEWdU&a?3DzN9BRbp5KwY4hEmcj5#fykMjJ=AuWz5GSMY-d@H7DMEh3M@9n2G552Us$$k9cD=3TxwWe86!x#Zyhe'

    @classmethod
    def _generate_aks(cls, deviceid):
        deviceid = deviceid.encode('utf-8')
        # add 1 hour and then drop minute and secs
        ts_1hour = int((time_seconds(hours=9) // 3600 + 1) * 3600)
        time_struct = time.gmtime(ts_1hour)
        ts_1hour_str = str(ts_1hour).encode('utf-8')

        tmp = None

        def mix_once(nonce):
            nonlocal tmp
            h = hmac.new(cls._SECRETKEY, digestmod=hashlib.sha256)
            h.update(nonce)
            tmp = h.digest()

        def mix_tmp(count):
            nonlocal tmp
            for i in range(count):
                mix_once(tmp)

        def mix_twist(nonce):
            nonlocal tmp
            mix_once(base64.urlsafe_b64encode(tmp).rstrip(b'=') + nonce)

        mix_once(cls._SECRETKEY)
        mix_tmp(time_struct.tm_mon)
        mix_twist(deviceid)
        mix_tmp(time_struct.tm_mday % 5)
        mix_twist(ts_1hour_str)
        mix_tmp(time_struct.tm_hour % 5)

        return base64.urlsafe_b64encode(tmp).rstrip(b'=').decode('utf-8')

    def _get_device_token(self):
        if self._USERTOKEN:
            return self._USERTOKEN

        AbemaTVBaseIE._DEVICE_ID = str(uuid.uuid4())
        aks = self._generate_aks(self._DEVICE_ID)
        user_data = self._download_json(
            'https://api.abema.io/v1/users', None, note='Authorizing',
            data=json.dumps({
                'deviceId': self._DEVICE_ID,
                'applicationKeySecret': aks,
            }).encode('utf-8'),
            headers={
                'Content-Type': 'application/json',
            })
        AbemaTVBaseIE._USERTOKEN = user_data['token']

        # don't allow adding it 2 times or more, though it's guarded
        remove_opener(self._downloader, AbemaLicenseHandler)
        add_opener(self._downloader, AbemaLicenseHandler(self))

        return self._USERTOKEN

    def _get_media_token(self, invalidate=False, to_show=True):
        if not invalidate and self._MEDIATOKEN:
            return self._MEDIATOKEN

        AbemaTVBaseIE._MEDIATOKEN = self._download_json(
            'https://api.abema.io/v1/media/token', None, note='Fetching media token' if to_show else False,
            query={
                'osName': 'android',
                'osVersion': '6.0.1',
                'osLang': 'ja_JP',
                'osTimezone': 'Asia/Tokyo',
                'appId': 'tv.abema',
                'appVersion': '3.27.1'
            }, headers={
                'Authorization': f'bearer {self._get_device_token()}',
            })['token']

        return self._MEDIATOKEN

    def _call_api(self, endpoint, video_id, query=None, note='Downloading JSON metadata'):
        return self._download_json(
            f'https://api.abema.io/{endpoint}', video_id, query=query or {},
            note=note,
            headers={
                'Authorization': f'bearer {self._get_device_token()}',
            })

    def _extract_breadcrumb_list(self, webpage, video_id):
        for jld in re.finditer(
                r'(?is)

',
                webpage):
            jsonld = self._parse_json(jld.group('json_ld'), video_id, fatal=False)
            if traverse_obj(jsonld, '@type') != 'BreadcrumbList':
                continue
            items = traverse_obj(jsonld, ('itemListElement', ..., 'name'))
            if items:
                return items
        return []

class AbemaTVIE(AbemaTVBaseIE):
    _VALID_URL = r'https?://abema\.tv/(?P<type>now-on-air|video/episode|channels/.+?/slots)/(?P<id>[^?/]+)'
    _NETRC_MACHINE = 'abematv'
    _TESTS = [{
        'url': 'https://abema.tv/video/episode/194-25_s2_p1',
        'info_dict': {
            'id': '194-25_s2_p1',
            'title': '第1話 「チーズケーキ」　「モーニング再び」',
            'series': '異世界食堂２',
            'series_number': 2,
            'episode': '第1話 「チーズケーキ」　「モーニング再び」',
            'episode_number': 1,
        },
        'skip': 'expired',
    }, {
        'url': 'https://abema.tv/channels/anime-live2/slots/E8tvAnMJ7a9a5d',
        'info_dict': {
            'id': 'E8tvAnMJ7a9a5d',
            'title': 'ゆるキャン△ SEASON２ 全話一挙【無料ビデオ72時間】',
            'series': 'ゆるキャン△ SEASON２',
            'episode': 'ゆるキャン△ SEASON２ 全話一挙【無料ビデオ72時間】',
            'series_number': 2,
            'episode_number': 1,
            'description': 'md5:9c5a3172ae763278f9303922f0ea5b17',
        },
        'skip': 'expired',
    }, {
        'url': 'https://abema.tv/video/episode/87-877_s1282_p31047',
        'info_dict': {
            'id': 'E8tvAnMJ7a9a5d',
            'title': '第5話『光射す』',
            'description': 'md5:56d4fc1b4f7769ded5f923c55bb4695d',
            'thumbnail': r're:https://hayabusa\.io/.+',
            'series': '相棒',
            'episode': '第5話『光射す』',
        },
        'skip': 'expired',
    }, {
        'url': 'https://abema.tv/now-on-air/abema-anime',
        'info_dict': {
            'id': 'abema-anime',
            # this varies
            # 'title': '女子高生の無駄づかい 全話一挙【無料ビデオ72時間】',
            'description': 'md5:55f2e61f46a17e9230802d7bcc913d5f',
            'is_live': True,
        },
        'skip': 'things varies over time',
    }]
    _TIMETABLE = None

    def _perform_login(self, username, password):
        if '@' in username:  # don't strictly check if it's email address or not
            ep, method = 'user/email', 'email'
        else:
            ep, method = 'oneTimePassword', 'userId'

        login_response = self._download_json(
            f'https://api.abema.io/v1/auth/{ep}', None, note='Logging in',
            data=json.dumps({
                method: username,
                'password': password
            }).encode('utf-8'), headers={
                'Authorization': f'bearer {self._get_device_token()}',
                'Origin': 'https://abema.tv',
                'Referer': 'https://abema.tv/',
                'Content-Type': 'application/json',
            })

        AbemaTVBaseIE._USERTOKEN = login_response['token']
        self._get_media_token(True)

    def _real_extract(self, url):
        # starting download using infojson from this extractor is undefined behavior,
        # and never be fixed in the future; you must trigger downloads by directly specifying URL.
        # (unless there's a way to hook before downloading by extractor)
        video_id, video_type = self._match_valid_url(url).group('id', 'type')
        headers = {
            'Authorization': 'Bearer ' + self._get_device_token(),
        }
        video_type = video_type.split('/')[-1]

        webpage = self._download_webpage(url, video_id)
        canonical_url = self._search_regex(
            r'<link\s+rel="canonical"\s*href="(.+?)"', webpage, 'canonical URL',
            default=url)
        info = self._search_json_ld(webpage, video_id, default={})

        title = self._search_regex(
            r'(.+?)', webpage, 'title', default=None)
        if not title:
            jsonld = None
            for jld in re.finditer(
                    r'(?is)(?:)?',
                    webpage):
                jsonld = self._parse_json(jld.group('json_ld'), video_id, fatal=False)
                if jsonld:
                    break
            if jsonld:
                title = jsonld.get('caption')
        if not title and video_type == 'now-on-air':
            if not self._TIMETABLE:
                # cache the timetable because it goes to 5MiB in size (!!)
                self._TIMETABLE = self._download_json(
                    'https://api.abema.io/v1/timetable/dataSet?debug=false', video_id,
                    headers=headers)
            now = time_seconds(hours=9)
            title = traverse_obj(
                self._TIMETABLE,
                ('slots',
                 # find the time slice from the requested channel in timetable
                 lambda _, v: v['channelId'] == video_id and v['startAt'] <= now and now < v['endAt'],
                 'title'), get_all=False)

        # read breadcrumb on top of page
        breadcrumb = self._extract_breadcrumb_list(webpage, video_id)
        if breadcrumb:
            # breadcrumb list translates to: (e.g. 1st test for this IE)
            # Home > Anime (genre) > Isekai Shokudo 2 (series name) > Episode 1 "Cheese cakes" "Morning again" (episode title)
            # hence this works
            info['series'] = breadcrumb[-2]
            info['episode'] = breadcrumb[-1]
            if not title:
                title = info['episode']

        description = self._html_search_regex(
            (r'
(.+?)
<div',
             r'(.+?)
<div',),
            webpage, 'description', default=None, group=1)
        if not description:
            og_desc = self._html_search_meta(
                ('description', 'og:description', 'twitter:description'), webpage)
            if og_desc:
                description = re.sub(r'''(?sx)
                    ^(.+?)(?:
                        アニメの動画を無料で見るならABEMA！| # anime
                        等、.+ # applies for most of categories
                    )?
                ''', r'\1', og_desc)

        # canonical URL may contain series and episode number
        mobj = re.search(r's(\d+)_p(\d+)$', canonical_url)
        if mobj:
            seri = int_or_none(mobj.group(1), default=float('inf'))
            epis = int_or_none(mobj.group(2), default=float('inf'))
            info['series_number'] = seri if seri < 100 else None
            # some anime like Detective Conan (though not available in AbemaTV)
            # has more than 1000 episodes (1026 as of 2021/11/15)
            info['episode_number'] = epis if epis < 2000 else None

        is_live, m3u8_url = False, None
        if video_type == 'now-on-air':
            is_live = True
            onair_channels = self._download_json(
                'https://api.abema.io/v1/channels', video_id)
            m3u8_url = traverse_obj(onair_channels, ('channels', lambda _, v: v['id'] == video_id, 'playback', 'hls'), get_all=False)
            if not m3u8_url:
                raise ExtractorError(f'Cannot find on-air {video_id} channel.', expected=True)
        elif video_type == 'episode':
            api_response = self._download_json(
                f'https://api.abema.io/v1/video/programs/{video_id}', video_id,
                note='Checking playability',
                headers=headers)
            ondemand_types = traverse_obj(api_response, ('terms', ..., 'onDemandType'), default=[])
            if 3 not in ondemand_types:
                self.report_warning('This is a premium-only stream')

            m3u8_url = f'https://vod-abematv.akamaized.net/program/{video_id}/playlist.m3u8'
        elif video_type == 'slots':
            api_response = self._download_json(
                f'https://api.abema.io/v1/media/slots/{video_id}', video_id,
                note='Checking playability',
                headers=headers)
            if not traverse_obj(api_response, ('slot', 'flags', 'timeshiftFree'), default=False):
                self.report_warning('This is a premium-only stream')

            m3u8_url = f'https://vod-abematv.akamaized.net/slot/{video_id}/playlist.m3u8'
        else:
            raise ExtractorError('Unreachable')

        if is_live:
            self.report_warning('This is a livestream; downloading livestreams from AbemaTV is an experimental feature.')
            self.report_warning('Open bug report at https://github.com/ytdl-patched/ytdl-patched/issues?q= for unplayable file')
        formats = self._extract_m3u8_formats(
            m3u8_url, video_id, ext='mp4', live=is_live)

        def aug_predicate(info_dict, dl):
            from ..downloader.external import ExternalFD
            return info_dict.get('is_live') or isinstance(dl, ExternalFD)

        from ..postprocessor.metadataparser import MetadataParserPP

        info.update({
            'id': video_id,
            'title': title,
            'description': description,
            'formats': formats,
            'is_live': is_live,
            'augments': [{
                'key': 'http_server',
                'condition': aug_predicate,
                'routes': [{
                    'route': r're:/hls/(?P<hex_url>[0-9a-fA-F]+)',
                    'callback': self._LICENSE_HANDLER.augment_hls_handler,
                }, {
                    'route': r're:/key/(?P<ticket>[^/&?]+)',
                    'callback': self._LICENSE_HANDLER.augment_key_handler,
                }]
            }, {
                'key': 'metadata_editor',
                'condition': aug_predicate,
                'actions': [
                    # neat hack here: `replace` here can be a function
                    # https://docs.python.org/3/library/re.html#re.sub
                    # the first arg is an extension of MetadataParserPP, to get info_dict
                    (MetadataParserPP.Actions.REPLACE, 'url', r'^.+$',
                        lambda info, m: f'http://localhost:{info["_httpserverport"]}/hls/{codecs.encode(m.group(0).encode("utf-8"), "hex").decode()}'),
                ],
            }],
        })
        return info

class AbemaTVTitleIE(AbemaTVBaseIE):
    _VALID_URL = r'https?://abema\.tv/video/title/(?P<id>[^?/]+)'
    _PAGE_SIZE = 25

    _TESTS = [{
        'url': 'https://abema.tv/video/title/90-1597',
        'info_dict': {
            'id': '90-1597',
            'title': 'シャッフルアイランド',
        },
        'playlist_mincount': 2,
    }, {
        'url': 'https://abema.tv/video/title/193-132',
        'info_dict': {
            'id': '193-132',
            'title': '真心が届く~僕とスターのオフィス・ラブ!?~',
        },
        'playlist_mincount': 16,
    }, {
        'url': 'https://abema.tv/video/title/25-102',
        'info_dict': {
            'id': '25-102',
            'title': 'ソードアート・オンライン アリシゼーション',
        },
        'playlist_mincount': 24,
    }]

    def _fetch_page(self, playlist_id, series_version, page):
        programs = self._call_api(
            f'v1/video/series/{playlist_id}/programs', playlist_id,
            note=f'Downloading page {page + 1}',
            query={
                'seriesVersion': series_version,
                'offset': str(page * self._PAGE_SIZE),
                'order': 'seq',
                'limit': str(self._PAGE_SIZE),
            })
        yield from (
            self.url_result(f'https://abema.tv/video/episode/{x}')
            for x in traverse_obj(programs, ('programs', ..., 'id'), default=[]))

    def _entries(self, playlist_id, series_version):
        return OnDemandPagedList(
            functools.partial(self._fetch_page, playlist_id, series_version),
            self._PAGE_SIZE)

    def _real_extract(self, url):
        playlist_id = self._match_id(url)
        series_info = self._call_api(f'v1/video/series/{playlist_id}', playlist_id)

        return self.playlist_result(
            self._entries(playlist_id, series_info['version']), playlist_id=playlist_id,
            playlist_title=series_info.get('title'),
            playlist_description=series_info.get('content'))


code:
import re

from .common import InfoExtractor

class AcademicEarthCourseIE(InfoExtractor):
    _VALID_URL = r'^https?://(?:www\.)?academicearth\.org/playlists/(?P<id>[^?#/]+)'
    IE_NAME = 'AcademicEarth:Course'
    _TEST = {
        'url': 'http://academicearth.org/playlists/laws-of-nature/',
        'info_dict': {
            'id': 'laws-of-nature',
            'title': 'Laws of Nature',
            'description': 'Introduce yourself to the laws of nature with these free online college lectures from Yale, Harvard, and MIT.',
        },
        'playlist_count': 3,
    }

    def _real_extract(self, url):
        playlist_id = self._match_id(url)

        webpage = self._download_webpage(url, playlist_id)
        title = self._html_search_regex(
            r'
]*?>(.*?)
', webpage, 'title')
        description = self._html_search_regex(
            r'
]*?>(.*?)
',
            webpage, 'description', fatal=False)
        urls = re.findall(
            r'
\s*?',
            webpage)
        entries = [self.url_result(u) for u in urls]

        return {
            '_type': 'playlist',
            'id': playlist_id,
            'title': title,
            'description': description,
            'entries': entries,
        }

repo_name: Lesmiscore/ytdl


code:
from .common import InfoExtractor
from ..utils import (
    clean_html,
    clean_podcast_url,
    int_or_none,
    parse_iso8601,
)

class ACastBaseIE(InfoExtractor):
    def _extract_episode(self, episode, show_info):
        title = episode['title']
        info = {
            'id': episode['id'],
            'display_id': episode.get('episodeUrl'),
            'url': clean_podcast_url(episode['url']),
            'title': title,
            'description': clean_html(episode.get('description') or episode.get('summary')),
            'thumbnail': episode.get('image'),
            'timestamp': parse_iso8601(episode.get('publishDate')),
            'duration': int_or_none(episode.get('duration')),
            'filesize': int_or_none(episode.get('contentLength')),
            'season_number': int_or_none(episode.get('season')),
            'episode': title,
            'episode_number': int_or_none(episode.get('episode')),
        }
        info.update(show_info)
        return info

    def _extract_show_info(self, show):
        return {
            'creator': show.get('author'),
            'series': show.get('title'),
        }

    def _call_api(self, path, video_id, query=None):
        return self._download_json(
            'https://feeder.acast.com/api/v1/shows/' + path, video_id, query=query)

class ACastIE(ACastBaseIE):
    IE_NAME = 'acast'
    _VALID_URL = r'''(?x)
                    https?://
                        (?:
                            (?:(?:embed|www)\.)?acast\.com/|
                            play\.acast\.com/s/
                        )
                        (?P<channel>[^/]+)/(?P<id>[^/#?]+)
                    '''
    _TESTS = [{
        'url': 'https://www.acast.com/sparpodcast/2.raggarmordet-rosterurdetforflutna',
        'md5': 'f5598f3ad1e4776fed12ec1407153e4b',
        'info_dict': {
            'id': '2a92b283-1a75-4ad8-8396-499c641de0d9',
            'ext': 'mp3',
            'title': '2. Raggarmordet - Röster ur det förflutna',
            'description': 'md5:a992ae67f4d98f1c0141598f7bebbf67',
            'timestamp': 1477346700,
            'upload_date': '20161024',
            'duration': 2766,
            'creator': 'Anton Berg & Martin Johnson',
            'series': 'Spår',
            'episode': '2. Raggarmordet - Röster ur det förflutna',
        }
    }, {
        'url': 'http://embed.acast.com/adambuxton/ep.12-adam-joeschristmaspodcast2015',
        'only_matching': True,
    }, {
        'url': 'https://play.acast.com/s/rattegangspodden/s04e09styckmordetihelenelund-del2-2',
        'only_matching': True,
    }, {
        'url': 'https://play.acast.com/s/sparpodcast/2a92b283-1a75-4ad8-8396-499c641de0d9',
        'only_matching': True,
    }]

    def _real_extract(self, url):
        channel, display_id = self._match_valid_url(url).groups()
        episode = self._call_api(
            '%s/episodes/%s' % (channel, display_id),
            display_id, {'showInfo': 'true'})
        return self._extract_episode(
            episode, self._extract_show_info(episode.get('show') or {}))

class ACastChannelIE(ACastBaseIE):
    IE_NAME = 'acast:channel'
    _VALID_URL = r'''(?x)
                    https?://
                        (?:
                            (?:www\.)?acast\.com/|
                            play\.acast\.com/s/
                        )
                        (?P<id>[^/#?]+)
                    '''
    _TESTS = [{
        'url': 'https://www.acast.com/todayinfocus',
        'info_dict': {
            'id': '4efc5294-5385-4847-98bd-519799ce5786',
            'title': 'Today in Focus',
            'description': 'md5:c09ce28c91002ce4ffce71d6504abaae',
        },
        'playlist_mincount': 200,
    }, {
        'url': 'http://play.acast.com/s/ft-banking-weekly',
        'only_matching': True,
    }]

    @classmethod
    def suitable(cls, url):
        return False if ACastIE.suitable(url) else super(ACastChannelIE, cls).suitable(url)

    def _real_extract(self, url):
        show_slug = self._match_id(url)
        show = self._call_api(show_slug, show_slug)
        show_info = self._extract_show_info(show)
        entries = []
        for episode in (show.get('episodes') or []):
            entries.append(self._extract_episode(episode, show_info))
        return self.playlist_result(
            entries, show.get('id'), show.get('title'), show.get('description'))


code:
from .common import InfoExtractor
from ..utils import (
    float_or_none,
    format_field,
    int_or_none,
    traverse_obj,
    parse_codecs,
    parse_qs,
)

class AcFunVideoBaseIE(InfoExtractor):
    def _extract_metadata(self, video_id, video_info):
        playjson = self._parse_json(video_info['ksPlayJson'], video_id)

        formats, subtitles = [], {}
        for video in traverse_obj(playjson, ('adaptationSet', 0, 'representation')):
            fmts, subs = self._extract_m3u8_formats_and_subtitles(video['url'], video_id, 'mp4', fatal=False)
            formats.extend(fmts)
            self._merge_subtitles(subs, target=subtitles)
            for f in fmts:
                f.update({
                    'fps': float_or_none(video.get('frameRate')),
                    'width': int_or_none(video.get('width')),
                    'height': int_or_none(video.get('height')),
                    'tbr': float_or_none(video.get('avgBitrate')),
                    **parse_codecs(video.get('codecs', ''))
                })

        self._sort_formats(formats)
        return {
            'id': video_id,
            'formats': formats,
            'subtitles': subtitles,
            'duration': float_or_none(video_info.get('durationMillis'), 1000),
            'timestamp': int_or_none(video_info.get('uploadTime'), 1000),
            'http_headers': {'Referer': 'https://www.acfun.cn/'},
        }

class AcFunVideoIE(AcFunVideoBaseIE):
    _VALID_URL = r'https?://www\.acfun\.cn/v/ac(?P<id>[_\d]+)'

    _TESTS = [{
        'url': 'https://www.acfun.cn/v/ac35457073',
        'info_dict': {
            'id': '35457073',
            'ext': 'mp4',
            'duration': 174.208,
            'timestamp': 1656403967,
            'title': '1 8 岁 现 状',
            'description': '“赶紧回去！班主任查班了！”',
            'uploader': '锤子game',
            'uploader_id': '51246077',
            'thumbnail': r're:^https?://.*\.(jpg|jpeg)',
            'upload_date': '20220628',
            'like_count': int,
            'view_count': int,
            'comment_count': int,
            'tags': list,
        },
    }, {
        # example for len(video_list) > 1
        'url': 'https://www.acfun.cn/v/ac35468952_2',
        'info_dict': {
            'id': '35468952_2',
            'ext': 'mp4',
            'title': '【动画剧集】Rocket & Groot Season 1（2022）/火箭浣熊与格鲁特第1季 P02 S01E02 十拿九穩',
            'duration': 90.459,
            'uploader': '比令',
            'uploader_id': '37259967',
            'upload_date': '20220629',
            'timestamp': 1656479962,
            'tags': list,
            'like_count': int,
            'view_count': int,
            'comment_count': int,
            'thumbnail': r're:^https?://.*\.(jpg|jpeg)',
            'description': 'md5:67583aaf3a0f933bd606bc8a2d3ebb17',
        }
    }]

    def _real_extract(self, url):
        video_id = self._match_id(url)

        webpage = self._download_webpage(url, video_id)
        json_all = self._search_json(r'window.videoInfo\s*=\s*', webpage, 'videoInfo', video_id)

        title = json_all.get('title')
        video_list = json_all.get('videoList') or []
        video_internal_id = traverse_obj(json_all, ('currentVideoInfo', 'id'))
        if video_internal_id and len(video_list) > 1:
            part_idx, part_video_info = next(
                (idx + 1, v) for (idx, v) in enumerate(video_list)
                if v['id'] == video_internal_id)
            title = f'{title} P{part_idx:02d} {part_video_info["title"]}'

        return {
            **self._extract_metadata(video_id, json_all['currentVideoInfo']),
            'title': title,
            'thumbnail': json_all.get('coverUrl'),
            'description': json_all.get('description'),
            'uploader': traverse_obj(json_all, ('user', 'name')),
            'uploader_id': traverse_obj(json_all, ('user', 'href')),
            'tags': traverse_obj(json_all, ('tagList', ..., 'name')),
            'view_count': int_or_none(json_all.get('viewCount')),
            'like_count': int_or_none(json_all.get('likeCountShow')),
            'comment_count': int_or_none(json_all.get('commentCountShow')),
        }

class AcFunBangumiIE(AcFunVideoBaseIE):
    _VALID_URL = r'https?://www\.acfun\.cn/bangumi/(?P<id>aa[_\d]+)'

    _TESTS = [{
        'url': 'https://www.acfun.cn/bangumi/aa6002917_36188_1745457?ac=2',
        'info_dict': {
            'id': 'aa6002917_36188_1745457__2',
            'ext': 'mp4',
            'title': '【7月】租借女友 水原千鹤角色曲『DATE』特别PV',
            'upload_date': '20200916',
            'timestamp': 1600243813,
            'duration': 92.091,
        },
    }, {
        'url': 'https://www.acfun.cn/bangumi/aa5023171_36188_1750645',
        'info_dict': {
            'id': 'aa5023171_36188_1750645',
            'ext': 'mp4',
            'title': '红孩儿之趴趴蛙寻石记 第5话 ',
            'duration': 760.0,
            'season': '红孩儿之趴趴蛙寻石记',
            'season_id': 5023171,
            'season_number': 1,  # series has only 1 season
            'episode': 'Episode 5',
            'episode_number': 5,
            'upload_date': '20181223',
            'timestamp': 1545552185,
            'thumbnail': r're:^https?://.*\.(jpg|jpeg|png)',
            'comment_count': int,
        },
    }, {
        'url': 'https://www.acfun.cn/bangumi/aa6065485_36188_1885061',
        'info_dict': {
            'id': 'aa6065485_36188_1885061',
            'ext': 'mp4',
            'title': '叽歪老表（第二季） 第5话 坚不可摧',
            'season': '叽歪老表（第二季）',
            'season_number': 2,
            'season_id': 6065485,
            'episode': '坚不可摧',
            'episode_number': 5,
            'upload_date': '20220324',
            'timestamp': 1648082786,
            'duration': 105.002,
            'thumbnail': r're:^https?://.*\.(jpg|jpeg|png)',
            'comment_count': int,
        },
    }]

    def _real_extract(self, url):
        video_id = self._match_id(url)
        ac_idx = parse_qs(url).get('ac', [None])[-1]
        video_id = f'{video_id}{format_field(ac_idx, template="__%s")}'

        webpage = self._download_webpage(url, video_id)
        json_bangumi_data = self._search_json(r'window.bangumiData\s*=\s*', webpage, 'bangumiData', video_id)

        if ac_idx:
            video_info = json_bangumi_data['hlVideoInfo']
            return {
                **self._extract_metadata(video_id, video_info),
                'title': video_info.get('title'),
            }

        video_info = json_bangumi_data['currentVideoInfo']

        season_id = json_bangumi_data.get('bangumiId')
        season_number = season_id and next((
            idx for idx, v in enumerate(json_bangumi_data.get('relatedBangumis') or [], 1)
            if v.get('id') == season_id), 1)

        json_bangumi_list = self._search_json(
            r'window\.bangumiList\s*=\s*', webpage, 'bangumiList', video_id, fatal=False)
        video_internal_id = int_or_none(traverse_obj(json_bangumi_data, ('currentVideoInfo', 'id')))
        episode_number = video_internal_id and next((
            idx for idx, v in enumerate(json_bangumi_list.get('items') or [], 1)
            if v.get('videoId') == video_internal_id), None)

        return {
            **self._extract_metadata(video_id, video_info),
            'title': json_bangumi_data.get('showTitle'),
            'thumbnail': json_bangumi_data.get('image'),
            'season': json_bangumi_data.get('bangumiTitle'),
            'season_id': season_id,
            'season_number': season_number,
            'episode': json_bangumi_data.get('title'),
            'episode_number': episode_number,
            'comment_count': int_or_none(json_bangumi_data.get('commentCount')),
        }


code:
import base64
import binascii
import json
import os
import random

from .common import InfoExtractor
from ..aes import aes_cbc_decrypt_bytes, unpad_pkcs7
from ..compat import (
    compat_HTTPError,
    compat_b64decode,
)
from ..utils import (
    ass_subtitles_timecode,
    bytes_to_intlist,
    bytes_to_long,
    ExtractorError,
    float_or_none,
    int_or_none,
    intlist_to_bytes,
    long_to_bytes,
    pkcs1pad,
    strip_or_none,
    try_get,
    unified_strdate,
    urlencode_postdata,
)

class ADNIE(InfoExtractor):
    IE_DESC = 'Anime Digital Network'
    _VALID_URL = r'https?://(?:www\.)?animedigitalnetwork\.fr/video/[^/]+/(?P<id>\d+)'
    _TEST = {
        'url': 'http://animedigitalnetwork.fr/video/blue-exorcist-kyoto-saga/7778-episode-1-debut-des-hostilites',
        'md5': '0319c99885ff5547565cacb4f3f9348d',
        'info_dict': {
            'id': '7778',
            'ext': 'mp4',
            'title': 'Blue Exorcist - Kyôto Saga - Episode 1',
            'description': 'md5:2f7b5aa76edbc1a7a92cedcda8a528d5',
            'series': 'Blue Exorcist - Kyôto Saga',
            'duration': 1467,
            'release_date': '20170106',
            'comment_count': int,
            'average_rating': float,
            'season_number': 2,
            'episode': 'Début des hostilités',
            'episode_number': 1,
        }
    }

    _NETRC_MACHINE = 'animedigitalnetwork'
    _BASE_URL = 'http://animedigitalnetwork.fr'
    _API_BASE_URL = 'https://gw.api.animedigitalnetwork.fr/'
    _PLAYER_BASE_URL = _API_BASE_URL + 'player/'
    _HEADERS = {}
    _LOGIN_ERR_MESSAGE = 'Unable to log in'
    _RSA_KEY = (0x9B42B08905199A5CCE2026274399CA560ECB209EE9878A708B1C0812E1BB8CB5D1FB7441861147C1A1F2F3A0476DD63A9CAC20D3E983613346850AA6CB38F16DC7D720FD7D86FC6E5B3D5BBC72E14CD0BF9E869F2CEA2CCAD648F1DCE38F1FF916CEFB2D339B64AA0264372344BC775E265E8A852F88144AB0BD9AA06C1A4ABB, 65537)
    _POS_ALIGN_MAP = {
        'start': 1,
        'end': 3,
    }
    _LINE_ALIGN_MAP = {
        'middle': 8,
        'end': 4,
    }

    def _get_subtitles(self, sub_url, video_id):
        if not sub_url:
            return None

        enc_subtitles = self._download_webpage(
            sub_url, video_id, 'Downloading subtitles location', fatal=False) or '{}'
        subtitle_location = (self._parse_json(enc_subtitles, video_id, fatal=False) or {}).get('location')
        if subtitle_location:
            enc_subtitles = self._download_webpage(
                subtitle_location, video_id, 'Downloading subtitles data',
                fatal=False, headers={'Origin': 'https://animedigitalnetwork.fr'})
        if not enc_subtitles:
            return None

        # http://animedigitalnetwork.fr/components/com_vodvideo/videojs/adn-vjs.min.js
        dec_subtitles = unpad_pkcs7(aes_cbc_decrypt_bytes(
            compat_b64decode(enc_subtitles[24:]),
            binascii.unhexlify(self._K + '7fac1178830cfe0c'),
            compat_b64decode(enc_subtitles[:24])))
        subtitles_json = self._parse_json(dec_subtitles.decode(), None, fatal=False)
        if not subtitles_json:
            return None

        subtitles = {}
        for sub_lang, sub in subtitles_json.items():
            ssa = '''[Script Info]
ScriptType:V4.00
[V4 Styles]
Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,TertiaryColour,BackColour,Bold,Italic,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,AlphaLevel,Encoding
Style: Default,Arial,18,16777215,16777215,16777215,0,-1,0,1,1,0,2,20,20,20,0,0
[Events]
Format: Marked,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text'''
            for current in sub:
                start, end, text, line_align, position_align = (
                    float_or_none(current.get('startTime')),
                    float_or_none(current.get('endTime')),
                    current.get('text'), current.get('lineAlign'),
                    current.get('positionAlign'))
                if start is None or end is None or text is None:
                    continue
                alignment = self._POS_ALIGN_MAP.get(position_align, 2) + self._LINE_ALIGN_MAP.get(line_align, 0)
                ssa += os.linesep + 'Dialogue: Marked=0,%s,%s,Default,,0,0,0,,%s%s' % (
                    ass_subtitles_timecode(start),
                    ass_subtitles_timecode(end),
                    '{\\a%d}' % alignment if alignment != 2 else '',
                    text.replace('\n', '\\N').replace('<i>', '{\\i1}').replace('</i>', '{\\i0}'))

            if sub_lang == 'vostf':
                sub_lang = 'fr'
            subtitles.setdefault(sub_lang, []).extend([{
                'ext': 'json',
                'data': json.dumps(sub),
            }, {
                'ext': 'ssa',
                'data': ssa,
            }])
        return subtitles

    def _perform_login(self, username, password):
        try:
            access_token = (self._download_json(
                self._API_BASE_URL + 'authentication/login', None,
                'Logging in', self._LOGIN_ERR_MESSAGE, fatal=False,
                data=urlencode_postdata({
                    'password': password,
                    'rememberMe': False,
                    'source': 'Web',
                    'username': username,
                })) or {}).get('accessToken')
            if access_token:
                self._HEADERS = {'authorization': 'Bearer ' + access_token}
        except ExtractorError as e:
            message = None
            if isinstance(e.cause, compat_HTTPError) and e.cause.code == 401:
                resp = self._parse_json(
                    e.cause.read().decode(), None, fatal=False) or {}
                message = resp.get('message') or resp.get('code')
            self.report_warning(message or self._LOGIN_ERR_MESSAGE)

    def _real_extract(self, url):
        video_id = self._match_id(url)
        video_base_url = self._PLAYER_BASE_URL + 'video/%s/' % video_id
        player = self._download_json(
            video_base_url + 'configuration', video_id,
            'Downloading player config JSON metadata',
            headers=self._HEADERS)['player']
        options = player['options']

        user = options['user']
        if not user.get('hasAccess'):
            self.raise_login_required()

        token = self._download_json(
            user.get('refreshTokenUrl') or (self._PLAYER_BASE_URL + 'refresh/token'),
            video_id, 'Downloading access token', headers={
                'x-player-refresh-token': user['refreshToken']
            }, data=b'')['token']

        links_url = try_get(options, lambda x: x['video']['url']) or (video_base_url + 'link')
        self._K = ''.join([random.choice('0123456789abcdef') for _ in range(16)])
        message = bytes_to_intlist(json.dumps({
            'k': self._K,
            't': token,
        }))

        # Sometimes authentication fails for no good reason, retry with
        # a different random padding
        links_data = None
        for _ in range(3):
            padded_message = intlist_to_bytes(pkcs1pad(message, 128))
            n, e = self._RSA_KEY
            encrypted_message = long_to_bytes(pow(bytes_to_long(padded_message), e, n))
            authorization = base64.b64encode(encrypted_message).decode()

            try:
                links_data = self._download_json(
                    links_url, video_id, 'Downloading links JSON metadata', headers={
                        'X-Player-Token': authorization
                    }, query={
                        'freeWithAds': 'true',
                        'adaptive': 'false',
                        'withMetadata': 'true',
                        'source': 'Web'
                    })
                break
            except ExtractorError as e:
                if not isinstance(e.cause, compat_HTTPError):
                    raise e

                if e.cause.code == 401:
                    # This usually goes away with a different random pkcs1pad, so retry
                    continue

                error = self._parse_json(e.cause.read(), video_id)
                message = error.get('message')
                if e.cause.code == 403 and error.get('code') == 'player-bad-geolocation-country':
                    self.raise_geo_restricted(msg=message)
                raise ExtractorError(message)
        else:
            raise ExtractorError('Giving up retrying')

        links = links_data.get('links') or {}
        metas = links_data.get('metadata') or {}
        sub_url = (links.get('subtitles') or {}).get('all')
        video_info = links_data.get('video') or {}
        title = metas['title']

        formats = []
        for format_id, qualities in (links.get('streaming') or {}).items():
            if not isinstance(qualities, dict):
                continue
            for quality, load_balancer_url in qualities.items():
                load_balancer_data = self._download_json(
                    load_balancer_url, video_id,
                    'Downloading %s %s JSON metadata' % (format_id, quality),
                    fatal=False) or {}
                m3u8_url = load_balancer_data.get('location')
                if not m3u8_url:
                    continue
                m3u8_formats = self._extract_m3u8_formats(
                    m3u8_url, video_id, 'mp4', 'm3u8_native',
                    m3u8_id=format_id, fatal=False)
                if format_id == 'vf':
                    for f in m3u8_formats:
                        f['language'] = 'fr'
                formats.extend(m3u8_formats)
        self._sort_formats(formats)

        video = (self._download_json(
            self._API_BASE_URL + 'video/%s' % video_id, video_id,
            'Downloading additional video metadata', fatal=False) or {}).get('video') or {}
        show = video.get('show') or {}

        return {
            'id': video_id,
            'title': title,
            'description': strip_or_none(metas.get('summary') or video.get('summary')),
            'thumbnail': video_info.get('image') or player.get('image'),
            'formats': formats,
            'subtitles': self.extract_subtitles(sub_url, video_id),
            'episode': metas.get('subtitle') or video.get('name'),
            'episode_number': int_or_none(video.get('shortNumber')),
            'series': show.get('title'),
            'season_number': int_or_none(video.get('season')),
            'duration': int_or_none(video_info.get('duration') or video.get('duration')),
            'release_date': unified_strdate(video.get('releaseDate')),
            'average_rating': float_or_none(video.get('rating') or metas.get('rating')),
            'comment_count': int_or_none(video.get('commentsCount')),
        }


code:
from .common import InfoExtractor
from ..compat import (
    compat_parse_qs,
    compat_urlparse,
)

class AdobeConnectIE(InfoExtractor):
    _VALID_URL = r'https?://\w+\.adobeconnect\.com/(?P<id>[\w-]+)'

    def _real_extract(self, url):
        video_id = self._match_id(url)
        webpage = self._download_webpage(url, video_id)
        title = self._html_extract_title(webpage)
        qs = compat_parse_qs(self._search_regex(r"swfUrl\s*=\s*'([^']+)'", webpage, 'swf url').split('?')[1])
        is_live = qs.get('isLive', ['false'])[0] == 'true'
        formats = []
        for con_string in qs['conStrings'][0].split(','):
            formats.append({
                'format_id': con_string.split('://')[0],
                'app': compat_urlparse.quote('?' + con_string.split('?')[1] + 'flvplayerapp/' + qs['appInstance'][0]),
                'ext': 'flv',
                'play_path': 'mp4:' + qs['streamName'][0],
                'rtmp_conn': 'S:' + qs['ticket'][0],
                'rtmp_live': is_live,
                'url': con_string,
            })

        return {
            'id': video_id,
            'title': title,
            'formats': formats,
            'is_live': is_live,
        }