""" YouTube Video Downloader Service Отдельный микросервис для скачивания видео с YouTube """ import os import logging import time from pathlib import Path from flask import Flask, request, jsonify from flask_cors import CORS import yt_dlp import uuid import re # Настройка логирования logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) app = Flask(__name__) CORS(app) # Разрешаем CORS для взаимодействия с основным ботом # Директория для временных файлов DOWNLOADS_DIR = Path('downloads') DOWNLOADS_DIR.mkdir(exist_ok=True) def _safe_filename(title: str) -> str: """Создает безопасное имя файла""" safe_title = re.sub(r'[<>:"/\\|?*]', '', title)[:100] return str(DOWNLOADS_DIR / f'{uuid.uuid4()}_{safe_title}.%(ext)s') def _is_valid_cookies_file(cookies_path: Path) -> bool: """Проверяет, что файл cookies существует и содержит валидные YouTube-куки""" if not cookies_path.exists(): return False try: with open(cookies_path, 'r', encoding='utf-8', errors='ignore') as f: content = f.read() # Простая проверка: есть ли YouTube-куки в файле has_youtube_domain = '.youtube.com' in content or 'youtube.com' in content if not has_youtube_domain: logger.warning("YouTube: в файле cookies нет YouTube-куков") return False # Проверяем строки с данными (не комментарии) lines = [line.strip() for line in content.split('\n') if line.strip() and not line.strip().startswith('#')] if len(lines) == 0: return False # Проверяем наличие YouTube-куков и важных параметров youtube_cookies_count = 0 important_cookies_count = 0 current_time = int(time.time()) for line in lines: # Формат Netscape cookie: domain, flag, path, secure, expiration, name, value # Разделяем по табуляции, но учитываем что значения могут содержать табы parts = line.split('\t') if len(parts) >= 7: # Должно быть минимум 7 полей domain = parts[0].strip() expiration = parts[4].strip() cookie_name = parts[5].strip() if len(parts) > 5 else '' # Проверяем домен if '.youtube.com' in domain or domain == 'youtube.com': youtube_cookies_count += 1 # Проверяем срок действия (0 означает сессионный куки, это нормально) is_valid = True if expiration != '0': try: exp_time = int(expiration) if exp_time < current_time: is_valid = False logger.debug(f"YouTube: просроченный куки {cookie_name} (истек: {exp_time})") except (ValueError, OverflowError): pass # Проверяем наличие важных куков if is_valid and any(important in cookie_name for important in ['VISITOR_INFO1_LIVE', '__Secure-3PSID', 'PREF', '__Secure-YNID', 'YSC']): important_cookies_count += 1 if youtube_cookies_count > 0: if important_cookies_count > 0: logger.debug(f"YouTube: найдены валидные куки ({youtube_cookies_count} YouTube-куков, {important_cookies_count} важных)") else: logger.warning(f"YouTube: найдены YouTube-куки ({youtube_cookies_count}), но нет важных параметров") return True else: logger.warning("YouTube: в файле cookies нет валидных YouTube-куков") return False except Exception as e: logger.warning(f"Ошибка при проверке файла cookies: {e}") return False def download_youtube_video(url: str, max_retries: int = 3) -> Path: """ Скачивает видео с YouTube с умной стратегией использования куков: 1. Сначала пробуем БЕЗ куков (работает в большинстве случаев) 2. Если нужны куки (18+, приватные, проверка на бота) - пробуем файл куков 3. Если файл не работает - пробуем автоматически из браузера (если доступен) """ cookies_file = os.getenv('YOUTUBE_COOKIES_FILE', 'youtube_cookies.txt') cookies_file_path = Path(cookies_file) # Проверяем наличие валидного файла куков (но не используем сразу) cookies_file_valid = _is_valid_cookies_file(cookies_file_path) user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' # Определяем, это Shorts или обычное видео is_shorts = '/shorts/' in url player_clients = ['android', 'ios', 'web'] if is_shorts else ['android', 'web'] def create_ydl_opts(use_cookies_from_file: bool = False, use_browser_cookies: str = None): """Создает опции для yt-dlp""" opts = { 'quiet': False, 'no_warnings': False, 'user_agent': user_agent, 'socket_timeout': 60, 'extractor_args': { 'youtube': { 'player_client': player_clients, 'player_skip': ['webpage'], }, }, 'http_headers': { 'User-Agent': user_agent, 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'en-us,en;q=0.5', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', }, } if use_browser_cookies: # Автоматически получаем куки из браузера (не нужен файл) opts['cookiesfrombrowser'] = (use_browser_cookies,) logger.info(f"YouTube: используем автоматические куки из браузера {use_browser_cookies}") elif use_cookies_from_file and cookies_file_valid: opts['cookiefile'] = str(cookies_file_path.absolute()) logger.info(f"YouTube: используем куки из файла {cookies_file_path.absolute()}") else: logger.info(f"YouTube: работаем БЕЗ куков (в большинстве случаев это работает)") return opts # Стратегии попыток: без куков -> файл куков -> браузер куки strategies = [ {'name': 'без куков', 'opts': create_ydl_opts(use_cookies_from_file=False)}, ] if cookies_file_valid: strategies.append({ 'name': 'куки из файла', 'opts': create_ydl_opts(use_cookies_from_file=True) }) # Добавляем стратегии с браузерами (в Docker обычно нет браузеров, но пробуем) # yt-dlp автоматически обработает ошибку, если браузер недоступен for browser in ['firefox', 'chrome', 'chromium']: strategies.append({ 'name': f'куки из браузера {browser}', 'opts': create_ydl_opts(use_browser_cookies=browser) }) last_error = None for attempt in range(max_retries): for strategy in strategies: try: logger.info(f"YouTube: попытка {attempt + 1}/{max_retries}, стратегия: {strategy['name']}") # Получаем информацию о видео with yt_dlp.YoutubeDL(strategy['opts'].copy()) as ydl: info = ydl.extract_info(url, download=False) video_title = info.get('title', 'video') if info else 'video' logger.info(f"YouTube: получена информация о видео: {video_title}") # Настройки для скачивания с более гибким форматом format_options = [ 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best', 'best[ext=mp4]/best', 'bestvideo+bestaudio/best', 'best', ] # Используем ту же стратегию куков для скачивания for format_option in format_options: download_opts = strategy['opts'].copy() download_opts.update({ 'format': format_option, 'outtmpl': _safe_filename(video_title), }) logger.info(f"YouTube: скачивание (формат: {format_option}, стратегия: {strategy['name']})") try: with yt_dlp.YoutubeDL(download_opts) as ydl: ydl.download([url]) # Находим скачанный файл downloaded_files = list(DOWNLOADS_DIR.glob('*')) if downloaded_files: downloaded_files.sort(key=lambda x: x.stat().st_mtime, reverse=True) logger.info(f"YouTube: успешно скачано стратегией '{strategy['name']}'") return downloaded_files[0] else: raise Exception("Файл не был найден после скачивания") except Exception as download_error: error_str = str(download_error).lower() # Если ошибка "bot" или "sign in" - пробуем следующую стратегию if any(keyword in error_str for keyword in ['bot', 'sign in', 'cookies']): logger.warning(f"YouTube: ошибка с куками для формата {format_option}, пробуем следующую стратегию...") break # Переходим к следующей стратегии # Если ошибка формата - пробуем следующий формат if 'format is not available' in error_str or 'requested format' in error_str: logger.warning(f"YouTube: формат {format_option} недоступен, пробуем следующий...") continue # Другая ошибка - логируем и пробуем следующий формат logger.warning(f"YouTube: ошибка формата {format_option}: {str(download_error)[:100]}") continue # Если все форматы не подошли для этой стратегии, пробуем следующую continue except Exception as e: error_str = str(e).lower() logger.warning(f"YouTube: стратегия '{strategy['name']}' не сработала: {str(e)[:100]}") # Если ошибка "bot" или "sign in" - пробуем следующую стратегию if any(keyword in error_str for keyword in ['bot', 'sign in', 'cookies']): continue # Для других ошибок - пробуем следующую стратегию или следующую попытку last_error = e # Если все стратегии не сработали, делаем паузу перед следующей попыткой if attempt < max_retries - 1: time.sleep((attempt + 1) * 2) raise last_error or Exception("Не удалось скачать видео ни с одной стратегией") @app.route('/health', methods=['GET']) def health(): """Health check endpoint""" return jsonify({'status': 'ok', 'service': 'youtube-downloader'}), 200 @app.route('/download/stream', methods=['POST']) def download_stream(): """Скачивает видео с YouTube и возвращает бинарные данные""" try: data = request.get_json() if not data or 'url' not in data: return jsonify({'error': 'URL is required'}), 400 url = data['url'] logger.info(f"Получен запрос на скачивание (stream): {url}") # Проверяем, что это YouTube URL if 'youtube.com' not in url and 'youtu.be' not in url: return jsonify({'error': 'Only YouTube URLs are supported'}), 400 # Скачиваем видео video_path = download_youtube_video(url) logger.info(f"Видео скачано: {video_path}") # Читаем файл и отправляем with open(video_path, 'rb') as f: video_data = f.read() # Безопасное имя файла без кириллицы для заголовка safe_filename = video_path.name.encode('ascii', 'ignore').decode('ascii') or 'youtube_video.mp4' if not safe_filename.endswith(('.mp4', '.webm', '.mkv')): safe_filename = 'youtube_video.mp4' # Определяем content-type content_type = 'video/mp4' if video_path.suffix == '.webm': content_type = 'video/webm' elif video_path.suffix == '.mkv': content_type = 'video/x-matroska' # Удаляем временный файл video_path.unlink() return video_data, 200, { 'Content-Type': content_type, 'Content-Disposition': f'attachment; filename="{safe_filename}"' } except Exception as e: error_str = str(e) logger.error(f"Ошибка при скачивании: {error_str}") # Улучшаем сообщение об ошибке error_msg = error_str if 'cookies' in error_str.lower() or 'bot' in error_str.lower() or 'sign in' in error_str.lower(): error_msg = ( f"{error_str}\n\n" "💡 Совет: YouTube требует авторизацию для этого видео (18+, приватное и т.д.). " "Попробуйте открыть видео в браузере, авторизоваться, затем повторить запрос." ) return jsonify({'error': error_msg}), 500 if __name__ == '__main__': port = int(os.getenv('PORT', 5000)) # Внутренний порт контейнера host = os.getenv('HOST', '0.0.0.0') logger.info(f"Запуск YouTube Downloader сервиса на {host}:{port}") app.run(host=host, port=port, debug=False)