Update app.py

app.py

@@ -2,116 +2,197 @@ import gradio as gr
 import spacy
 import pandas as pd
 from docx import Document
-from io import BytesIO
 import tempfile
 import os
 import multiprocessing as mp
 import psutil
-import time
 from datetime import datetime
+from typing import List, Dict
+from dataclasses import dataclass
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
-# Cargar el modelo de SpaCy en español
-nlp = spacy.load('zh_core_web_trf')
-
-def get_system_status():
-    cpu_usage = psutil.cpu_percent()
-    memory = psutil.virtual_memory()
-    return f"CPU: {cpu_usage}% | RAM: {memory.percent}% | Último update: {datetime.now().strftime('%H:%M:%S')}"
-
-def extract_names_from_text(text):
-    print(f'{len(text)}/n/n')
-    doc = nlp(text)
-    persons = [ent.text for ent in doc.ents if ent.label_ == 'PERSON']
-    return persons
-
-def split_text(text, max_length=100000):
-    result = []
-    current_chunk = []
-    current_length = 0
-    paragraphs = text.split('\n')
-
-    for paragraph in paragraphs:
-        paragraph_length = len(paragraph) + 1
-        if current_length + paragraph_length <= max_length:
-            current_chunk.append(paragraph)
-            current_length += paragraph_length
-        else:
+@dataclass
+class ProcessingResult:
+    filename: str
+    names: List[str]
+    status: str
+    error: str = None
+
+class SystemMonitor:
+    @staticmethod
+    def get_status() -> str:
+        cpu_usage = psutil.cpu_percent()
+        memory = psutil.virtual_memory()
+        return f"CPU: {cpu_usage}% | RAM: {memory.percent}% | Último update: {datetime.now().strftime('%H:%M:%S')}"
+
+class TextProcessor:
+    def __init__(self, model_name: str = 'zh_core_web_trf'):
+        self.nlp = spacy.load(model_name)
+        self.max_chunk_size = 100000
+
+    def extract_names(self, text: str) -> List[str]:
+        doc = self.nlp(text)
+        return [ent.text for ent in doc.ents if ent.label_ == 'PERSON']
+
+    def split_text(self, text: str) -> List[str]:
+        result = []
+        current_chunk = []
+        current_length = 0
+        
+        for paragraph in text.split('\n'):
+            paragraph_length = len(paragraph) + 1
+            if current_length + paragraph_length <= self.max_chunk_size:
+                current_chunk.append(paragraph)
+                current_length += paragraph_length
+            else:
+                result.append('\n'.join(current_chunk))
+                current_chunk = [paragraph]
+                current_length = paragraph_length
+
+        if current_chunk:
             result.append('\n'.join(current_chunk))
-            current_chunk = [paragraph]
-            current_length = paragraph_length
-
-    if current_chunk:
-        result.append('\n'.join(current_chunk))
-
-    return result
-
-def extract_names_from_fragments(fragments):
-    with mp.Pool(processes=4) as pool:
-        results = pool.map(extract_names_from_text, fragments)
-    return results
-
-def extract_names_from_docx(docx_file, progress=gr.Progress()):
-    # Inicializar variables de progreso
-    progress(0, desc="Iniciando procesamiento...")
-    
-    # Cargar el archivo DOCX
-    document = Document(docx_file)
-    full_text = []
-    for para in document.paragraphs:
-        full_text.append(para.text)
-    
-    progress(0.2, desc="Documento cargado, preparando texto...")
-    
-    # Unir todo el texto
-    text = ' '.join(full_text)
-    
-    # Dividir el texto en fragmentos
-    text_fragments = split_text(text)
-    progress(0.3, desc=f"Texto dividido en {len(text_fragments)} fragmentos...")
-    
-    # Extraer los nombres de cada fragmento en paralelo
-    all_persons = []
-    for i, fragment_results in enumerate(extract_names_from_fragments(text_fragments)):
-        all_persons.extend(fragment_results)
-        progress((0.3 + (0.5 * (i+1)/len(text_fragments))), 
-                desc=f"Procesando fragmento {i+1} de {len(text_fragments)}...")
-    
-    # Eliminar duplicados
-    all_persons = list(set(all_persons))
-    progress(0.9, desc="Preparando resultados...")
-    
-    # Crear un DataFrame
-    df = pd.DataFrame(all_persons, columns=['Nombres'])
-    
-    # Crear un archivo temporal para guardar el Excel
-    temp_dir = tempfile.mkdtemp()
-    temp_file_path = os.path.join(temp_dir, "nombres_personas.xlsx")
-    
-    # Guardar el DataFrame en un archivo Excel
-    with pd.ExcelWriter(temp_file_path, engine='openpyxl') as writer:
-        df.to_excel(writer, index=False)
-    
-    progress(1.0, desc="¡Procesamiento completado!")
-    return temp_file_path
-
-# Interfaz de Gradio
-with gr.Blocks() as demo:
-    gr.Markdown("# Extractor de Nombres")
-    gr.Markdown("Sube un archivo .docx y extrae los nombres de las personas usando NLP con SpaCy.")
-    
-    # Componente de estado del sistema (keepalive)
-    system_status = gr.Textbox(label="Estado del Sistema", value="Inicializando...")
-    
-    # Componentes principales
-    file_input = gr.File(file_types=[".docx"])
-    output_file = gr.File(label="Archivo de resultados")
-    
-    # Botón de proceso
-    process_btn = gr.Button("Procesar Documento")
-    process_btn.click(fn=extract_names_from_docx, inputs=file_input, outputs=output_file)
-    
-    # Actualización periódica del estado del sistema
-    demo.load(get_system_status, None, system_status, every=5)
-
-# Iniciar la aplicación
-demo.launch()
+
+        return result
+
+class DocumentProcessor:
+    def __init__(self, text_processor: TextProcessor):
+        self.text_processor = text_processor
+        self.num_processes = mp.cpu_count()
+
+    def process_document(self, file_path: str, progress=None) -> ProcessingResult:
+        try:
+            if progress:
+                progress(0.1, desc=f"Procesando {os.path.basename(file_path)}...")
+
+            # Cargar documento
+            document = Document(file_path)
+            text = ' '.join(para.text for para in document.paragraphs)
+
+            if progress:
+                progress(0.3, desc="Dividiendo texto en fragmentos...")
+
+            # Dividir texto en fragmentos
+            fragments = self.text_processor.split_text(text)
+
+            if progress:
+                progress(0.5, desc="Extrayendo nombres...")
+
+            # Procesar fragmentos en paralelo
+            with mp.Pool(processes=self.num_processes) as pool:
+                all_names = []
+                for names in pool.imap(self.text_processor.extract_names, fragments):
+                    all_names.extend(names)
+
+            if progress:
+                progress(0.8, desc="Finalizando procesamiento...")
+
+            return ProcessingResult(
+                filename=os.path.basename(file_path),
+                names=list(set(all_names)),
+                status="success"
+            )
+
+        except Exception as e:
+            return ProcessingResult(
+                filename=os.path.basename(file_path),
+                names=[],
+                status="error",
+                error=str(e)
+            )
+
+class ResultsExporter:
+    @staticmethod
+    def export_to_excel(results: List[ProcessingResult]) -> str:
+        # Crear DataFrame con todos los resultados
+        data = []
+        for result in results:
+            for name in result.names:
+                data.append({
+                    'Archivo': result.filename,
+                    'Nombre': name,
+                    'Estado': result.status,
+                    'Error': result.error
+                })
+
+        df = pd.DataFrame(data)
+
+        # Guardar a archivo temporal
+        temp_dir = tempfile.mkdtemp()
+        temp_file_path = os.path.join(temp_dir, "nombres_extraidos.xlsx")
+        
+        with pd.ExcelWriter(temp_file_path, engine='openpyxl') as writer:
+            df.to_excel(writer, index=False)
+
+        return temp_file_path
+
+class NameExtractorApp:
+    def __init__(self):
+        self.text_processor = TextProcessor()
+        self.document_processor = DocumentProcessor(self.text_processor)
+        self.system_monitor = SystemMonitor()
+        self.results_exporter = ResultsExporter()
+
+    def process_files(self, files: List[tempfile._TemporaryFileWrapper], progress=None) -> str:
+        if progress:
+            progress(0, desc="Iniciando procesamiento...")
+
+        results = []
+        total_files = len(files)
+
+        # Procesar archivos en paralelo usando ThreadPoolExecutor
+        with ThreadPoolExecutor(max_workers=min(total_files, os.cpu_count() * 2)) as executor:
+            future_to_file = {
+                executor.submit(self.document_processor.process_document, file.name): file
+                for file in files
+            }
+
+            for i, future in enumerate(as_completed(future_to_file)):
+                result = future.result()
+                results.append(result)
+                if progress:
+                    progress((i + 1) / total_files, 
+                            desc=f"Procesado {i + 1} de {total_files} archivos...")
+
+        if progress:
+            progress(0.9, desc="Generando archivo de resultados...")
+
+        # Exportar resultados
+        output_file = self.results_exporter.export_to_excel(results)
+
+        if progress:
+            progress(1.0, desc="¡Procesamiento completado!")
+
+        return output_file
+
+    def create_interface(self):
+        with gr.Blocks() as demo:
+            gr.Markdown("# Extractor de Nombres - Procesamiento Paralelo")
+            gr.Markdown("Sube uno o varios archivos .docx para extraer nombres de personas usando NLP.")
+            
+            # Estado del sistema
+            system_status = gr.Textbox(label="Estado del Sistema", value="Inicializando...")
+            
+            # Entrada y salida
+            file_input = gr.File(file_types=[".docx"], multiple=True)
+            output_file = gr.File(label="Archivo de resultados")
+            
+            # Botón de proceso
+            process_btn = gr.Button("Procesar Documentos")
+            process_btn.click(
+                fn=self.process_files,
+                inputs=file_input,
+                outputs=output_file
+            )
+            
+            # Actualización del estado del sistema
+            demo.load(self.system_monitor.get_status, None, system_status, every=5)
+
+        return demo
+
+def main():
+    app = NameExtractorApp()
+    demo = app.create_interface()
+    demo.launch()
+
+if __name__ == "__main__":
+    main()