Compare commits

..
2 Commits
Author SHA1 Message Date
YoursFunny ae1f5a8c8c rename 2024-06-20 02:48:06 +08:00
YoursFunny 38adbf0cb8 use cached property 2024-06-20 02:40:35 +08:00
2 changed files with 17 additions and 17 deletions
+5 -5
View File
@@ -36,7 +36,7 @@ async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
if query == "": if query == "":
return return
logger.info(f"Query: {query}") logger.info(f"Query: {query}")
async with TGTweet(query) as tweet: async with TelegramTweet(query) as tweet:
result = list(tweet.inline_query_generator) result = list(tweet.inline_query_generator)
await update.inline_query.answer(result) await update.inline_query.answer(result)
@@ -45,8 +45,8 @@ async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
async def url_media(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: async def url_media(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
url = update.message.text url = update.message.text
logger.info(f"Receiving url: {url}") logger.info(f"Receiving url: {url}")
async with TGTweet(url) as tweet: async with TelegramTweet(url) as tweet:
media = list(tweet.pm_media_generator) media = list(tweet.message_media_generator)
message_to_send = await update.effective_message.reply_media_group( message_to_send = await update.effective_message.reply_media_group(
media, media,
caption=tweet.message_text, caption=tweet.message_text,
@@ -204,7 +204,7 @@ async def post_init(application: Application) -> None:
DESCRIPTION = "A bot to fetch tweets from Twitter." DESCRIPTION = "A bot to fetch tweets from Twitter."
await application.bot.set_my_description(DESCRIPTION) await application.bot.set_my_description(DESCRIPTION)
await application.bot.set_my_short_description(DESCRIPTION) await application.bot.set_my_short_description(DESCRIPTION)
TGTweet.init_client() TelegramTweet.init_client()
async def post_stop(application: Application) -> None: async def post_stop(application: Application) -> None:
@@ -212,7 +212,7 @@ async def post_stop(application: Application) -> None:
async def post_shutdown(application: Application) -> None: async def post_shutdown(application: Application) -> None:
await TGTweet.close_client() await TelegramTweet.close_client()
def main(): def main():
+12 -12
View File
@@ -14,6 +14,15 @@ from common import get_logger, x_media_regex, x_tco_regex, x_url_regex
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Generator, TypedDict from typing import Generator, TypedDict
class TweetInfo(TypedDict):
tweetID: str
user_name: str
user_screen_name: str
text: str
media_extended: list[dict]
possibly_sensitive: bool
logger = get_logger(__name__) logger = get_logger(__name__)
@@ -135,16 +144,6 @@ class Tweet:
return self._sensitive return self._sensitive
class TGTweet(Tweet):
class TweetInfo(TypedDict):
tweetID: str
user_name: str
user_screen_name: str
text: str
media_extended: list[dict]
possibly_sensitive: bool
class ProcessTweet: class ProcessTweet:
__slots__ = ('_httpx_client', '_url', '_tweet') __slots__ = ('_httpx_client', '_url', '_tweet')
@@ -189,7 +188,8 @@ class ProcessTweet:
] ]
class TelegramTweet: class TelegramTweet(Tweet):
_httpx_client: AsyncClient
def __init__(self, url: str): def __init__(self, url: str):
self._url: str = url self._url: str = url
@@ -293,7 +293,7 @@ class TelegramTweet:
) )
@property @property
def pm_media_generator(self) -> Generator[InputMediaPhoto | InputMediaVideo | tuple[str, bool], None, None]: def message_media_generator(self) -> Generator[InputMediaPhoto | InputMediaVideo | tuple[str, bool], None, None]:
for tweet_media in self.media: for tweet_media in self.media:
logger.info(str(tweet_media)) logger.info(str(tweet_media))
if tweet_media.type == "image": if tweet_media.type == "image":