diff --git a/main.py b/main.py index 8680b02..d136bd2 100644 --- a/main.py +++ b/main.py @@ -64,7 +64,10 @@ async def url_media(update: Update, context: CustomContext, url: str) -> None: return media = tweet.message_media_result() if not media: - await update.effective_message.reply_text("No media found or media type is not supported.") + await update.effective_message.reply_text( + "No media found or media type is not supported.", + reply_to_message_id=update.message.message_id, + ) return message_to_send = await update.effective_message.reply_media_group( media, @@ -277,7 +280,8 @@ def main(): handlers = [ InlineQueryHandler(inline_query), - MessageHandler((filters.Regex(regex.x_url) | filters.Regex(regex.pixiv_url)) & filters.ChatType.PRIVATE, + MessageHandler((filters.Regex(regex.x_url) | filters.Regex(regex.pixiv_url)) | filters.Regex( + regex.bsky_url) & filters.ChatType.PRIVATE, handel_url_media), CommandHandler("set_forward_channel", cmd_set_forward_channel), CommandHandler("remove_forward_channel", cmd_remove_forward_channel), diff --git a/utils/bsky.py b/utils/bsky.py new file mode 100644 index 0000000..3068fc8 --- /dev/null +++ b/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']}") diff --git a/utils/regex.py b/utils/regex.py index a82de13..fadc131 100644 --- a/utils/regex.py +++ b/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/(.+)") diff --git a/utils/telegram.py b/utils/telegram.py index 08fac28..e3198ff 100644 --- a/utils/telegram.py +++ b/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 diff --git a/utils/types.py b/utils/types.py index d100c7b..e966d73 100644 --- a/utils/types.py +++ b/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 +})