mirror of
https://github.com/tropicoo/yt-dlp-bot.git
synced 2025-03-01 09:12:58 +08:00
62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
import logging
|
|
|
|
from pyrogram.enums import ParseMode
|
|
from pyrogram.types import Message
|
|
from yt_shared.emoji import SUCCESS_EMOJI
|
|
|
|
from bot.core.bot import VideoBot
|
|
from bot.core.service import UrlParser, UrlService
|
|
from bot.core.utils import bold, get_user_id
|
|
|
|
|
|
class TelegramCallback:
|
|
_MSG_SEND_OK = f'{SUCCESS_EMOJI} {bold("{count}URL{plural} sent for download")}'
|
|
_MSG_SEND_FAIL = f'🛑 {bold("Failed to send URL for download")}'
|
|
|
|
def __init__(self) -> None:
|
|
self._log = logging.getLogger(self.__class__.__name__)
|
|
self._url_parser = UrlParser()
|
|
self._url_service = UrlService()
|
|
|
|
@staticmethod
|
|
async def on_start(client: VideoBot, message: Message) -> None:
|
|
await message.reply(
|
|
bold('Send video URL to start processing'),
|
|
parse_mode=ParseMode.HTML,
|
|
reply_to_message_id=message.id,
|
|
)
|
|
|
|
async def on_message(self, client: VideoBot, message: Message) -> None:
|
|
"""Receive video URL and send to the download worker."""
|
|
self._log.debug('Received Telegram Message: %s', message)
|
|
urls = message.text.splitlines()
|
|
user = client.allowed_users[get_user_id(message)]
|
|
if user.use_url_regex_match:
|
|
urls = self._url_parser.filter_urls(
|
|
urls=urls, regexes=client.conf.telegram.url_validation_regexes
|
|
)
|
|
if not urls:
|
|
self._log.debug('No urls to download, skipping message')
|
|
return
|
|
|
|
urls = self._url_parser.parse_urls(urls=urls, message=message, user=user)
|
|
await self._url_service.process_urls(urls)
|
|
await self._send_acknowledge_message(message=message, url_count=len(urls))
|
|
|
|
async def _send_acknowledge_message(
|
|
self,
|
|
message: Message,
|
|
url_count: int,
|
|
) -> None:
|
|
await message.reply(
|
|
text=self._format_acknowledge_text(url_count),
|
|
parse_mode=ParseMode.HTML,
|
|
reply_to_message_id=message.id,
|
|
)
|
|
|
|
def _format_acknowledge_text(self, url_count: int) -> str:
|
|
is_multiple = url_count > 1
|
|
return self._MSG_SEND_OK.format(
|
|
count=f'{url_count} ' if is_multiple else '',
|
|
plural='s' if is_multiple else '',
|
|
)
|