diff --git a/common.py b/common.py index d65e978..28c96bc 100644 --- a/common.py +++ b/common.py @@ -14,7 +14,7 @@ if WEBHOOK: WEBHOOK_CERT = os.getenv("WEBHOOK_CERT", "cert/cert.pem") WEBHOOK_SECRET_TOKEN = os.getenv("WEBHOOK_SECRET_TOKEN") -x_url_regex = re.compile(r"^(?:https?://)(?:www\.|mobile\.|)(?:x|twitter)\.com/.+/status/(\d+)") +x_url_regex = re.compile(r"^(?:https?://)(?:www\.|mobile\.|)(?:x|twitter)\.com/(.+)/status/(\d+)") x_media_regex = re.compile(r"^(?:https?://)(pbs|video)\.twimg\.com/(.*)") x_tco_regex = re.compile(r"(?:https?://)t\.co/.+$", re.M) diff --git a/main.py b/main.py index 58a56bb..4926d1a 100644 --- a/main.py +++ b/main.py @@ -9,7 +9,8 @@ from telegram.ext import ( filters, InlineQueryHandler, PicklePersistence, - MessageHandler, CommandHandler + MessageHandler, + CommandHandler ) import common @@ -32,7 +33,8 @@ async def reply_media(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non async with TGTweet(context.bot_data['client'], url) as tweet: media = list(tweet.pm_media_generator) message_sent = await update.effective_message.reply_media_group( - media, caption=tweet.message_text, + media, + caption=tweet.message_text, reply_to_message_id=update.message.message_id ) if 'forward_channel_id' in context.user_data: diff --git a/tweet.py b/tweet.py index 35a6989..562723d 100644 --- a/tweet.py +++ b/tweet.py @@ -11,7 +11,7 @@ from telegram import ( from common import x_url_regex, x_media_regex, x_tco_regex twimg_url = 'https://pbs.twimg.com/' -vx_api_url = 'https://api.vxtwitter.com/status/' +vx_api_url = 'https://api.vxtwitter.com/status/{0}/status/{1}' message_raw_text = """{url} {author}: {text} @@ -108,25 +108,25 @@ class TGTweet(Tweet): def __init__(self, session: ClientSession, url: str): self._session: ClientSession = session self._url: str = url - self._id: str = self._tweet_id - assert self._id + self._api_param: tuple[str] = self._tweet_id + assert self._api_param async def __aenter__(self): - self._tweet: dict = await self._fetch_tweet(self._id) + self._tweet: dict = await self._fetch_tweet(self._api_param) super().__init__(*self._init_properties) return self async def __aexit__(self, exc_type, exc_val, exc_tb): pass - async def _fetch_tweet(self, tweet_id: str) -> dict: - return await fetch_json(self._session, vx_api_url + tweet_id) + async def _fetch_tweet(self, api_param: tuple[str]) -> dict: + return await fetch_json(self._session, vx_api_url.format(*api_param)) @property - def _tweet_id(self) -> str | None: + def _tweet_id(self) -> tuple[str] | None: match = x_url_regex.match(self._url) if match: - return match.group(1).strip('/') + return match.groups() return None @property