turns-00053.parquet:47702
bddd8359375b8477fa51a54ddegenerate_repetitionAbsentFinal dense release
USER
Полностью код обнови
ASSISTANT
Конечно! Вот обновлённый код для приложения загрузки видео с YouTube с улучшенной обработкой ошибки "HTTP 429: Слишком много запросов", а также увеличенным временем ожидания для дебаунса:
```python
import sys
import datetime
import logging
import time
import random
from PyQt6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QLineEdit, QPushButton, QLabel,
QComboBox, QFileDialog, QMessageBox, QProgressBar, QTextEdit, QProgressDialog
)
from PyQt6.QtCore import Qt, QThread, pyqtSignal, QTimer
from pytube import YouTube, exceptions
# Настройка логирования
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
class LogHandler(logging.Handler):
def __init__(self, log_widget):
super().__init__()
self.log_widget = log_widget
def emit(self, record):
msg = self.format(record)
self.log_widget.append(msg)
class VideoInfoThread(QThread):
info_signal = pyqtSignal(str)
error_signal = pyqtSignal(str)
def __init__(self, url):
super().__init__()
self.url = url
def run(self):
retry_count = 0
while retry_count < 5: # Разрешить максимум 5 попыток
try:
yt = YouTube(self.url)
title = yt.title
self.info_signal.emit(title)
return # Выход из цикла, если успешно
except Exception as e:
if "429" in str(e): # Проверяем наличие ошибки 429
retry_count += 1
wait_time = (2 ** retry_count) + random.uniform(0, 1) # Экспоненциальное ожидание с джиттером
logging.warning(f'HTTP 429: Слишком много запросов. Повторная попытка через {wait_time:.2f} секунд...')
time.sleep(wait_time) # Ждем перед повторной попыткой
else:
self.error_signal.emit(f'Ошибка: {e}')
return
self.error_signal.emit('Не удалось получить информацию о видео после нескольких попыток.')
class DownloadThread(QThread):
progress_signal = pyqtSignal(int)
speed_signal = pyqtSignal(str)
time_signal = pyqtSignal(str)
error_signal = pyqtSignal(str)
finished_signal = pyqtSignal()
def __init__(self, url, quality, path):
super().__init__()
self.url = url
self.quality = quality
self.path = path
self._is_running = True
def run(self):
retry_count = 0
while retry_count < 5:
try:
yt = YouTube(self.url, on_progress_callback=self.on_progress)
if self.quality == 'Выберите качество':
stream = yt.streams.get_highest_resolution()
else:
stream = yt.streams.filter(res=self.quality).first()
if stream is None:
self.error_signal.emit(f'Не удалось найти видео с качеством {self.quality}')
return
stream.download(self.path)
self.finished_signal.emit()
return # Выход из цикла, если успешно
except exceptions.PytubeError as e:
self.error_signal.emit(f'Ошибка Pytube: {e}')
return
except Exception as e:
if "429" in str(e):
retry_count += 1
wait_time = (2 ** retry_count) + random.uniform(0, 1) # Экспоненциальное ожидание с джиттером
logging.warning(f'HTTP 429: Слишком много запросов. Повторная попытка через {wait_time:.2f} секунд...')
time.sleep(wait_time) # Ждем перед повторной попыткой
else:
self.error_signal.emit(f'Ошибка: {e}')
return
self.error_signal.emit('Не удалось загрузить видео после нескольких попыток.')
def on_progress(self, stream, chunk, bytes_remaining):
if not self._is_running:
return
total_size = stream.filesize
bytes_downloaded = total_size - bytes_remaining
percentage = (bytes_downloaded / total_size) * 100
self.progress_signal.emit(int(percentage))
speed = bytes_downloaded / (self.elapsed_time() or 1)
speed_str = f"{speed / (1024 * 1024):.2f} MB/s"
self.speed_signal.emit(speed_str)
time_remaining = bytes_remaining / speed if speed > 0 else 0
time_str = f"{time_remaining / 60:.2f} мин"
self.time_signal.emit(time_str)
def elapsed_time(self):
return (datetime.datetime.now() - self.start_time).total_seconds()
def start(self):
self.start_time = datetime.datetime.now()
super().start()
def stop(self):
self._is_running = False
class YouTubeDownloader(QWidget):
def __init__(self):
super().__init__()
self.initUI()
self.debounce_timer = QTimer(self)
self.debounce_timer.timeout.connect(self.on_debounce_timeout)
self.last_url = None
def initUI(self):
self.setWindowTitle('YouTube Downloader')
self.setGeometry(100, 100, 800, 600)
layout = QVBoxLayout()
# URL input
self.url_input = QLineEdit(self)
self.url_input.setPlaceholderText('Введите URL видео с YouTube')
self.url_input.textChanged.connect(self.update_video_info_debounced)
layout.addWidget(self.url_input)
# Title Label
self.title_label = QLabel(self)
layout.addWidget(self.title_label)
# Quality selection
self.quality_combo = QComboBox(self)
self.quality_combo.addItems(['Выберите качество'])
layout.addWidget(self.quality_combo)
# Path selection
self.path_button = QPushButton('Выбрать путь сохранения', self)
self.path_button.clicked.connect(self.select_path)
layout.addWidget(self.path_button)
self.path_label = QLabel(self)
self.path_label.setText('Путь для сохранения: ')
layout.addWidget(self.path_label)
# Download button
self.download_button = QPushButton('Начать загрузку', self)
self.download_button.clicked.connect(self.start_download)
layout.addWidget(self.download_button)
# Cancel download button
self.cancel_button = QPushButton('Отмена загрузки', self)
self.cancel_button.clicked.connect(self.cancel_download)
layout.addWidget(self.cancel_button)
# Progress bar
self.progress_bar = QProgressBar(self)
layout.addWidget(self.progress_bar)
# Speed and time labels
self.speed_label = QLabel(self)
self.speed_label.setText('Скорость: ')
layout.addWidget(self.speed_label)
self.time_label = QLabel(self)
self.time_label.setText('Оставшееся время: ')
layout.addWidget(self.time_label)
# Log output
self.log_output = QTextEdit(self)
self.log_output.setReadOnly(True)
layout.addWidget(self.log_output)
self.setLayout(layout)
# Apply CSS
self.setStyleSheet("""
QWidget {
font-family: 'Arial', sans-serif;
font-size: 14px;
background-color: #f0f0f0;
}
QLineEdit, QComboBox {
padding: 10px;
margin: 10px;
border: 2px solid #ccc;
border-radius: 5px;
background-color: #fff;
color: #333;
}
QLineEdit:focus, QComboBox:focus {
border-color: #e74c3c;
}
QPushButton {
padding: 10px 20px;
margin: 10px;
border: none;
border-radius: 5px;
background-color: #e74c3c;
color: #fff;
font-weight: bold;
text-transform: uppercase;
letter-spacing: 1px;
}
QPushButton:hover {
background-color: #c0392b;
}
QLabel {
padding: 10px;
color: #333;
}
QProgressBar {
border: 2px solid #ccc;
border-radius: 5px;
text-align: center;
background-color: #fff;
}
QProgressBar::chunk {
background-color: #e74c3c;
border-radius: 5px;
}
QTextEdit {
border: 2px solid #ccc;
border-radius: 5px;
background-color: #fff;
color: #333;
padding: 10px;
}
""")
# Настройка логирования в UI
log_handler = LogHandler(self.log_output)
logging.getLogger().addHandler(log_handler)
def update_video_info_debounced(self):
if self.debounce_timer.isActive():
self.debounce_timer.stop()
self.debounce_timer.start(1000) # Увеличьте до 1000 мс (1 секунды)
def on_debounce_timeout(self):
url = self.url_input.text()
if url and url != self.last_url:
self.last_url = url
self.title_label.clear()
self.video_info_thread = VideoInfoThread(url)
self.video_info_thread.info_signal.connect(self.set_video_info)
self.video_info_thread.error_signal.connect(self.set_error_info)
self.video_info_thread.start()
else:
self.title_label.clear()
def set_video_info(self, title):
self.title_label.setText(f'Название: {title}')
def set_error_info(self, error_message):
logging.error(error_message)
QMessageBox.warning(self, 'Ошибка', error_message)
def select_path(self):
path = QFileDialog.getExistingDirectory(self, "Выберите папку для сохранения")
if path:
self.path_label.setText(f'Путь для сохранения: {path}')
def start_download(self):
url = self.url_input.text()
quality = self.quality_combo.currentText()
path = self.path_label.text().replace('Путь для сохранения: ', '')
if not url or not path:
QMessageBox.warning(self, 'Ошибка', 'Пожалуйста, заполните все поля')
return
try:
yt = YouTube(url)
available_qualities = [stream.resolution for stream in yt.streams.filter(progressive=True, file_extension='mp4')]
if not available_qualities:
QMessageBox.warning(self, 'Ошибка', 'Нет доступных потоков для этого видео')
return
self.quality_combo.clear()
self.quality_combo.addItems(['Выберите качество'] + available_qualities)
if quality == 'Выберите качество':
stream = yt.streams.get_highest_resolution()
else:
stream = yt.streams.filter(res=quality, progressive=True, file_extension='mp4').first()
if not stream:
QMessageBox.warning(self, 'Ошибка', f'Выбранное качество {quality} недоступно для этого видео')
return
self.download_thread = DownloadThread(url, quality, path)
self.download_thread.progress_signal.connect(self.update_progress)
self.download_thread.speed_signal.connect(self.update_speed)
self.download_thread.time_signal.connect(self.update_time)
self.download_thread.error_signal.connect(self.handle_download_error)
self.download_thread.finished_signal.connect(self.download_finished)
self.download_thread.start()
self.progress_dialog = QProgressDialog("Загрузка видео...", "Отмена", 0, 100, self)
self.progress_dialog.setWindowModality(Qt.WindowModality.WindowModal)
self.progress_dialog.setAutoClose(False)
self.progress_dialog.setAutoReset(False)
self.progress_dialog.canceled.connect(self.cancel_download)
self.progress_dialog.setValue(0)
except Exception as e:
self.log_output.append(f'Ошибка при загрузке видео: {e}')
QMessageBox.warning(self, 'Ошибка', f'Произошла ошибка: {e}')
def update_progress(self, value):
logging.debug(f'Обновление прогресса: {value}%')
self.progress_bar.setValue(value)
self.progress_dialog.setValue(value)
def update_speed(self, speed):
logging.debug(f'Обновление скорости: {speed}')
self.speed_label.setText(f'Скорость: {speed}')
def update_time(self, time):
logging.debug(f'Обновление оставшегося времени: {time}')
self.time_label.setText(f'Оставшееся время: {time}')
def handle_download_error(self, error_message):
self.log_output.append(error_message)
QMessageBox.warning(self, 'Ошибка', error_message)
def download_finished(self):
self.log_output.append('Видео успешно скачано!')
QMessageBox.information(self, 'Успех', 'Видео успешно скачано!')
self.progress_dialog.reset()
def cancel_download(self):
if hasattr(self, 'download_thread') and self.download_thread.isRunning():
self.download_thread.stop()
self.download_thread.quit()
self.download_thread.wait()
self.log_output.append('Загрузка отменена пользователем.')
def cleanup_threads(self):
if hasattr(self, 'video_info_thread') and self.video_info_thread.isRunning():
self.video_info_thread.quit()
self.video_info_thread.wait()
if hasattr(self, 'download_thread') and self.download_thread.isRunning():
self.download_thread.stop()
self.download_thread.quit()
self.download_thread.wait()
if __name__ == '__main__':
app = QApplication(sys.argv)
ex = YouTubeDownloader()
ex.show()
app.aboutToQuit.connect(ex.cleanup_threads)
sys.exit(app.exec())
```
### Ключевые Изменения
1. **Улучшенная Обработка Ошибок 429**: Я добавил рандомизированный джиттер к времени ожидания при повторной попытке запроса, что уменьшает вероятность повторного столкновения с ограничениями YouTube.
2. **Увеличение Времени Ожидания (Debounce)**: Время ожидания для проверки информации о видео увеличено до 1000 мс (1 секунды), чтобы уменьшить количество запросов к YouTube при вводе текста.
3. **Управление Графическим Интерфейсом**: В интерфейсе все элементы остались без изменений, он по-прежнему прост и понятен для пользователя.
### Установка и Запуск Код
1. Убедитесь, что у вас установлены необходимые библиотеки:
- `PyQt6` и `pytube` (можно установить через pip):
```bash
pip install PyQt6 pytube
```
2. Сохраните код в файл (например, `youtube_downloader.py`).
3. Запустите файл в вашей среде Python:
```bash
python youtube_downloader.py
```
Этот код должен улучшить надежность работы вашего приложения загрузки видео с YouTube и уменьшить количество возникновений ошибок HTTP 429. Если у вас есть дополнительные вопросы или нужны доработки, пожалуйста, дайте знать!