change to x.com in url template

.gitignore

@@ -4,3 +4,10 @@ cert/
 data/
 docker-compose.yml
 utils/x.py
+
+.env
+
+
+# Added by cargo
+
+/target

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.12-slim-bullseye
+FROM python:3.13-slim-bullseye
 
 LABEL maintainer="admin@yoursfunny.top"
 

common.py

@@ -9,7 +9,7 @@ except ImportError:
     uvloop = None
 
 BOT_TOKEN = os.getenv("BOT_TOKEN")
-ADMIN = [int(i) for i in os.getenv("BOT_ADMIN").split(",")]
+ADMIN = [int(i) for i in os.getenv("BOT_ADMIN", "").split(",") if i]
 
 PIXIV_REFRESH_TOKEN = os.getenv("PIXIV_REFRESH_TOKEN")
 

main.py

@@ -28,8 +28,10 @@ def send_action(action):
     def decorator(func):
         @wraps(func)
         async def command_func(update: Update, context: CustomContext, *args, **kwargs):
-            await update.effective_chat.send_action(action)
-            return await func(update, context, *args, **kwargs)
+            try:
+                await update.effective_chat.send_action(action)
+            finally:
+                return await func(update, context, *args, **kwargs)
 
         return command_func
 
@@ -62,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,
@@ -132,7 +137,7 @@ async def edit_message(update: Update, context: CustomContext) -> bool:
 
 
 async def handle_message(update: Update, context: CustomContext) -> None:
-    if edit_message(update, context):
+    if await edit_message(update, context):
         return
     if not (urls := extract_urls(update.message)):
         return
@@ -157,6 +162,22 @@ async def query_template(update: Update, context: CustomContext) -> None:
     await _edit_message.forward[0].edit_caption(context.chat_data.template[name])
 
 
+@send_action(ChatAction.TYPING)
+async def cmd_start(update: Update, context: CustomContext) -> None:
+    await update.effective_message.reply_text(
+        "Welcome to the Twitter Fetcher Bot!\n"
+        "You can use this bot to fetch tweets from Twitter and forward them to a channel.\n"
+        "Use /set_forward_channel to set a channel to forward tweets.\n"
+        "Use /remove_forward_channel to remove the channel.\n"
+        "Use /edit_before_forward to enable or disable edit before forward.\n"
+        "Use /set_template to set a template for the forwarded message.\n"
+        "Use /bot_dict to see the bot's data.\n"
+        "Use /clear_edit_message to clear the edit message cache.\n"
+        "You can also reply to a message with a tweet URL to fetch the tweet and forward it to the channel.\n"
+        "You can also use inline query to search for tweets."
+    )
+
+
 @send_action(ChatAction.TYPING)
 async def cmd_set_forward_channel(update: Update, context: CustomContext) -> None:
     if not context.args:
@@ -232,6 +253,12 @@ async def cmd_user_dict(update: Update, context: CustomContext) -> None:
     await update.effective_message.reply_text(html.escape(str(context.chat_data)), disable_web_page_preview=True)
 
 
+@send_action(ChatAction.TYPING)
+async def cmd_clear_edit_message(update: Update, context: CustomContext) -> None:
+    context.chat_data.edit_message.clear()
+    await update.effective_message.reply_text("Edit message cleared.")
+
+
 async def post_init(application: Application) -> None:
     # commands = [
     #     BotCommand('start', CMD_START),
@@ -246,7 +273,8 @@ async def post_init(application: Application) -> None:
 
 
 async def post_stop(application: Application) -> None:
-    await application.bot.send_message(common.ADMIN[0], "Shutting down...")
+    if common.ADMIN:
+        await application.bot.send_message(common.ADMIN[0], "Shutting down...")
 
 
 async def post_shutdown(application: Application) -> None:
@@ -274,8 +302,10 @@ 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("start", cmd_start),
         CommandHandler("set_forward_channel", cmd_set_forward_channel),
         CommandHandler("remove_forward_channel", cmd_remove_forward_channel),
         CommandHandler("edit_before_forward", cmd_edit_before_forward),
@@ -284,6 +314,7 @@ def main():
         CallbackQueryHandler(query_forward_message, pattern="forward"),
         CallbackQueryHandler(query_template, pattern=r"^template\|"),
         CommandHandler("bot_dict", cmd_user_dict),
+        CommandHandler("clear_edit_message", cmd_clear_edit_message),
     ]
 
     application.add_handlers(handlers)

requirements.txt

@@ -1,4 +1,4 @@
-python-telegram-bot[webhooks]~=21.4
+python-telegram-bot[webhooks]~=22.5
 httpx[http2]~=0.27
-uvloop~=0.19.0; sys_platform != 'win32'
+uvloop~=0.22; sys_platform != 'win32'
 async-pixiv @ git+https://github.com/TheFunny/async-pixiv@main

utils/bsky.py

@@ -0,0 +1,163 @@
+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'
+
+SENSITIVE_TAG = {'sexual', 'nudity', 'porn', 'graphic-media'}
+
+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=self._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']}")
+
+    @property
+    def _sensitive(self) -> bool:
+        return any(
+            tag in label['val']
+            for label in self._bsky['labels']
+            for tag in SENSITIVE_TAG
+        )

utils/net.py

@@ -11,8 +11,8 @@ async def close_client(_client: AsyncClient) -> None:
     return await _client.aclose()
 
 
-async def fetch_json(_client: AsyncClient, url: str) -> dict:
-    response = await _client.get(url)
+async def fetch_json(_client: AsyncClient, url: str, params: dict = None) -> dict:
+    response = await _client.get(url, params=params)
     assert response.is_success, f"Failed to fetch {url}, status code {response.status_code}"
     return response.json()
 
@@ -33,5 +33,5 @@ class NetClient:
         return cls._httpx_client
 
     @classmethod
-    async def fetch_json(cls, url: str) -> dict:
-        return await fetch_json(cls._httpx_client, url)
+    async def fetch_json(cls, url: str, params: dict = None) -> dict:
+        return await fetch_json(cls._httpx_client, url, params)

utils/regex.py

@@ -1,8 +1,11 @@
 import re
 
-x_url = re.compile(r"^(?:https?://)?(?:www\.|mobile\.)?(?:x|twitter|fixvx|vxtwitter)\.com/(.+)/status/(\d+)")
+x_url = re.compile(
+    r"^(?:https?://)?(?:www\.|mobile\.)?(?:x|twitter|fixvx|vxtwitter|fixupx|fxtwitter)\.com/(.+)/status/(\d+)")
 x_media_url = re.compile(r"^(?:https?://)?(pbs|video)\.twimg\.com/(.*)")
 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,90 @@ 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
+                # )
+                yield
+
+    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/tweet.py

@@ -84,7 +84,7 @@ class Tweet:
 
     @cached_property
     def url(self) -> str:
-        return f"https://twitter.com/{self._author_id}/status/{self._id}"
+        return f"https://x.com/{self._author_id}/status/{self._id}"
 
     @property
     def author(self) -> str:
@@ -92,7 +92,7 @@ class Tweet:
 
     @cached_property
     def author_url(self) -> str:
-        return f"https://twitter.com/{self._author_id}"
+        return f"https://x.com/{self._author_id}"
 
     @property
     def text(self) -> str:
@@ -136,7 +136,7 @@ class ProcessTweet:
     @property
     def _tweet_text(self) -> str:
         match = x_tco_url.search(self._tweet['text'])
-        return self._tweet['text'][:match.start()].strip(" ") if match else self._tweet['text']
+        return self._tweet['text'][:match.start()].strip() if match else self._tweet['text']
 
     @property
     def _tweet_media(self) -> list[TweetMedia]:

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,59 @@ 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
+    labels: list[BskyLabel]
+
+
+class BskyAuthor(TypedDict):
+    handle: str
+    displayName: str
+
+
+class BskyPostRecord(TypedDict):
+    text: str
+
+
+class BskyLabel(TypedDict):
+    val: 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
+})