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 .bsky import ProcessBsky
from .logger import get_logger
from .pixiv import ProcessPixiv
from .regex import bsky_url, pixiv_url, x_url
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:
def __init__(self, url: str):
self._url = url
async def __aenter__(self):
if x_url.match(self._url):
async with TelegramTweet(self._url) as tweet:
return tweet
elif PIXIV_REFRESH_TOKEN and pixiv_url.match(self._url):
async with TelegramPixiv(self._url) as pixiv:
return pixiv
elif bsky_url.match(self._url):
async with TelegramBsky(self._url) as bsky:
return bsky
else:
return None # TODO add raise and catch
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
@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(i for i in self.inline_query_generator() if i)
def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]:
return tuple(i for i in self.message_media_generator() if i)
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
class TelegramBsky:
message_raw_text = message_raw_text_tweet
__slots__ = ('_url', '_bsky', '__dict__')
def __init__(self, url: str):
self._url: str = url
async def __aenter__(self):
async with ProcessBsky(self._url) as bsky:
self._bsky = bsky
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
@property
def url(self) -> str:
return self._bsky.url
@cached_property
def message_text(self) -> str:
bsky = self._bsky
return self.message_raw_text.format(
url=bsky.url,
author_url=bsky.author_url,
author=html.escape(bsky.author),
text=html.escape(bsky.text)
)
def inline_query_result(self) -> tuple[TypeInlineQueryResult, ...]:
return tuple(i for i in self.inline_query_generator() if i)
def message_media_result(self) -> tuple[TypeMessageMediaResult, ...]:
return tuple(i for i in self.message_media_generator() if i)
def inline_query_generator(self) -> Generator[TypeInlineQueryResult, None, None]:
bsky = self._bsky
for bsky_media in bsky.media:
logger.info(str(bsky_media))
if bsky_media.type == "image":
yield InlineQueryResultPhoto(
id=str(uuid4()),
photo_url=bsky_media.url,
thumbnail_url=bsky_media.thumb,
caption=self.message_text
)
elif bsky_media.type == "video":
# yield InlineQueryResultVideo(
# id=str(uuid4()),
# video_url=bsky_media.url,
# mime_type="video/mp4",
# thumbnail_url=bsky_media.thumb,
# title=bsky.text,
# caption=self.message_text
# )
yield
elif bsky_media.type == "external":
# yield InlineQueryResultVideo(
# id=str(uuid4()),
# video_url=bsky_media.url,
# mime_type="image/gif",
# thumbnail_url=bsky_media.thumb,
# title=bsky.text,
# caption=self.message_text
# )
yield
def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]:
bsky = self._bsky
for bsky_media in bsky.media:
logger.info(str(bsky_media))
if bsky_media.type == "image":
yield InputMediaPhoto(
media=bsky_media.url,
has_spoiler=bsky.sensitive
)
elif bsky_media.type == "video":
# yield InputMediaVideo(
# media=bsky_media.url,
# has_spoiler=bsky.sensitive,
# thumbnail=bsky_media.thumb
# )
yield
elif bsky_media.type == "external":
yield bsky_media.url, bsky.sensitive