Как настроить Reko для автоматической обработки YouTube-плейлистов

Проблема: ручная обработка плейлистов убивает время

Если вы работаете с образовательным контентом, анализируете YouTube-каналы или создаете материалы на основе видео, вы знаете, насколько мучительна ручная обработка плейлистов. Скачать десятки видео, расшифровать их, структурировать информацию — на это уходят часы, если не дни. Особенно когда нужно обрабатывать регулярно обновляемые плейлисты.

Решение: Reko как локальный автоматизированный пайплайн

Reko — это open-source инструмент для автоматической обработки видео, который можно развернуть локально. Его ключевые преимущества:

• Полная локальность: все данные остаются на вашем компьютере
• Поддержка локальных моделей: транскрибация через Whisper, Falcon или другие модели
• Автоматизация: обработка целых плейлистов одной командой
• Гибкость: можно настроить под свои нужды

В этой статье я покажу, как настроить Reko для автоматической обработки YouTube-плейлистов от начала до конца — от установки до получения готовых транскриптов.

1Подготовка окружения

Перед началом убедитесь, что у вас установлены:

• Python 3.8+
• FFmpeg (обязательно для работы с видео)
• Минимум 8 ГБ оперативной памяти (лучше 16+ для локальных моделей)
• Достаточно места на диске (каждое видео занимает место)

Установите FFmpeg:

# Для Ubuntu/Debian
sudo apt update && sudo apt install ffmpeg

# Для macOS
brew install ffmpeg

# Для Windows (через Chocolatey)
choco install ffmpeg

2Установка Reko и зависимостей

Клонируйте репозиторий Reko и установите зависимости:

git clone https://github.com/reko-project/reko.git
cd reko
pip install -r requirements.txt

Для работы с YouTube потребуется дополнительная установка:

pip install yt-dlp
pip install whisper-openai-api

3Базовая настройка конфигурации

Создайте файл конфигурации config.yaml в корне проекта:

# config.yaml
storage:
  download_path: "./downloads"
  transcripts_path: "./transcripts"

processing:
  max_video_length: 3600  # максимальная длина видео в секундах (1 час)
  language: "ru"         # язык для транскрибации (ru, en, auto)
  model_size: "medium"   # размер модели Whisper (tiny, base, small, medium, large)

output:
  format: "txt"          # формат вывода (txt, srt, json)
  include_timestamps: true
  separate_speakers: false

youtube:
  quality: "best"        # качество видео для скачивания
  extract_audio: true    # извлекать только аудио (экономит место)
  playlist_end: null     # номер последнего видео в плейлисте (null = все)

4Создание скрипта для автоматической обработки

Теперь создадим основной скрипт обработки. Этот скрипт будет:

1. Получать список видео из плейлиста
2. Скачивать их (или только аудио)
3. Запускать транскрибацию через Whisper
4. Сохранять результаты в структурированном виде

Создайте файл process_playlist.py:

#!/usr/bin/env python3
"""
Автоматическая обработка YouTube-плейлистов через Reko
"""

import os
import sys
import yaml
import subprocess
from pathlib import Path
import yt_dlp as youtube_dlp
import whisper
from datetime import datetime

class PlaylistProcessor:
    def __init__(self, config_path="config.yaml"):
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = yaml.safe_load(f)
        
        # Создаем директории
        Path(self.config['storage']['download_path']).mkdir(parents=True, exist_ok=True)
        Path(self.config['storage']['transcripts_path']).mkdir(parents=True, exist_ok=True)
        
        # Загружаем модель Whisper
        print(f"Загрузка модели Whisper {self.config['processing']['model_size']}...")
        self.model = whisper.load_model(self.config['processing']['model_size'])
        
    def get_playlist_videos(self, playlist_url):
        """Получаем информацию о видео в плейлисте"""
        ydl_opts = {
            'quiet': True,
            'extract_flat': True,
            'force_generic_extractor': False,
        }
        
        with youtube_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(playlist_url, download=False)
            
            if 'entries' in info:
                videos = []
                for entry in info['entries']:
                    if entry:
                        videos.append({
                            'id': entry.get('id'),
                            'title': entry.get('title'),
                            'url': entry.get('url') or f"https://youtube.com/watch?v={entry.get('id')}",
                            'duration': entry.get('duration'),
                        })
                return videos
            return []
    
    def download_video(self, video_info):
        """Скачиваем видео (или только аудио)"""
        video_id = video_info['id']
        output_path = os.path.join(
            self.config['storage']['download_path'],
            f"{video_id}.{'mp3' if self.config['youtube']['extract_audio'] else 'mp4'}"
        )
        
        # Если файл уже существует, пропускаем скачивание
        if os.path.exists(output_path):
            print(f"Файл уже существует: {output_path}")
            return output_path
        
        ydl_opts = {
            'format': 'bestaudio/best' if self.config['youtube']['extract_audio'] else 'best',
            'outtmpl': output_path.replace('.mp3', '').replace('.mp4', ''),
            'quiet': False,
        }
        
        try:
            with youtube_dlp.YoutubeDL(ydl_opts) as ydl:
                ydl.download([video_info['url']])
            print(f"Скачано: {video_info['title']}")
            return output_path
        except Exception as e:
            print(f"Ошибка при скачивании {video_info['title']}: {e}")
            return None
    
    def transcribe_audio(self, audio_path):
        """Транскрибируем аудио через Whisper"""
        if not audio_path or not os.path.exists(audio_path):
            return None
            
        try:
            result = self.model.transcribe(
                audio_path,
                language=self.config['processing']['language'],
                fp16=False  # Для CPU лучше отключить
            )
            return result['text']
        except Exception as e:
            print(f"Ошибка транскрибации {audio_path}: {e}")
            return None
    
    def save_transcript(self, video_info, transcript_text):
        """Сохраняем транскрипт в файл"""
        if not transcript_text:
            return
            
        # Очищаем название файла от недопустимых символов
        safe_title = "".join(c for c in video_info['title'] if c.isalnum() or c in (' ', '-', '_')).rstrip()
        filename = f"{video_info['id']}_{safe_title[:50]}.{self.config['output']['format']}"
        output_file = os.path.join(self.config['storage']['transcripts_path'], filename)
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(f"# {video_info['title']}\n")
            f.write(f"# URL: {video_info['url']}\n")
            f.write(f"# Дата обработки: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
            f.write(transcript_text)
        
        print(f"Транскрипт сохранен: {output_file}")
        return output_file
    
    def process_playlist(self, playlist_url, limit=None):
        """Основной метод обработки плейлиста"""
        print(f"Получение информации о плейлисте: {playlist_url}")
        videos = self.get_playlist_videos(playlist_url)
        
        if not videos:
            print("Не удалось получить видео из плейлиста")
            return
        
        print(f"Найдено видео: {len(videos)}")
        
        # Ограничиваем количество видео, если указано
        if limit:
            videos = videos[:limit]
        
        processed_count = 0
        for i, video in enumerate(videos, 1):
            print(f"\nОбработка видео {i}/{len(videos)}: {video['title']}")
            
            # Проверяем длину видео
            if video.get('duration') and video['duration'] > self.config['processing']['max_video_length']:
                print(f"Пропускаем (длинее {self.config['processing']['max_video_length']} сек): {video['title']}")
                continue
            
            # Скачиваем
            audio_path = self.download_video(video)
            if not audio_path:
                continue
            
            # Транскрибируем
            transcript = self.transcribe_audio(audio_path)
            
            # Сохраняем
            if transcript:
                self.save_transcript(video, transcript)
                processed_count += 1
        
        print(f"\nГотово! Обработано видео: {processed_count}/{len(videos)}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Использование: python process_playlist.py  [лимит]")
        sys.exit(1)
    
    playlist_url = sys.argv[1]
    limit = int(sys.argv[2]) if len(sys.argv) > 2 else None
    
    processor = PlaylistProcessor()
    processor.process_playlist(playlist_url, limit)

5Запуск и использование

Теперь можно запустить обработку плейлиста:

# Обработать весь плейлист
python process_playlist.py "https://www.youtube.com/playlist?list=PL1234567890"

# Обработать только первые 5 видео
python process_playlist.py "https://www.youtube.com/playlist?list=PL1234567890" 5

Скрипт автоматически:

1. Скачает все видео из плейлиста (или только аудио)
2. Пропустит слишком длинные видео (настраивается в конфиге)
3. Транскрибирует через локальную модель Whisper
4. Сохранит результаты в папку transcripts

6Расширенная настройка: постобработка транскриптов

Сырые транскрипты можно улучшить. Добавим постобработку:

# postprocess.py
import re
from typing import List

class TranscriptProcessor:
    @staticmethod
    def clean_transcript(text: str) -> str:
        """Очистка транскрипта от артефактов"""
        # Убираем повторяющиеся пробелы
        text = re.sub(r'\s+', ' ', text)
        
        # Восстанавливаем пунктуацию в конце предложений
        text = re.sub(r'([а-яА-Яa-zA-Z])(\s+[А-ЯA-Z])', lambda m: m.group(1) + '. ' + m.group(2), text)
        
        # Убираем типичные артефакты ASR
        artifacts = [
            '\ufeff', '\u200b', '\u200e', '\u200f',
            '[Music]', '[Applause]', '[Laughter]'
        ]
        for artifact in artifacts:
            text = text.replace(artifact, '')
        
        return text.strip()
    
    @staticmethod
    def split_into_chunks(text: str, max_chunk_size: int = 2000) -> List[str]:
        """Разбиваем длинный текст на чанки для дальнейшей обработки"""
        sentences = re.split(r'([.!?]\\s+)', text)
        chunks = []
        current_chunk = ""
        
        for i in range(0, len(sentences), 2):
            sentence = sentences[i] + (sentences[i+1] if i+1 < len(sentences) else '')
            if len(current_chunk) + len(sentence) <= max_chunk_size:
                current_chunk += sentence
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks

# Использование в основном скрипте
processor = TranscriptProcessor()
cleaned_text = processor.clean_transcript(raw_transcript)
chunks = processor.split_into_chunks(cleaned_text)

Нюансы и возможные проблемы

Интеграция с другими инструментами

Обработанные транскрипты можно использовать дальше:

• Семантический поиск: используйте семантический пайплайн для LLM для индексации транскриптов
• Анализ контента: подавайте транскрипты в локальные LLM для суммаризации или анализа
• Создание контента: используйте транскрипты для написания статей на основе видео
• Обучение моделей: соберите датасет для тренировки собственных моделей

Автоматизация через планировщик

Для регулярной обработки можно настроить cron (Linux/macOS) или планировщик задач (Windows):

# Ежедневная обработка нового контента в плейлисте
# Добавьте в crontab -e
0 2 * * * cd /path/to/reko && python process_playlist.py "https://youtube.com/playlist?list=YOUR_PLAYLIST_ID" >> /var/log/reko.log 2>&1

FAQ: Частые вопросы

Можно ли обрабатывать приватные плейлисты?

Да, если у вас есть доступ к плейлисту. yt-dlp поддерживает cookies — можно экспортировать cookies из браузера и использовать их.

Как ускорить обработку?

1. Используйте GPU для Whisper
2. Обрабатывайте несколько видео параллельно
3. Используйте модель меньшего размера
4. Скачивайте только аудио (экономит время и место)

Можно ли использовать другие модели для транскрибации?

Да, Whisper — не единственный вариант. Можно интегрировать другие ASR-системы или использовать облачные API, но тогда потеряете локальность.

Как обрабатывать очень длинные видео?

Whisper сам разбивает длинные аудио на сегменты. Но если видео длиннее 1-2 часов, лучше увеличить лимит в конфиге или обрабатывать частями.

Заключение

Настройка Reko для автоматической обработки YouTube-плейлистов — это мощный инструмент для любого, кто работает с видео-контентом. Вы получаете:

• Полную автономность (все данные локально)
• Масштабируемость (от одного видео до сотен)
• Гибкость (можно адаптировать под любые нужды)
• Бесплатность (только ваше железо)

Этот подход особенно полезен для создания датасетов, анализа конкурентов, подготовки материалов для обучения моделей или просто для архивации контента с возможностью быстрого поиска.