【开源】别让小鸡吃灰 - 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

支持

给大佬点赞👍

帮顶

鸡脚给你

好贴帮顶

这个支持下载不允许转发不允许下载的视频么？

不错不错

bd

支持

star

MARK

强！

你说得对但是我选择打包成docker放容器云，小鸡继续吃灰

Nice pussy

好办法支持

大佬点赞

建议docker一键

@byte #14 不支持哦 大伙要是需求比较强烈可以研究一下 不过TG下载自己服务器上的内容好像有大小限制

@cabbage #25 docker版我有空弄一个

点赞

策略上如果网盘支持rclone，就没必要用cd2

支持

@BlueSkyXN #29 rclone原生不支持国内的一些网盘 另外用改版的rclone或者alist速度也会有问题

帮顶

帮顶 感谢分享

能下载tg的视频吗

牛

绑定

很好

@brains #27 支持弄一个docker版

@brains #0 很需要

点赞

这个看起来不错，插个眼，以后能用上

支持大佬

这不是一般的强，是超强！

大佬牛逼

已经点亮了

@brains #26 那如果是 TG 下载的话，可以使用这个方案：
https://github.com/tangyoha/telegram_media_downloader

可以直接下载限制分享或下载的视频，自带 rclone 支持。

@byte #48 这项目之前研究过 下载存储在TG上的视频 需要高级API 申请不能说麻烦 但也不简单 另外有下载大小限制 除非另外开个项目把这两个分开

好东西鸡腿送上

提个建议大佬，支持转发tg视频，支持转存到onedrive

@brains #0 教程能不能再详细点，老实说我用不来

@brains #50 API 申请我到没注意，我拿一个新的号来登录的，似乎也没有很麻烦？

另外说的下载大小限制是什么呢？我看我用这个下载的最大的 TG 视频是 3.8GB

能保存pikpak吗

bd

@brains #0

mark

bd

有点意思

? 项目好像没了

...好像是连号都没了

@neek #60 不知道为啥会显示404。。。

@brains #0 cd2可以基于alist吗，如果能融合下115bot这个项目就完美了'只需要一个机器人了

@brains #61 你自己看到的吗

这不得点个star

@neek #63 我自己看得到

佬，网盘失踪了

更新网盘：https://gofile.io/d/yK3meL

@maimaige #66 看帖子置顶 更新了个下载链接

老哥，这个机器人能不能支持下 下载阿里的视频转存到115呀

谢谢分享，收藏先

@bacon159 #69 直接用CloudDrive2或者alist将阿里云盘的资源秒传到115

@brains #71 就是因为无法秒传 才需要vps手动搬的，，，能秒传就不需要这个机器人了

@bacon159 #72 不应该啊 确定CloudDrive2无法秒传么

@brains #73 对 有些国漫资源秒传都是需要有人搬运的 可能过了一天都没人搬运这样 只能自己搬运

@bacon159 #74 那阿里云盘没开第三方应用权益包的话 挂载CloudDrive2想下到本地也会限速吧 只能开了vip下载到本地再上传到115？

@brains #75 限速400k每秒这样 下载了再上传到115如果在vps是没有感知的 肯定比等别人一天后才搬运要快哈哈

@bacon159 #76 那用chatgpt写个搬运脚本啊 两个网盘都挂载到CloudDrive2上 然后从阿里云盘直接用脚本复制到115

牛批

好东西

感谢分享，等一个docker

这玩意不错，谢谢楼主了

马克

BD

谢谢分享

可以直接把tg视频转发给他，让他下载吗

bd

好东西

强！

支持

链接好像失效了

保存为 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()

@poliste #90 直接把代码发到帖子里了 看置顶

bd