Spaces:
Sleeping
Sleeping
| 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) |