【开源】别让小鸡吃灰 - TG机器人，下载Youtube、X等网站视频转存到115\阿里云盘等网盘

保存为 xxx.py

import os
import shlex
import subprocess
import logging
from tqdm import tqdm
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, CallbackContext
from telegram.ext.filters import TEXT
from urllib.parse import urlparse
from datetime import datetime

# 替换为你的 Telegram 机器人 API Token 
API_TOKEN = '1234567890:beTK1P4NtjnVnLs-gtLHgCQJXyoI0gWgRWg'
# 下载目录
DOWNLOAD_DIR = '/Down'
# 允许使用的用户 ID 
ALLOWED_USER_ID = 1234567890  # 替换为你想允许的用户 ID 
# 网盘挂载路径
MOUNT_PATH = '/115/Server'
# Rclone 配置的项目名称 rclone config 可以查看
RCLONE_NAME = '115'

# 确保下载目录存在
os.makedirs(DOWNLOAD_DIR, exist_ok=True)

# 配置日志记录
LOGGING_ENABLED = os.getenv('LOGGING_ENABLED', 'true').lower() == 'true'
if LOGGING_ENABLED:
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
else:
    logging.basicConfig(level=logging.CRITICAL)  # 只记录严重错误

logger = logging.getLogger(__name__)

# /start 命令的处理函数
async def start(update: Update, context: CallbackContext):
    if update.message.from_user.id != ALLOWED_USER_ID:
        await update.message.reply_text('您无权使用此机器人。')
        return
    await update.message.reply_text('你好！向我发送视频 URL，我将为您下载该视频。')

# 处理收到的 URL
async def handle_message(update: Update, context: CallbackContext):
    if update.message.from_user.id != ALLOWED_USER_ID:
        await update.message.reply_text('您无权使用此机器人。')
        return

    url = update.message.text
    chat_id = update.message.chat_id

    # 解析 URL 中的域名
    domain = urlparse(url).netloc
    domain_dir = os.path.join(DOWNLOAD_DIR, domain)

    # 确保域名对应的目录存在
    os.makedirs(domain_dir, exist_ok=True)
    current_date = datetime.now().strftime('%Y-%m-%d')

    download_command = f'yt-dlp -o {shlex.quote(domain_dir + "/%(title).30s_" + current_date + ".%(ext)s")} {shlex.quote(url)}'

    # 发送下载开始消息
    await update.message.reply_text(f'开始下载：{url}')

    try:
        # 运行 yt-dlp 命令并捕获输出
        process = subprocess.Popen(shlex.split(download_command), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding='utf-8')

        progress_message = await update.message.reply_text('下载进度：0%')
        last_progress = 0

        while True:
            output = process.stdout.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                logger.info(output.strip())
                # 解析进度信息
                if '[download]' in output and '%' in output:
                    progress = output.split('%')[0].split()[-1].strip()
                    try:
                        progress = float(progress)
                        if progress >= last_progress + 10:
                            last_progress = progress
                            await progress_message.edit_text(f'下载进度：{progress}%')
                    except ValueError:
                        pass

        # 捕获下载过程中的错误信息
        stderr_output = process.stderr.read()
        if process.returncode == 0:
            await update.message.reply_text('下载完成！')
            logger.info(f'Download complete for {url}')
             # 同步文件到115:/115/Server
            sync_command = f'rclone move {shlex.quote(DOWNLOAD_DIR)} {RCLONE_NAME}:{shlex.quote(MOUNT_PATH)}'
            sync_process = subprocess.run(shlex.split(sync_command), capture_output=True, text=True)
            
            if sync_process.returncode == 0:
                await update.message.reply_text('文件同步完成！')
                logger.info('File sync complete')
            else:
                await update.message.reply_text(f'文件同步错误：{sync_process.stderr.strip()}')
                logger.error(f'File sync error: {sync_process.stderr.strip()}')

        else:
            await update.message.reply_text(f'Error: {stderr_output.strip()}')
            logger.error(f'Error: {stderr_output.strip()}')
    except Exception as e:
        await update.message.reply_text(f'Error: {str(e)}')
        logger.error(f'Exception occurred: {str(e)}')

def main():
    application = ApplicationBuilder().token(API_TOKEN).build()
    application.add_handler(CommandHandler("start", start))
    application.add_handler(MessageHandler(TEXT, handle_message))  # 更正后的过滤器

    # 运行机器人
    application.run_polling()

if __name__ == '__main__':
    main()

支持

点个star

bd

不错👍

绑定

bd

前排支持

bd

支持

给大佬点赞👍