296 lines
14 KiB
Python
296 lines
14 KiB
Python
from nicegui import ui, run
|
|
import os
|
|
import shutil
|
|
import json
|
|
import asyncio
|
|
import datetime
|
|
from collections import deque
|
|
|
|
# --- CONFIGURAÇÕES DE DIRETÓRIOS ---
|
|
SRC_ROOT = "/downloads"
|
|
DST_ROOT = "/media"
|
|
CONFIG_PATH = "/app/data/presets.json"
|
|
LOG_PATH = "/app/data/history.log"
|
|
|
|
class DeployManager:
|
|
def __init__(self):
|
|
self.src_path = SRC_ROOT
|
|
self.dst_path = DST_ROOT
|
|
self.selected_items = []
|
|
self.container = None
|
|
self.presets = self.load_presets()
|
|
self.pendencies = []
|
|
self.logs = self.load_logs_from_file()
|
|
|
|
# --- GERENCIAMENTO DE LOGS ---
|
|
def load_logs_from_file(self):
|
|
if not os.path.exists(LOG_PATH):
|
|
return []
|
|
try:
|
|
with open(LOG_PATH, 'r', encoding='utf-8') as f:
|
|
last_lines = list(deque(f, maxlen=50))
|
|
return [line.strip() for line in reversed(last_lines)]
|
|
except:
|
|
return []
|
|
|
|
def add_log(self, message, type="info"):
|
|
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
full_msg = f"[{timestamp}] {message}"
|
|
self.logs.insert(0, full_msg)
|
|
if len(self.logs) > 50: self.logs.pop()
|
|
try:
|
|
with open(LOG_PATH, 'a', encoding='utf-8') as f:
|
|
f.write(full_msg + "\n")
|
|
except: pass
|
|
|
|
# --- 1. PERSISTÊNCIA ---
|
|
def load_presets(self):
|
|
if os.path.exists(CONFIG_PATH):
|
|
try:
|
|
with open(CONFIG_PATH, 'r') as f: return json.load(f)
|
|
except: return {}
|
|
return {}
|
|
|
|
def save_preset(self, name):
|
|
if not name: return
|
|
self.presets[name] = {'src': self.src_path, 'dst': self.dst_path}
|
|
with open(CONFIG_PATH, 'w') as f: json.dump(self.presets, f)
|
|
ui.notify(f'Preset "{name}" salvo!')
|
|
self.refresh()
|
|
|
|
def delete_preset(self, name):
|
|
if name in self.presets:
|
|
del self.presets[name]
|
|
with open(CONFIG_PATH, 'w') as f: json.dump(self.presets, f)
|
|
self.refresh()
|
|
|
|
# --- 2. DIÁLOGO ---
|
|
def confirm_preset_execution(self, name, paths):
|
|
src, dst = paths['src'], paths['dst']
|
|
if not os.path.exists(src):
|
|
ui.notify(f'Erro: Origem não existe: {src}', type='negative')
|
|
return
|
|
|
|
try: count = len([f for f in os.listdir(src) if not f.startswith('.')])
|
|
except: count = 0
|
|
|
|
with ui.dialog() as dialog, ui.card():
|
|
ui.label(f'Executar: {name}?').classes('text-xl font-bold text-blue-900')
|
|
ui.label(f'Origem: {src}').classes('text-xs bg-gray-100 p-1 w-full break-all')
|
|
ui.label(f'Destino: {dst}').classes('text-xs bg-gray-100 p-1 w-full break-all')
|
|
ui.label(f'{count} itens encontrados.').classes('font-bold text-green-700 mt-2') if count > 0 else None
|
|
|
|
async def execute_action():
|
|
dialog.close()
|
|
await asyncio.sleep(0.1)
|
|
await self.move_process_from_preset(paths)
|
|
|
|
with ui.row().classes('w-full justify-end mt-4'):
|
|
ui.button('Cancelar', on_click=dialog.close).props('flat')
|
|
ui.button('CONFIRMAR', on_click=execute_action).props('color=green')
|
|
dialog.open()
|
|
|
|
# --- 3. MOVIMENTAÇÃO BLINDADA (SAFE MERGE) ---
|
|
async def move_process(self, items_to_move, target_folder):
|
|
"""
|
|
Nova lógica: Nunca move pastas inteiras.
|
|
Sempre cria a estrutura no destino e move apenas arquivos.
|
|
"""
|
|
for item_path in items_to_move:
|
|
if not os.path.exists(item_path): continue
|
|
|
|
name = os.path.basename(item_path)
|
|
destination = os.path.join(target_folder, name)
|
|
|
|
# --- CASO 1: É UMA PASTA? ---
|
|
if os.path.isdir(item_path):
|
|
# Não verifica se existe ou não. Simplesmente garante que a pasta
|
|
# exista no destino (mkdir -p) e entra nela.
|
|
try:
|
|
if not os.path.exists(destination):
|
|
os.makedirs(destination, exist_ok=True)
|
|
self.apply_permissions(destination) # Garante permissão na nova pasta
|
|
|
|
# Pega conteúdo e RECURSIVIDADE (mergulha na pasta)
|
|
sub_items = [os.path.join(item_path, f) for f in os.listdir(item_path)]
|
|
await self.move_process(sub_items, destination)
|
|
|
|
# Limpeza: Se a pasta de origem ficou vazia, remove ela
|
|
if not os.listdir(item_path):
|
|
os.rmdir(item_path)
|
|
|
|
except Exception as e:
|
|
self.add_log(f"❌ Erro na pasta {name}: {e}", "negative")
|
|
continue
|
|
|
|
# --- CASO 2: É UM ARQUIVO? ---
|
|
# Se já existe o arquivo exato no destino -> Pendência
|
|
if os.path.exists(destination):
|
|
self.add_log(f"⚠️ Conflito de arquivo: {name}", "warning")
|
|
self.pendencies.append({'name': name, 'src': item_path, 'dst': destination})
|
|
self.refresh()
|
|
continue
|
|
|
|
# Se não existe -> Move o arquivo
|
|
try:
|
|
await run.cpu_bound(shutil.move, item_path, destination)
|
|
self.apply_permissions(destination)
|
|
self.add_log(f"✅ Arquivo movido: {name}", "positive")
|
|
except Exception as e:
|
|
self.add_log(f"❌ Erro arquivo {name}: {e}", "negative")
|
|
|
|
self.selected_items = []
|
|
self.refresh()
|
|
|
|
async def move_process_from_preset(self, paths):
|
|
src, dst = paths['src'], paths['dst']
|
|
if os.path.exists(src):
|
|
items = [os.path.join(src, f) for f in os.listdir(src)]
|
|
await self.move_process(items, dst)
|
|
else: ui.notify('Origem não encontrada!', type='negative')
|
|
|
|
def apply_permissions(self, path):
|
|
try:
|
|
if os.path.isdir(path): os.system(f'chmod -R 777 "{path}"')
|
|
else: os.chmod(path, 0o777)
|
|
except: pass
|
|
|
|
# --- 4. AÇÕES PENDÊNCIAS ---
|
|
async def handle_all_pendencies(self, action):
|
|
temp_list = list(self.pendencies)
|
|
for i in range(len(temp_list)):
|
|
await self.handle_pendency(0, action, refresh=False)
|
|
self.refresh()
|
|
|
|
async def handle_pendency(self, index, action, refresh=True):
|
|
if index >= len(self.pendencies): return
|
|
item = self.pendencies.pop(index)
|
|
|
|
if action == 'replace':
|
|
try:
|
|
# Como agora só arquivos caem aqui, remove e substitui
|
|
if os.path.isdir(item['dst']):
|
|
await run.cpu_bound(shutil.rmtree, item['dst'])
|
|
else:
|
|
await run.cpu_bound(os.remove, item['dst'])
|
|
|
|
await run.cpu_bound(shutil.move, item['src'], item['dst'])
|
|
self.apply_permissions(item['dst'])
|
|
self.add_log(f"🔄 Substituído: {item['name']}")
|
|
except Exception as e:
|
|
self.add_log(f"❌ Erro ao substituir {item['name']}: {e}", "negative")
|
|
|
|
if refresh: self.refresh()
|
|
|
|
# --- 5. NAVEGAÇÃO ---
|
|
def render_breadcrumbs(self, current_path, root_dir, nav_callback):
|
|
with ui.row().classes('items-center gap-1 bg-gray-100 p-1 rounded w-full mb-2'):
|
|
ui.button('🏠', on_click=lambda: nav_callback(root_dir)).props('flat dense size=sm')
|
|
rel = os.path.relpath(current_path, root_dir)
|
|
if rel != '.':
|
|
acc = root_dir
|
|
for part in rel.split(os.sep):
|
|
ui.label('/')
|
|
acc = os.path.join(acc, part)
|
|
ui.button(part, on_click=lambda p=acc: nav_callback(p)).props('flat dense no-caps size=sm')
|
|
if current_path != root_dir:
|
|
ui.space()
|
|
parent = os.path.dirname(current_path)
|
|
ui.button(icon='arrow_upward', on_click=lambda: nav_callback(parent)).props('flat round dense size=sm color=primary')
|
|
|
|
def navigate_src(self, path):
|
|
if os.path.exists(path) and os.path.isdir(path):
|
|
self.src_path = path; self.refresh()
|
|
|
|
def navigate_dst(self, path):
|
|
if os.path.exists(path) and os.path.isdir(path):
|
|
self.dst_path = path; self.refresh()
|
|
|
|
# --- 6. UI ---
|
|
def refresh(self):
|
|
if self.container:
|
|
self.container.clear()
|
|
with self.container: self.render_layout()
|
|
|
|
def render_layout(self):
|
|
with ui.row().classes('w-full bg-blue-50 p-3 rounded-lg items-center shadow-sm'):
|
|
ui.icon('bolt', color='blue').classes('text-2xl')
|
|
ui.label('SMART DEPLOYS:').classes('font-bold text-blue-900 mr-4')
|
|
for name, paths in self.presets.items():
|
|
with ui.button_group().props('rounded'):
|
|
ui.button(name, on_click=lambda n=name, p=paths: self.confirm_preset_execution(n, p)).props('color=blue-6')
|
|
ui.button(on_click=lambda n=name: self.delete_preset(n)).props('icon=delete color=red-4')
|
|
ui.button('Salvar Favorito', on_click=self.prompt_save_preset).props('flat icon=add_circle color=green-7').classes('ml-auto')
|
|
|
|
with ui.row().classes('w-full gap-6 mt-4'):
|
|
with ui.column().classes('flex-grow w-1/2'):
|
|
ui.label('📂 ORIGEM').classes('text-lg font-bold text-blue-700')
|
|
self.render_breadcrumbs(self.src_path, SRC_ROOT, self.navigate_src)
|
|
self.render_file_list(self.src_path, is_source=True)
|
|
with ui.column().classes('flex-grow w-1/2'):
|
|
ui.label('🎯 DESTINO').classes('text-lg font-bold text-green-700')
|
|
self.render_breadcrumbs(self.dst_path, DST_ROOT, self.navigate_dst)
|
|
self.render_file_list(self.dst_path, is_source=False)
|
|
|
|
with ui.row().classes('w-full gap-6 mt-6'):
|
|
with ui.card().classes('flex-grow h-64 bg-orange-50 border-orange-200 shadow-none'):
|
|
with ui.row().classes('w-full items-center border-b pb-2'):
|
|
ui.label(f'⚠️ Pendências ({len(self.pendencies)})').classes('font-bold text-orange-900 text-lg')
|
|
if self.pendencies:
|
|
ui.button('SUBSTITUIR TODOS', on_click=lambda: self.handle_all_pendencies('replace')).props('color=green-8 size=sm icon=done_all')
|
|
ui.button('IGNORAR TODOS', on_click=lambda: self.handle_all_pendencies('ignore')).props('color=grey-7 size=sm icon=clear_all')
|
|
with ui.scroll_area().classes('w-full h-full'):
|
|
for i, p in enumerate(self.pendencies):
|
|
with ui.row().classes('w-full items-center p-2 border-b bg-white rounded mb-1'):
|
|
ui.label(p['name']).classes('flex-grow text-xs font-medium')
|
|
ui.button(icon='swap_horiz', on_click=lambda idx=i: self.handle_pendency(idx, 'replace')).props('flat dense color=green')
|
|
ui.button(icon='close', on_click=lambda idx=i: self.handle_pendency(idx, 'ignore')).props('flat dense color=red')
|
|
|
|
with ui.card().classes('flex-grow h-64 bg-slate-900 text-slate-200 shadow-none'):
|
|
ui.label('📜 Logs').classes('font-bold border-b border-slate-700 w-full pb-2')
|
|
with ui.scroll_area().classes('w-full h-full'):
|
|
for log in self.logs:
|
|
color = 'text-red-400' if '❌' in log else 'text-green-400' if '✅' in log else 'text-slate-300'
|
|
ui.label(f"> {log}").classes(f'text-[10px] font-mono leading-tight {color}')
|
|
|
|
ui.button('INICIAR MOVIMENTAÇÃO', on_click=lambda: self.move_process(self.selected_items, self.dst_path))\
|
|
.classes('w-full py-6 mt-4 text-xl font-black shadow-lg')\
|
|
.props('color=green-7 icon=forward')\
|
|
.bind_enabled_from(self, 'selected_items', backward=lambda x: len(x) > 0)
|
|
|
|
def render_file_list(self, path, is_source):
|
|
try:
|
|
entries = sorted(os.scandir(path), key=lambda e: (not e.is_dir(), e.name.lower()))
|
|
with ui.scroll_area().classes('h-[400px] border-2 rounded-lg bg-white w-full shadow-inner'):
|
|
if not entries: ui.label('Pasta vazia').classes('p-4 text-gray-400 italic')
|
|
for entry in entries:
|
|
is_selected = entry.path in self.selected_items
|
|
bg = "bg-blue-100 border-blue-200" if is_selected else "hover:bg-gray-50 border-gray-100"
|
|
with ui.row().classes(f'w-full p-2 border-b items-center {bg} transition-colors cursor-pointer') as r:
|
|
if is_source: ui.checkbox(value=is_selected, on_change=lambda e, p=entry.path: self.toggle_selection(p)).props('dense')
|
|
icon = 'folder' if entry.is_dir() else 'movie' if entry.name.lower().endswith(('.mkv','.mp4')) else 'description'
|
|
ui.icon(icon, color='amber-500' if entry.is_dir() else 'blue-grey-400')
|
|
ui.label(entry.name).classes('text-sm flex-grow truncate select-none')
|
|
if entry.is_dir(): r.on('click', lambda p=entry.path: self.navigate_src(p) if is_source else self.navigate_dst(p))
|
|
elif is_source: r.on('click', lambda p=entry.path: self.toggle_selection(p))
|
|
except: ui.label('Erro ao acessar diretório.').classes('text-red-500 p-4 font-bold')
|
|
|
|
def prompt_save_preset(self):
|
|
with ui.dialog() as d, ui.card().classes('p-6'):
|
|
ui.label('Criar Preset').classes('text-lg font-bold')
|
|
name_input = ui.input('Nome')
|
|
with ui.row().classes('w-full justify-end mt-4'):
|
|
ui.button('Cancelar', on_click=d.close).props('flat')
|
|
ui.button('SALVAR', on_click=lambda: [self.save_preset(name_input.value), d.close()]).props('color=green')
|
|
d.open()
|
|
|
|
def toggle_selection(self, path):
|
|
if path in self.selected_items: self.selected_items.remove(path)
|
|
else: self.selected_items.append(path)
|
|
self.refresh()
|
|
|
|
def create_ui():
|
|
os.makedirs("/app/data", exist_ok=True)
|
|
dm = DeployManager()
|
|
dm.container = ui.column().classes('w-full h-full p-4 max-w-7xl mx-auto')
|
|
dm.refresh() |