Бот для скачивания рилс и TikTok в Telegram на Python

Как сделать Telegram-бота для скачивания рилс Instagram и видео TikTok?#

Нужен бот на Python и aiogram 3. Он находит ссылку Instagram или TikTok в сообщении, запрашивает у метода download API файлы, скачивает их и загружает в чат. Скрипт ниже занимает около 120 строк и тратит один запрос API на одну ссылку. StoriesFly — наш API, поэтому используем его; аккаунт Instagram или TikTok, логин и cookies не нужны.

Бот работает с публичными рилс, постами и ссылками IGTV, а также с видео TikTok (ссылки www, m, а также короткие vm и vt). Видео TikTok приходят без водяного знака. Одно видео или фото бот присылает как есть, карусель — альбомом до 10 элементов.

Что понадобится#

• Python 3.10 или новее (в коде используется запись bytes | None).
• Токен Telegram-бота.
• Ключ StoriesFly API.

Токен бота. Откройте в Telegram @BotFather, отправьте /newbot, выберите имя и username, заканчивающийся на bot. BotFather пришлёт токен вида 123456:ABC.... Не публикуйте его.

Ключ API. Войдите через Telegram на /developers/dashboard и создайте ключ вида sf_live_.... Новый аккаунт получает 100 бесплатных запросов (один раз на Telegram-аккаунт), этого хватит, чтобы проверить бота. Документация на английском: /developers/docs.

Установка и запуск.

pip install "aiogram>=3,<4" aiohttp
export BOT_TOKEN="123456:ABC..."
export STORIESFLY_KEY="sf_live_..."
python reels_bot.py

Откройте бота в Telegram, нажмите /start и отправьте ссылку, например https://www.instagram.com/reel/XXXXXXXXXXX/ или https://www.tiktok.com/@user/video/123....

Как устроен бот#

Поиск ссылки#

LINK_RE = re.compile(
    r"https?://(?:www\.)?instagram\.com/(?:p|reels?|tv)/[\w-]+"
    r"|https?://(?:www\.|m\.|vm\.|vt\.)?tiktok\.com/[^\s]+"
)

Первая половина шаблона принимает /p/ (посты), /reel/ и /reels/ (рилс) и /tv/ (IGTV). Она обрывается на коде поста, поэтому ссылка с трекинг-параметрами вроде ?igsh=... тоже подходит, а параметры отбрасываются. Вторая половина принимает ссылки TikTok на хостах www, m, vm и vt и берёт всё до ближайшего пробела. Обработчик использует F.text.regexp(LINK_RE): сработает любое сообщение со ссылкой, даже если вокруг есть текст. Остальные сообщения попадают в последний обработчик, который просит прислать ссылку.

Запрос /v1/download#

async with session.get(
    f"{API}/download", params={"url": url}, headers={"X-Api-Key": KEY}
) as r:

GET /v1/download принимает одну ссылку url и работает для обеих платформ. Метод отдаёт только файлы, без метрик. Поля ответа:

Ответ выглядит так (значения условные, сокращено):

{
  "platform": "tiktok",
  "id": "7300000000000000000",
  "url": "https://www.tiktok.com/@user/video/7300000000000000000",
  "author": "sample_user",
  "caption": "Sample caption",
  "media_type": "video",
  "thumbnail_url": "https://...",
  "media": [
    { "type": "video", "url": "https://...", "quality": "hd", "thumbnail_url": "https://..." }
  ],
  "audio_url": "https://..."
}

Для видео TikTok первый элемент media — лучшее качество без водяного знака. Поэтому, когда media_type равен video, бот берёт только первый элемент:

files = post["media"]
if post["media_type"] == "video":
    files = files[:1]  # the first entry is the best quality

Подпись читается как (post.get("caption") or "")[:1000] or None, поэтому пост Instagram, где она null, уходит без подписи.

Ошибки приходят в JSON: {"error": "...", "code": "not_found"}. fetch_download при любом статусе, кроме 200, выбрасывает LookupError(code), а обработчик превращает код в сообщение.

Почему бот скачивает файл и загружает его сам#

Ссылки на медиа в ответе через какое-то время перестают работать, поэтому их нужно использовать сразу. Проще всего передать URL в answer_video, и Telegram сам скачает файл. Но это работает только для небольших файлов: лимит на файлы, которые бот отправляет по ссылке, ниже лимита 50 МБ на файлы, которые бот загружает (в документации Bot API для ссылок указаны 5 МБ для фото и 20 МБ для остальных файлов). Видео часто больше.

Поэтому бот скачивает байты через fetch_file и отправляет их как BufferedInputFile. Сначала проверяется Content-Length: если файл больше 50 МБ, бот не загружает его, а присылает ссылку.

Ограничение кода: если сервер не передал Content-Length, проверка размера пройдёт и файл целиком прочитается в память. Рилс и видео TikTok обычно сильно меньше лимита, для небольшого бота этого достаточно.

Карусели как альбомы#

for i, m in enumerate(files[:10]):  # an album holds up to 10

В альбом Telegram помещается до 10 элементов, поэтому список обрезается до 10. Каждый элемент становится InputMediaVideo или InputMediaPhoto в зависимости от m["type"]. Подпись добавляется только к первому элементу: Telegram показывает подпись альбома по первому. Подпись обрезается до 1000 символов, это меньше лимита Telegram для подписи к медиа. Если остался один элемент, бот отправляет его через answer_video или answer_photo, а не альбомом.

Ошибки в понятных сообщениях#

ERRORS = {
    "not_found": "This post does not exist or was deleted.",
    "private_account": "The account is private, so its posts are not available.",
    "bad_request": "That does not look like an Instagram or TikTok post link.",
    ...
    "upstream_unavailable": "Could not get this one right now. Try again in a minute.",
}

Код ошибки API сопоставляется с фразой для пользователя. Для неизвестного кода бот отвечает «Something went wrong, try later.» Чаще всего встречаются такие коды:

На 429 бот не повторяет запрос, а просит пользователя попробовать позже. Если ботом пользуются многие сразу, добавьте небольшую очередь или паузу.

Тексты сообщений в коде на английском; при желании переведите их в словаре ERRORS.

Полный скрипт#

Сохраните как reels_bot.py.

"""Telegram bot: send it an Instagram reel/post or a TikTok link, get the video back.

Usage:
    pip install "aiogram>=3,<4" aiohttp
    BOT_TOKEN=123:abc STORIESFLY_KEY=sf_live_... python reels_bot.py
"""
import asyncio
import os
import re

import aiohttp
from aiogram import Bot, Dispatcher, F
from aiogram.filters import CommandStart
from aiogram.types import (
    BufferedInputFile,
    InputMediaPhoto,
    InputMediaVideo,
    Message,
)

API = "https://storiesfly.com/api/v1"
KEY = os.environ["STORIESFLY_KEY"]
BOT_TOKEN = os.environ["BOT_TOKEN"]

# A bot may upload files up to 50 MB; bigger videos are sent as a link.
MAX_UPLOAD = 50 * 1024 * 1024
LINK_RE = re.compile(
    r"https?://(?:www\.)?instagram\.com/(?:p|reels?|tv)/[\w-]+"
    r"|https?://(?:www\.|m\.|vm\.|vt\.)?tiktok\.com/[^\s]+"
)

ERRORS = {
    "not_found": "This post does not exist or was deleted.",
    "private_account": "The account is private, so its posts are not available.",
    "bad_request": "That does not look like an Instagram or TikTok post link.",
    "rate_limited": "Too many requests at once. Try again in a few seconds.",
    "credits_exhausted": "The bot is out of API requests for now.",
    "upstream_unavailable": "Could not get this one right now. Try again in a minute.",
}

dp = Dispatcher()


async def fetch_download(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(
        f"{API}/download", params={"url": url}, headers={"X-Api-Key": KEY}
    ) as r:
        data = await r.json()
        if r.status != 200:
            raise LookupError(data.get("code", "error"))
        return data


async def fetch_file(session: aiohttp.ClientSession, url: str) -> bytes | None:
    async with session.get(url) as r:
        r.raise_for_status()
        size = int(r.headers.get("Content-Length", 0))
        if size > MAX_UPLOAD:
            return None
        return await r.read()


@dp.message(CommandStart())
async def start(message: Message) -> None:
    await message.answer("Send me a link to an Instagram reel or post, or a TikTok video.")


@dp.message(F.text.regexp(LINK_RE))
async def on_link(message: Message, session: aiohttp.ClientSession) -> None:
    link = LINK_RE.search(message.text).group(0)
    await message.bot.send_chat_action(message.chat.id, "upload_video")

    try:
        post = await fetch_download(session, link)
    except LookupError as e:
        await message.answer(ERRORS.get(str(e), "Something went wrong, try later."))
        return

    files = post["media"]
    if post["media_type"] == "video":
        files = files[:1]  # the first entry is the best quality
    caption = (post.get("caption") or "")[:1000] or None

    items = []
    for i, m in enumerate(files[:10]):  # an album holds up to 10
        data = await fetch_file(session, m["url"])
        if data is None:  # over 50 MB: give the link instead
            await message.answer(f"The video is too big to upload, here is the link:\n{m['url']}")
            continue
        is_video = m["type"] == "video"
        name = f"{post['id'] or 'post'}_{i}.{'mp4' if is_video else 'jpg'}"
        file = BufferedInputFile(data, filename=name)
        cap = caption if not items else None
        items.append(
            InputMediaVideo(media=file, caption=cap)
            if is_video
            else InputMediaPhoto(media=file, caption=cap)
        )

    if len(items) == 1:
        one = items[0]
        if isinstance(one, InputMediaVideo):
            await message.answer_video(one.media, caption=one.caption)
        else:
            await message.answer_photo(one.media, caption=one.caption)
    elif items:
        await message.answer_media_group(items)


@dp.message()
async def other(message: Message) -> None:
    await message.answer("I need a link like https://www.instagram.com/reel/... or https://www.tiktok.com/@user/video/...")


async def main() -> None:
    bot = Bot(BOT_TOKEN)
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120)) as session:
        await dp.start_polling(bot, session=session)


if __name__ == "__main__":
    asyncio.run(main())

Как запустить на сервере#

В режиме polling публичный адрес не нужен, подойдёт любой небольшой VPS. Создайте пользователя и virtualenv, положите скрипт в /opt/reels-bot, а секреты храните в env-файле с правами chmod 600.

# /etc/systemd/system/reels-bot.service
[Unit]
Description=Reels and TikTok Telegram bot
After=network-online.target

[Service]
WorkingDirectory=/opt/reels-bot
EnvironmentFile=/opt/reels-bot/.env
ExecStart=/opt/reels-bot/venv/bin/python reels_bot.py
Restart=always
RestartSec=5
User=reelsbot

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now reels-bot
journalctl -u reels-bot -f

В .env две строки: BOT_TOKEN=... и STORIESFLY_KEY=..., без кавычек. Если нужен Docker, возьмите образ python:3.12-slim, установите оба пакета через pip, передайте те же переменные через --env-file и добавьте --restart unless-stopped.

Сколько это стоит#

Бот делает один запрос API на одну ссылку. Скачивание самого файла запросом API не считается. У download своя фиксированная цена за скачивание, ниже обычного запроса данных, причём скачивание TikTok стоит дешевле, чем Instagram. Пока у вас есть бесплатные запросы, одно скачивание тратит один из 100. Ответы «не найдено» и «закрытый аккаунт» списываются как данные, поэтому ссылка на удалённый пост тоже стоит одного скачивания. Актуальные цены смотрите на /developers/pricing.

Если нужны просмотры, лайки и другие цифры, вызывайте /v1/post для Instagram или /v1/tiktok/video для TikTok. Это обычные запросы данных.

Лимит частоты — 1 запрос в секунду до первого пополнения. В каждом ответе есть заголовки X-Balance-Remaining и X-RateLimit-Remaining: логируйте их, чтобы видеть остаток.

Как добавить сторис#

Для сторис подход тот же. Вызовите GET /v1/user/stories?username=.... В ответе stories[] с полями media_type (video или photo), media_url, taken_at и video_duration. Добавьте команду /stories username, скачайте каждый media_url так же и отправьте альбомом. Подробный разбор — в статье Как скачать сторис Instagram на Python. Для закрытых аккаунтов вернётся private_account.

Важно#

Это работает только с публичными данными. Соблюдайте правила Instagram и TikTok и местное законодательство и отправляйте людям только тот контент, который они вправе сохранять: свой или с разрешения автора.