Compare commits

..
14 Commits
Author SHA1 Message Date
YoursFunny f660bc2c5e fix edit message url 2024-05-13 00:38:36 +08:00
YoursFunny 7747dab4a3 fix edit message caption 2024-05-13 00:32:17 +08:00
YoursFunny f2fac6a9e0 fix edit message 2024-05-13 00:24:03 +08:00
YoursFunny e3dcb6e59b fix send gif 2024-05-13 00:19:24 +08:00
YoursFunny a1d1dc5ea3 fix template replace 2024-05-13 00:00:08 +08:00
YoursFunny e629899a72 fix missing return 2024-05-12 23:51:19 +08:00
YoursFunny d079271ef9 fix action 2024-05-12 23:42:18 +08:00
YoursFunny 1526c7900c add set template 2024-05-12 23:34:11 +08:00
YoursFunny 79d68b2e82 fix int admin id 2024-04-28 01:57:10 +08:00
YoursFunny e700ae65ae add message user dict debug 2024-04-28 01:36:36 +08:00
YoursFunny 9ac6065366 remove dict key if disable edit forward 2024-04-28 01:33:50 +08:00
YoursFunny 40877b52eb fix confirm 2024-04-28 01:26:43 +08:00
YoursFunny 0ca9b55ea8 try edit reply markup 2024-04-28 01:18:35 +08:00
YoursFunny ef17b05c14 fix html escape only message 2024-04-28 01:12:02 +08:00
3 changed files with 69 additions and 24 deletions
+2 -2
View File
@@ -3,7 +3,7 @@ import os
import re import re
BOT_TOKEN = os.getenv("BOT_TOKEN") BOT_TOKEN = os.getenv("BOT_TOKEN")
ADMIN = os.getenv("BOT_ADMIN").split(",") ADMIN = [int(i) for i in os.getenv("BOT_ADMIN").split(",")]
WEBHOOK = os.getenv("WEBHOOK", False) WEBHOOK = os.getenv("WEBHOOK", False)
if WEBHOOK: if WEBHOOK:
@@ -17,7 +17,7 @@ if WEBHOOK:
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_media_regex = re.compile(r"^(?:https?://)(pbs|video)\.twimg\.com/(.*)")
x_tco_regex = re.compile(r"(?:https?://)t\.co/.+$", re.M) x_tco_regex = re.compile(r"(?:https?://)t\.co/.+$", re.M)
message_url_regex = re.compile(r"\{.+\}", re.S) message_url_regex = re.compile(r"\[.+]", re.S)
logging.basicConfig( logging.basicConfig(
level=os.getenv("LOG_LEVEL", "WARNING"), level=os.getenv("LOG_LEVEL", "WARNING"),
+56 -18
View File
@@ -2,7 +2,7 @@ import html
from functools import wraps from functools import wraps
from aiohttp import ClientSession from aiohttp import ClientSession
from telegram import Update, Chat, InlineKeyboardMarkup, InlineKeyboardButton, ForceReply, Message from telegram import Update, Chat, InlineKeyboardMarkup, InlineKeyboardButton, Message
from telegram.constants import ParseMode, ChatAction, ChatType from telegram.constants import ParseMode, ChatAction, ChatType
from telegram.ext import ( from telegram.ext import (
Application, Application,
@@ -52,12 +52,19 @@ async def url_media(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
media, media,
caption=tweet.message_text, caption=tweet.message_text,
reply_to_message_id=update.message.message_id, reply_to_message_id=update.message.message_id,
) if not tweet.is_single_gif else await update.effective_message.reply_animation(
media[0][0],
caption=tweet.message_text,
reply_to_message_id=update.message.message_id,
has_spoiler=media[0][1]
) )
url = tweet.url url = tweet.url
if context.user_data.get('edit_before_forward', False): if context.user_data.get('edit_before_forward', False):
message_reply = await update.effective_message.reply_text( message_reply = await update.effective_message.reply_text(
"Reply to edit message.", "Reply to edit message. [URL]",
reply_markup=ForceReply(selective=True, input_field_placeholder="{URL}"), reply_markup=InlineKeyboardMarkup.from_button(
InlineKeyboardButton("↩️ Confirm", callback_data="forward")
),
reply_to_message_id=update.message.message_id, reply_to_message_id=update.message.message_id,
) )
context.user_data['message_reply'] = message_reply context.user_data['message_reply'] = message_reply
@@ -87,20 +94,28 @@ async def edit_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
return return
if update.message.reply_to_message != context.user_data['message_reply']: if update.message.reply_to_message != context.user_data['message_reply']:
return return
update_text = update.message.text template = context.user_data.get('template', None)
match = common.message_url_regex.search(update_text) message_url = '<a href="{0}">{1}</a>'
if match: url = context.user_data['message_url']
match = match.span() if template:
update_text = update_text[:match[0]] + '<a href="{0}">{1}</a>'.format( update_text = template.replace("[]", message_url.format(
context.user_data['message_url'], update_text[match[0] + 1:match[1] - 1] url,
) + update_text[match[1]:] html.escape(update.message.text)
))
else:
update_text = html.escape(update.message.text)
match = common.message_url_regex.search(update_text)
if match:
match = match.span()
update_text = update_text[:match[0]] + message_url.format(
url,
update_text[match[0] + 1:match[1] - 1]
) + update_text[match[1]:]
message_to_send = context.user_data['message_to_send'] message_to_send = context.user_data['message_to_send']
await message_to_send[0].edit_caption(html.escape(update_text)) if isinstance(message_to_send, tuple):
await context.user_data['message_reply'].edit_message_reply_markup( await message_to_send[0].edit_text(update_text)
InlineKeyboardMarkup.from_button( else:
InlineKeyboardButton("↩️ Confirm", callback_data="forward") await message_to_send.edit_caption(update_text)
)
)
async def query_forward_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: async def query_forward_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
@@ -156,12 +171,33 @@ async def cmd_edit_before_forward(update: Update, context: ContextTypes.DEFAULT_
ebf_status = context.user_data.get('edit_before_forward', False) ebf_status = context.user_data.get('edit_before_forward', False)
if ebf_status: if ebf_status:
context.user_data['edit_before_forward'] = False context.user_data['edit_before_forward'] = False
context.user_data.pop('message_reply', None)
context.user_data.pop('message_to_send', None)
context.user_data.pop('message_url', None)
await update.effective_message.reply_text("Disable edit before forward.") await update.effective_message.reply_text("Disable edit before forward.")
return return
context.user_data['edit_before_forward'] = True context.user_data['edit_before_forward'] = True
await update.effective_message.reply_text("Enable edit before forward.") await update.effective_message.reply_text("Enable edit before forward.")
@send_action(ChatAction.TYPING)
async def cmd_set_template(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
reply = update.effective_message.reply_to_message
if not reply:
await update.effective_message.reply_text("Please reply to a message to set as template.")
return
if '[]' not in reply.text_html:
await update.effective_message.reply_text("Please reply to a message with [] to set as template.")
return
context.user_data['template'] = reply.text_html
await update.effective_message.reply_text("Template set.")
@send_action(ChatAction.TYPING)
async def cmd_user_dict(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await update.effective_message.reply_text(str(context.user_data))
async def post_init(application: Application) -> None: async def post_init(application: Application) -> None:
# commands = [ # commands = [
# BotCommand('start', CMD_START), # BotCommand('start', CMD_START),
@@ -194,8 +230,8 @@ def main():
.build() .build()
) )
# user_filter = filters.User() user_filter = filters.User()
# user_filter.add_user_ids(common.admin) user_filter.add_user_ids(common.ADMIN)
handlers = [ handlers = [
MessageHandler(filters.Regex(common.x_url_regex) & filters.ChatType.PRIVATE, url_media), MessageHandler(filters.Regex(common.x_url_regex) & filters.ChatType.PRIVATE, url_media),
@@ -203,8 +239,10 @@ def main():
CommandHandler("set_forward_channel", cmd_set_forward_channel), CommandHandler("set_forward_channel", cmd_set_forward_channel),
CommandHandler("remove_forward_channel", cmd_remove_forward_channel), CommandHandler("remove_forward_channel", cmd_remove_forward_channel),
CommandHandler("edit_before_forward", cmd_edit_before_forward), CommandHandler("edit_before_forward", cmd_edit_before_forward),
CommandHandler("set_template", cmd_set_template),
MessageHandler(~filters.COMMAND & filters.ChatType.PRIVATE, edit_message), MessageHandler(~filters.COMMAND & filters.ChatType.PRIVATE, edit_message),
CallbackQueryHandler(query_forward_message, pattern="forward"), CallbackQueryHandler(query_forward_message, pattern="forward"),
CommandHandler("bot_dict", cmd_user_dict, filters=user_filter),
] ]
application.add_handlers(handlers) application.add_handlers(handlers)
+11 -4
View File
@@ -7,8 +7,7 @@ from telegram import (
InlineQueryResultVideo, InlineQueryResultVideo,
InlineQueryResultMpeg4Gif, InlineQueryResultMpeg4Gif,
InputMediaPhoto, InputMediaPhoto,
InputMediaVideo, InputMediaVideo
InputMediaAnimation
) )
from common import x_url_regex, x_media_regex, x_tco_regex, logger from common import x_url_regex, x_media_regex, x_tco_regex, logger
@@ -125,6 +124,7 @@ class TGTweet(Tweet):
def __init__(self, url: str): def __init__(self, url: str):
self._url: str = url self._url: str = url
self._api_param: tuple[str] = self._tweet_id self._api_param: tuple[str] = self._tweet_id
self._is_single_gif: bool = False
assert self._api_param assert self._api_param
async def __aenter__(self): async def __aenter__(self):
@@ -179,6 +179,10 @@ class TGTweet(Tweet):
sensitive = self._tweet['possibly_sensitive'] sensitive = self._tweet['possibly_sensitive']
return id, author, author_id, text, media, sensitive return id, author, author_id, text, media, sensitive
@property
def is_single_gif(self) -> bool:
return self._is_single_gif
@property @property
def message_text(self) -> str: def message_text(self) -> str:
return message_raw_text.format( return message_raw_text.format(
@@ -219,7 +223,7 @@ class TGTweet(Tweet):
) )
@property @property
def pm_media_generator(self) -> Generator[InputMediaPhoto | InputMediaVideo | InputMediaAnimation, None, None]: def pm_media_generator(self) -> Generator[InputMediaPhoto | InputMediaVideo | tuple[str, bool], None, None]:
for tweet_media in 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":
@@ -234,7 +238,10 @@ class TGTweet(Tweet):
thumbnail=tweet_media.thumb thumbnail=tweet_media.thumb
) )
elif tweet_media.type == "gif": elif tweet_media.type == "gif":
yield InputMediaAnimation( if len(self.media) == 1:
self._is_single_gif = True
yield tweet_media.url, self.sensitive
yield InputMediaVideo(
media=tweet_media.url, media=tweet_media.url,
has_spoiler=self.sensitive, has_spoiler=self.sensitive,
thumbnail=tweet_media.thumb thumbnail=tweet_media.thumb