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 import html
from functools import wraps from functools import wraps
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@@ -8,7 +10,7 @@ from telegram.ext import (ApplicationBuilder, CallbackQueryHandler, CommandHandl
InlineQueryHandler, MessageHandler, PicklePersistence, filters) InlineQueryHandler, MessageHandler, PicklePersistence, filters)
import common import common
from tweet import TGTweet from tweet import TelegramTweet
if TYPE_CHECKING: if TYPE_CHECKING:
from telegram import Chat, Message, Update from telegram import Chat, Message, Update
+15 -8
View File
@@ -1,5 +1,9 @@
from __future__ import annotations
import html 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 httpx import AsyncClient
from telegram import (InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, InputMediaPhoto, 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 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__) logger = get_logger(__name__)
@@ -18,15 +25,15 @@ message_raw_text = """{url}
""" """
def create_client() -> 'AsyncClient': def create_client() -> AsyncClient:
return AsyncClient(http2=True) return AsyncClient(http2=True)
async def close_client(_client: 'AsyncClient') -> None: async def close_client(_client: AsyncClient) -> None:
await _client.aclose() 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}") logger.info(f"Fetching {url}")
response = await _client.get(url) response = await _client.get(url)
assert response.status_code == response.is_success, f"Failed to fetch {url}, status code {response.status_code}" 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[ def inline_query_generator(self) -> Generator[
InlineQueryResultPhoto | InlineQueryResultVideo | InlineQueryResultMpeg4Gif, None, None InlineQueryResultPhoto | InlineQueryResultVideo | InlineQueryResultMpeg4Gif, None, None
]: ]:
for i, tweet_media in enumerate(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":
yield InlineQueryResultPhoto( yield InlineQueryResultPhoto(
id=str(i), id=str(uuid4()),
photo_url=tweet_media.url, photo_url=tweet_media.url,
thumbnail_url=tweet_media.thumb, thumbnail_url=tweet_media.thumb,
caption=self.message_text caption=self.message_text
) )
elif tweet_media.type == "video": elif tweet_media.type == "video":
yield InlineQueryResultVideo( yield InlineQueryResultVideo(
id=str(i), id=str(uuid4()),
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,
@@ -222,7 +229,7 @@ class TGTweet(Tweet):
) )
elif tweet_media.type == "gif": elif tweet_media.type == "gif":
yield InlineQueryResultMpeg4Gif( yield InlineQueryResultMpeg4Gif(
id=str(i), id=str(uuid4()),
mpeg4_url=tweet_media.url, mpeg4_url=tweet_media.url,
thumbnail_url=tweet_media.thumb, thumbnail_url=tweet_media.thumb,
caption=self.message_text caption=self.message_text