mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-23 23:32:05 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7846873b9c
|
||
|
|
6d06e28598
|
||
|
|
4083b09093
|
||
|
|
8fb53e8412
|
||
|
|
5c0ee8b491
|
||
|
|
af561ed0a0
|
+1
-1
@@ -27,5 +27,5 @@
|
||||
.python-version
|
||||
LICENSE
|
||||
README.md
|
||||
pers_data
|
||||
data/
|
||||
cert/
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
**/.idea
|
||||
__pycache__/
|
||||
cert/
|
||||
pers_data
|
||||
data/
|
||||
docker-compose.yml
|
||||
x.py
|
||||
|
||||
@@ -16,6 +16,6 @@ services:
|
||||
WEBHOOK_SECRET_TOKEN: 'secret-token'
|
||||
# LOG_LEVEL: 'WARNING'
|
||||
volumes:
|
||||
- ./pers_data: /app/pers_data
|
||||
- ./data: /app/data
|
||||
# - ./cert:/app/cert
|
||||
container_name: tgxmb
|
||||
|
||||
@@ -22,7 +22,7 @@ async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
|
||||
if query == "":
|
||||
return
|
||||
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)
|
||||
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
|
||||
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)
|
||||
message_sent = await update.effective_message.reply_media_group(
|
||||
media,
|
||||
@@ -92,7 +92,7 @@ async def post_init(application: Application) -> None:
|
||||
DESCRIPTION = "A bot to fetch tweets from Twitter."
|
||||
await application.bot.set_my_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:
|
||||
@@ -100,12 +100,12 @@ async def post_stop(application: Application) -> None:
|
||||
|
||||
|
||||
async def post_shutdown(application: Application) -> None:
|
||||
await application.bot_data['client'].close()
|
||||
await TGTweet.close_session()
|
||||
|
||||
|
||||
def main():
|
||||
defaults = Defaults(parse_mode=ParseMode.HTML, allow_sending_without_reply=True)
|
||||
persistence = PicklePersistence(filepath='pers_data')
|
||||
persistence = PicklePersistence(filepath='data/pers.pkl')
|
||||
application = (ApplicationBuilder()
|
||||
.token(common.BOT_TOKEN)
|
||||
.defaults(defaults)
|
||||
|
||||
@@ -4,8 +4,10 @@ from aiohttp import ClientSession
|
||||
from telegram import (
|
||||
InlineQueryResultPhoto,
|
||||
InlineQueryResultVideo,
|
||||
InlineQueryResultMpeg4Gif,
|
||||
InputMediaPhoto,
|
||||
InputMediaVideo
|
||||
InputMediaVideo,
|
||||
InputMediaAnimation
|
||||
)
|
||||
|
||||
from common import x_url_regex, x_media_regex, x_tco_regex, logger
|
||||
@@ -48,6 +50,10 @@ class TweetMedia:
|
||||
return f"{twimg_url}{self._uri}?format=jpg&name=4096x4096"
|
||||
case "video":
|
||||
return self._url
|
||||
case "gif":
|
||||
return self._url
|
||||
case _:
|
||||
return self._url
|
||||
|
||||
@property
|
||||
def thumb(self) -> str:
|
||||
@@ -56,6 +62,10 @@ class TweetMedia:
|
||||
return f"{twimg_url}{self._uri}?format=jpg&name=thumb"
|
||||
case "video":
|
||||
return self._thumb
|
||||
case "gif":
|
||||
return self._thumb
|
||||
case _:
|
||||
return self._thumb
|
||||
|
||||
@property
|
||||
def type(self) -> str:
|
||||
@@ -109,8 +119,9 @@ class Tweet:
|
||||
|
||||
|
||||
class TGTweet(Tweet):
|
||||
def __init__(self, session: ClientSession, url: str):
|
||||
self._session: ClientSession = session
|
||||
_session: ClientSession
|
||||
|
||||
def __init__(self, url: str):
|
||||
self._url: str = url
|
||||
self._api_param: tuple[str] = self._tweet_id
|
||||
assert self._api_param
|
||||
@@ -123,6 +134,14 @@ class TGTweet(Tweet):
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
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:
|
||||
return await fetch_json(self._session, vx_api_url.format(*api_param))
|
||||
|
||||
@@ -177,8 +196,6 @@ class TGTweet(Tweet):
|
||||
id=str(i),
|
||||
photo_url=tweet_media.url,
|
||||
thumbnail_url=tweet_media.thumb,
|
||||
title=self.url,
|
||||
description=self.text,
|
||||
caption=self.message_text if not i else None
|
||||
)
|
||||
elif tweet_media.type == "video":
|
||||
@@ -187,8 +204,14 @@ class TGTweet(Tweet):
|
||||
video_url=tweet_media.url,
|
||||
mime_type="video/mp4",
|
||||
thumbnail_url=tweet_media.thumb,
|
||||
title=self.url,
|
||||
description=self.text,
|
||||
title=self.text,
|
||||
caption=self.message_text if not i else None
|
||||
)
|
||||
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
|
||||
)
|
||||
|
||||
@@ -207,3 +230,9 @@ class TGTweet(Tweet):
|
||||
has_spoiler=self.sensitive,
|
||||
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