from datetime import datetime, timedelta INSTRUCCIONES_USUARIO =""" # 📋 Sistema de Revisión de Documentos **📦 Instrucciones para preparar el archivo ZIP:** 1. **Crea una Carpeta para cada trabajador** (por ejemplo: "Juan Pérez", "María López", etc.) 2. **Dentro de cada Carpeta**, coloca: - La cédula del trabajador (el nombre debe contener "cedula" o "cédula") - Todos los demás documentos del trabajador (contratos, afiliafonasa, etc.) 3. **Comprime todas las carpetas en un solo archivo ZIP** 4. **Sube el archivo ZIP** usando el botón de abajo **Ejemplo de estructura:** ``` Documentos.zip ├── Juan Perez/ │ ├── cedula.pdf │ ├── contrato.pdf │ └── afiliafonasa.pdf ├── Maria Lopez/ │ ├── cedula.pdf │ ├── contrato.pdf │ └── certificado.pdf └── Pedro Garcia/ ├── cedula.pdf └── contrato.pdf ``` ⚠️ **Importante:** Cada Carpeta debe tener un archivo que contenga "cedula" o "cédula" en su nombre. """ prompt_cedula = """ Eres un revisor de acreditación de personal para una minera chilena. Se te entrega un Documento de identificación Nacional de un trabajador en formato PDF, enviado por el contratista para acreditación. Tu tarea: - Verifica la vigencia del Documento. - Verifica que se vea el frente y el reverso del Documento en la misma hoja. - Verifica que exista una firma en el Documento. - Verifica que exista una foto de la cara del trabajador en el Documento. - Extrae la fecha de vencimiento del Documento. - Identifica el RUT o RUN del trabajador, bajo la imagen de su foto. - Si todo está correcto, responde EXACTAMENTE en este formato (sin agregar texto adicional): "Acreditado: [Nombre], [Apellido], [RUT], [Fecha_Vencimiento]" Ejemplo: "Acreditado: Juan, Pérez González, 12.345.678-9, 31/12/2028" - Si hay diferencias, responde: "Rechazado: [razón detallada]" No agregues texto adicional, saludos, ni explicaciones fuera del formato solicitado. """ fecha_actual = datetime.now().strftime("%d/%m/%Y") dos_semanas_atras = (datetime.now() - timedelta(weeks=10)).strftime("%d/%m/%Y") prompt_documento_general = """ Eres un revisor de documentos de acreditación de personal para una minera chilena. Se te entregan dos documentos PDF: 1. La cédula de identidad del trabajador (como referencia para revisar firma). 2. Un Documento enviado por el contratista para ser acreditado. Datos del trabajador a verificar: - Nombre: {nombre} - Apellido: {apellido} - RUT: {rut} - Fecha actual: {fecha_actual} Tu tarea: - Verifica que el Documento contenga el nombre completo ({nombre} {apellido}) del trabajador. - Verifica que el Documento contenga el RUT ({rut}) del trabajador. - Rechazar la aprobación en el caso de que las firmas que estén presentes en el Documento y que sean del trabajador, no sean lo suficientemente similares a la firma en la cédula de identidad como para asumir que es él quien firma. - Verifica que el Documento tenga una fecha reciente (mayor a """+f"{dos_semanas_atras}"+"""). - Verifica que el Documento esté completo y legible. - Extrae la fecha del Documento. - Si todo está correcto, responde EXACTAMENTE en este formato (sin agregar texto adicional): "Acreditado: [Nombre], [Apellido], [RUT], [Fecha_Documento], [Estado_Firma]" - Si el Documento son examenes médicos, solo verifica [Nombre], [Apellido] y [Fecha_Documento] Donde [Estado_Firma] puede ser: "Firma Coincide" o "Firma Similar" o "Firma No Verificable" Ejemplo: "Acreditado: Juan, Pérez González, 12.345.678-9, 15/10/2024, Firma Coincide" - Si hay diferencias, EXACTAMENTE en este formato (sin agregar texto adicional): "Rechazado: [razón detallada explicando qué no coincide]" No agregues texto adicional, saludos, ni explicaciones fuera del formato solicitado. """ # pip install google-genai gradio import gradio as gr from google import genai from google.genai import types import os import time from datetime import datetime, timedelta import tempfile import shutil from pathlib import Path from collections import defaultdict import zipfile import analytics fecha_actual = datetime.now().strftime("%d/%m/%Y") dos_semanas_atras = (datetime.now() - timedelta(weeks=2)).strftime("%d/%m/%Y") # Variables globales para almacenar datos del trabajador y cédula datos_trabajador = { 'nombre': None, 'apellido': None, 'rut': None, 'fecha_vencimiento': None, 'cedula_path': None } API_BREVO = os.environ.get("API_BREVO") G_API_KEY = os.environ.get("G_API_KEY") client = genai.Client(api_key=G_API_KEY) def revisar_cedula(cedula_file): """Procesa la cédula y extrae datos del trabajador""" global datos_trabajador try: # Leer bytes de la cédula with open(cedula_file, "rb") as f: cedula_bytes = f.read() # Guardar path de la cédula para comparaciones futuras datos_trabajador['cedula_path'] = cedula_file parts = [ types.Part.from_text(text=prompt_cedula), types.Part.from_bytes(data=cedula_bytes, mime_type="application/pdf") ] response = client.models.generate_content( model="gemini-2.5-flash", contents=parts, config=types.GenerateContentConfig(response_modalities=["TEXT"]), ) result = response.text.strip() analytics.enviar_evento(email_us,'Revisión cédula') if result.lower().startswith("acreditado"): try: txt_str = result.split(":", 1)[1].strip() partes = [p.strip() for p in txt_str.split(",")] if len(partes) >= 4: datos_trabajador['nombre'] = partes[0] datos_trabajador['apellido'] = partes[1] datos_trabajador['rut'] = partes[2] datos_trabajador['fecha_vencimiento'] = partes[3] detalle = (f"✓ Nombre: {datos_trabajador['nombre']}\n" f"✓ Apellido: {datos_trabajador['apellido']}\n" f"✓ RUT: {datos_trabajador['rut']}\n" f"✓ Vencimiento: {datos_trabajador['fecha_vencimiento']}") return os.path.basename(cedula_file), "✅ Pre-Acreditado", detalle else: return os.path.basename(cedula_file), "❌ Error", f"Formato incorrecto en respuesta. Se esperaban 4 campos, se recibieron {len(partes)}: {result}" except Exception as e: return os.path.basename(cedula_file), "❌ Error", f"No se pudieron extraer los datos: {str(e)}\nRespuesta recibida: {result}" else: return os.path.basename(cedula_file), "❌ Rechazado", result except Exception as e: return os.path.basename(cedula_file), "❌ Error", f"Error al procesar cédula: {str(e)}" def revisar_documento_general(documento_file): """Revisa cualquier Documento comparándolo con la cédula""" global datos_trabajador if not all([datos_trabajador['nombre'], datos_trabajador['apellido'], datos_trabajador['rut'], datos_trabajador['cedula_path']]): return os.path.basename(documento_file), "⚠️ Pendiente", "Debe procesarse primero la cédula" try: prompt = prompt_documento_general.format( nombre=datos_trabajador['nombre'], apellido=datos_trabajador['apellido'], rut=datos_trabajador['rut'], fecha_actual=fecha_actual ) with open(datos_trabajador['cedula_path'], "rb") as f: cedula_bytes = f.read() with open(documento_file, "rb") as f: documento_bytes = f.read() parts = [ types.Part.from_text(text=prompt), types.Part.from_bytes(data=cedula_bytes, mime_type="application/pdf"), types.Part.from_bytes(data=documento_bytes, mime_type="application/pdf") ] response = client.models.generate_content( model="gemini-2.5-flash", contents=parts, config=types.GenerateContentConfig(response_modalities=["TEXT"]), ) result = response.text.strip() analytics.enviar_evento(email_us,'Revisión') if result.lower().startswith("acreditado"): try: txt_str = result.split(":", 1)[1].strip() partes = [p.strip() for p in txt_str.split(",")] if len(partes) >= 5: nombre = partes[0] apellido = partes[1] rut = partes[2] fecha_doc = partes[3] estado_firma = partes[4] detalle = (f"✓ Nombre: {nombre}\n" f"✓ Apellido: {apellido}\n" f"✓ RUT: {rut}\n" f"✓ Fecha Documento: {fecha_doc}\n" f"✓ Estado Firma: {estado_firma}") return os.path.basename(documento_file), "✅ Pre-Acreditado", detalle else: return os.path.basename(documento_file), "✅ Pre-Acreditado", result except Exception as e: return os.path.basename(documento_file), "✅ Pre-Acreditado", result else: return os.path.basename(documento_file), "❌ Posible rechazo", result.replace(f'no es mayor a {dos_semanas_atras}', 'es posible que sea muy antigua')\ .replace(f'no es mayor al {dos_semanas_atras}', 'es posible que sea muy antigua')\ .replace(f'no es posterior a {dos_semanas_atras}', 'es posible que sea muy antigua')\ .replace(f'no es posterior al {dos_semanas_atras}', 'es posible que sea muy antigua')\ .replace('Rechazado: ','') except Exception as e: return os.path.basename(documento_file), "❌ Error", f"Error al procesar Documento: {str(e)}" def validar_acceso(email): import re global email_us patron_email = r'^\+569[0-9]+$' if not email or not email.strip(): return { pantalla_acceso: gr.update(visible=True), pantalla_principal: gr.update(visible=False), mensaje_error: gr.update(value="⚠️ Por favor ingresa un número de WhatsApp para recibir los resultados", visible=True), acceso_granted: False, email_usuario: "" } if not re.match(patron_email, email.strip()): return { pantalla_acceso: gr.update(visible=True), pantalla_principal: gr.update(visible=False), mensaje_error: gr.update(value="⚠️ Por favor ingresa un número de WhatsApp válido", visible=True), acceso_granted: False, email_usuario: "" } email_us = email analytics.enviar_evento(email,'Sign up') return { pantalla_acceso: gr.update(visible=False), pantalla_principal: gr.update(visible=True), mensaje_error: gr.update(visible=False), acceso_granted: True, email_usuario: email } def extraer_archivos_de_zip(zip_path): """ Extrae archivos del ZIP y los organiza por Carpeta. Retorna un diccionario: {nombre_carpeta: [lista_de_rutas_pdf]} """ carpetas = defaultdict(list) temp_dir = tempfile.mkdtemp() try: with zipfile.ZipFile(zip_path, 'r') as zip_ref: # Extraer todos los archivos zip_ref.extractall(temp_dir) # Recorrer el directorio extraído for root, dirs, files in os.walk(temp_dir): for file in files: if file.lower().endswith('.pdf'): archivo_path = os.path.join(root, file) # Obtener la Carpeta relativa carpeta_relativa = os.path.relpath(root, temp_dir) # Si está en la raíz, usar "Raíz" if carpeta_relativa == '.': carpeta_nombre = "Raíz" else: # Usar solo el primer nivel de Carpeta carpeta_nombre = carpeta_relativa.split(os.sep)[0] carpetas[carpeta_nombre].append(archivo_path) return carpetas, temp_dir except Exception as e: # Limpiar en caso de error if os.path.exists(temp_dir): shutil.rmtree(temp_dir) raise e from concurrent.futures import ThreadPoolExecutor, as_completed import threading def revisar_zip_stream(archivo_zip, email_usuario, progress=gr.Progress()): global datos_trabajador, resultados_acreditacion # ✅ Reiniciar resultados de forma segura resultados_acreditacion = {} resultados_acreditacion['Carpeta'] = [] resultados_acreditacion['Documento'] = [] resultados_acreditacion['Estado'] = [] resultados_acreditacion['Observaciones'] = [] if not archivo_zip: return [] zip_path = archivo_zip if isinstance(archivo_zip, str) else archivo_zip.name if not zip_path.lower().endswith('.zip'): return [] resumen_carpetas = [] temp_dir = None resultados_lock = threading.Lock() try: progress(0, desc="📦 Extrayendo archivo ZIP...") carpetas_dict, temp_dir = extraer_archivos_de_zip(zip_path) if not carpetas_dict: return [] carpetas_ordenadas = sorted(carpetas_dict.items()) total_carpetas = len(carpetas_ordenadas) for idx_carpeta, (nombre_carpeta, archivos_carpeta) in enumerate(carpetas_ordenadas, 1): progress(idx_carpeta / total_carpetas, desc=f"📂 Revisando carpeta: \"{nombre_carpeta}\"") # Reiniciar datos del trabajador datos_trabajador = { 'nombre': None, 'apellido': None, 'rut': None, 'fecha_vencimiento': None, 'cedula_path': None } # Buscar cédula cedula = next((p for p in archivos_carpeta if 'cedula' in p.lower() or 'cédula' in p.lower() or ' ci.pdf' in p.lower()), None) if not cedula: resultados_acreditacion['Carpeta'].append(nombre_carpeta) resultados_acreditacion['Documento'].append("-") resultados_acreditacion['Estado'].append("❌ Error") resultados_acreditacion['Observaciones'].append("No se encontró ninguna cédula en esta Carpeta.") resumen_carpetas.append([nombre_carpeta, "❌ Error", "No se encontró cédula"]) continue # Procesar cédula nombre, Estado, detalle = revisar_cedula(cedula) resultados_acreditacion['Carpeta'].append(nombre_carpeta) resultados_acreditacion['Documento'].append(os.path.basename(cedula)) resultados_acreditacion['Estado'].append(Estado) resultados_acreditacion['Observaciones'].append(detalle) if Estado.startswith("❌"): resumen_carpetas.append([nombre_carpeta, "❌ Posible rechazo", "Error en cédula"]) continue # Procesar documentos generales documentos_a_procesar = [ p for p in archivos_carpeta if 'cedula' not in os.path.basename(p).lower() and 'cédula' not in os.path.basename(p).lower() and ' ci.pdf' not in os.path.basename(p).lower() ] tiene_rechazo = False documentos_procesados = 0 total_docs = len(documentos_a_procesar) def procesar_doc(pdf_path, idx_doc): """Revisa un documento y actualiza el progreso.""" doc_name = os.path.basename(pdf_path) progress( (idx_carpeta - 1 + (idx_doc / max(1, total_docs))) / total_carpetas, desc=f"📂 Carpeta: '{nombre_carpeta}' → 📄 Documento {idx_doc}/{total_docs}: {doc_name}" ) nombre, Estado, detalle = revisar_documento_general(pdf_path) with resultados_lock: resultados_acreditacion['Carpeta'].append(nombre_carpeta) resultados_acreditacion['Documento'].append(nombre) resultados_acreditacion['Estado'].append(Estado) resultados_acreditacion['Observaciones'].append(detalle) return Estado # ✅ Paralelización por carpeta si hay más de 1 documento if total_docs >= 2: print(f"🧩 Procesando {total_docs} documentos en paralelo dentro de '{nombre_carpeta}'") with ThreadPoolExecutor(max_workers=min(4, total_docs)) as executor: futures = {executor.submit(procesar_doc, pdf, i+1): pdf for i, pdf in enumerate(documentos_a_procesar)} for future in as_completed(futures): try: estado = future.result() if "rechazo" in estado.lower() or "error" in estado.lower(): tiene_rechazo = True documentos_procesados += 1 except Exception as e: print(f"⚠️ Error en revisión paralela de {futures[future]}: {e}") else: # Procesamiento secuencial (solo 1 documento) for idx_doc, pdf_path in enumerate(documentos_a_procesar, 1): estado = procesar_doc(pdf_path, idx_doc) if "rechazo" in estado.lower() or "error" in estado.lower(): tiene_rechazo = True documentos_procesados += 1 resumen_carpetas.append([ nombre_carpeta, "⚠️ Posible rechazo" if tiene_rechazo else "✅ Pre-acreditada", f"{documentos_procesados + 1} documentos revisados" ]) progress(1.0, desc=f"🎉 Completado - {total_carpetas} Carpeta(s) revisada(s)") # ✅ Reemplazar global al final resultados_acreditacion = resultados_acreditacion.copy() # Envío de correo import brevo email = brevo.BrevoEmail(API_BREVO) email.enviar( destinatario_email='emiliano@prosync.cl', asunto=f"Resultados pre-acreditación: {email_usuario}", data=resultados_acreditacion, numero_documentos=len(resultados_acreditacion['Documento']), numero_carpetas=len(set(resultados_acreditacion['Carpeta'])) ) return resumen_carpetas finally: if temp_dir and os.path.exists(temp_dir): shutil.rmtree(temp_dir) def habilitar_boton(archivo): """Habilita el botón solo si hay un archivo cargado""" if archivo is None: return gr.Button("🔍 Iniciar revisión", variant="primary", interactive=False) else: return gr.Button("🔍 Iniciar revisión", variant="primary", interactive=True) with gr.Blocks(title="Revisión de Documentos de Acreditación de Personal") as demo: acceso_granted = gr.State(False) email_usuario = gr.State("") # PANTALLA DE ACCESO with gr.Column(visible=True) as pantalla_acceso: gr.Markdown(""" # 🔐 Sistema de revisión documentos de Acreditación de Personal Por favor, ingresa tu tu WhatsApp para recibir los resultados. """) email_input = gr.Textbox( label="WhatsApp", placeholder="Deja tu WhatsApp para recibir los resultados", type="email", value="+569" ) acceder_btn = gr.Button("Acceder al Sistema", variant="primary", size="lg") mensaje_error = gr.Markdown(visible=False) with gr.Column(visible=False) as pantalla_principal: gr.Markdown(""" # 📋 Sistema de Revisión de Documentos Sube un archivo ZIP con las carpetas de documentos para revisar. """) archivo_zip_input = gr.File( label="📦 Sube el archivo ZIP con las carpetas de documentos", file_types=[".zip"], file_count="single" ) revisar_btn = gr.Button("🔍 Iniciar revisión", variant="primary", size="lg") # Tabla de resumen por carpetas tabla_resumen = gr.Dataframe( headers=["Carpeta", "Estado", "Detalle"], datatype=["str", "str", "str"], interactive=False, wrap=True, label=f"📊 Resumen por Carpeta (Información completa en tu WhatsApp)" ) # Habilitar botón cuando se cargue el archivo archivo_zip_input.change( fn=habilitar_boton, inputs=[archivo_zip_input], outputs=revisar_btn ) def activar_carga(): return gr.Button("⏳ Procesando...", interactive=False) def desactivar_carga(): return gr.Button("🔍 Iniciar revisión", variant="primary", interactive=True) revisar_btn.click( fn=activar_carga, outputs=revisar_btn ).then( fn=revisar_zip_stream, inputs=[archivo_zip_input, email_usuario], outputs=tabla_resumen ).then( fn=desactivar_carga, outputs=revisar_btn ) acceder_btn.click( fn=validar_acceso, inputs=[email_input], outputs=[pantalla_acceso, pantalla_principal, mensaje_error, acceso_granted, email_usuario] ) if __name__ == "__main__": demo.launch(share=True)