Как извлечь контент из закрытых Telegram-каналов через Browser Automation

MTProto бессилен, когда владелец включил запрет копирования. Playwright + Neet-Nestor + Telegram WebK — единственный путь. Полное руководство с кодом.

Проблема: noforwards и ограничения MTProto

Ты подписан на закрытый Telegram-канал с ценным обучающим контентом. Платная подписка, куча видео, PDF-гайды, аудио-уроки. Ты хочешь загрузить всё это в свою RAG-систему, чтобы AI-агент мог искать по этим материалам. Логично? Логично.

Открываешь Telethon, пишешь client.download_media(message)... и получаешь ошибку. Владелец канала включил noforwards=True — запрет пересылки и копирования. MTProto API честно соблюдает это ограничение: метод messages.forwardMessages возвращает ошибку, download_media() тоже не работает для защищённого контента.

Но Telegram Web всё равно отображает этот контент. Видео воспроизводится, PDF рендерится, аудио играет. Данные физически приходят в браузер. И вот тут начинается интересное: если данные уже в браузере — их можно перехватить.

Browser automation — единственный рабочий подход для restricted-каналов. Не хак, не эксплойт, а работа с тем, что уже отображается на вашем экране. Как делать конспект с лекции — только автоматизированно.

WebK vs WebZ: какой Telegram Web автоматизировать

У Telegram два веб-клиента: WebK (web.telegram.org/k/) и WebZ (web.telegram.org/a/). Оба полнофункциональные, оба отображают restricted-контент. Но для автоматизации разница есть:

Ключевые DOM-селекторы (WebK)

Telegram Web использует виртуальный скролл — в DOM находятся только видимые сообщения плюс небольшой буфер. Чтобы загрузить историю, нужно прокручивать вверх и собирать данные порциями.

Авторизация через Playwright: QR и session persistence

Первый шаг — авторизоваться в Telegram Web через Playwright. Есть два способа:

Способ 1: QR-код (рекомендуется)

Открываете web.telegram.org/k/ в headful-режиме Playwright, видите QR-код, сканируете с телефона. Просто и надёжно. После авторизации сохраняете session.

from playwright.async_api import async_playwright

async def authorize():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080}
        )
        page = await context.new_page()
        await page.goto("https://web.telegram.org/k/")

        # Ждём пока пользователь отсканирует QR
        print("Отсканируйте QR-код с телефона...")
        await page.wait_for_selector('.bubbles-inner', timeout=120000)

        # Сохраняем сессию
        await context.storage_state(path="telegram_session.json")
        print("Сессия сохранена!")
        await browser.close()

Session Persistence: не авторизуемся каждый раз

Это критически важный момент. Сессия Telegram Web живёт ~30 дней. Сохраняйте storage_state после каждого сеанса и загружайте при следующем:

# Загрузка сохранённой сессии
context = await browser.new_context(
    storage_state="telegram_session.json",
    viewport={"width": 1920, "height": 1080}
)

# После работы — сохраняем обновлённую сессию
await context.storage_state(path="telegram_session.json")

Извлечение текста: скроллинг и DOM-парсинг

Базовая задача — собрать все текстовые посты из канала. Telegram Web подгружает сообщения по мере скроллинга, поэтому нужно прокручивать вверх и собирать данные порциями.

async def extract_text_posts(page, max_posts=500):
    """Извлечь текстовые посты через скроллинг"""
    all_messages = {}
    no_new_count = 0

    for i in range(max_posts // 10):
        # Извлекаем видимые сообщения
        messages = await page.evaluate('''
            Array.from(document.querySelectorAll('.bubble'))
            .map(el => ({
                id: el.dataset.mid,
                text: el.querySelector('.message')?.textContent || '',
                date: el.querySelector('.time')?.textContent || ''
            }))
            .filter(m => m.id && m.text)
        ''')

        prev_count = len(all_messages)
        for msg in messages:
            all_messages[msg['id']] = msg

        # Проверяем, появились ли новые
        if len(all_messages) == prev_count:
            no_new_count += 1
            if no_new_count > 5:
                break  # Дошли до начала канала
        else:
            no_new_count = 0

        # Скроллим вверх
        await page.evaluate(
            'document.querySelector(".bubbles-inner").scrollTop = 0'
        )
        await page.wait_for_timeout(1500)

    return list(all_messages.values())

Извлечение комментариев

Если у постов есть комментарии, их можно получить, кликнув на «Comments» под постом:

async def extract_comments(page, message_id):
    """Извлечь комментарии под конкретным постом"""
    selector = f'.bubble[data-mid="{message_id}"] .replies-footer'
    reply_btn = await page.query_selector(selector)
    if not reply_btn:
        return []

    await reply_btn.click()
    await page.wait_for_timeout(2000)

    comments = await page.evaluate('''
        Array.from(document.querySelectorAll('.bubbles-inner .bubble'))
        .map(el => ({
            author: el.querySelector('.peer-title')?.textContent || '',
            text: el.querySelector('.message')?.textContent || '',
            date: el.querySelector('.time')?.textContent || ''
        }))
        .filter(c => c.text)
    ''')

    return comments

Извлечение видео: Neet-Nestor и network intercept

Видео — самый ценный и самый сложный для извлечения формат. Telegram Web использует streaming через Service Worker и хранит медиа в IndexedDB/Cache API. Прямых URL нет — данные идут через blob: URLs.

Метод 1: Neet-Nestor Telegram Media Downloader (⭐ приоритет)

Neet-Nestor Telegram Media Downloader — userscript с 3500+ звёздами на GitHub. Добавляет кнопку скачивания даже в restricted-каналах. Поддерживает фото, видео, GIF, аудио.

Принцип работы: скрипт перехватывает blob URL из внутреннего состояния Telegram Web. Браузер всё равно скачивает видео для воспроизведения — скрипт просто забирает этот blob и создаёт ссылку на скачивание.

Для автоматизации через Playwright — инжектируем скрипт при загрузке страницы:

# Скачиваем скрипт заранее
# https://greasyfork.org/en/scripts/446342

with open('telegram_media_downloader.js', 'r') as f:
    script = f.read()

# Инжектируем при каждой загрузке страницы
await page.add_init_script(script)

# Или после загрузки конкретной страницы:
await page.evaluate(script)

Метод 2: Перехват Network Requests

Playwright позволяет перехватывать все сетевые запросы. Когда видео начинает воспроизводиться, данные идут через Service Worker — их можно перехватить:

async def intercept_video(route, request):
    if 'video' in request.resource_type or '.mp4' in request.url:
        response = await route.fetch()
        body = await response.body()
        filename = f'video_{hash(request.url)}.mp4'
        with open(filename, 'wb') as f:
            f.write(body)
        print(f"Сохранено: {filename} ({len(body)} bytes)")
    await route.continue_()

await page.route('**/*', intercept_video)

Метод 3: Screen Recording (гарантированный fallback)

Если ничего не работает, Playwright умеет записывать видео страницы. Качество ниже, но работает всегда:

# Playwright записывает видео страницы
context = await browser.new_context(
    record_video_dir="./videos/",
    record_video_size={"width": 1280, "height": 720}
)
page = await context.new_page()

# Навигация к видео, запуск воспроизведения...
await page.click('.media-video')
# Ждём окончания
await page.wait_for_timeout(video_duration_ms)

await context.close()  # Видео сохранится автоматически

Для транскрибации достаточно 480p — главное чистый звук, а не картинка.

Извлечение PDF: перехват blob и fallback через OCR

Telegram Web скачивает PDF целиком в браузер и рендерит через встроенный viewer. Файл хранится в памяти как ArrayBuffer/Blob. Neet-Nestor поддерживает скачивание документов из restricted-каналов.

Метод 1: Network Intercept для PDF

pdfs = []

async def intercept_pdf(route, request):
    response = await route.fetch()
    content_type = response.headers.get('content-type', '')
    if 'pdf' in content_type or request.url.endswith('.pdf'):
        body = await response.body()
        filename = f'document_{len(pdfs)}.pdf'
        pdfs.append(filename)
        with open(filename, 'wb') as f:
            f.write(body)
        print(f"PDF сохранён: {filename}")
    await route.continue_()

await page.route('**/*', intercept_pdf)

# Кликаем на PDF для открытия preview
await page.click('.document')

Метод 2: Screenshot + OCR (fallback)

Если PDF невозможно перехватить, скриншотим каждую страницу и распознаём текст:

import pytesseract
from PIL import Image

async def screenshot_pdf_pages(page):
    pages_text = []
    page_num = 1
    while True:
        screenshot = await page.screenshot(
            clip={"x": 100, "y": 50, "width": 800, "height": 1100}
        )
        # Сохраняем скриншот
        img_path = f'pdf_page_{page_num}.png'
        with open(img_path, 'wb') as f:
            f.write(screenshot)

        # OCR
        text = pytesseract.image_to_string(
            Image.open(img_path), lang='rus+eng'
        )
        pages_text.append(text)

        # Следующая страница
        next_btn = await page.query_selector('.pdf-next-page')
        if not next_btn:
            break
        await next_btn.click()
        await page.wait_for_timeout(500)
        page_num += 1

    return '\n\n---\n\n'.join(pages_text)

Извлечение аудио и транскрибация через Whisper

Аудиофайлы извлекаются аналогично видео: Neet-Nestor, network intercept или Web Audio API capture.

async def intercept_audio(route, request):
    if (request.resource_type == 'media'
        or '.ogg' in request.url
        or '.mp3' in request.url):
        response = await route.fetch()
        body = await response.body()
        with open(f'audio_{hash(request.url)}.ogg', 'wb') as f:
            f.write(body)
    await route.continue_()

await page.route('**/*', intercept_audio)

Транскрибация через Whisper

После извлечения аудио/видео — транскрибация. Два варианта:

# Локальный Whisper (бесплатно, но медленно)
# pip install openai-whisper
whisper audio.ogg --model large-v3 --language ru --output_format txt

# Через API (быстро, $0.006/мин)
import openai
client = openai.OpenAI()
with open("audio.ogg", "rb") as f:
    transcript = client.audio.transcriptions.create(
        model="whisper-1",
        file=f,
        language="ru"
    )
print(transcript.text)

Anti-detection: stealth, rate limits, human-like behavior

Telegram Web может обнаружить автоматизацию. Вот как минимизировать риски:

Stealth-настройки Playwright

from playwright.async_api import async_playwright
from playwright_stealth import stealth_async  # pip install playwright-stealth

browser = await p.chromium.launch(headless=False)  # Headful безопаснее!
context = await browser.new_context(
    viewport={"width": 1920, "height": 1080},
    user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
               "AppleWebKit/537.36 (KHTML, like Gecko) "
               "Chrome/132.0.0.0 Safari/537.36",
    locale="ru-RU",
    timezone_id="Europe/Moscow"
)
page = await context.new_page()
await stealth_async(page)

Stealth-патчи маскируют: navigator.webdriver = false, правильные navigator.plugins, Chrome runtime properties, WebGL vendor, canvas fingerprint.

Human-like поведение

import random, asyncio

async def human_scroll(page):
    """Прокрутка с человеческой задержкой"""
    await page.mouse.wheel(0, random.randint(300, 800))
    await asyncio.sleep(random.uniform(1.5, 4.0))

async def human_click(page, selector):
    """Клик с небольшим смещением"""
    el = await page.query_selector(selector)
    box = await el.bounding_box()
    x = box['x'] + random.uniform(5, box['width'] - 5)
    y = box['y'] + random.uniform(5, box['height'] - 5)
    await page.mouse.move(x, y, steps=random.randint(5, 15))
    await asyncio.sleep(random.uniform(0.1, 0.3))
    await page.mouse.click(x, y)

Безопасные rate limits

RAG Pipeline: от сырого контента к knowledge base

Извлечённый контент бесполезен в сыром виде. Цель — построить knowledge base, по которой AI-агент может искать ответы.

Архитектура pipeline

Telegram Web (Playwright)
    ├── Текст постов → Markdown
    ├── Комментарии → Markdown
    ├── Видео → Whisper → Транскрипты (.md)
    ├── Аудио → Whisper → Транскрипты (.md)
    └── PDF → Скачивание / OCR → Текст (.md)
         │
         ▼
    Markdown файлы (первичное хранилище)
         │
         ▼
    Chunking (500-1000 токенов, overlap 100)
         │
         ▼
    Embeddings (text-embedding-3-small / Jina v3)
         │
         ▼
    Vector DB (ChromaDB / SQLite + sqlite-vss)
         │
         ▼
    RAG Query API → AI-агенты

Структура хранения

knowledge-base/
├── channel-name/
│   ├── posts/
│   │   ├── 2024-01-15_post_123.md
│   │   └── 2024-01-16_post_124.md
│   ├── transcripts/
│   │   ├── 2024-01-20_video_stream.md
│   │   └── 2024-01-25_audio_lesson.md
│   └── pdfs/
│       ├── guide.pdf
│       └── guide.md  (извлечённый текст)
├── chroma_db/  (vector index)
└── metadata.json

Chunking и embedding

from langchain.text_splitter import RecursiveCharacterTextSplitter
import chromadb

# Chunking
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)
chunks = splitter.split_text(document_text)

# Embedding + индексация
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("knowledge")

collection.add(
    documents=chunks,
    metadatas=[{
        "source": "channel-name",
        "type": "post",  # post / video_transcript / pdf
        "date": "2024-01-15",
        "message_id": "123"
    } for _ in chunks],
    ids=[f"chunk_{i}" for i in range(len(chunks))]
)

# Поиск
results = collection.query(
    query_texts=["Стратегии продвижения YouTube"],
    n_results=10,
    where={"source": "channel-name"}
)

Юридическая и этическая сторона

Важный вопрос: законно ли это? Короткий ответ: для личного использования — да.

Пошаговый план запуска

Фаза 1: Настройка (день 1-2)

• ☐ Установить зависимости: pip install playwright playwright-stealth chromadb
• ☐ playwright install chromium
• ☐ Авторизовать Telegram Web через QR
• ☐ Сохранить telegram_session.json
• ☐ Протестировать Neet-Nestor на одном видео

Фаза 2: Извлечение текста (день 2-3)

• ☐ Скрапить все текстовые посты + комментарии
• ☐ Скачать PDF через network intercept
• ☐ Извлечь текст из PDF (PyPDF2 / pdfplumber)
• ☐ Сохранить всё в Markdown

Фаза 3: Медиа (день 3-5)

• ☐ Скачать аудиофайлы (Neet-Nestor / intercept)
• ☐ Скачать видео
• ☐ Транскрибировать через Whisper

Фаза 4: RAG Pipeline (день 5-7)

• ☐ Chunking всех текстов
• ☐ Embedding через text-embedding-3-small
• ☐ Индексация в ChromaDB
• ☐ API для AI-агентов

Фаза 5: Автообновление (ongoing)

• ☐ Cron: раз в день проверять новые посты
• ☐ Инкрементальная загрузка
• ☐ Автоматическая транскрибация и индексация

Существующие инструменты

Не обязательно всё писать с нуля. Вот проверенные решения:

• Neet-Nestor Media Downloader — userscript, 3500+ stars, скачивание медиа из restricted-каналов
• rcdtool — CLI для скачивания restricted-контента через Telethon session
• Telegram Media Downloader (Improved) — улучшенная версия Neet-Nestor
• TeleCapt — Chrome-расширение для экспорта чатов в PDF с auto-scroll

Заключение

Browser automation — не хак и не взлом. Это автоматизация того, что вы и так видите на экране. Telegram Web загружает контент в ваш браузер — вы его сохраняете. Как конспект лекции, только быстрее.

Ключевые выводы:

• WebK (/k/) — лучший вариант для автоматизации, стабильные селекторы
• Neet-Nestor — решает 90% задач с видео и аудио из restricted-каналов
• Playwright + stealth — полный контроль над процессом
• Session persistence — авторизуемся один раз, работаем 30 дней
• Human-like behavior — рандомные паузы, рабочие часы, headful mode
• RAG pipeline — конечная цель: от сырого контента к searchable knowledge base