fix changes of updated async-pixiv module

common.py

@@ -13,7 +13,7 @@ ADMIN = [int(i) for i in os.getenv("BOT_ADMIN", "").split(",") if i]
 
 PIXIV_REFRESH_TOKEN = os.getenv("PIXIV_REFRESH_TOKEN")
 
-WEBHOOK = os.getenv("WEBHOOK", False)
+WEBHOOK = os.getenv("WEBHOOK").strip().lower() in ("true", "yes", "1")
 if WEBHOOK:
     WEBHOOK_LISTEN = os.getenv("WEBHOOK_LISTEN", "0.0.0.0")
     WEBHOOK_PORT = int(os.getenv("WEBHOOK_PORT", 8443))

main.py

@@ -162,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:
@@ -237,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),
@@ -283,6 +305,7 @@ def main():
         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),
@@ -291,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.6
+python-telegram-bot[webhooks]~=22.8
 httpx[http2]~=0.27
-uvloop~=0.19.0; sys_platform != 'win32'
-async-pixiv @ git+https://github.com/TheFunny/async-pixiv@main
+uvloop~=0.22; sys_platform != 'win32'
+async-pixiv~=1.1.2

utils/bsky.py

@@ -11,6 +11,7 @@ if TYPE_CHECKING:
 
 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__')
@@ -101,7 +102,7 @@ class ProcessBsky:
             author_id=self._bsky['author']['handle'],
             text=self._bsky['record']['text'],
             media=self._bsky_media,
-            sensitive=False  # TODO: implement sensitive
+            sensitive=self._sensitive
         )
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
@@ -151,4 +152,12 @@ class ProcessBsky:
                     )
                 ]
             case _:
-                raise NotImplementedError(f"Unknown Bsky embed type: {embed['type']}")
+                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/pixiv.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from typing import Literal, TYPE_CHECKING
 
 from async_pixiv import PixivClient
-from async_pixiv.error import ApiError
+from async_pixiv.error import APIError
 
 if TYPE_CHECKING:
     from async_pixiv.model.illust import Illust
@@ -73,27 +73,27 @@ class Pixiv:
 
     @property
     def is_nsfw(self) -> bool:
-        return self._illust.is_nsfw
+        return self._illust.sanity_level > 5
 
     @property
     def is_ai(self) -> bool:
-        return self._illust.ai_type.value == 2
+        return self._illust.ai_type == 2
 
     @property
     def images(self) -> list[PixivMedia]:
         if self.is_multiple_pages:
             return [
                 PixivMedia(
-                    url=page.image_urls.original,
-                    thumb=page.image_urls.medium
+                    url=str(page.image_urls.original),
+                    thumb=str(page.image_urls.medium)
                 )
                 for page in self._illust.meta_pages
             ]
         else:
             return [
                 PixivMedia(  # should observe if image_url.original is always None for single page
-                    url=self._illust.image_urls.original or self._illust.meta_single_page.original,
-                    thumb=self._illust.image_urls.medium
+                    url=str(self._illust.image_urls.original or self._illust.meta_single_page.original),
+                    thumb=str(self._illust.image_urls.medium)
                 )
             ]
 
@@ -133,7 +133,7 @@ class ProcessPixiv(_ProcessPixiv):
         illust_id = self._parse_illust_id()
         try:
             return (await self._client.ILLUST.detail(illust_id)).illust
-        except ApiError:
+        except APIError:
             await self.refresh_token()
             return (await self._client.ILLUST.detail(illust_id)).illust  # TODO use retry here
 

utils/regex.py

@@ -1,10 +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/(.+)")
+bsky_url = re.compile(r"^(?:https?://)?bsky\.app/profile/(.+)/post/(.+)")

utils/telegram.py

@@ -258,14 +258,15 @@ class TelegramBsky:
                 # )
                 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 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

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

@@ -31,6 +31,7 @@ class BskyPost(TypedDict):
     author: BskyAuthor
     record: BskyPostRecord
     embed: BskyEmbedImages | BskyEmbedVideo | BskyEmbedExternal
+    labels: list[BskyLabel]
 
 
 class BskyAuthor(TypedDict):
@@ -42,6 +43,10 @@ class BskyPostRecord(TypedDict):
     text: str
 
 
+class BskyLabel(TypedDict):
+    val: str
+
+
 class BskyEmbedImage(TypedDict):
     thumb: str
     fullsize: str