new bsky support

utils/bsky.py

@@ -0,0 +1,154 @@
+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']}")

utils/regex.py

@@ -6,3 +6,5 @@ x_tco_url = re.compile(r"(?:https?://)?t\.co/.+$", re.M)
 message_url = re.compile(r"\[.+]", re.S)
 
 pixiv_url = re.compile(r"^(?:https?://)?(?:www\.)?pixiv\.net/(?:en/)?(?:artworks/|i/)(\d+)")
+
+bsky_url = re.compile(r"^(?:https?://)?(?:bsky)\.app/profile/(.+)/post/(.+)")

utils/telegram.py

@@ -9,9 +9,10 @@ from telegram import InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQu
     InputMediaVideo
 
 from common import PIXIV_REFRESH_TOKEN
+from .bsky import ProcessBsky
 from .logger import get_logger
 from .pixiv import ProcessPixiv
-from .regex import pixiv_url, x_url
+from .regex import bsky_url, pixiv_url, x_url
 from .tweet import ProcessTweet
 
 if TYPE_CHECKING:
@@ -39,6 +40,9 @@ class Telegram:
         elif PIXIV_REFRESH_TOKEN and pixiv_url.match(self._url):
             async with TelegramPixiv(self._url) as pixiv:
                 return pixiv
+        elif bsky_url.match(self._url):
+            async with TelegramBsky(self._url) as bsky:
+                return bsky
         else:
             return None  # TODO add raise and catch
 
@@ -61,7 +65,7 @@ class TelegramTweet:
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         pass
 
-    @cached_property
+    @property
     def url(self) -> str:
         return self._tweet.url
 
@@ -165,10 +169,10 @@ class TelegramPixiv:
         )
 
     def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]:
-        return tuple(self.inline_query_generator())
+        return tuple(i for i in self.inline_query_generator() if i)
 
     def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]:
-        return tuple(self.message_media_generator())
+        return tuple(i for i in self.message_media_generator() if i)
 
     def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]:
         pixiv = self._pixiv
@@ -195,3 +199,89 @@ class TelegramPixiv:
                 )
             else:
                 yield
+
+
+class TelegramBsky:
+    message_raw_text = message_raw_text_tweet
+    __slots__ = ('_url', '_bsky', '__dict__')
+
+    def __init__(self, url: str):
+        self._url: str = url
+
+    async def __aenter__(self):
+        async with ProcessBsky(self._url) as bsky:
+            self._bsky = bsky
+            return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    @property
+    def url(self) -> str:
+        return self._bsky.url
+
+    @cached_property
+    def message_text(self) -> str:
+        bsky = self._bsky
+        return self.message_raw_text.format(
+            url=bsky.url,
+            author_url=bsky.author_url,
+            author=html.escape(bsky.author),
+            text=html.escape(bsky.text)
+        )
+
+    def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]:
+        return tuple(i for i in self.inline_query_generator() if i)
+
+    def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]:
+        return tuple(i for i in self.message_media_generator() if i)
+
+    def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]:
+        bsky = self._bsky
+        for bsky_media in bsky.media:
+            logger.info(str(bsky_media))
+            if bsky_media.type == "image":
+                yield InlineQueryResultPhoto(
+                    id=str(uuid4()),
+                    photo_url=bsky_media.url,
+                    thumbnail_url=bsky_media.thumb,
+                    caption=self.message_text
+                )
+            elif bsky_media.type == "video":
+                # yield InlineQueryResultVideo(
+                #     id=str(uuid4()),
+                #     video_url=bsky_media.url,
+                #     mime_type="video/mp4",
+                #     thumbnail_url=bsky_media.thumb,
+                #     title=bsky.text,
+                #     caption=self.message_text
+                # )
+                yield
+            elif bsky_media.type == "external":
+                yield InlineQueryResultVideo(
+                    id=str(uuid4()),
+                    video_url=bsky_media.url,
+                    mime_type="image/gif",
+                    thumbnail_url=bsky_media.thumb,
+                    title=bsky.text,
+                    caption=self.message_text
+                )
+
+    def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]:
+        bsky = self._bsky
+        for bsky_media in bsky.media:
+            logger.info(str(bsky_media))
+            if bsky_media.type == "image":
+                yield InputMediaPhoto(
+                    media=bsky_media.url,
+                    has_spoiler=bsky.sensitive
+                )
+            elif bsky_media.type == "video":
+                # yield InputMediaVideo(
+                #     media=bsky_media.url,
+                #     has_spoiler=bsky.sensitive,
+                #     thumbnail=bsky_media.thumb
+                # )
+                yield
+            elif bsky_media.type == "external":
+                yield bsky_media.url, bsky.sensitive

utils/types.py

@@ -1,4 +1,6 @@
-from typing import TypedDict
+from __future__ import annotations
+
+from typing import Literal, TypedDict
 
 from telegram import InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, InputMediaPhoto, \
     InputMediaVideo
@@ -15,3 +17,54 @@ class TweetInfo(TypedDict):
     text: str
     media_extended: list[dict]
     possibly_sensitive: bool
+
+
+class BskyInfo(TypedDict):
+    thread: BskyThread
+
+
+class BskyThread(TypedDict):
+    post: BskyPost
+
+
+class BskyPost(TypedDict):
+    author: BskyAuthor
+    record: BskyPostRecord
+    embed: BskyEmbedImages | BskyEmbedVideo | BskyEmbedExternal
+
+
+class BskyAuthor(TypedDict):
+    handle: str
+    displayName: str
+
+
+class BskyPostRecord(TypedDict):
+    text: str
+
+
+class BskyEmbedImage(TypedDict):
+    thumb: str
+    fullsize: str
+
+
+BskyEmbedImages = TypedDict('BskyEmbedImages', {
+    '$type': Literal['app.bsky.embed.images#view'],
+    'images': list[BskyEmbedImage]
+})
+
+BskyEmbedVideo = TypedDict('BskyEmbedVideo', {
+    '$type': Literal['app.bsky.embed.video#view'],
+    'playlist': str,
+    'thumbnail': str
+})
+
+
+class BskyEmbedExternalItem(TypedDict):
+    uri: str
+    thumb: str
+
+
+BskyEmbedExternal = TypedDict('BskyEmbedExternal', {
+    '$type': Literal['app.bsky.embed.external#view'],
+    'external': BskyEmbedExternalItem
+})