Repository: yt-dlp/yt-dlp
File: yt_dlp/extractor/bluesky.py (Python)

Code before:
        meta = self._download_json(
            'https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread',
            video_id, headers={'Content-Type': 'application/json'}, query={
                'uri': f'at://{handle}/app.bsky.feed.post/{video_id}',
                'depth': 1000 if getcomments else 0,
                'parentHeight': 1000 if getcomments else 0,
            })['thread']
        post = meta.get('post')

        did = traverse_obj(post, ('author', 'did'))
        record_embed = traverse_obj(post, ('record', 'embed', ('media', None)), get_all=False)
        post_type = record_embed.get('$type') if record_embed else None
        quoted_post = traverse_obj(post, ('embed', 'record', ('record', None)), get_all=False)
        quoted_type = traverse_obj(quoted_post, ('value', 'embed', ('media', None), '$type'), get_all=False)
        quoted_media = traverse_obj(quoted_post, ('embeds', 0, ('media', None)), get_all=False)

        if post_type == 'app.bsky.embed.external':
            return self.url_result(traverse_obj(
                post, ('embed', ('media', None), 'external', 'uri'), get_all=False)
                or traverse_obj(record_embed, ('external', 'uri')))
        elif post_type == 'app.bsky.embed.video':
            formats, subs = self._extract_m3u8_formats_and_subtitles(
                traverse_obj(post, ('embed', ('media', None), 'playlist'), get_all=False),
                video_id, 'mp4', 'm3u8_native', m3u8_id='hls', fatal=False,
                note='Downloading m3u8 information', errnote='Unable to download m3u8 information')
            if blob_cid := traverse_obj(record_embed, ('video', 'ref', '$link'), ('video', 'cid')):
                endpoint = self.get_service_endpoint(did, video_id)
                formats.append({
                    'format_id': 'blob',
                    'url': f'{endpoint}/xrpc/com.atproto.sync.getBlob?did={did}&cid={blob_cid}',
                    **traverse_obj(record_embed, {
                        'ext': ('video', 'mimeType', {mimetype2ext}),
                        'width': ('aspectRatio', 'width', {int_or_none}),
                        'height': ('aspectRatio', 'height', {int_or_none}),
                        'filesize': ('video', 'size', {int_or_none}),
                    }),
                })
            video_info = {
                'formats': formats,
                'subtitles': subs,
                **traverse_obj(post, {
                    'thumbnail': ('embed', 'thumbnail', {url_or_none}),
                    'alt_title': ('embed', 'alt'),
                }),
            }
        elif quoted_type == 'app.bsky.embed.external':
            return self.url_result(traverse_obj(quoted_media, ('external', 'uri')))
        elif quoted_type == 'app.bsky.embed.video':
            formats, subs = self._extract_m3u8_formats_and_subtitles(
                quoted_media.get('playlist'), video_id, 'mp4', 'm3u8_native', m3u8_id='hls', fatal=False,
                note='Downloading m3u8 information', errnote='Unable to download m3u8 information')

Review comment (security):
For the video extract method, this is what I came up with:
```suggestion

    @staticmethod
    def _build_profile_url(path):
        return format_field(path, None, 'https://bsky.app/profile/%s', default=None)

    def _extract_videos(self, root, video_id, embed_path='embed', record_path='record', record_subpath='embed'):
        embed_path = variadic(embed_path, (str, bytes, dict, set))
        record_path = variadic(record_path, (str, bytes, dict, set))
        record_subpath = variadic(record_subpath, (str, bytes, dict, set))

        entries = []
        if external_uri := traverse_obj(root, (((*record_path, *record_subpath), embed_path), 'external', 'uri', {url_or_none}, any)):
            entries.append(self.url_result(external_uri))

        if playlist := traverse_obj(root, (*embed_path, 'playlist', {url_or_none})):
            formats, subtitles = self._extract_m3u8_formats_and_subtitles(
                playlist, video_id, 'mp4', m3u8_id='hls', fatal=False)
        else:
            return entries

        video_cid = traverse_obj(
            root, (*embed_path, 'cid', {str}),
            (*record_path, *record_subpath, 'video', 'ref', '$link', {str}))
        did = traverse_obj(root, ('author', 'did', {str}))
        if did and video_cid:
            endpoint = self.get_service_endpoint(did, video_id)
            formats.append({
                'format_id': 'blob',
                'url': update_url_query(self._BLOB_URL % endpoint, {'did': did, 'cid': video_cid}),
                **traverse_obj(root, (*embed_path, 'aspectRatio', {
                    'width': ('width', {int_or_none}),
                    'height': ('height', {int_or_none}),
                })),
                **traverse_obj(root, (*record_path, *record_subpath, 'video', {
                    'filesize': ('size', {int_or_none}),
                    'ext': ('mimeType', {mimetype2ext}),
                })),
            })
            for sub_data in traverse_obj(root, (
                    *record_path, *record_subpath, 'captions', lambda _, v: v['file']['ref']['$link'])):
                subtitles.setdefault(sub_data.get('lang') or 'und', []).append({
                    'url': update_url_query(self._BLOB_URL % endpoint, {'did': did, 'cid': sub_data['file']['ref']['$link']}),
                    'ext': traverse_obj(sub_data, ('file', 'mimeType', {mimetype2ext})),
                })

        entries.append({
            'id': video_id,
            'formats': formats,
            'subtitles': subtitles,
            **traverse_obj(root, {
                'id': ('uri', {lambda x: x.split('/')[-1] or None}, {str}),
                'thumbnail': (*embed_path, 'thumbnail', {url_or_none}),
                'alt_title': (*embed_path, 'alt', {str}),
                'uploader': ('author', 'displayName', {str}),
                'uploader_id': ('author', 'handle', {str}),
                'uploader_url': ('author', 'handle', {self._build_profile_url}),
                'channel_id': ('author', 'did', {str}),
                'channel_url': ('author', 'did', {self._build_profile_url}),
                'like_count': ('likeCount', {int_or_none}),
                'repost_count': ('repostCount', {int_or_none}),
                'comment_count': ('replyCount', {int_or_none}),
                'timestamp': ('indexedAt', {parse_iso8601}),
                'tags': ('labels', ..., 'val', {str}, all, {lambda x: list(set(x))}),
                'age_limit': ('labels', ..., 'val', {lambda x: 18 if x in ('sexual', 'porn', 'graphic-media') else None}, any),
                'description': (*record_path, 'text', {str}),
                'title': (*record_path, 'text', {lambda x: truncate_string(x, 50)}),
            }),
        })
        return entries

```
This requires the const `_BLOB_URL = '%s/xrpc/com.atproto.sync.getBlob'` to be added.

Some notes on the implementation/differences to yours:
- I tried to combine the different paths with paths to the record/embed being able to be passed
- I adjusted the uploader/channel(_id/_url) logic. I think like this it makes sense, and is somewhat in line with twitter.
- For the tags, I removed the `language`. Potentially, we could return this in the `language` field. I also deduplicated the tags.
- For the title, I think it makes more sense to go with the same value as the description. To not have the same issue as with the twitter extractor (potentially very long filenames that are not supported by the target filesystem) I would truncate the string.

Code after:
                'channel_url': 'https://bsky.app/profile/did:plc:6ttyq36rhiyed7wu3ws7dmqj',
                'description': r're:(?s)hearing people on twitter say that bluesky .{93}',
                'tags': [],
                'alt_title': 'md5:9b1ee1937fb3d1a81e932f9ec14d560e',
                'uploader': 'T9',
                'channel_id': 'did:plc:6ttyq36rhiyed7wu3ws7dmqj',
                'thumbnail': r're:https://video.bsky.app/watch/.*\.jpg$',
                'timestamp': 1729731642,
                'comment_count': int,
                'repost_count': int,
            },
        }],
    }]
    _BLOB_URL_TMPL = '{}/xrpc/com.atproto.sync.getBlob'

    def _get_service_endpoint(self, did, video_id):
        if did.startswith('did:web:'):
            url = f'https://{did[8:]}/.well-known/did.json'
        else:
            url = f'https://plc.directory/{did}'
        services = self._download_json(
            url, video_id, 'Fetching service endpoint', 'Falling back to bsky.social', fatal=False)
        return traverse_obj(
            services, ('service', lambda _, x: x['type'] == 'AtprotoPersonalDataServer',
                       'serviceEndpoint', {url_or_none}, any)) or 'https://bsky.social'

    def _real_extract(self, url):
        handle, video_id = self._match_valid_url(url).group('handle', 'id')

        post = self._download_json(
            'https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread',
            video_id, query={
                'uri': f'at://{handle}/app.bsky.feed.post/{video_id}',
                'depth': 0,
                'parentHeight': 0,
            })['thread']['post']

        entries = []
        # app.bsky.embed.video.view/app.bsky.embed.external.view
        entries.extend(self._extract_videos(post, video_id))
        # app.bsky.embed.recordWithMedia.view
        entries.extend(self._extract_videos(
            post, video_id, embed_path=('embed', 'media'), record_subpath=('embed', 'media')))
        # app.bsky.embed.record.view
        if nested_post := traverse_obj(post, ('embed', 'record', ('record', None), {dict}, any)):
            entries.extend(self._extract_videos(
                nested_post, video_id, embed_path=('embeds', 0), record_path='value'))

        if not entries:
            raise ExtractorError('No video could be found in this post', expected=True)
        if len(entries) == 1: