Телеграм бот - делаем кружки из видео

С помощью этого бота можно самому делать кружки из видео, быстро и бесплатно

import os
import random
from moviepy.editor import VideoFileClip
from telegram import Update, InputFile
from telegram.ext import ApplicationBuilder, MessageHandler, CommandHandler, ContextTypes, filters

# === Настройки ===
BOT_TOKEN = ''
OWNER_ID =

# Хранилище статистики и качества
user_stats = {}
user_quality = {}

# === Вспомогательные функции ===
def get_bitrate(quality: str):
return {
'low': '300k',
'medium': '800k',
'high': '1500k'
}.get(quality, None) # None = оригинал

def is_authorized(user_id):
return user_id == OWNER_ID

def get_random_phrase(stage: str):
phrases = {
'start': [
" Получаю видосик!",
" Поймал видео!",
" Загрузка пошла!"
],
'processing': [
" Колдую над кружком...",
" Делаю магию видео!",
" Превращаю в кружочек..."
],
'done': [
" Готово! Лови!",
" Кружок готов!",
" Вуаля!"
]
}
return random.choice(phrases.get(stage, ["..."]))

# === Команды ===
async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not is_authorized(update.effective_user.id): return
await update.message.reply_text(
"/help — показать это сообщение\n"
"/setquality low|medium|high — установить качество\n"
"/status — показать статистику\n\n"
"Просто отправь мне видео — я сделаю из него кружочек!"
)

async def set_quality(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not is_authorized(update.effective_user.id): return
args = context.args
if not args or args[0] not in ['low', 'medium', 'high']:
await update.message.reply_text(" Укажи качество: /setquality low|medium|high")
return
user_quality[update.effective_user.id] = args[0]
await update.message.reply_text(f" Качество установлено: {args[0]}")

async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not is_authorized(update.effective_user.id): return
count = user_stats.get(update.effective_user.id, 0)
quality = user_quality.get(update.effective_user.id, "original")
await update.message.reply_text(
f" Обработано видео: {count}\n Текущее качество: {quality}"
)

# === Обработка видео ===
async def handle_video(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
if not is_authorized(user_id):
await update.message.reply_text(" У тебя нет доступа к этому боту.")
return

video = update.message.video
if not video:
await update.message.reply_text(" Отправь обычное видео, а не файл.")
return

await update.message.reply_text(get_random_phrase('start'))

# Скачивание
input_path = "input.mp4"
output_path = "output.mp4"
file = await context.bot.get_file(video.file_id)
await file.download_to_drive(input_path)

await update.message.reply_text(get_random_phrase('processing'))

try:
clip = VideoFileClip(input_path).resize(height=360)

# Применяем битрейт, если есть
user_bitrate = get_bitrate(user_quality.get(user_id, "original"))

clip.write_videofile(
output_path,
codec="libx264",
audio_codec="aac",
bitrate=user_bitrate,
threads=4,
logger=None
)

with open(output_path, "rb") as f:
await context.bot.send_video_note(chat_id=update.message.chat_id, video_note=f)

await update.message.reply_text(get_random_phrase('done'))

# Статистика
user_stats[user_id] = user_stats.get(user_id, 0) + 1

except Exception as e:
await update.message.reply_text(f" Ошибка: {e}")

finally:
for file in [input_path, output_path]:
if os.path.exists(file): os.remove(file)

# === Текстовые сообщения ===
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
if is_authorized(update.effective_user.id):
await update.message.reply_text(" Просто отправь видео, и я сделаю кружок!")
else:
await update.message.reply_text(" У тебя нет доступа.")

# === Запуск ===
def main():
app = ApplicationBuilder().token(BOT_TOKEN).build()

app.add_handler(CommandHandler("help", help_cmd))
app.add_handler(CommandHandler("setquality", set_quality))
app.add_handler(CommandHandler("status", status))
app.add_handler(MessageHandler(filters.VIDEO, handle_video))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

print(" Бот запущен!")
app.run_polling()

if __name__ == "__main__":
main()

Python

import os

import random

from moviepy.editor import VideoFileClip

from telegram import Update, InputFile

from telegram.ext import ApplicationBuilder, MessageHandler, CommandHandler, ContextTypes, filters



# === Настройки ===

BOT_TOKEN = ''

OWNER_ID =



# Хранилище статистики и качества

user_stats = {}

user_quality = {}



# === Вспомогательные функции ===

def get_bitrate(quality: str):

    return {

        'low': '300k',

        'medium': '800k',

        'high': '1500k'

    }.get(quality, None)  # None = оригинал



def is_authorized(user_id):

    return user_id == OWNER_ID



def get_random_phrase(stage: str):

    phrases = {

        'start': [

            " Получаю видосик!",

            " Поймал видео!",

            " Загрузка пошла!"

        ],

        'processing': [

            " Колдую над кружком...",

            " Делаю магию видео!",

            " Превращаю в кружочек..."

        ],

        'done': [

            " Готово! Лови!",

            " Кружок готов!",

            " Вуаля!"

        ]

    }

    return random.choice(phrases.get(stage, ["..."]))



# === Команды ===

async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE):

    if not is_authorized(update.effective_user.id): return

    await update.message.reply_text(

        "/help — показать это сообщение\n"

        "/setquality low|medium|high — установить качество\n"

        "/status — показать статистику\n\n"

        "Просто отправь мне видео — я сделаю из него кружочек!"

    )



async def set_quality(update: Update, context: ContextTypes.DEFAULT_TYPE):

    if not is_authorized(update.effective_user.id): return

    args = context.args

    if not args or args[0] not in ['low', 'medium', 'high']:

        await update.message.reply_text(" Укажи качество: /setquality low|medium|high")

        return

    user_quality[update.effective_user.id] = args[0]

    await update.message.reply_text(f" Качество установлено: {args[0]}")



async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):

    if not is_authorized(update.effective_user.id): return

    count = user_stats.get(update.effective_user.id, 0)

    quality = user_quality.get(update.effective_user.id, "original")

    await update.message.reply_text(

        f" Обработано видео: {count}\n Текущее качество: {quality}"

    )



# === Обработка видео ===

async def handle_video(update: Update, context: ContextTypes.DEFAULT_TYPE):

    user_id = update.effective_user.id

    if not is_authorized(user_id):

        await update.message.reply_text(" У тебя нет доступа к этому боту.")

        return



    video = update.message.video

    if not video:

        await update.message.reply_text(" Отправь обычное видео, а не файл.")

        return



    await update.message.reply_text(get_random_phrase('start'))



    # Скачивание

    input_path = "input.mp4"

    output_path = "output.mp4"

    file = await context.bot.get_file(video.file_id)

    await file.download_to_drive(input_path)



    await update.message.reply_text(get_random_phrase('processing'))



    try:

        clip = VideoFileClip(input_path).resize(height=360)



        # Применяем битрейт, если есть

        user_bitrate = get_bitrate(user_quality.get(user_id, "original"))



        clip.write_videofile(

            output_path,

            codec="libx264",

            audio_codec="aac",

            bitrate=user_bitrate,

            threads=4,

            logger=None

        )



        with open(output_path, "rb") as f:

            await context.bot.send_video_note(chat_id=update.message.chat_id, video_note=f)



        await update.message.reply_text(get_random_phrase('done'))



        # Статистика

        user_stats[user_id] = user_stats.get(user_id, 0) + 1



    except Exception as e:

        await update.message.reply_text(f" Ошибка: {e}")



    finally:

        for file in [input_path, output_path]:

            if os.path.exists(file): os.remove(file)



# === Текстовые сообщения ===

async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):

    if is_authorized(update.effective_user.id):

        await update.message.reply_text(" Просто отправь видео, и я сделаю кружок!")

    else:

        await update.message.reply_text(" У тебя нет доступа.")



# === Запуск ===

def main():

    app = ApplicationBuilder().token(BOT_TOKEN).build()



    app.add_handler(CommandHandler("help", help_cmd))

    app.add_handler(CommandHandler("setquality", set_quality))

    app.add_handler(CommandHandler("status", status))

    app.add_handler(MessageHandler(filters.VIDEO, handle_video))

    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))



    print(" Бот запущен!")

    app.run_polling()



if __name__ == "__main__":

    main()