mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-23 23:32:05 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4083b09093
|
||
|
|
8fb53e8412
|
@@ -22,7 +22,7 @@ async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
|
|||||||
if query == "":
|
if query == "":
|
||||||
return
|
return
|
||||||
common.logger.info(f"Query: {query}")
|
common.logger.info(f"Query: {query}")
|
||||||
async with TGTweet(context.bot_data['client'], query) as tweet:
|
async with TGTweet(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)
|
||||||
|
|
||||||
@@ -32,7 +32,7 @@ async def reply_media(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non
|
|||||||
# url = update.message.text.split(" ")[0]
|
# url = update.message.text.split(" ")[0]
|
||||||
url = update.message.text
|
url = update.message.text
|
||||||
common.logger.info(f"Receiving url: {url}")
|
common.logger.info(f"Receiving url: {url}")
|
||||||
async with TGTweet(context.bot_data['client'], url) as tweet:
|
async with TGTweet(url) as tweet:
|
||||||
media = list(tweet.pm_media_generator)
|
media = list(tweet.pm_media_generator)
|
||||||
message_sent = await update.effective_message.reply_media_group(
|
message_sent = await update.effective_message.reply_media_group(
|
||||||
media,
|
media,
|
||||||
@@ -92,7 +92,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)
|
||||||
application.bot_data['client'] = ClientSession()
|
TGTweet.set_session(ClientSession())
|
||||||
|
|
||||||
|
|
||||||
async def post_stop(application: Application) -> None:
|
async def post_stop(application: Application) -> None:
|
||||||
@@ -100,7 +100,7 @@ async def post_stop(application: Application) -> None:
|
|||||||
|
|
||||||
|
|
||||||
async def post_shutdown(application: Application) -> None:
|
async def post_shutdown(application: Application) -> None:
|
||||||
await application.bot_data['client'].close()
|
await TGTweet.close_session()
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ from telegram import (
|
|||||||
InlineQueryResultPhoto,
|
InlineQueryResultPhoto,
|
||||||
InlineQueryResultVideo,
|
InlineQueryResultVideo,
|
||||||
InputMediaPhoto,
|
InputMediaPhoto,
|
||||||
InputMediaVideo
|
InputMediaVideo, InputMediaAnimation, InlineQueryResultMpeg4Gif
|
||||||
)
|
)
|
||||||
|
|
||||||
from common import x_url_regex, x_media_regex, x_tco_regex, logger
|
from common import x_url_regex, x_media_regex, x_tco_regex, logger
|
||||||
@@ -109,8 +109,9 @@ class Tweet:
|
|||||||
|
|
||||||
|
|
||||||
class TGTweet(Tweet):
|
class TGTweet(Tweet):
|
||||||
def __init__(self, session: ClientSession, url: str):
|
_session: ClientSession
|
||||||
self._session: ClientSession = session
|
|
||||||
|
def __init__(self, url: str):
|
||||||
self._url: str = url
|
self._url: str = url
|
||||||
self._api_param: tuple[str] = self._tweet_id
|
self._api_param: tuple[str] = self._tweet_id
|
||||||
assert self._api_param
|
assert self._api_param
|
||||||
@@ -123,6 +124,14 @@ class TGTweet(Tweet):
|
|||||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def set_session(cls, session: ClientSession) -> None:
|
||||||
|
cls._session = session
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
async def close_session(cls) -> None:
|
||||||
|
await cls._session.close()
|
||||||
|
|
||||||
async def _fetch_tweet(self, api_param: tuple[str]) -> dict:
|
async def _fetch_tweet(self, api_param: tuple[str]) -> dict:
|
||||||
return await fetch_json(self._session, vx_api_url.format(*api_param))
|
return await fetch_json(self._session, vx_api_url.format(*api_param))
|
||||||
|
|
||||||
@@ -177,8 +186,6 @@ class TGTweet(Tweet):
|
|||||||
id=str(i),
|
id=str(i),
|
||||||
photo_url=tweet_media.url,
|
photo_url=tweet_media.url,
|
||||||
thumbnail_url=tweet_media.thumb,
|
thumbnail_url=tweet_media.thumb,
|
||||||
title=self.url,
|
|
||||||
description=self.text,
|
|
||||||
caption=self.message_text if not i else None
|
caption=self.message_text if not i else None
|
||||||
)
|
)
|
||||||
elif tweet_media.type == "video":
|
elif tweet_media.type == "video":
|
||||||
@@ -187,8 +194,13 @@ class TGTweet(Tweet):
|
|||||||
video_url=tweet_media.url,
|
video_url=tweet_media.url,
|
||||||
mime_type="video/mp4",
|
mime_type="video/mp4",
|
||||||
thumbnail_url=tweet_media.thumb,
|
thumbnail_url=tweet_media.thumb,
|
||||||
title=self.url,
|
caption=self.message_text if not i else None
|
||||||
description=self.text,
|
)
|
||||||
|
elif tweet_media.type == "gif":
|
||||||
|
yield InlineQueryResultMpeg4Gif(
|
||||||
|
id=str(i),
|
||||||
|
mpeg4_url=tweet_media.url,
|
||||||
|
thumbnail_url=tweet_media.thumb,
|
||||||
caption=self.message_text if not i else None
|
caption=self.message_text if not i else None
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -207,3 +219,9 @@ class TGTweet(Tweet):
|
|||||||
has_spoiler=self.sensitive,
|
has_spoiler=self.sensitive,
|
||||||
thumbnail=tweet_media.thumb
|
thumbnail=tweet_media.thumb
|
||||||
)
|
)
|
||||||
|
elif tweet_media.type == "gif":
|
||||||
|
yield InputMediaAnimation(
|
||||||
|
media=tweet_media.url,
|
||||||
|
has_spoiler=self.sensitive,
|
||||||
|
thumbnail=tweet_media.thumb
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user