TelegramTwitterMediaBot/utils/bsky.py

from __future__ import annotations

from functools import cached_property
from typing import TYPE_CHECKING

from utils.net import NetClient
from utils.regex import bsky_url

if TYPE_CHECKING:
    from utils.types import BskyEmbedImages, BskyInfo, BskyEmbedVideo, BskyEmbedExternal

bsky_api_url = 'https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread'


class BskyMedia:
    __slots__ = ('_url', '_thumb', '_type', '__dict__')

    def __init__(self, url: str, thumb: str, media_type: str):
        self._url: str = url
        self._thumb: str = thumb
        self._type: str = media_type

    def __str__(self):
        return f"BskyMedia(url={self.url}, thumb={self.thumb}, type={self.type})"

    @property
    def url(self) -> str:
        return self._url

    @property
    def thumb(self) -> str:
        return self._thumb

    @property
    def type(self) -> str:
        return self._type


class Bsky:
    __slots__ = ('_id', '_author', '_author_id', '_text', '_media', '_sensitive', '__dict__')

    def __init__(
            self,
            id: str,
            author: str,
            author_id: str,
            text: str,
            media: list[BskyMedia],
            sensitive: bool = False
    ):
        self._id: str = id
        self._author: str = author
        self._author_id: str = author_id
        self._text: str = text
        self._media: list[BskyMedia] = media
        self._sensitive: bool = sensitive

    @property
    def id(self) -> str:
        return self._id

    @cached_property
    def url(self) -> str:
        return f"https://bsky.app/profile/{self._author_id}/post/{self._id}"

    @property
    def author(self) -> str:
        return self._author

    @cached_property
    def author_url(self) -> str:
        return f"https://bsky.app/profile/{self._author_id}"

    @property
    def text(self) -> str:
        return self._text

    @property
    def media(self) -> list[BskyMedia]:
        return self._media

    @property
    def sensitive(self) -> bool:
        return self._sensitive


class ProcessBsky:
    __slots__ = ('_url', '_id', '_bsky')

    def __init__(self, url: str):
        self._url: str = url

    async def __aenter__(self):
        bsky = await self._fetch_bsky()
        if not bsky['thread'].get('post'):
            raise ValueError(f"BSky post not found: {bsky}")
        self._bsky = bsky['thread']['post']
        return Bsky(
            id=self._id,
            author=self._bsky['author']['displayName'],
            author_id=self._bsky['author']['handle'],
            text=self._bsky['record']['text'],
            media=self._bsky_media,
            sensitive=False  # TODO: implement sensitive
        )

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass

    async def _fetch_bsky(self) -> BskyInfo:
        match = bsky_url.match(self._url)
        if not match:
            raise ValueError(f"Invalid Bsky URL: {self._url}")
        auther_id, self._id = match.groups()
        return await NetClient.fetch_json(
            bsky_api_url,
            params={'uri': f'at://{auther_id}/app.bsky.feed.post/{self._id}', 'depth': 0}
        )

    @property
    def _bsky_media(self) -> list[BskyMedia]:
        if not self._bsky.get('embed'):
            return []
        match (embed := self._bsky['embed'])['$type']:
            case 'app.bsky.embed.images#view':
                embed: BskyEmbedImages
                return [
                    BskyMedia(
                        url=image['fullsize'],
                        thumb=image['thumb'],
                        media_type='image'
                    )
                    for image in embed['images']
                ]
            case 'app.bsky.embed.video#view':
                embed: BskyEmbedVideo
                return [
                    BskyMedia(
                        url=embed['playlist'],
                        thumb=embed['thumbnail'],
                        media_type='video'
                    )
                ]
            case 'app.bsky.embed.external#view':
                embed: BskyEmbedExternal
                return [
                    BskyMedia(
                        url=embed['external']['uri'],
                        thumb=embed['external']['thumb'],
                        media_type='external'
                    )
                ]
            case _:
                raise NotImplementedError(f"Unknown Bsky embed type: {embed['type']}")