| import os |
| import shutil |
| import time |
| from pathlib import Path |
| import ai_processor as ai |
| import database as db |
|
|
| |
| BASE_DIR = Path(os.path.dirname(os.path.abspath(__file__))) |
| INPUT_DIR = BASE_DIR / "input_files" |
| PROCESSED_DIR = BASE_DIR / "procesados" |
| FAILED_DIR = BASE_DIR / "fallidos" |
|
|
| def setup_dirs(): |
| """Crea las carpetas necesarias si no existen.""" |
| for d in [INPUT_DIR, PROCESSED_DIR, FAILED_DIR]: |
| d.mkdir(parents=True, exist_ok=True) |
| |
| |
| db.init_db() |
| print(f"Directorios listos: {INPUT_DIR.name}/, {PROCESSED_DIR.name}/, {FAILED_DIR.name}/") |
|
|
| def procesar_archivo(file_path: Path): |
| """Procesa un archivo individual y lo envía a la base de datos.""" |
| print(f"\nProcesando: {file_path.name}") |
| |
| res = ai.extraer_lista(str(file_path)) |
| |
| if not res.get("procesable"): |
| print(f"Error al procesar {file_path.name}: {res.get('error', 'Causa desconocida')}") |
| |
| dest = FAILED_DIR / file_path.name |
| shutil.move(str(file_path), str(dest)) |
| return False |
| |
| personas = res.get("personas", []) |
| if not personas: |
| print(f"No se detectaron personas legibles en {file_path.name}.") |
| dest = FAILED_DIR / file_path.name |
| shutil.move(str(file_path), str(dest)) |
| return False |
|
|
| print(f"Se extrajeron {len(personas)} persona(s). Guardando en base de datos...") |
| |
| |
| resultado_db = db.agregar_personas_bulk( |
| personas, |
| hospital_default="Lista Automatizada", |
| fuente="agente_automatico", |
| saltar_duplicados=True |
| ) |
| |
| agregados = resultado_db.get("agregados", 0) |
| omitidos = resultado_db.get("omitidos", 0) |
| print(f"Resultados: {agregados} agregados, {omitidos} omitidos (duplicados).") |
| |
| |
| dest = PROCESSED_DIR / file_path.name |
| try: |
| shutil.move(str(file_path), str(dest)) |
| print(f"Archivo movido a {PROCESSED_DIR.name}/") |
| except Exception as e: |
| print(f"Error al mover el archivo: {e}") |
| |
| return True |
|
|
| def correr_agente(modo_continuo=False): |
| setup_dirs() |
| print("Agente de extraccion iniciado.") |
| |
| while True: |
| archivos = [] |
| for ext in ai.TIPOS_IMAGEN | ai.TIPOS_PDF | ai.TIPOS_EXCEL: |
| archivos.extend(INPUT_DIR.glob(f"*{ext}")) |
| archivos.extend(INPUT_DIR.glob(f"*{ext.upper()}")) |
| |
| archivos = list(set(archivos)) |
| |
| if archivos: |
| print(f"\nEncontrados {len(archivos)} archivo(s) para procesar.") |
| for file_path in archivos: |
| procesar_archivo(file_path) |
| time.sleep(2) |
| |
| if not modo_continuo: |
| print("\nProcesamiento por lotes finalizado.") |
| break |
| |
| print("\nEsperando nuevos archivos en 'input_files/'... (Presiona Ctrl+C para salir)") |
| time.sleep(10) |
|
|
| if __name__ == "__main__": |
| import argparse |
| parser = argparse.ArgumentParser(description="Agente automático de procesamiento de listas") |
| parser.add_argument("--continuo", action="store_true", help="Mantener el agente ejecutándose esperando archivos nuevos.") |
| args = parser.parse_args() |
| |
| correr_agente(modo_continuo=args.continuo) |
|
|
