Compare commits

..
2 Commits
Author SHA1 Message Date
YoursFunny 55ae73ee6b use uuid for query 2024-06-20 02:21:14 +08:00
YoursFunny d6589fec5b fix annotations 2024-06-20 02:20:10 +08:00
2 changed files with 18 additions and 9 deletions
+3 -1
View File
@@ -1,3 +1,5 @@
from __future__ import annotations
import html
from functools import wraps
from typing import TYPE_CHECKING
@@ -8,7 +10,7 @@ from telegram.ext import (ApplicationBuilder, CallbackQueryHandler, CommandHandl
InlineQueryHandler, MessageHandler, PicklePersistence, filters)
import common
from tweet import TGTweet
from tweet import TelegramTweet
if TYPE_CHECKING:
from telegram import Chat, Message, Update
+15 -8
View File
@@ -1,5 +1,9 @@
from __future__ import annotations
import html
from typing import Generator
from functools import cached_property
from typing import TYPE_CHECKING
from uuid import uuid4
from httpx import AsyncClient
from telegram import (InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, InputMediaPhoto,
@@ -7,6 +11,9 @@ from telegram import (InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQ
from common import get_logger, x_media_regex, x_tco_regex, x_url_regex
if TYPE_CHECKING:
from typing import Generator, TypedDict
logger = get_logger(__name__)
@@ -18,15 +25,15 @@ message_raw_text = """{url}
"""
def create_client() -> 'AsyncClient':
def create_client() -> AsyncClient:
return AsyncClient(http2=True)
async def close_client(_client: 'AsyncClient') -> None:
async def close_client(_client: AsyncClient) -> None:
await _client.aclose()
async def fetch_json(_client: 'AsyncClient', url: str) -> dict:
async def fetch_json(_client: AsyncClient, url: str) -> dict:
logger.info(f"Fetching {url}")
response = await _client.get(url)
assert response.status_code == response.is_success, f"Failed to fetch {url}, status code {response.status_code}"
@@ -202,18 +209,18 @@ class TGTweet(Tweet):
def inline_query_generator(self) -> Generator[
InlineQueryResultPhoto | InlineQueryResultVideo | InlineQueryResultMpeg4Gif, None, None
]:
for i, tweet_media in enumerate(self.media):
for tweet_media in self.media:
logger.info(str(tweet_media))
if tweet_media.type == "image":
yield InlineQueryResultPhoto(
id=str(i),
id=str(uuid4()),
photo_url=tweet_media.url,
thumbnail_url=tweet_media.thumb,
caption=self.message_text
)
elif tweet_media.type == "video":
yield InlineQueryResultVideo(
id=str(i),
id=str(uuid4()),
video_url=tweet_media.url,
mime_type="video/mp4",
thumbnail_url=tweet_media.thumb,
@@ -222,7 +229,7 @@ class TGTweet(Tweet):
)
elif tweet_media.type == "gif":
yield InlineQueryResultMpeg4Gif(
id=str(i),
id=str(uuid4()),
mpeg4_url=tweet_media.url,
thumbnail_url=tweet_media.thumb,
caption=self.message_text