from nicegui import ui, run import os import threading import time import yt_dlp # --- CONFIGURAÇÕES --- DOWNLOAD_DIR = "/downloads/ytdlp" # --- WORKER (BACKEND) --- class DownloadWorker(threading.Thread): def __init__(self, url, format_type, status_callback): super().__init__() self.url = url self.format_type = format_type self.callback = status_callback # Função para atualizar o estado na Interface self.daemon = True self.stop_requested = False def progress_hook(self, d): if self.stop_requested: raise Exception("Cancelado pelo usuário") if d['status'] == 'downloading': total = d.get('total_bytes') or d.get('total_bytes_estimate') or 0 downloaded = d.get('downloaded_bytes', 0) pct = int((downloaded / total) * 100) if total > 0 else 0 speed = d.get('speed', 0) or 0 speed_str = f"{speed / 1024 / 1024:.2f} MiB/s" filename = os.path.basename(d.get('filename', 'Baixando...')) eta = d.get('_eta_str', '?') # Atualiza estado em memória via callback self.callback({ "running": True, "file": filename, "progress": pct, "log": f"Baixando: {speed_str} | ETA: {eta}", "status": "downloading" }) elif d['status'] == 'finished': self.callback({ "running": True, "file": "Processando...", "progress": 99, "log": "Convertendo/Juntando arquivos (ffmpeg)...", "status": "processing" }) def run(self): if not os.path.exists(DOWNLOAD_DIR): os.makedirs(DOWNLOAD_DIR, exist_ok=True) ydl_opts = { 'outtmpl': f'{DOWNLOAD_DIR}/%(title)s.%(ext)s', 'progress_hooks': [self.progress_hook], 'nocheckcertificate': True, 'ignoreerrors': False, # Mudado para False para pegarmos os erros reais 'ffmpeg_location': '/usr/bin/ffmpeg', 'writethumbnail': True, # Garante metadados no arquivo final 'addmetadata': True, } if self.format_type == 'best': ydl_opts['format'] = 'bestvideo+bestaudio/best' ydl_opts['merge_output_format'] = 'mkv' elif self.format_type == 'audio': ydl_opts['format'] = 'bestaudio/best' ydl_opts['postprocessors'] = [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '192' }] elif self.format_type == '1080p': ydl_opts['format'] = 'bestvideo[height<=1080]+bestaudio/best[height<=1080]' ydl_opts['merge_output_format'] = 'mkv' try: self.callback({"running": True, "file": "Iniciando...", "progress": 0, "log": "Conectando...", "status": "starting"}) with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([self.url]) self.callback({"running": False, "file": "Concluído!", "progress": 100, "log": "Download finalizado com sucesso.", "status": "success"}) except Exception as e: msg = str(e) if "Cancelado" in msg: log_msg = "Download cancelado pelo usuário." else: log_msg = f"Erro: {msg}" self.callback({"running": False, "file": "Erro/Parado", "progress": 0, "log": log_msg, "status": "error"}) # --- FUNÇÃO AUXILIAR DE METADADOS (IO BOUND) --- def fetch_meta(url): try: ydl_opts = {'quiet': True, 'nocheckcertificate': True, 'ignoreerrors': True} with yt_dlp.YoutubeDL(ydl_opts) as ydl: return ydl.extract_info(url, download=False) except: return None # --- INTERFACE (FRONTEND) --- class DownloaderInterface: def __init__(self): self.container = None self.timer = None self.worker = None # Estado Local (Em memória) self.state = { "running": False, "file": "Aguardando...", "progress": 0, "log": "---", "status": "idle" } # Elementos UI self.url_input = None self.fmt_select = None self.btn_check = None self.btn_download = None self.btn_stop = None self.btn_reset = None self.preview_card = None self.preview_img = None self.preview_title = None self.status_card = None self.lbl_file = None self.progress_bar = None self.lbl_log = None def update_state(self, new_data): """Callback chamada pelo Worker (thread) para atualizar o dict de estado.""" self.state.update(new_data) async def check_url(self): url = self.url_input.value if not url: ui.notify('Insira uma URL primeiro!', type='warning') return self.btn_check.props('loading') self.lbl_log.text = "Buscando informações do vídeo..." # Roda em thread separada para não travar a UI info = await run.io_bound(fetch_meta, url) self.btn_check.props(remove='loading') if info and 'title' in info: self.preview_card.visible = True self.preview_title.text = info.get('title', 'Sem título') self.preview_img.set_source(info.get('thumbnail', '')) self.btn_download.enable() self.status_card.visible = True self.lbl_log.text = "Vídeo encontrado. Pronto para baixar." else: ui.notify('Não foi possível obter dados do vídeo. Verifique o link.', type='negative') self.lbl_log.text = "Erro ao buscar metadados." def start_download(self): url = self.url_input.value fmt = self.fmt_select.value # Reset visual self.state['progress'] = 0 self.btn_download.disable() self.btn_check.disable() self.url_input.disable() self.btn_reset.visible = False # Inicia Worker self.worker = DownloadWorker(url, fmt, self.update_state) self.worker.start() ui.notify('Download iniciado!') def stop_download(self): if self.worker and self.worker.is_alive(): self.worker.stop_requested = True self.worker.join(timeout=1.0) ui.notify('Solicitação de cancelamento enviada.') def reset_ui(self): """Reseta a interface para um novo download""" self.url_input.value = '' self.url_input.enable() self.btn_check.enable() self.btn_download.disable() self.preview_card.visible = False self.status_card.visible = False self.btn_reset.visible = False self.lbl_log.text = '---' self.state = {"running": False, "file": "Aguardando...", "progress": 0, "log": "---", "status": "idle"} def ui_update_loop(self): """Timer que atualiza os elementos visuais com base no self.state""" # Sincroniza dados da memória com os componentes self.lbl_file.text = f"Arquivo: {self.state.get('file')}" self.progress_bar.value = self.state.get('progress', 0) / 100 self.lbl_log.text = self.state.get('log') status = self.state.get('status') is_running = self.state.get('running', False) # Controle de visibilidade do botão Cancelar if self.btn_stop: self.btn_stop.visible = is_running # Tratamento de finalização/erro para mostrar botão de "Novo" if status in ['success', 'error'] and not is_running: self.btn_reset.visible = True if status == 'error': self.lbl_log.classes('text-red-500', remove='text-gray-500') else: self.lbl_log.classes('text-green-600', remove='text-gray-500') else: self.lbl_log.classes('text-gray-500', remove='text-red-500 text-green-600') def render(self): ui.label('📺 YouTube Downloader (Docker)').classes('text-xl font-bold mb-2') # --- ÁREA DE INPUT --- with ui.card().classes('w-full p-4 mb-4'): with ui.row().classes('w-full items-center gap-2'): self.url_input = ui.input('URL do Vídeo').classes('flex-grow').props('clearable placeholder="https://..."') self.btn_check = ui.button('Verificar', on_click=self.check_url).props('icon=search color=secondary') with ui.row().classes('items-center mt-2 gap-4'): self.fmt_select = ui.select( {'best': 'Melhor Qualidade (MKV)', '1080p': 'Limitado a 1080p (MKV)', 'audio': 'Apenas Áudio (MP3)'}, value='best', label='Formato' ).classes('w-64') self.btn_download = ui.button('Baixar Agora', on_click=self.start_download)\ .props('icon=download color=primary').classes('w-40') self.btn_download.disable() # Começa desabilitado até verificar # --- PREVIEW (Melhoria 7) --- self.preview_card = ui.card().classes('w-full p-2 mb-4 bg-gray-100 flex-row gap-4 items-center') self.preview_card.visible = False with self.preview_card: self.preview_img = ui.image().classes('w-32 h-24 rounded object-cover') with ui.column(): ui.label('Vídeo Detectado:').classes('text-xs text-gray-600 uppercase font-bold') self.preview_title = ui.label('').classes('font-bold text-md leading-tight') # --- STATUS E MONITORAMENTO --- self.status_card = ui.card().classes('w-full p-4') self.status_card.visible = False with self.status_card: with ui.row().classes('w-full justify-between items-center'): ui.label('Status do Processo').classes('font-bold') self.btn_reset = ui.button('Baixar Outro', on_click=self.reset_ui)\ .props('icon=refresh flat color=primary').classes('text-sm') self.btn_reset.visible = False self.lbl_file = ui.label('Aguardando...') self.progress_bar = ui.linear_progress(value=0).classes('w-full my-2') self.lbl_log = ui.label('---').classes('text-sm text-gray-500 font-mono') with ui.row().classes('w-full justify-end mt-2'): self.btn_stop = ui.button('🛑 Cancelar Download', on_click=self.stop_download).props('color=red flat') # Timer para atualizar UI a partir do estado em memória self.timer = ui.timer(0.5, self.ui_update_loop) # --- INICIALIZADOR --- def create_ui(): dl = DownloaderInterface() dl.container = ui.column().classes('w-full h-full p-4 max-w-4xl mx-auto') with dl.container: dl.render()