yt-dlp/yt_dlp/extractor/line.py

from .common import InfoExtractor
from ..compat import compat_str
from ..utils import (
    format_field,
    int_or_none,
    str_or_none,
)


class LineLiveBaseIE(InfoExtractor):
    _API_BASE_URL = 'https://live-api.line-apps.com/web/v4.0/channel/'

    def _parse_broadcast_item(self, item):
        broadcast_id = compat_str(item['id'])
        title = item['title']
        is_live = item.get('isBroadcastingNow')

        thumbnails = []
        for thumbnail_id, thumbnail_url in (item.get('thumbnailURLs') or {}).items():
            if not thumbnail_url:
                continue
            thumbnails.append({
                'id': thumbnail_id,
                'url': thumbnail_url,
            })

        channel = item.get('channel') or {}
        channel_id = str_or_none(channel.get('id'))

        return {
            'id': broadcast_id,
            'title': title,
            'thumbnails': thumbnails,
            'timestamp': int_or_none(item.get('createdAt')),
            'channel': channel.get('name'),
            'channel_id': channel_id,
            'channel_url': format_field(channel_id, template='https://live.line.me/channels/%s'),
            'duration': int_or_none(item.get('archiveDuration')),
            'view_count': int_or_none(item.get('viewerCount')),
            'comment_count': int_or_none(item.get('chatCount')),
            'is_live': is_live,
        }


class LineLiveIE(LineLiveBaseIE):
    _VALID_URL = r'https?://live\.line\.me/channels/(?P<channel_id>\d+)/broadcast/(?P<id>\d+)'
    _TESTS = [{
        'url': 'https://live.line.me/channels/5833718/broadcast/18373277',
        'md5': '2c15843b8cb3acd55009ddcb2db91f7c',
        'info_dict': {
            'id': '18373277',
            'title': '2021/12/05 （15分犬）定例譲渡会🐶',
            'ext': 'mp4',
            'timestamp': 1638674925,
            'upload_date': '20211205',
            'thumbnail': 'md5:e1f5817e60f4a72b7e43377cf308d7ef',
            'channel_url': 'https://live.line.me/channels/5833718',
            'channel': 'Yahooニュース掲載🗞プロフ見てね🐕🐕',
            'channel_id': '5833718',
            'duration': 937,
            'view_count': int,
            'comment_count': int,
            'is_live': False,
        }
    }, {
        # archiveStatus == 'DELETED'
        'url': 'https://live.line.me/channels/4778159/broadcast/16378488',
        'only_matching': True,
    }]

    def _real_extract(self, url):
        channel_id, broadcast_id = self._match_valid_url(url).groups()
        broadcast = self._download_json(
            self._API_BASE_URL + '%s/broadcast/%s' % (channel_id, broadcast_id),
            broadcast_id)
        item = broadcast['item']
        info = self._parse_broadcast_item(item)
        protocol = 'm3u8' if info['is_live'] else 'm3u8_native'
        formats = []
        for k, v in (broadcast.get(('live' if info['is_live'] else 'archived') + 'HLSURLs') or {}).items():
            if not v:
                continue
            if k == 'abr':
                formats.extend(self._extract_m3u8_formats(
                    v, broadcast_id, 'mp4', protocol,
                    m3u8_id='hls', fatal=False))
                continue
            f = {
                'ext': 'mp4',
                'format_id': 'hls-' + k,
                'protocol': protocol,
                'url': v,
            }
            if not k.isdigit():
                f['vcodec'] = 'none'
            formats.append(f)
        if not formats:
            archive_status = item.get('archiveStatus')
            if archive_status != 'ARCHIVED':
                self.raise_no_formats('this video has been ' + archive_status.lower(), expected=True)
        self._sort_formats(formats)
        info['formats'] = formats
        return info


class LineLiveChannelIE(LineLiveBaseIE):
    _VALID_URL = r'https?://live\.line\.me/channels/(?P<id>\d+)(?!/broadcast/\d+)(?:[/?&#]|$)'
    _TEST = {
        'url': 'https://live.line.me/channels/5893542',
        'info_dict': {
            'id': '5893542',
            'title': 'いくらちゃんだよぉ🦒',
            'description': 'md5:4d418087973ad081ceb1b3481f0b1816',
        },
        'playlist_mincount': 29
    }

    def _archived_broadcasts_entries(self, archived_broadcasts, channel_id):
        while True:
            for row in (archived_broadcasts.get('rows') or []):
                share_url = str_or_none(row.get('shareURL'))
                if not share_url:
                    continue
                info = self._parse_broadcast_item(row)
                info.update({
                    '_type': 'url',
                    'url': share_url,
                    'ie_key': LineLiveIE.ie_key(),
                })
                yield info
            if not archived_broadcasts.get('hasNextPage'):
                return
            archived_broadcasts = self._download_json(
                self._API_BASE_URL + channel_id + '/archived_broadcasts',
                channel_id, query={
                    'lastId': info['id'],
                })

    def _real_extract(self, url):
        channel_id = self._match_id(url)
        channel = self._download_json(self._API_BASE_URL + channel_id, channel_id)
        return self.playlist_result(
            self._archived_broadcasts_entries(channel.get('archivedBroadcasts') or {}, channel_id),
            channel_id, channel.get('title'), channel.get('information'))