javier
/
nicegui_app
mirror of https://github.com/JavMB/nicegui_app.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

first commit

This commit is contained in:
javiermengual 2025-12-05 18:52:29 +01:00
commit 6883e8a628
7 changed files with 792 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
.gitignore vendored Normal file
View File

@ -0,0 +1,30 @@
# Virtual Environment
venv/
env/
ENV/
.venv
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
# Database
*.db
*.sqlite3
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Backups
backups/

59
README.md Normal file
View File

@ -0,0 +1,59 @@
# PSP Dashboard
Dashboard web para monitorizar recursos del sistema. Básicamente es un proyecto para probar NiceGUI y jugar con procesos, threads y cosillas de Python.
## Qué hace
- Muestra hora del sistema en tiempo real
- Tráfico de red (bajada/subida)
- Uso de CPU y RAM con gráficos circulares
- Contador de tareas activas
- Editor de notas que se guarda en SQLite
- Ejecutor de comandos (threads y procesos)
- Sistema de alarmas
- Web scraping básico
- Reproductor de MP3
- Backups automáticos
## Cómo ejecutarlo
Necesitas Python 3.9 o superior.
```bash
# Clonar o descargar el proyecto
cd nicegui_app
# Crear entorno virtual
python3 -m venv venv
source venv/bin/activate # En Windows: venv\Scripts\activate
# Instalar dependencias
pip install -r requirements.txt
# Ejecutar
python main.py
```
Abre el navegador en `http://localhost:8080`
## Estructura
```
.
├── main.py # UI y layout principal
├── services.py # Lógica, threads, procesos
├── dashboard.db # Base de datos SQLite
└── requirements.txt
```
## Dependencias
- NiceGUI para la interfaz web
- psutil para métricas del sistema
- requests y beautifulsoup4 para scraping
## Notas
El proyecto usa multiprocessing y threading para demostrar concurrencia. La BD se inicializa automáticamente al arrancar.
Los backups se guardan en la carpeta `backups/` si usas esa función.

BIN
dashboard.db-shm Normal file
View File

Binary file not shown.

0
dashboard.db-wal Normal file
View File

410
main.py Normal file
View File

@ -0,0 +1,410 @@
from nicegui import ui, app
import psutil
import datetime
import multiprocessing
import subprocess
import os
from services import (
init_db, get_note_content, save_note_content_process,
tarea_backup_process, tarea_alarma_thread,
trafico_red, tareas_activas,
tarea_scraping_thread, tarea_reproducir_mp3_thread,
ejecutar_en_thread, ejecutar_en_proceso
)
import time
# Inicializar DB
init_db()
# =============================================================================
# UI LAYOUT
# =============================================================================
@ui.page('/')
def main():
# Estilos globales
ui.add_head_html('<link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css" rel="stylesheet">')
ui.colors(primary='#5898d4') # Un azul tipo VS Code
# Estado local para las alarmas de este cliente
# Usamos una lista para comunicar el thread con la UI
alarmas_completadas = []
scraping_resultados = []
mp3_estado = []
with ui.header().classes('items-center justify-between bg-slate-800 text-white'):
ui.label('PSP Dashboard').classes('text-xl font-bold')
ui.label('NiceGUI + Python').classes('text-sm opacity-70')
# Contenedor principal
with ui.column().classes('w-full p-4 gap-4'):
# --- FILA 1: RELOJ Y RED ---
with ui.row().classes('w-full gap-4 items-stretch'):
# Reloj
with ui.card().classes('flex-1 w-0'):
ui.label('Hora del sistema').classes('text-gray-500 text-sm')
lbl_clock = ui.label().classes('text-4xl font-mono')
# Timer para actualizar reloj
ui.timer(1.0, lambda: lbl_clock.set_text(datetime.datetime.now().strftime('%H:%M:%S')))
# Red
with ui.card().classes('flex-1 w-0 items-end'):
ui.label('Tráfico de red').classes('text-gray-500 text-sm')
with ui.row().classes('gap-6 mt-2 justify-end'):
# Bajada
with ui.column().classes('items-end'):
with ui.row().classes('items-center gap-2 mb-1'):
ui.label('Bajada').classes('text-xs text-gray-600')
ui.icon('download', size='sm').classes('text-green-600')
lbl_rx = ui.label('0 KB/s').classes('text-2xl font-bold text-green-600')
# Subida
with ui.column().classes('items-end'):
with ui.row().classes('items-center gap-2 mb-1'):
ui.label('Subida').classes('text-xs text-gray-600')
ui.icon('upload', size='sm').classes('text-blue-600')
lbl_tx = ui.label('0 KB/s').classes('text-2xl font-bold text-blue-600')
def update_net():
lbl_rx.set_text(f"{trafico_red['bajada_kbps']:.1f} KB/s")
lbl_tx.set_text(f"{trafico_red['subida_kbps']:.1f} KB/s")
ui.timer(1.0, update_net)
# --- FILA 2: RECURSOS ---
with ui.card().classes('w-full'):
ui.label('Recursos del sistema').classes('text-gray-500 text-sm mb-2')
with ui.row().classes('w-full justify-around'):
# CPU
with ui.column().classes('items-center'):
ui.label('CPU').classes('text-xs mb-2')
with ui.element('div').classes('relative'):
progress_cpu = ui.circular_progress(value=0, show_value=False, color='red').props('size=80px thickness=0.15')
lbl_cpu = ui.label('0%').classes('text-lg font-bold absolute top-1/2 left-1/2 transform -translate-x-1/2 -translate-y-1/2')
# RAM
with ui.column().classes('items-center'):
ui.label('RAM').classes('text-xs mb-2')
with ui.element('div').classes('relative'):
progress_ram = ui.circular_progress(value=0, show_value=False, color='purple').props('size=80px thickness=0.15')
lbl_ram = ui.label('0%').classes('text-lg font-bold absolute top-1/2 left-1/2 transform -translate-x-1/2 -translate-y-1/2')
# Threads
with ui.column().classes('items-center'):
ui.label('Tareas Activas').classes('text-xs mb-2')
lbl_threads = ui.label('0').classes('text-3xl font-mono')
lbl_threads_detalle = ui.label('').classes('text-xs text-gray-500 mt-1')
def update_resources():
cpu = psutil.cpu_percent()
ram = psutil.virtual_memory().percent
progress_cpu.set_value(cpu / 100)
lbl_cpu.set_text(f"{cpu:.0f}%")
progress_ram.set_value(ram / 100)
lbl_ram.set_text(f"{ram:.0f}%")
# Contamos tareas activas en los pools
total_tareas = tareas_activas['threads'] + tareas_activas['procesos']
lbl_threads.set_text(str(total_tareas))
lbl_threads_detalle.set_text(f"{tareas_activas['threads']} threads • {tareas_activas['procesos']} procesos")
ui.timer(2.0, update_resources)
# --- T1: MULTIPROCESOS ---
ui.label('T1 · Multiprocesos').classes('text-lg font-bold mt-4')
with ui.row().classes('w-full gap-4 items-stretch'):
# Lanzar Apps
with ui.card().classes('flex-1 min-w-0 flex flex-col'):
ui.label('Lanzar Apps').classes('text-gray-500 text-sm')
sel_app = ui.select(['Safari', 'Google Chrome'], value='Safari').classes('w-full')
inp_url = ui.input(placeholder='https://...').classes('w-full')
def launch():
try:
subprocess.Popen(['open', '-a', sel_app.value, inp_url.value])
ui.notify(f'Lanzando {sel_app.value}...', type='positive')
except Exception as e:
ui.notify(f'Error: {e}', type='negative')
with ui.element('div').classes('flex-grow'):
pass
ui.button('Lanzar', on_click=launch).classes('w-full mt-2 bg-cyan-600')
# Backup
with ui.card().classes('flex-1 min-w-0 flex flex-col'):
ui.label('Backup').classes('text-gray-500 text-sm')
inp_source = ui.input(placeholder='Ruta (vacío=actual)').classes('w-full')
def do_backup():
origen = inp_source.value or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
destino = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'backups')
# Lanzar proceso usando el pool (no bloqueante)
ejecutar_en_proceso(tarea_backup_process, origen, destino)
ui.notify('Backup iniciado en segundo plano...', type='info')
with ui.element('div').classes('flex-grow'):
pass
ui.button('Crear ZIP', on_click=do_backup).classes('w-full mt-2 bg-indigo-600')
# Editor
with ui.card().classes('flex-1 min-w-0 flex flex-col'):
ui.label('Editor (Persistencia)').classes('text-gray-500 text-sm')
# Cargar contenido inicial
txt_editor = ui.textarea(value=get_note_content()).classes('w-full flex-grow').props('rows=3')
def save_editor():
content = txt_editor.value
# Lanzar proceso usando el pool (no bloqueante)
ejecutar_en_proceso(save_note_content_process, content)
ui.notify('Guardando notas...', type='info')
ui.button('Guardar', on_click=save_editor).classes('w-full mt-2 bg-teal-600')
# --- T2: MULTIHILOS ---
ui.label('T2 · Multihilos').classes('text-lg font-bold mt-4')
with ui.row().classes('w-full gap-4 items-stretch'):
# Alarma
with ui.card().classes('flex-1 min-w-0 flex flex-col'):
ui.label('Alarma').classes('text-gray-500 text-sm')
with ui.column().classes('w-full flex-grow items-center justify-center'):
# Reloj visual con progreso circular
with ui.element('div').classes('relative mb-3'):
alarm_progress = ui.circular_progress(value=0, show_value=False, color='orange').props('size=100px thickness=0.1')
with ui.column().classes('absolute top-1/2 left-1/2 transform -translate-x-1/2 -translate-y-1/2 items-center'):
ui.icon('alarm', size='xl').classes('text-orange-500')
lbl_alarm_countdown = ui.label('').classes('text-xs font-mono text-gray-600')
# Controles compactos
with ui.row().classes('w-full gap-2'):
inp_sec = ui.number(label='Segundos', value=10, min=1, max=300).classes('flex-1')
inp_msg = ui.input(label='Mensaje', value='¡Tiempo!').classes('flex-1')
# Variables para el countdown
alarm_state = {'tiempo_restante': 0, 'tiempo_total': 0, 'activa': False}
def update_alarm_display():
if alarm_state['activa'] and alarm_state['tiempo_restante'] > 0:
alarm_state['tiempo_restante'] -= 0.1
if alarm_state['tiempo_restante'] < 0:
alarm_state['tiempo_restante'] = 0
# Actualizar progreso circular (invertido: empieza en 1 y va a 0)
progreso = alarm_state['tiempo_restante'] / alarm_state['tiempo_total']
alarm_progress.set_value(1 - progreso)
# Mostrar tiempo restante
segundos = int(alarm_state['tiempo_restante'])
lbl_alarm_countdown.set_text(f"{segundos}s")
elif not alarm_state['activa']:
lbl_alarm_countdown.set_text('')
alarm_progress.set_value(0)
ui.timer(0.1, update_alarm_display)
def check_completed_alarms():
# Revisamos si hay alarmas en la lista compartida
while alarmas_completadas:
msg = alarmas_completadas.pop(0)
alarm_state['activa'] = False
alarm_state['tiempo_restante'] = 0
lbl_alarm_countdown.set_text('¡ALARMA!')
ui.notify(f"⏰ ALARMA: {msg}", type='warning', close_button=True, timeout=0)
ui.run_javascript('new Audio("https://actions.google.com/sounds/v1/alarms/beep_short.ogg").play()')
# Timer que revisa la cola de alarmas cada 500ms
ui.timer(0.5, check_completed_alarms)
def start_alarm():
sec = inp_sec.value
msg = inp_msg.value
if sec <= 0 or sec > 300:
ui.notify('Tiempo debe estar entre 1 y 300 segundos', type='negative')
return
# Lógica del thread de alarma usando el pool
def thread_logic(s, m, lista_salida):
time.sleep(s)
lista_salida.append(m)
# Ejecutar en el pool de threads (no bloqueante)
ejecutar_en_thread(thread_logic, sec, msg, alarmas_completadas)
# Iniciar countdown visual
alarm_state['tiempo_total'] = sec
alarm_state['tiempo_restante'] = sec
alarm_state['activa'] = True
ui.notify(f'Alarma configurada para {sec}s', type='positive')
ui.button('⏰ Iniciar Alarma', on_click=start_alarm).classes('w-full mt-2 bg-orange-500')
# Scraping
with ui.card().classes('flex-1 min-w-0 flex flex-col'):
ui.label('Web Scraping').classes('text-gray-500 text-sm')
inp_scraping_url = ui.input(placeholder='https://example.com').classes('w-full')
with ui.column().classes('w-full flex-grow mt-2'):
lbl_scraping_titulo = ui.label('').classes('text-xs font-bold text-gray-700')
lbl_scraping_stats = ui.label('').classes('text-xs text-gray-600')
lbl_scraping_desc = ui.label('').classes('text-xs text-gray-500 italic')
def check_scraping_results():
while scraping_resultados:
resultado = scraping_resultados.pop(0)
if 'error' in resultado:
lbl_scraping_titulo.set_text("❌ Error")
lbl_scraping_stats.set_text(str(resultado['error'])[:60])
lbl_scraping_desc.set_text('')
ui.notify(f"Error en scraping: {resultado['error']}", type='negative')
else:
lbl_scraping_titulo.set_text(f"✅ {resultado['titulo'][:40]}...")
lbl_scraping_stats.set_text(f"{resultado['num_links']} links • {resultado['num_imagenes']} imgs • {resultado['num_parrafos']} párrafos")
lbl_scraping_desc.set_text(resultado.get('descripcion', '')[:80])
ui.notify(f"Scraping completado: {resultado['titulo']}", type='positive')
ui.timer(0.5, check_scraping_results)
def start_scraping():
url = inp_scraping_url.value
if not url or not url.startswith('http'):
ui.notify('URL inválida', type='negative')
return
# Ejecutar en el pool de threads (no bloqueante)
ejecutar_en_thread(tarea_scraping_thread, url, scraping_resultados)
lbl_scraping_titulo.set_text('🔄 Scraping en progreso...')
lbl_scraping_stats.set_text('')
lbl_scraping_desc.set_text('')
ui.notify('Scraping iniciado...', type='info')
ui.button('Scrapear', on_click=start_scraping).classes('w-full mt-2 bg-purple-600')
# Reproductor MP3
with ui.card().classes('flex-1 min-w-0 flex flex-col'):
ui.label('Reproductor MP3').classes('text-gray-500 text-sm')
# Variable para almacenar la ruta del archivo
mp3_file_path = {'path': ''}
with ui.column().classes('w-full flex-grow'):
lbl_mp3_filename = ui.label('Sin archivo seleccionado').classes('text-xs text-gray-500 mb-2')
# Upload component para seleccionar/arrastrar archivo
async def handle_upload(e):
print("[Upload] Evento recibido")
try:
# El archivo viene en e.file (FileUpload object)
# Leer contenido del archivo (es async!)
content = await e.file.read()
# Guardar archivo temporalmente
temp_path = f"/tmp/{e.file.name}"
with open(temp_path, 'wb') as f:
f.write(content)
mp3_file_path['path'] = temp_path
lbl_mp3_filename.set_text(f"{e.file.name}")
ui.notify(f'Archivo cargado: {e.file.name}', type='positive')
print(f"[Upload] Guardado en: {temp_path}")
except Exception as ex:
print(f"[Upload] Error: {ex}")
ui.notify(f'Error al cargar: {ex}', type='negative')
ui.upload(
on_upload=handle_upload,
auto_upload=True,
max_files=1,
label='Arrastra MP3 aquí o haz clic'
).props('accept=".mp3,audio/mpeg" color="pink"').classes('w-full')
lbl_mp3_estado = ui.label('').classes('text-xs text-gray-600 mt-2')
def check_mp3_status():
while mp3_estado:
estado = mp3_estado.pop(0)
if estado['status'] == 'completed':
lbl_mp3_estado.set_text("Reproducción completada")
ui.notify('Reproducción finalizada', type='positive')
elif estado['status'] == 'error':
lbl_mp3_estado.set_text(f"Error: {estado['error']}")
ui.notify(f"Error: {estado['error']}", type='negative')
ui.timer(0.5, check_mp3_status)
def play_mp3():
archivo = mp3_file_path['path']
print(f"[Play] Intentando reproducir: '{archivo}'")
print(f"[Play] Estado mp3_file_path: {mp3_file_path}")
if not archivo:
print("[Play] No hay archivo en mp3_file_path")
ui.notify('Por favor, sube un archivo MP3 primero', type='warning')
return
if not os.path.exists(archivo):
print(f"[Play] Archivo no existe: {archivo}")
ui.notify('Archivo no encontrado', type='negative')
return
print(f"[Play] Lanzando reproducción de: {archivo}")
# Ejecutar en el pool de threads (no bloqueante)
ejecutar_en_thread(tarea_reproducir_mp3_thread, archivo, mp3_estado)
lbl_mp3_estado.set_text('Reproduciendo...')
ui.notify('Reproducción iniciada...', type='info')
ui.button('Reproducir', on_click=play_mp3).classes('w-full mt-2 bg-pink-600')
# --- T3: SOCKETS ---
ui.label('T3 · Sockets').classes('text-lg font-bold mt-4')
with ui.row().classes('w-full gap-4 items-stretch'):
# Chat
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Chat').classes('text-gray-400')
# TCP Server
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('TCP Server').classes('text-gray-400')
# Hueco
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Placeholder').classes('text-gray-400')
# --- T4: SERVICIOS ---
ui.label('T4 · Servicios').classes('text-lg font-bold mt-4')
with ui.row().classes('w-full gap-4 items-stretch'):
# HTTP Client
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('HTTP Client').classes('text-gray-400')
# Hueco
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Placeholder').classes('text-gray-400')
# Hueco
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Placeholder').classes('text-gray-400')
# --- T5: SEGURIDAD ---
ui.label('T5 · Seguridad').classes('text-lg font-bold mt-4')
with ui.row().classes('w-full gap-4 items-stretch'):
# Hash SHA256
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Hash SHA256').classes('text-gray-400')
# Hueco
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Placeholder').classes('text-gray-400')
# Hueco
with ui.card().classes('flex-1 min-w-0 bg-gray-50 flex flex-col'):
ui.label('Placeholder').classes('text-gray-400')
if __name__ in {"__main__", "__mp_main__"}:
ui.run(title='PSP Dashboard', port=8080, reload=True)

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
nicegui>=1.4.0
psutil>=5.9.0
requests>=2.31.0
beautifulsoup4>=4.12.0

289
services.py Normal file
View File

@ -0,0 +1,289 @@
import os
import time
import shutil
import threading
import multiprocessing
import sqlite3
import datetime
import psutil
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import asyncio
# =============================================================================
# BASE DE DATOS (SQLite simple para el editor)
# =============================================================================
DB_PATH = os.path.join(os.path.dirname(__file__), 'dashboard.db')
def init_db():
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, content TEXT)''')
# Crear nota inicial si no existe
c.execute('SELECT count(*) FROM notes WHERE id=1')
if c.fetchone()[0] == 0:
c.execute('INSERT INTO notes (id, content) VALUES (1, "")')
conn.commit()
conn.close()
def get_note_content():
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute('SELECT content FROM notes WHERE id=1')
res = c.fetchone()
conn.close()
return res[0] if res else ""
def save_note_content_process(content):
"""Función que ejecutará el proceso independiente"""
# Simulamos carga
time.sleep(1)
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
c.execute('UPDATE notes SET content = ? WHERE id=1', (content,))
conn.commit()
conn.close()
print(f"[Proceso Editor] Notas guardadas: {len(content)} chars")
# =============================================================================
# VARIABLES GLOBALES Y ESTADO
# =============================================================================
trafico_red = {
'bajada_kbps': 0.0,
'subida_kbps': 0.0,
'total_bajada': 0,
'total_subida': 0,
}
# Para rastrear threads creados por nosotros
mis_threads = []
# Para contar tareas activas - simplificado
tareas_activas = {
'threads': 0,
'procesos': 0
}
# =============================================================================
# POOL DE THREADS Y PROCESOS
# =============================================================================
# Crear un pool global de threads para ejecutar tareas en paralelo sin bloqueo
thread_pool = ThreadPoolExecutor(max_workers=10, thread_name_prefix="TaskWorker")
process_pool = ProcessPoolExecutor(max_workers=4)
# =============================================================================
# TAREAS EN SEGUNDO PLANO (THREADS)
# =============================================================================
def monitor_red_loop():
global trafico_red
if not psutil: return
anterior = psutil.net_io_counters()
tiempo_anterior = time.time()
while True:
time.sleep(1)
actual = psutil.net_io_counters()
tiempo_actual = time.time()
segundos = tiempo_actual - tiempo_anterior
if segundos < 0.001: segundos = 0.001
bajada = (actual.bytes_recv - anterior.bytes_recv) / segundos / 1024
subida = (actual.bytes_sent - anterior.bytes_sent) / segundos / 1024
# Actualizar valores del diccionario existente en lugar de reasignarlo
trafico_red['bajada_kbps'] = bajada
trafico_red['subida_kbps'] = subida
trafico_red['total_bajada'] = actual.bytes_recv
trafico_red['total_subida'] = actual.bytes_sent
anterior = actual
tiempo_anterior = tiempo_actual
# Arrancar monitor de red al importar
t_red = threading.Thread(target=monitor_red_loop, daemon=True, name="MonitorRed")
t_red.start()
mis_threads.append(t_red)
def tarea_alarma_thread(segundos, mensaje, callback_fin):
"""Espera y luego llama al callback (que actualizará la UI)"""
time.sleep(segundos)
if callback_fin:
callback_fin(mensaje)
# =============================================================================
# TAREAS MULTIPROCESO
# =============================================================================
def tarea_backup_process(origen, destino):
time.sleep(2) # Simular trabajo pesado
if not os.path.exists(destino):
os.makedirs(destino)
nombre_archivo = f"backup_{int(time.time())}"
ruta_completa = os.path.join(destino, nombre_archivo)
shutil.make_archive(ruta_completa, 'zip', origen)
print(f"[Proceso Backup] Creado en {ruta_completa}.zip")
# =============================================================================
# SCRAPING WEB (THREAD)
# =============================================================================
def tarea_scraping_thread(url, lista_resultados):
"""Realiza scraping de una URL y guarda el resultado en la lista compartida"""
try:
import requests
from bs4 import BeautifulSoup
print(f"[Thread Scraping] Scrapeando {url}...")
# Delay mínimo para simular trabajo (hace visible el contador)
time.sleep(1)
# Headers para simular un navegador real
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
}
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
# Parsear HTML con BeautifulSoup
soup = BeautifulSoup(response.text, 'html.parser')
# Extraer información
titulo = soup.title.string.strip() if soup.title else "Sin título"
# Contar elementos
num_links = len(soup.find_all('a'))
num_imagenes = len(soup.find_all('img'))
num_parrafos = len(soup.find_all('p'))
# Extraer metadescripción si existe
meta_desc = ""
meta_tag = soup.find('meta', attrs={'name': 'description'})
if meta_tag and meta_tag.get('content'):
meta_desc = meta_tag['content'][:100] + "..." if len(meta_tag['content']) > 100 else meta_tag['content']
resultado = {
'url': url,
'titulo': titulo,
'descripcion': meta_desc,
'num_links': num_links,
'num_imagenes': num_imagenes,
'num_parrafos': num_parrafos,
'longitud': len(response.text),
'status_code': response.status_code,
'timestamp': datetime.datetime.now().strftime('%H:%M:%S')
}
lista_resultados.append(resultado)
print(f"[Thread Scraping] {titulo} - {num_links} links, {num_imagenes} imgs")
except Exception as e:
lista_resultados.append({
'url': url,
'error': str(e),
'timestamp': datetime.datetime.now().strftime('%H:%M:%S')
})
print(f"[Thread Scraping] Error: {e}")
# =============================================================================
# REPRODUCTOR MP3 (THREAD)
# =============================================================================
def tarea_reproducir_mp3_thread(archivo_mp3, lista_estado):
"""Reproduce un archivo MP3 usando pygame.mixer"""
try:
import pygame
import os
print("[Thread MP3] Iniciando reproducción...")
print(f"[Thread MP3] Archivo: {archivo_mp3}")
print(f"[Thread MP3] Existe archivo? {os.path.exists(archivo_mp3)}")
# Inicializar pygame mixer
pygame.mixer.init()
print("[Thread MP3] pygame.mixer inicializado")
# Cargar y reproducir el MP3
pygame.mixer.music.load(archivo_mp3)
print("[Thread MP3] Archivo cargado, reproduciendo...")
pygame.mixer.music.play()
# Esperar a que termine la reproducción
while pygame.mixer.music.get_busy():
time.sleep(0.1)
print("[Thread MP3] Reproducción finalizada exitosamente")
lista_estado.append({'status': 'completed', 'file': archivo_mp3})
# Limpiar
pygame.mixer.quit()
except Exception as e:
lista_estado.append({'status': 'error', 'error': str(e)})
print(f"[Thread MP3] Error: {e}")
try:
pygame.mixer.quit()
except:
pass
# =============================================================================
# FUNCIONES WRAPPER PARA EJECUTAR TAREAS DE FORMA NO BLOQUEANTE
# =============================================================================
def ejecutar_en_thread(funcion, *args, **kwargs):
"""
Ejecuta una función en el pool de threads de forma no bloqueante.
Retorna un Future que puede ser cancelado o monitoreado.
Incrementa el contador de tareas activas.
"""
global tareas_activas
# Incrementar ANTES de enviar al pool
tareas_activas['threads'] += 1
print(f"[Contador] Threads: {tareas_activas['threads']} (+1)")
# Wrapper para decrementar contador cuando termine
def wrapper():
try:
resultado = funcion(*args, **kwargs)
print("[Thread] Completado exitosamente")
return resultado
except Exception as e:
print(f"[Thread] Error: {e}")
raise
finally:
tareas_activas['threads'] -= 1
print(f"[Contador] Threads: {tareas_activas['threads']} (-1)")
future = thread_pool.submit(wrapper)
return future
def ejecutar_en_proceso(funcion, *args, **kwargs):
"""
Ejecuta una función en el pool de procesos de forma no bloqueante.
Retorna un Future que puede ser cancelado o monitoreado.
Incrementa el contador de tareas activas.
IMPORTANTE: Usa un thread auxiliar para monitorear el proceso y actualizar
el contador cuando termine, ya que los callbacks de ProcessPoolExecutor
pueden no ejecutarse correctamente en todos los contextos.
"""
global tareas_activas
tareas_activas['procesos'] += 1
print(f"[Contador] Procesos: {tareas_activas['procesos']} (+1)")
future = process_pool.submit(funcion, *args, **kwargs)
# Usamos un thread auxiliar para esperar el resultado y decrementar
def monitor_proceso():
try:
future.result() # Espera a que termine (bloqueante en este thread)
print("[Proceso] Completado exitosamente")
except Exception as e:
print(f"[Proceso] Error: {e}")
finally:
global tareas_activas
tareas_activas['procesos'] -= 1
print(f"[Contador] Procesos: {tareas_activas['procesos']} (-1)")
# Lanzar thread monitor (no cuenta en el pool de threads)
monitor_thread = threading.Thread(target=monitor_proceso, daemon=True, name="MonitorProceso")
monitor_thread.start()
return future