92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
import os
|
|
import json
|
|
import requests
|
|
import logging
|
|
from telegram import Update
|
|
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackContext
|
|
|
|
# Set up logging
|
|
logging.basicConfig(
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
level=logging.DEBUG
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Function to start the bot
|
|
def start(update: Update, context: CallbackContext) -> None:
|
|
logger.debug("Received /start command.")
|
|
update.message.reply_text('Hello! I will download YouTube links in mp3.')
|
|
|
|
# Function to handle messages
|
|
def handle_message(update: Update, context: CallbackContext) -> None:
|
|
message_text = update.message.text
|
|
logger.debug(f"Received message: {message_text}")
|
|
|
|
# Check if the message contains a YouTube link
|
|
if "youtube.com" in message_text or "youtu.be" in message_text:
|
|
url = message_text.strip() # Get the URL from the message
|
|
logger.debug(f"Detected YouTube link: {url}")
|
|
|
|
# Prepare the JSON payload
|
|
payload = {
|
|
"url": url,
|
|
"quality": "best",
|
|
"format": "mp3",
|
|
"playlist_strict_mode": False,
|
|
"auto_start": True
|
|
}
|
|
|
|
# Read the MeTube URL from environment variable
|
|
metube_url = os.getenv('METUBE_URL')
|
|
logger.debug(f"Using MeTube URL: {metube_url}")
|
|
|
|
# Send a POST request
|
|
try:
|
|
response = requests.post(f'{metube_url}/add', json=payload)
|
|
logger.debug(f"POST request sent. Response status code: {response.status_code}")
|
|
|
|
# Check the response status
|
|
if response.status_code == 200:
|
|
update.message.reply_text(f"Download started for {url}")
|
|
logger.info(f"Download started for {url}")
|
|
else:
|
|
update.message.reply_text("Failed to start download.")
|
|
logger.error(f"Failed to start download for {url}. Response: {response.text}")
|
|
except Exception as e:
|
|
logger.exception("An error occurred while sending the POST request.")
|
|
update.message.reply_text("An error occurred while trying to start the download.")
|
|
else:
|
|
# Echo the message if it's not a YouTube link
|
|
update.message.reply_text('No YouTube link provided. Please use youtube.com or youtu.be')
|
|
logger.warning("No valid YouTube link provided.")
|
|
|
|
def main() -> None:
|
|
# Read the Telegram API token from environment variable
|
|
api_token = os.getenv('TELEGRAM_API_TOKEN')
|
|
|
|
# Check if the API token is set
|
|
if not api_token:
|
|
logger.error("TELEGRAM_API_TOKEN environment variable not set.")
|
|
return
|
|
|
|
updater = Updater(api_token)
|
|
|
|
# Get the dispatcher to register handlers
|
|
dispatcher = updater.dispatcher
|
|
|
|
# Register command and message handlers
|
|
dispatcher.add_handler(CommandHandler("start", start))
|
|
dispatcher.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_message))
|
|
|
|
# Start the Bot
|
|
updater.start_polling()
|
|
logger.info("Bot started and polling for messages.")
|
|
|
|
# Run the bot until you send a signal to stop
|
|
updater.idle()
|
|
logger.info("Bot stopped.")
|
|
|
|
if __name__ == '__main__':
|
|
main()
|