diff --git a/utils/pixiv.py b/utils/pixiv.py index cc76ba7..bd18a35 100644 --- a/utils/pixiv.py +++ b/utils/pixiv.py @@ -1,24 +1,12 @@ from __future__ import annotations -import html -from typing import Generator, Literal, TYPE_CHECKING -from uuid import uuid4 +from typing import Literal, TYPE_CHECKING from async_pixiv import PixivClient from async_pixiv.error import ApiError -from telegram import InlineQueryResultPhoto, InputMediaPhoto - -from utils.logger import get_logger if TYPE_CHECKING: from async_pixiv.model.illust import Illust - from utils.types import TypeInlineQueryResult, TypeMessageMediaResult - -logger = get_logger(__name__) - -message_raw_text = """{text} / {author} -{tags} -""" class PixivMedia: @@ -110,11 +98,9 @@ class Pixiv: ] -class ProcessPixiv: +class _ProcessPixiv: _client: PixivClient - __slots__ = ('_url', '_illust') - @classmethod async def init_client(cls, token: str) -> None: cls._client = PixivClient() @@ -129,6 +115,10 @@ class ProcessPixiv: async def refresh_token(cls): await cls._client.login_with_token(cls._token) + +class ProcessPixiv(_ProcessPixiv): + __slots__ = ('_url', '_illust') + def __init__(self, url: str): self._url: str = url @@ -149,65 +139,3 @@ class ProcessPixiv: def _parse_illust_id(self) -> int: return int(self._url.split("/")[-1]) - - -class TelegramPixiv: - __slots__ = ('_url', '_pixiv') - - def __init__(self, url: str): - self._url = url - - async def __aenter__(self): - async with ProcessPixiv(self._url) as pixiv: - self._pixiv = pixiv - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - pass - - @property - def url(self) -> str: - return self._pixiv.url - - @property - def message_text(self) -> str: - pixiv = self._pixiv - return message_raw_text.format( - url=pixiv.url, - author_url=pixiv.author_url, - author=html.escape(pixiv.author), - text=html.escape(pixiv.title), - tags=html.escape(" ".join(f"#{name}" for name in pixiv.tags)) - ) - - def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]: - return tuple(self.inline_query_generator()) - - def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]: - return tuple(self.message_media_generator()) - - def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]: - pixiv = self._pixiv - for media in pixiv.images: - logger.info(str(media)) - if pixiv.type in ("illust", "manga"): - yield InlineQueryResultPhoto( - id=str(uuid4()), - photo_url=media.large, - thumbnail_url=media.thumb, - caption=self.message_text - ) - else: - yield - - def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]: - pixiv = self._pixiv - for media in pixiv.images: - logger.info(str(media)) - if pixiv.type in ("illust", "manga"): - yield InputMediaPhoto( - media=media.large, - has_spoiler=pixiv.is_nsfw - ) - else: - yield diff --git a/utils/telegram.py b/utils/telegram.py index b53b5a2..08fac28 100644 --- a/utils/telegram.py +++ b/utils/telegram.py @@ -1,9 +1,31 @@ from __future__ import annotations +import html +from functools import cached_property +from typing import Generator, TYPE_CHECKING +from uuid import uuid4 + +from telegram import InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, InputMediaPhoto, \ + InputMediaVideo + from common import PIXIV_REFRESH_TOKEN -from .pixiv import TelegramPixiv +from .logger import get_logger +from .pixiv import ProcessPixiv from .regex import pixiv_url, x_url -from .tweet import TelegramTweet +from .tweet import ProcessTweet + +if TYPE_CHECKING: + from .types import TypeInlineQueryResult, TypeMessageMediaResult + +logger = get_logger(__name__) + +message_raw_text_tweet = """{url} +{author}: {text} +""" + +message_raw_text_pixiv = """{text} / {author} +{tags} +""" class Telegram: @@ -22,3 +44,154 @@ class Telegram: async def __aexit__(self, exc_type, exc_val, exc_tb): pass + + +class TelegramTweet: + message_raw_text = message_raw_text_tweet + __slots__ = ('_url', '_tweet', '__dict__') + + def __init__(self, url: str): + self._url: str = url + + async def __aenter__(self): + async with ProcessTweet(self._url) as tweet: + self._tweet = tweet + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + pass + + @cached_property + def url(self) -> str: + return self._tweet.url + + @cached_property + def message_text(self) -> str: + tweet = self._tweet + return self.message_raw_text.format( + url=tweet.url, + author_url=tweet.author_url, + author=html.escape(tweet.author), + text=html.escape(tweet.text) + ) + + def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]: + return tuple(self.inline_query_generator()) + + def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]: + return tuple(self.message_media_generator()) + + def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]: + tweet = self._tweet + for tweet_media in tweet.media: + logger.info(str(tweet_media)) + if tweet_media.type == "image": + yield InlineQueryResultPhoto( + 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(uuid4()), + video_url=tweet_media.url, + mime_type="video/mp4", + thumbnail_url=tweet_media.thumb, + title=tweet.text, + caption=self.message_text + ) + elif tweet_media.type == "gif": + yield InlineQueryResultMpeg4Gif( + id=str(uuid4()), + mpeg4_url=tweet_media.url, + thumbnail_url=tweet_media.thumb, + caption=self.message_text + ) + + def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]: + tweet = self._tweet + for tweet_media in tweet.media: + logger.info(str(tweet_media)) + if tweet_media.type == "image": + yield InputMediaPhoto( + media=tweet_media.url, + has_spoiler=tweet.sensitive + ) + elif tweet_media.type == "video": + yield InputMediaVideo( + media=tweet_media.url, + has_spoiler=tweet.sensitive, + thumbnail=tweet_media.thumb + ) + elif tweet_media.type == "gif": + if len(tweet.media) == 1: + yield tweet_media.url, tweet.sensitive + yield InputMediaVideo( + media=tweet_media.url, + has_spoiler=tweet.sensitive, + thumbnail=tweet_media.thumb + ) + + +class TelegramPixiv: + message_raw_text = message_raw_text_pixiv + __slots__ = ('_url', '_pixiv') + + def __init__(self, url: str): + self._url = url + + async def __aenter__(self): + async with ProcessPixiv(self._url) as pixiv: + self._pixiv = pixiv + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + pass + + @property + def url(self) -> str: + return self._pixiv.url + + @property + def message_text(self) -> str: + pixiv = self._pixiv + return self.message_raw_text.format( + url=pixiv.url, + author_url=pixiv.author_url, + author=html.escape(pixiv.author), + text=html.escape(pixiv.title), + tags=html.escape(" ".join(f"#{name}" for name in pixiv.tags)) + ) + + def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]: + return tuple(self.inline_query_generator()) + + def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]: + return tuple(self.message_media_generator()) + + def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]: + pixiv = self._pixiv + for media in pixiv.images: + logger.info(str(media)) + if pixiv.type in ("illust", "manga"): + yield InlineQueryResultPhoto( + id=str(uuid4()), + photo_url=media.large, + thumbnail_url=media.thumb, + caption=self.message_text + ) + else: + yield + + def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]: + pixiv = self._pixiv + for media in pixiv.images: + logger.info(str(media)) + if pixiv.type in ("illust", "manga"): + yield InputMediaPhoto( + media=media.large, + has_spoiler=pixiv.is_nsfw + ) + else: + yield diff --git a/utils/tweet.py b/utils/tweet.py index c41551e..905edda 100644 --- a/utils/tweet.py +++ b/utils/tweet.py @@ -1,30 +1,17 @@ from __future__ import annotations -import html from functools import cached_property -from typing import Generator, TYPE_CHECKING -from uuid import uuid4 +from typing import TYPE_CHECKING -from telegram import (InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, InputMediaPhoto, - InputMediaVideo) - -from .logger import get_logger from .net import NetClient from .regex import x_media_url, x_tco_url, x_url if TYPE_CHECKING: - from .types import TweetInfo, TypeInlineQueryResult, TypeMessageMediaResult - -logger = get_logger(__name__) - + from .types import TweetInfo twimg_url = 'https://pbs.twimg.com/' vx_api_url = 'https://api.vxtwitter.com/{0}/status/{1}' -message_raw_text = """{url} -{author}: {text} -""" - class TweetMedia: __slots__ = ('_url', '_thumb', '_type', '__dict__') @@ -163,88 +150,3 @@ class ProcessTweet: ] -class TelegramTweet: - __slots__ = ('_url', '_tweet', '__dict__') - - def __init__(self, url: str): - self._url: str = url - - async def __aenter__(self): - async with ProcessTweet(self._url) as tweet: - self._tweet = tweet - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - pass - - @cached_property - def url(self) -> str: - return self._tweet.url - - @cached_property - def message_text(self) -> str: - tweet = self._tweet - return message_raw_text.format( - url=tweet.url, - author_url=tweet.author_url, - author=html.escape(tweet.author), - text=html.escape(tweet.text) - ) - - def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]: - return tuple(self.inline_query_generator()) - - def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]: - return tuple(self.message_media_generator()) - - def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]: - tweet = self._tweet - for tweet_media in tweet.media: - logger.info(str(tweet_media)) - if tweet_media.type == "image": - yield InlineQueryResultPhoto( - 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(uuid4()), - video_url=tweet_media.url, - mime_type="video/mp4", - thumbnail_url=tweet_media.thumb, - title=tweet.text, - caption=self.message_text - ) - elif tweet_media.type == "gif": - yield InlineQueryResultMpeg4Gif( - id=str(uuid4()), - mpeg4_url=tweet_media.url, - thumbnail_url=tweet_media.thumb, - caption=self.message_text - ) - - def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]: - tweet = self._tweet - for tweet_media in tweet.media: - logger.info(str(tweet_media)) - if tweet_media.type == "image": - yield InputMediaPhoto( - media=tweet_media.url, - has_spoiler=tweet.sensitive - ) - elif tweet_media.type == "video": - yield InputMediaVideo( - media=tweet_media.url, - has_spoiler=tweet.sensitive, - thumbnail=tweet_media.thumb - ) - elif tweet_media.type == "gif": - if len(tweet.media) == 1: - yield tweet_media.url, tweet.sensitive - yield InputMediaVideo( - media=tweet_media.url, - has_spoiler=tweet.sensitive, - thumbnail=tweet_media.thumb - )