diff --git a/README.md b/README.md index 319b49f..fc89e70 100644 --- a/README.md +++ b/README.md @@ -16,9 +16,20 @@ Arquitectura de microservicios: cada servicio tiene su propia API, dependencias, | `database` | — | 5432 | PostgreSQL 16 | Solo red Docker | | `redis` | — | 6379 | Redis 7 | Solo red Docker | | `desktop_app` | local | Electron + React | Aplicación de escritorio | +| `bot-descarga` (API) | — | 8200 | Python/FastAPI | Control remoto de bots; solo red Docker | --- +## Bots externos (`bots/`) + +Agents para instalar **en PCs** (no Docker). Carpetas hermanas de `services/`. + +| Carpeta | Descripción | +|---------|-------------| +| [`bots/bot_descarga`](bots/bot_descarga) | Agent Selenium + MinIO; se conecta a la API `services/bot_descarga` vía gateway | + +Ver [`bots/README.md`](bots/README.md). + ## Flujo principal ``` @@ -105,10 +116,13 @@ Ver `.env.example` para la lista completa. ## Estructura del repositorio ``` +bots/ + bot_descarga/ — agent Selenium para PCs (instalación externa) → README services/ api_gateway/ — proxy HTTP (Python) → README extraction/ — LLM + jobs + conteo BD (Python) → README cleaning/ — limpieza de datos (Python) → README + bot_descarga/ — API control remoto de bots → README count-api/ — OCR + ONNX stateless (Go) → README db/ — migraciones Alembic → README desktop_app/ — app Electron + React → README @@ -145,3 +159,5 @@ Esquema completo: [`docs/database.dbml`](docs/database.dbml) - [`services/services-count-facturas-medicas-serving/README.md`](services/services-count-facturas-medicas-serving/README.md) - [`services/db/README.md`](services/db/README.md) - [`services/desktop_app/README.md`](services/desktop_app/README.md) +- [`services/bot_descarga/README.md`](services/bot_descarga/README.md) +- [`bots/bot_descarga/README.md`](bots/bot_descarga/README.md) diff --git a/bots/README.md b/bots/README.md new file mode 100644 index 0000000..ec48413 --- /dev/null +++ b/bots/README.md @@ -0,0 +1,9 @@ +# Bots externos + +Agents que se **instalan en PCs** (Chrome, Imaging, etc.). No van en Docker ni en `services/`. + +| Carpeta | Rol | +|---------|-----| +| [`bot_descarga/`](bot_descarga/) | Agent Selenium: descarga PDFs y sube a MinIO; habla con `services/bot_descarga` vía gateway | + +La API / persistencia correspondiente vive en `services//`. diff --git a/bots/bot_descarga/.env.example b/bots/bot_descarga/.env.example new file mode 100644 index 0000000..02a2f4e --- /dev/null +++ b/bots/bot_descarga/.env.example @@ -0,0 +1,14 @@ +# Token hacia la API IPSExtractor (mismo que BOT_API_TOKEN del servidor) +BOT_API_TOKEN=dev-token-change-me +BOT_API_URL=https://pys1.builker.xyz/bot-descarga + +# Imaging (hostname interno; requiere VPN/DNS o IP alcanzable) +BOT_IMAGING_URL=http://serviciosimaging:8180/ + +# MinIO — bucket pys-cuentas-medicas: sin credenciales (PutObject anónimo) +MINIO_ENDPOINT=https://s3.builker.xyz +MINIO_BUCKET=pys-cuentas-medicas +MINIO_ANONYMOUS=1 +MINIO_ACCESS_KEY= +MINIO_SECRET_KEY= +MINIO_REGION=us-east-1 diff --git a/bots/bot_descarga/.gitignore b/bots/bot_descarga/.gitignore new file mode 100644 index 0000000..1bbbbc1 --- /dev/null +++ b/bots/bot_descarga/.gitignore @@ -0,0 +1,18 @@ +logs/* +!logs/.gitkeep +data/ +__pycache__/ +*.pyc +*.pyo +.venv/ +venv/ +env/ +descargas/ +*.egg-info/ +.pytest_cache/ +.mypy_cache/ +dist/ +build/ +.DS_Store +Thumbs.db +.env diff --git a/bots/bot_descarga/README.md b/bots/bot_descarga/README.md new file mode 100644 index 0000000..d0893ce --- /dev/null +++ b/bots/bot_descarga/README.md @@ -0,0 +1,80 @@ +# Bot de descarga (agent para PC remota) + +Agent Selenium que corre **fuera de Docker**, en la PC con Chrome e Imaging. +Vive en este monorepo bajo `bots/` (no es un microservicio de `services/`). + +- **No expone puertos.** Solo HTTP saliente al gateway IPSExtractor y PutObject a MinIO. +- La API de control está en [`services/bot_descarga`](../../services/bot_descarga) (Docker + PostgreSQL). + +## Requisitos en la PC + +- Python 3.11+ +- Google Chrome +- Acceso de red a Imaging (`BOT_IMAGING_URL`) y al gateway (`BOT_API_URL`) +- Token `BOT_API_TOKEN` (mismo que el servicio en Docker) + +## Instalación + +```bash +cd bots/bot_descarga +python -m venv .venv + +# Windows +.venv\Scripts\activate +# Linux/macOS +# source .venv/bin/activate + +pip install -r requirements.txt +cp .env.example .env +# Editar .env +``` + +## Arranque con GUI + +```bash +python run_gui.py +``` + +Configura URL API (ej. `https://pys1.builker.xyz/bot-descarga` o `http://127.0.0.1:8189/bot-descarga`), Worker ID y token. Pulsa **Conectar agent**. Queda idle hasta que el desktop asigne un job. + +## Arranque solo agent (CLI) + +```bash +python -m bot_descarga.worker.run \ + --api-url http://127.0.0.1:8189/bot-descarga \ + --worker-id PC-01 \ + --token $BOT_API_TOKEN +``` + +## Arquitectura + +``` +Desktop IPSExtractor → Gateway :8189 → services/bot_descarga (Docker) + ↑ +PC bot (esta carpeta) ────────┘ HTTP outbound + └─ PutObject PDFs → MinIO pys-cuentas-medicas/downloads/ +``` + +## Variables + +Ver [`.env.example`](.env.example): + +| Variable | Uso | +|----------|-----| +| `BOT_API_URL` | Base del API vía gateway (`…/bot-descarga`) | +| `BOT_API_TOKEN` | Bearer compartido con el servicio | +| `BOT_IMAGING_URL` | URL del sistema Imaging (requiere VPN/DNS) | +| `MINIO_*` | Bucket de PDFs; por defecto anónimo | + +## Debug local (sin API remota) + +```bash +python main.py --excel datos.xlsx --carpeta descargas +``` + +## Tests + +```bash +cd bots/bot_descarga +python -m pytest tests/ -q +``` diff --git a/bots/bot_descarga/bot_descarga/__init__.py b/bots/bot_descarga/bot_descarga/__init__.py new file mode 100644 index 0000000..b8140b4 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/__init__.py @@ -0,0 +1,3 @@ +"""Bot de Descargas PDF — automatización de facturas desde Imaging.""" + +__version__ = "2.0.0" diff --git a/bots/bot_descarga/bot_descarga/audit/__init__.py b/bots/bot_descarga/bot_descarga/audit/__init__.py new file mode 100644 index 0000000..b6be235 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/audit/__init__.py @@ -0,0 +1,10 @@ +from bot_descarga.audit.events import BotEvent, EventType, create_event +from bot_descarga.audit.jsonl_logger import JsonlLogger, EventBus + +__all__ = [ + "BotEvent", + "EventType", + "create_event", + "JsonlLogger", + "EventBus", +] diff --git a/bots/bot_descarga/bot_descarga/audit/events.py b/bots/bot_descarga/bot_descarga/audit/events.py new file mode 100644 index 0000000..1233a7e --- /dev/null +++ b/bots/bot_descarga/bot_descarga/audit/events.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from enum import Enum +from typing import Any, Dict, Optional + + +class EventType(str, Enum): + RUN_STARTED = "run_started" + NIT_STARTED = "nit_started" + NIT_FINISHED = "nit_finished" + RADICADO_PROCESSING = "radicado_processing" + RADICADO_SKIPPED = "radicado_skipped" + RADICADO_SAVED = "radicado_saved" + RADICADO_FAILED = "radicado_failed" + BROWSER_RECONNECTED = "browser_reconnected" + RUN_FINISHED = "run_finished" + INFO = "info" + WARNING = "warning" + + +@dataclass +class BotEvent: + timestamp: str + event_type: EventType + message: str + radicado: Optional[str] = None + nit: Optional[str] = None + index: Optional[int] = None + total: Optional[int] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict: + data = asdict(self) + data["event_type"] = self.event_type.value + return data + + def display_message(self) -> str: + """Mensaje legible para consola / UI.""" + prefix = "" + if self.index is not None and self.total is not None: + prefix = f"[{self.index}/{self.total}] " + + icons = { + EventType.RUN_STARTED: "", + EventType.NIT_STARTED: "📂 ", + EventType.NIT_FINISHED: "🏁 ", + EventType.RADICADO_PROCESSING: "🔄 ", + EventType.RADICADO_SKIPPED: "✅ ", + EventType.RADICADO_SAVED: "✅ ", + EventType.RADICADO_FAILED: "❌ ", + EventType.BROWSER_RECONNECTED: "⚠️ ", + EventType.RUN_FINISHED: "", + EventType.INFO: "", + EventType.WARNING: "⚠️ ", + } + icon = icons.get(self.event_type, "") + return f"{prefix}{icon}{self.message}" + + +def create_event( + event_type: EventType, + message: str, + *, + radicado: Optional[str] = None, + nit: Optional[str] = None, + index: Optional[int] = None, + total: Optional[int] = None, + metadata: Optional[Dict[str, Any]] = None, +) -> BotEvent: + return BotEvent( + timestamp=datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"), + event_type=event_type, + message=message, + radicado=radicado, + nit=nit, + index=index, + total=total, + metadata=metadata or {}, + ) diff --git a/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py b/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py new file mode 100644 index 0000000..7bb5f8c --- /dev/null +++ b/bots/bot_descarga/bot_descarga/audit/jsonl_logger.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import json +from datetime import datetime +from pathlib import Path +from typing import Callable, List, Optional + +from bot_descarga.audit.events import BotEvent + + +class JsonlLogger: + """Escribe un evento por línea en un archivo .jsonl de auditoría.""" + + def __init__(self, logs_dir: Path | str = "logs"): + self.logs_dir = Path(logs_dir) + self.logs_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + self.path = self.logs_dir / f"audit_{stamp}.jsonl" + self._file = open(self.path, "a", encoding="utf-8") + + def write(self, event: BotEvent) -> None: + self._file.write(json.dumps(event.to_dict(), ensure_ascii=False) + "\n") + self._file.flush() + + def close(self) -> None: + if self._file and not self._file.closed: + self._file.close() + + def __enter__(self): + return self + + def __exit__(self, *args): + self.close() + + +EventCallback = Callable[[BotEvent], None] + + +class EventBus: + """Distribuye eventos a logger JSONL y callbacks (UI / consola).""" + + def __init__( + self, + jsonl_logger: Optional[JsonlLogger] = None, + callbacks: Optional[List[EventCallback]] = None, + ): + self.jsonl_logger = jsonl_logger + self.callbacks: List[EventCallback] = list(callbacks or []) + + def emit(self, event: BotEvent) -> None: + if self.jsonl_logger is not None: + self.jsonl_logger.write(event) + for cb in self.callbacks: + cb(event) + + def add_callback(self, callback: EventCallback) -> None: + self.callbacks.append(callback) + + def close(self) -> None: + if self.jsonl_logger is not None: + self.jsonl_logger.close() diff --git a/bots/bot_descarga/bot_descarga/config/__init__.py b/bots/bot_descarga/bot_descarga/config/__init__.py new file mode 100644 index 0000000..547ced9 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/config/__init__.py @@ -0,0 +1,3 @@ +from bot_descarga.config.settings import Settings, DEFAULT_SETTINGS + +__all__ = ["Settings", "DEFAULT_SETTINGS"] diff --git a/bots/bot_descarga/bot_descarga/config/settings.py b/bots/bot_descarga/bot_descarga/config/settings.py new file mode 100644 index 0000000..d79c887 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/config/settings.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path + + +@dataclass(frozen=True) +class Settings: + url_sistema: str = "http://serviciosimaging:8180/" + peso_minimo_bytes: int = 10_000 + max_intentos_click: int = 3 + max_intentos_pdf: int = 2 + timeout_descarga: int = 35 + max_reconexiones: int = 5 + wait_timeout: int = 10 + grupo: str = "CUENTAS MEDICAS" + perfil: str = "Cuentas Medicas" + logs_dir: Path = field(default_factory=lambda: Path("logs")) + + @classmethod + def from_env(cls) -> "Settings": + url = ( + os.getenv("BOT_IMAGING_URL") + or os.getenv("IMAGING_URL") + or cls.url_sistema + ).strip() + if url and not url.endswith("/"): + url = f"{url}/" + return cls(url_sistema=url or cls.url_sistema) + + +DEFAULT_SETTINGS = Settings.from_env() diff --git a/bots/bot_descarga/bot_descarga/core/__init__.py b/bots/bot_descarga/bot_descarga/core/__init__.py new file mode 100644 index 0000000..fa867a7 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/core/__init__.py @@ -0,0 +1,3 @@ +from bot_descarga.core.download_service import DownloadService + +__all__ = ["DownloadService"] diff --git a/bots/bot_descarga/bot_descarga/core/download_service.py b/bots/bot_descarga/bot_descarga/core/download_service.py new file mode 100644 index 0000000..d35169e --- /dev/null +++ b/bots/bot_descarga/bot_descarga/core/download_service.py @@ -0,0 +1,377 @@ +from __future__ import annotations + +from typing import Callable, Optional + +from bot_descarga.audit.events import EventType, create_event +from bot_descarga.audit.jsonl_logger import EventBus, JsonlLogger +from bot_descarga.config.settings import Settings, DEFAULT_SETTINGS +from bot_descarga.infrastructure.browser import BrowserManager, SESION_MUERTA +from bot_descarga.infrastructure.excel_reader import ExcelReader +from bot_descarga.infrastructure.file_manager import FileManager +from bot_descarga.infrastructure.imaging_client import ImagingClient +from bot_descarga.infrastructure.pdf_validator import PdfValidator +from bot_descarga.models.resultado import ( + DownloadResult, + NitStats, + ResultStatus, + RunSummary, +) + + +class DownloadService: + """Orquesta la descarga masiva de facturas PDF por radicado/NIT.""" + + def __init__( + self, + ruta_excel: str, + carpeta_descarga: str, + settings: Settings = DEFAULT_SETTINGS, + event_callback: Optional[Callable] = None, + should_cancel: Optional[Callable[[], bool]] = None, + on_pdf_saved: Optional[Callable[[str, str, str], None]] = None, + ): + self.ruta_excel = ruta_excel + self.carpeta_descarga = carpeta_descarga + self.settings = settings + self.should_cancel = should_cancel or (lambda: False) + self.on_pdf_saved = on_pdf_saved + + self.validator = PdfValidator(settings) + self.files = FileManager(carpeta_descarga, self.validator) + self.excel = ExcelReader() + self.browser = BrowserManager(settings) + + jsonl = JsonlLogger(settings.logs_dir) + callbacks = [event_callback] if event_callback else [] + self.bus = EventBus(jsonl_logger=jsonl, callbacks=callbacks) + + def _emit(self, event_type: EventType, message: str, **kwargs) -> None: + self.bus.emit(create_event(event_type, message, **kwargs)) + + def _log_text(self, message: str) -> None: + """Adaptador texto → evento INFO (para BrowserManager / ImagingClient).""" + self._emit(EventType.INFO, message) + + def _friendly_browser_error(self, exc: Exception) -> str: + raw = str(exc) + url = self.settings.url_sistema + if "ERR_NAME_NOT_RESOLVED" in raw: + return ( + f"No se pudo resolver el host de Imaging ({url}). " + "Revisa VPN/DNS o define BOT_IMAGING_URL. " + "Detalle: net::ERR_NAME_NOT_RESOLVED" + ) + if "ERR_CONNECTION_REFUSED" in raw or "ERR_CONNECTION_TIMED_OUT" in raw: + return f"Imaging no responde en {url}. Detalle: {raw.splitlines()[0]}" + return f"No se pudo abrir Imaging ({url}): {raw.splitlines()[0]}" + + def run(self) -> RunSummary: + summary = RunSummary() + reconexiones = 0 + nit_actual = None + + try: + filas = self.excel.read(self.ruta_excel) + summary.total = len(filas) + + if summary.total == 0: + self._emit( + EventType.RADICADO_FAILED, + "No se encontraron radicados en el Excel.", + ) + return summary + + self._emit( + EventType.RUN_STARTED, + f"Iniciando descarga de {summary.total} radicados", + total=summary.total, + metadata={ + "excel": self.ruta_excel, + "carpeta": self.carpeta_descarga, + "audit_log": str(self.bus.jsonl_logger.path), + "total": summary.total, + "url_sistema": self.settings.url_sistema, + }, + ) + + try: + driver, wait = self.browser.start( + self.carpeta_descarga, self._log_text + ) + except Exception as e: + summary.fatal_error = self._friendly_browser_error(e) + self._emit( + EventType.WARNING, + summary.fatal_error, + total=summary.total, + metadata={"motivo": str(e), "url_sistema": self.settings.url_sistema}, + ) + return summary + + imaging = ImagingClient( + driver, wait, self.settings, self.validator, self.files + ) + + idx = 0 + while idx < summary.total: + if self.should_cancel(): + summary.cancelled = True + self._emit( + EventType.WARNING, + "Cancelación solicitada — deteniendo proceso", + index=idx + 1, + total=summary.total, + ) + break + + record = filas[idx] + numero, nit = record.radicado, record.nit + + if nit not in summary.stats_por_nit: + summary.stats_por_nit[nit] = NitStats() + + stats: NitStats = summary.stats_por_nit[nit] + stats.total += 1 + + if nit != nit_actual: + if nit_actual is not None: + prev = summary.stats_por_nit[nit_actual] + self._emit( + EventType.NIT_FINISHED, + ( + f"FINALIZÓ NIT {nit_actual} | " + f"OK: {prev.ok} | " + f"Fallidos: {prev.fallidos} | " + f"Ya existían: {prev.omitidos}" + ), + nit=nit_actual, + metadata={ + "ok": prev.ok, + "fallidos": prev.fallidos, + "omitidos": prev.omitidos, + }, + ) + + nit_actual = nit + self._emit( + EventType.NIT_STARTED, + f"INICIANDO NIT {nit_actual}", + nit=nit_actual, + ) + + ruta_final = self.files.ruta_final(nit, numero) + existe, _, motivo_exist = self.files.exists_valid(nit, numero) + + if existe: + self._emit( + EventType.RADICADO_SKIPPED, + f"Ya existe: {nit}/{numero}.pdf", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + ) + summary.exitosos += 1 + stats.omitidos += 1 + reconexiones = 0 + idx += 1 + continue + + if motivo_exist not in ("no existe", "OK"): + self._emit( + EventType.WARNING, + f"Corrupto ({motivo_exist}), re-descargando: {nit}/{numero}.pdf", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + ) + + self._emit( + EventType.RADICADO_PROCESSING, + f"Procesando: {numero} | NIT: {nit}", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + ) + + if not self.browser.is_alive(): + try: + driver, wait = self.browser.reconnect( + self.carpeta_descarga, + reconexiones, + self._log_text, + ) + imaging.update_driver(driver, wait) + reconexiones += 1 + self._emit( + EventType.BROWSER_RECONNECTED, + f"Navegador reconectado (intento {reconexiones})", + metadata={"intento": reconexiones}, + ) + except RuntimeError as e: + self._emit(EventType.RADICADO_FAILED, str(e)) + break + continue + + try: + ok, motivo = imaging.search_and_download( + numero, + self.carpeta_descarga, + ruta_final, + self._log_text, + ) + + if ok: + if self.on_pdf_saved is not None: + try: + self.on_pdf_saved(nit, numero, ruta_final) + except Exception as upload_err: + self._emit( + EventType.WARNING, + f"Post-save falló: {upload_err}", + radicado=numero, + nit=nit, + ) + self._emit( + EventType.RADICADO_SAVED, + f"Guardado: {nit}/{numero}.pdf", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + metadata={"ruta": f"{nit}/{numero}.pdf"}, + ) + summary.exitosos += 1 + stats.ok += 1 + else: + self._emit( + EventType.RADICADO_FAILED, + f"{motivo}: {numero}", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + metadata={"motivo": motivo}, + ) + summary.fallidos.append( + DownloadResult( + radicado=numero, + nit=nit, + status=ResultStatus.FAILED, + motivo=motivo, + ) + ) + stats.fallidos += 1 + + reconexiones = 0 + idx += 1 + + except SESION_MUERTA as e: + self._emit( + EventType.WARNING, + f"Sesión perdida: {type(e).__name__}", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + ) + try: + driver, wait = self.browser.reconnect( + self.carpeta_descarga, + reconexiones, + self._log_text, + ) + imaging.update_driver(driver, wait) + reconexiones += 1 + self._emit( + EventType.BROWSER_RECONNECTED, + f"Navegador reconectado (intento {reconexiones})", + metadata={"intento": reconexiones}, + ) + except RuntimeError as re_err: + self._emit(EventType.RADICADO_FAILED, str(re_err)) + break + + except Exception as e: + self._emit( + EventType.RADICADO_FAILED, + f"Error con {numero}: {e}", + radicado=numero, + nit=nit, + index=idx + 1, + total=summary.total, + metadata={"motivo": str(e)}, + ) + summary.fallidos.append( + DownloadResult( + radicado=numero, + nit=nit, + status=ResultStatus.FAILED, + motivo=str(e), + ) + ) + stats.fallidos += 1 + reconexiones = 0 + idx += 1 + + if nit_actual is not None: + prev = summary.stats_por_nit[nit_actual] + self._emit( + EventType.NIT_FINISHED, + ( + f"FINALIZÓ NIT {nit_actual} | " + f"OK: {prev.ok} | " + f"Fallidos: {prev.fallidos} | " + f"Ya existían: {prev.omitidos}" + ), + nit=nit_actual, + metadata={ + "ok": prev.ok, + "fallidos": prev.fallidos, + "omitidos": prev.omitidos, + }, + ) + + finally: + self.browser.close() + + if summary.cancelled: + finish_msg = ( + f"PROCESO CANCELADO — {summary.exitosos}/{summary.total} exitosos" + ) + elif summary.fatal_error: + finish_msg = f"PROCESO FALLIDO — {summary.fatal_error}" + else: + finish_msg = ( + f"PROCESO TERMINADO — {summary.exitosos}/{summary.total} exitosos" + ) + + self._emit( + EventType.RUN_FINISHED, + finish_msg, + metadata={ + "exitosos": summary.exitosos, + "total": summary.total, + "fallidos": len(summary.fallidos), + "cancelled": summary.cancelled, + }, + ) + + if summary.fallidos: + self._emit( + EventType.INFO, + f"Fallidos ({len(summary.fallidos)}):", + ) + for item in summary.fallidos: + self._emit( + EventType.INFO, + f"NIT {item.nit} | {item.radicado}: {item.motivo}", + radicado=item.radicado, + nit=item.nit, + ) + + self.bus.close() + + return summary diff --git a/bots/bot_descarga/bot_descarga/infrastructure/__init__.py b/bots/bot_descarga/bot_descarga/infrastructure/__init__.py new file mode 100644 index 0000000..c7dad28 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/infrastructure/__init__.py @@ -0,0 +1,25 @@ +"""Infraestructura: Excel, PDF, archivos, navegador e Imaging.""" + +from bot_descarga.infrastructure.excel_reader import ExcelReader, limpiar_nombre_carpeta +from bot_descarga.infrastructure.pdf_validator import PdfValidator, esperar_pdf_completo +from bot_descarga.infrastructure.file_manager import FileManager + +__all__ = [ + "ExcelReader", + "limpiar_nombre_carpeta", + "PdfValidator", + "esperar_pdf_completo", + "FileManager", + "BrowserManager", + "ImagingClient", +] + + +def __getattr__(name: str): + if name == "BrowserManager": + from bot_descarga.infrastructure.browser import BrowserManager + return BrowserManager + if name == "ImagingClient": + from bot_descarga.infrastructure.imaging_client import ImagingClient + return ImagingClient + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/bots/bot_descarga/bot_descarga/infrastructure/browser.py b/bots/bot_descarga/bot_descarga/infrastructure/browser.py new file mode 100644 index 0000000..10a20fc --- /dev/null +++ b/bots/bot_descarga/bot_descarga/infrastructure/browser.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import time +from typing import Callable, Optional, Tuple + +from selenium import webdriver +from selenium.common.exceptions import ( + InvalidSessionIdException, + NoSuchWindowException, +) +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import Select, WebDriverWait + +from bot_descarga.config.settings import Settings, DEFAULT_SETTINGS + +SESION_MUERTA = (InvalidSessionIdException, NoSuchWindowException) + +LogFunc = Callable[[str], None] + + +class BrowserManager: + """Ciclo de vida del driver Chrome y sesión inicial en Imaging.""" + + def __init__(self, settings: Settings = DEFAULT_SETTINGS): + self.settings = settings + self.driver = None + self.wait = None + + def start(self, carpeta_descarga: str, log: LogFunc = print) -> Tuple: + options = webdriver.ChromeOptions() + + prefs = { + "printing.print_preview_sticky_settings.appState": + '{"recentDestinations":[{"id":"Save as PDF","origin":"local"}],' + '"selectedDestinationId":"Save as PDF","version":2}', + "savefile.default_directory": carpeta_descarga, + "download.default_directory": carpeta_descarga, + "download.prompt_for_download": False, + "download.directory_upgrade": True, + "plugins.always_open_pdf_externally": True, + } + + options.add_experimental_option("prefs", prefs) + options.add_argument("--kiosk-printing") + options.add_argument("--disable-infobars") + options.add_argument("--disable-extensions") + options.add_argument("--disable-gpu") + options.add_argument("--no-sandbox") + options.add_argument("--disable-dev-shm-usage") + options.add_argument("--disable-background-networking") + options.add_argument("--disable-default-apps") + options.add_argument("--disable-sync") + options.add_argument("--disable-translate") + options.add_argument("--disable-features=TranslateUI") + + self.driver = webdriver.Chrome(options=options) + self.wait = WebDriverWait(self.driver, self.settings.wait_timeout) + + self.driver.get(self.settings.url_sistema) + self._configurar_sesion(log) + + return self.driver, self.wait + + def _configurar_sesion(self, log: LogFunc) -> None: + try: + self.wait.until(EC.element_to_be_clickable((By.ID, "search"))).click() + log("Botón 'Buscar' clickeado") + except Exception as e: + log(f"No se pudo clicar 'Buscar': {e}") + + try: + grupo_el = self.wait.until( + EC.presence_of_element_located((By.ID, "groupSelect")) + ) + self.wait.until( + lambda d: len(grupo_el.find_elements(By.TAG_NAME, "option")) > 1 + ) + Select(grupo_el).select_by_visible_text(self.settings.grupo) + log(f"Grupo: {self.settings.grupo}") + except Exception as e: + log(f"Error groupSelect: {e}") + + try: + perfil_el = self.wait.until( + EC.presence_of_element_located((By.ID, "profileSelect")) + ) + self.wait.until( + lambda d: len(perfil_el.find_elements(By.TAG_NAME, "option")) > 1 + ) + Select(perfil_el).select_by_visible_text(self.settings.perfil) + log(f"Perfil: {self.settings.perfil} — esperando 2s...") + time.sleep(2) + except Exception as e: + log(f"Error profileSelect: {e}") + + def is_alive(self) -> bool: + if self.driver is None: + return False + try: + _ = self.driver.window_handles + return True + except Exception: + return False + + def reconnect( + self, + carpeta_descarga: str, + n_intento: int, + log: LogFunc = print, + ) -> Tuple: + if n_intento >= self.settings.max_reconexiones: + raise RuntimeError( + f"Chrome crasheó {n_intento} veces seguidas. Abortando." + ) + + self.close() + + espera = 3 * (n_intento + 1) + log( + f"Chrome caído — reconectando " + f"({n_intento + 1}/{self.settings.max_reconexiones}), espera {espera}s..." + ) + time.sleep(espera) + return self.start(carpeta_descarga, log) + + def close(self) -> None: + if self.driver is None: + return + try: + self.driver.quit() + except Exception: + pass + self.driver = None + self.wait = None diff --git a/bots/bot_descarga/bot_descarga/infrastructure/excel_reader.py b/bots/bot_descarga/bot_descarga/infrastructure/excel_reader.py new file mode 100644 index 0000000..9e8b8c8 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/infrastructure/excel_reader.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import re +from typing import List + +from openpyxl import load_workbook + +from bot_descarga.models.radicado import RadicadoRecord + +_HEADER_NAMES = ("radicado", "proceso", "numero", "número") + + +def limpiar_nombre_carpeta(valor) -> str: + valor = str(valor).strip() + valor = re.sub(r'[\\/:*?"<>|]', "_", valor) + return valor if valor else "SIN_NIT" + + +class ExcelReader: + """Lee radicados (col A) y NITs (col B) desde un Excel.""" + + def read(self, ruta_excel: str) -> List[RadicadoRecord]: + wb = load_workbook(ruta_excel, data_only=True) + ws = wb.active + filas: List[RadicadoRecord] = [] + + for fila in ws.iter_rows(values_only=True): + radicado = fila[0] if len(fila) > 0 else None + nit = fila[1] if len(fila) > 1 else None + + if not radicado or str(radicado).strip() in ("", "None"): + continue + + radicado = str(radicado).strip() + + if radicado.lower() in _HEADER_NAMES: + continue + + if nit is None or str(nit).strip() in ("", "None"): + nit = "SIN_NIT" + + nit = limpiar_nombre_carpeta(nit) + filas.append(RadicadoRecord(radicado=radicado, nit=nit)) + + return filas diff --git a/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py b/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py new file mode 100644 index 0000000..4ae9e25 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/infrastructure/file_manager.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import os +from typing import Tuple + +from bot_descarga.infrastructure.pdf_validator import PdfValidator + + +class FileManager: + """Gestión de carpetas por NIT y archivos PDF finales.""" + + def __init__(self, carpeta_descarga: str, validator: PdfValidator): + self.carpeta_descarga = carpeta_descarga + self.validator = validator + os.makedirs(carpeta_descarga, exist_ok=True) + + def ruta_final(self, nit: str, radicado: str) -> str: + carpeta_nit = os.path.join(self.carpeta_descarga, nit) + os.makedirs(carpeta_nit, exist_ok=True) + return os.path.join(carpeta_nit, radicado + ".pdf") + + def exists_valid(self, nit: str, radicado: str) -> Tuple[bool, str, str]: + """ + Returns (existe_y_valido, ruta_final, motivo). + Si existe pero está corrupto, lo elimina. + """ + ruta = self.ruta_final(nit, radicado) + + if not os.path.exists(ruta): + return False, ruta, "no existe" + + valido, motivo = self.validator.validate(ruta) + if valido: + return True, ruta, "OK" + + try: + os.remove(ruta) + except Exception: + pass + + return False, ruta, motivo + + def move_to_final(self, archivo_temp: str, ruta_final: str) -> None: + try: + os.replace(archivo_temp, ruta_final) + except Exception: + os.rename(archivo_temp, ruta_final) + + def remove_safe(self, ruta: str) -> None: + try: + if ruta and os.path.exists(ruta): + os.remove(ruta) + except Exception: + pass + + def list_pdfs(self) -> set: + return { + f for f in os.listdir(self.carpeta_descarga) + if f.lower().endswith(".pdf") + } diff --git a/bots/bot_descarga/bot_descarga/infrastructure/imaging_client.py b/bots/bot_descarga/bot_descarga/infrastructure/imaging_client.py new file mode 100644 index 0000000..adbfa88 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/infrastructure/imaging_client.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import os +import time +from typing import Callable, Optional, Tuple + +from selenium.webdriver.common.action_chains import ActionChains +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait + +from bot_descarga.config.settings import Settings, DEFAULT_SETTINGS +from bot_descarga.infrastructure.browser import SESION_MUERTA +from bot_descarga.infrastructure.file_manager import FileManager +from bot_descarga.infrastructure.pdf_validator import PdfValidator, esperar_pdf_completo + +LogFunc = Callable[[str], None] + + +class ImagingClient: + """Interacciones con el portal Imaging: búsqueda, factura y descarga.""" + + def __init__( + self, + driver, + wait: WebDriverWait, + settings: Settings = DEFAULT_SETTINGS, + validator: Optional[PdfValidator] = None, + files: Optional[FileManager] = None, + ): + self.driver = driver + self.wait = wait + self.settings = settings + self.validator = validator or PdfValidator(settings) + self.files = files + + def update_driver(self, driver, wait: WebDriverWait) -> None: + self.driver = driver + self.wait = wait + + def esperar_documento_listo(self, log: LogFunc = print) -> bool: + try: + WebDriverWait(self.driver, 8).until( + lambda d: d.execute_script("return document.readyState") == "complete" + ) + except Exception: + pass + + try: + WebDriverWait(self.driver, 8).until( + EC.invisibility_of_element_located((By.ID, "imagingSearchOverlay")) + ) + except Exception: + pass + + try: + WebDriverWait(self.driver, 8).until( + EC.presence_of_element_located((By.CLASS_NAME, "document")) + ) + return True + except Exception: + log("No se encontraron documentos en pantalla") + return False + + def esperar_visor_pdf(self) -> bool: + try: + WebDriverWait(self.driver, 8).until( + lambda d: d.execute_script("return document.readyState") == "complete" + ) + except Exception: + pass + time.sleep(0.8) + return True + + def buscar_factura_en_documentos(self, log: LogFunc = print): + docs = self.driver.find_elements(By.CLASS_NAME, "document") + + log("Documentos encontrados:") + for d in docs: + log(f" - {d.get_attribute('title')}") + + for d in docs: + titulo = d.get_attribute("title") or "" + titulo_limpio = titulo.lower().strip() + if "factura" in titulo_limpio and titulo_limpio.endswith(".pdf"): + return d + + return None + + def search(self, numero: str) -> None: + self.driver.switch_to.window(self.driver.window_handles[0]) + + self.wait.until(EC.element_to_be_clickable((By.ID, "search"))).click() + + campo = self.wait.until( + EC.presence_of_element_located((By.ID, "searchValue3")) + ) + campo.clear() + campo.send_keys(numero) + + self.wait.until( + EC.element_to_be_clickable((By.ID, "searchButton")) + ).click() + + def click_factura(self, factura) -> bool: + for _ in range(self.settings.max_intentos_click): + try: + ActionChains(self.driver).move_to_element(factura).click().perform() + return True + except SESION_MUERTA: + raise + except Exception: + time.sleep(0.4) + return False + + def download_pdf( + self, + carpeta_descarga: str, + ruta_final: str, + log: LogFunc = print, + ) -> Tuple[bool, str]: + """Ejecuta window.print y valida el PDF. Returns (ok, motivo).""" + for intento in range(self.settings.max_intentos_pdf): + archivos_antes = { + f for f in os.listdir(carpeta_descarga) + if f.lower().endswith(".pdf") + } + + self.driver.execute_script("window.print();") + + archivo_temp = esperar_pdf_completo( + carpeta_descarga, + archivos_antes, + timeout=self.settings.timeout_descarga, + ) + + if not archivo_temp or not os.path.exists(archivo_temp): + log( + f"Intento {intento + 1}/{self.settings.max_intentos_pdf}: sin archivo" + ) + time.sleep(0.5) + continue + + valido, motivo = self.validator.validate(archivo_temp) + + if valido: + if self.files is not None: + self.files.move_to_final(archivo_temp, ruta_final) + else: + try: + os.replace(archivo_temp, ruta_final) + except Exception: + os.rename(archivo_temp, ruta_final) + return True, "OK" + + log( + f"Intento {intento + 1}/{self.settings.max_intentos_pdf}: " + f"inválido → {motivo}" + ) + try: + os.remove(archivo_temp) + except Exception: + pass + time.sleep(0.8) + + return False, "PDF inválido repetido" + + def search_and_download( + self, + numero: str, + carpeta_descarga: str, + ruta_final: str, + log: LogFunc = print, + ) -> Tuple[bool, str]: + """ + Flujo completo: buscar → localizar factura → clic → print → guardar. + Returns (ok, motivo_error). + """ + self.search(numero) + + if not self.esperar_documento_listo(log): + return False, "Página no cargó" + + factura = self.buscar_factura_en_documentos(log) + if not factura: + return False, "Sin factura detectada" + + if not self.click_factura(factura): + return False, "Click fallido" + + log("Esperando visor...") + self.esperar_visor_pdf() + + return self.download_pdf(carpeta_descarga, ruta_final, log) diff --git a/bots/bot_descarga/bot_descarga/infrastructure/pdf_validator.py b/bots/bot_descarga/bot_descarga/infrastructure/pdf_validator.py new file mode 100644 index 0000000..7c918ae --- /dev/null +++ b/bots/bot_descarga/bot_descarga/infrastructure/pdf_validator.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import os +import time +from typing import Optional, Set, Tuple + +import pikepdf + +from bot_descarga.config.settings import Settings, DEFAULT_SETTINGS + + +def _validar_firma(ruta: str) -> bool: + try: + with open(ruta, "rb") as f: + return f.read(5) == b"%PDF-" + except Exception: + return False + + +def _validar_eof(ruta: str) -> bool: + try: + with open(ruta, "rb") as f: + f.seek(-1024, 2) + return b"%%EOF" in f.read() + except Exception: + return False + + +def _validar_pikepdf(ruta: str) -> bool: + try: + with pikepdf.open(ruta) as pdf: + _ = len(pdf.pages) + return True + except Exception: + return False + + +class PdfValidator: + def __init__(self, settings: Settings = DEFAULT_SETTINGS): + self.settings = settings + + def validate(self, ruta: str) -> Tuple[bool, str]: + try: + peso = os.path.getsize(ruta) + + if peso < self.settings.peso_minimo_bytes: + return False, f"Peso insuficiente ({peso} bytes)" + + if not _validar_firma(ruta): + return False, "Sin firma %PDF-" + + if not _validar_eof(ruta): + return False, "Sin %%EOF (truncado)" + + if not _validar_pikepdf(ruta): + return False, "pikepdf: estructura corrupta" + + return True, "OK" + + except Exception as e: + return False, str(e) + + +def esperar_pdf_completo( + carpeta: str, + archivos_antes: Set[str], + timeout: int = 35, +) -> Optional[str]: + inicio = time.time() + tamaño_anterior = -1 + estable = 0 + + while time.time() - inicio < timeout: + actuales = { + f for f in os.listdir(carpeta) + if f.lower().endswith(".pdf") + } + + nuevos = actuales - archivos_antes + + if nuevos: + ruta = os.path.join(carpeta, list(nuevos)[0]) + + if not os.path.exists(ruta): + time.sleep(0.2) + continue + + tamaño = os.path.getsize(ruta) + + if tamaño == tamaño_anterior and tamaño > 1000: + estable += 1 + else: + estable = 0 + + tamaño_anterior = tamaño + + if estable >= 2: + return ruta + + time.sleep(0.2) + + return None diff --git a/bots/bot_descarga/bot_descarga/models/__init__.py b/bots/bot_descarga/bot_descarga/models/__init__.py new file mode 100644 index 0000000..b4c653b --- /dev/null +++ b/bots/bot_descarga/bot_descarga/models/__init__.py @@ -0,0 +1,4 @@ +from bot_descarga.models.radicado import RadicadoRecord +from bot_descarga.models.resultado import DownloadResult, NitStats, RunSummary + +__all__ = ["RadicadoRecord", "DownloadResult", "NitStats", "RunSummary"] diff --git a/bots/bot_descarga/bot_descarga/models/radicado.py b/bots/bot_descarga/bot_descarga/models/radicado.py new file mode 100644 index 0000000..c31db63 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/models/radicado.py @@ -0,0 +1,7 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True) +class RadicadoRecord: + radicado: str + nit: str diff --git a/bots/bot_descarga/bot_descarga/models/resultado.py b/bots/bot_descarga/bot_descarga/models/resultado.py new file mode 100644 index 0000000..690e2a6 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/models/resultado.py @@ -0,0 +1,36 @@ +from dataclasses import dataclass, field +from enum import Enum +from typing import List, Optional + + +class ResultStatus(str, Enum): + SAVED = "saved" + SKIPPED = "skipped" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass +class DownloadResult: + radicado: str + nit: str + status: ResultStatus + motivo: str = "" + + +@dataclass +class NitStats: + total: int = 0 + ok: int = 0 + fallidos: int = 0 + omitidos: int = 0 + + +@dataclass +class RunSummary: + total: int = 0 + exitosos: int = 0 + cancelled: bool = False + fatal_error: Optional[str] = None + fallidos: List[DownloadResult] = field(default_factory=list) + stats_por_nit: dict = field(default_factory=dict) diff --git a/bots/bot_descarga/bot_descarga/ui/__init__.py b/bots/bot_descarga/bot_descarga/ui/__init__.py new file mode 100644 index 0000000..ef6ac4f --- /dev/null +++ b/bots/bot_descarga/bot_descarga/ui/__init__.py @@ -0,0 +1,4 @@ +from bot_descarga.ui.dashboard import BotDashboard +from bot_descarga.ui.settings_store import DEFAULT_API_URL, normalize_api_url + +__all__ = ["BotDashboard", "DEFAULT_API_URL", "normalize_api_url"] diff --git a/bots/bot_descarga/bot_descarga/ui/dashboard.py b/bots/bot_descarga/bot_descarga/ui/dashboard.py new file mode 100644 index 0000000..3e5e21e --- /dev/null +++ b/bots/bot_descarga/bot_descarga/ui/dashboard.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +import socket +import threading + +import customtkinter as ctk + +from bot_descarga.ui.settings_store import ( + DEFAULT_API_URL, + load_gui_settings, + normalize_api_url, + save_gui_settings, +) +from bot_descarga.worker.agent import WorkerAgent + +ctk.set_appearance_mode("dark") +ctk.set_default_color_theme("blue") + + +class BotDashboard(ctk.CTk): + """GUI del agent: configura API remota y conecta como worker idle.""" + + def __init__(self): + super().__init__() + + self.title("Bot de Descargas PDF — Agent") + self.geometry("920x680") + self.resizable(False, False) + + self._agent: WorkerAgent | None = None + self._agent_thread: threading.Thread | None = None + self._connected = False + + self._construir_ui() + self._cargar_settings() + + self.protocol("WM_DELETE_WINDOW", self._on_close) + + def _construir_ui(self): + root = ctk.CTkFrame(self, corner_radius=18) + root.pack(fill="both", expand=True, padx=18, pady=18) + + ctk.CTkLabel( + root, + text="Bot de Descargas PDF", + font=ctk.CTkFont(size=26, weight="bold"), + ).pack(pady=(18, 4)) + + ctk.CTkLabel( + root, + text="Agent remoto — idle hasta que el desktop asigne un Excel a este bot", + text_color="gray", + font=ctk.CTkFont(size=13), + ).pack(pady=(0, 18)) + + config = ctk.CTkFrame(root, corner_radius=14) + config.pack(fill="x", padx=24, pady=(0, 14)) + + ctk.CTkLabel( + config, + text="API remota (IPSExtractor)", + font=ctk.CTkFont(size=16, weight="bold"), + ).pack(anchor="w", padx=18, pady=(14, 8)) + + row_api = ctk.CTkFrame(config, fg_color="transparent") + row_api.pack(fill="x", padx=18, pady=(0, 8)) + ctk.CTkLabel(row_api, text="URL API", width=90, anchor="w").pack(side="left") + self.entry_api = ctk.CTkEntry( + row_api, + placeholder_text=DEFAULT_API_URL, + ) + self.entry_api.pack(side="left", fill="x", expand=True) + + row_id = ctk.CTkFrame(config, fg_color="transparent") + row_id.pack(fill="x", padx=18, pady=(0, 8)) + ctk.CTkLabel(row_id, text="Worker ID", width=90, anchor="w").pack(side="left") + self.entry_worker = ctk.CTkEntry( + row_id, + placeholder_text=socket.gethostname() or "PC-01", + ) + self.entry_worker.pack(side="left", fill="x", expand=True) + + row_token = ctk.CTkFrame(config, fg_color="transparent") + row_token.pack(fill="x", padx=18, pady=(0, 14)) + ctk.CTkLabel(row_token, text="Token", width=90, anchor="w").pack(side="left") + self.entry_token = ctk.CTkEntry( + row_token, + placeholder_text="BOT_API_TOKEN", + show="•", + ) + self.entry_token.pack(side="left", fill="x", expand=True, padx=(0, 8)) + self._token_visible = False + ctk.CTkButton( + row_token, + text="👁", + width=40, + command=self._toggle_token, + ).pack(side="left") + + status = ctk.CTkFrame(root, corner_radius=14) + status.pack(fill="x", padx=24, pady=(0, 14)) + + self.lbl_estado = ctk.CTkLabel( + status, + text="Estado: desconectado", + font=ctk.CTkFont(size=15, weight="bold"), + ) + self.lbl_estado.pack(anchor="w", padx=18, pady=(14, 2)) + + self.lbl_hint = ctk.CTkLabel( + status, + text="Configura la API y pulsa Conectar. Este PC no expone puertos.", + text_color="gray", + font=ctk.CTkFont(size=13), + ) + self.lbl_hint.pack(anchor="w", padx=18, pady=(0, 14)) + + botones = ctk.CTkFrame(root, fg_color="transparent") + botones.pack(fill="x", padx=24, pady=(0, 14)) + + self.btn_connect = ctk.CTkButton( + botones, + text="▶ Conectar agent", + height=42, + width=180, + font=ctk.CTkFont(size=14, weight="bold"), + command=self._toggle_connect, + ) + self.btn_connect.pack(side="left", padx=(0, 10)) + + ctk.CTkButton( + botones, + text="Guardar config", + height=42, + width=140, + command=self._guardar_settings, + ).pack(side="left", padx=(0, 10)) + + ctk.CTkButton( + botones, + text="Limpiar logs", + height=42, + width=130, + command=self._limpiar_logs, + ).pack(side="left") + + logs_frame = ctk.CTkFrame(root, corner_radius=14) + logs_frame.pack(fill="both", expand=True, padx=24, pady=(0, 18)) + + ctk.CTkLabel( + logs_frame, + text="Detalle del agent", + font=ctk.CTkFont(size=15, weight="bold"), + ).pack(anchor="w", padx=18, pady=(12, 6)) + + self.text_logs = ctk.CTkTextbox( + logs_frame, font=ctk.CTkFont(family="Courier", size=12) + ) + self.text_logs.pack(fill="both", expand=True, padx=18, pady=(0, 16)) + + def _cargar_settings(self): + s = load_gui_settings() + self.entry_api.insert(0, s.get("api_url") or DEFAULT_API_URL) + self.entry_worker.insert(0, s.get("worker_id") or socket.gethostname()) + self.entry_token.insert(0, s.get("token") or "") + + def _guardar_settings(self): + api = normalize_api_url(self.entry_api.get()) + self.entry_api.delete(0, "end") + self.entry_api.insert(0, api) + save_gui_settings(api, self.entry_worker.get(), self.entry_token.get()) + self._log("Configuración guardada en data/gui_settings.json") + + def _toggle_token(self): + self._token_visible = not self._token_visible + self.entry_token.configure(show="" if self._token_visible else "•") + + def _toggle_connect(self): + if self._connected: + self._desconectar() + else: + self._conectar() + + def _conectar(self): + api = normalize_api_url(self.entry_api.get()) + worker_id = self.entry_worker.get().strip() + token = self.entry_token.get().strip() + + if not worker_id: + self._log("Debes indicar un Worker ID.") + return + if not token: + self._log("Debes indicar el token (BOT_API_TOKEN).") + return + + self.entry_api.delete(0, "end") + self.entry_api.insert(0, api) + save_gui_settings(api, worker_id, token) + + self._agent = WorkerAgent( + api_url=api, + worker_id=worker_id, + token=token, + log_func=self._log, + ) + self._connected = True + self.btn_connect.configure(text="⏹ Desconectar", fg_color="#c0392b", hover_color="#a93226") + self.lbl_estado.configure(text=f"Estado: conectado como {worker_id}") + self.lbl_hint.configure(text=f"API: {api} — esperando jobs del desktop…") + self._set_config_enabled(False) + + self._agent_thread = threading.Thread( + target=self._run_agent, + daemon=True, + ) + self._agent_thread.start() + self._log(f"Conectando a {api}…") + + def _run_agent(self): + try: + assert self._agent is not None + self._agent.run_forever() + except Exception as e: + self._log(f"Error del agent: {e}") + finally: + self.after(0, self._on_agent_stopped) + + def _desconectar(self): + if self._agent: + self._agent.stop() + self._log("Solicitando desconexión…") + self.btn_connect.configure(state="disabled", text="Desconectando…") + + def _on_agent_stopped(self): + self._connected = False + self._agent = None + self.btn_connect.configure( + state="normal", + text="▶ Conectar agent", + fg_color=["#3B8ED0", "#1F6AA5"], + hover_color=["#36719F", "#144870"], + ) + self.lbl_estado.configure(text="Estado: desconectado") + self.lbl_hint.configure( + text="Configura la API y pulsa Conectar. Este PC no expone puertos." + ) + self._set_config_enabled(True) + + def _set_config_enabled(self, enabled: bool): + state = "normal" if enabled else "disabled" + self.entry_api.configure(state=state) + self.entry_worker.configure(state=state) + self.entry_token.configure(state=state) + + def _limpiar_logs(self): + self.text_logs.delete("1.0", "end") + + def _log(self, mensaje: str): + def _escribir(): + self.text_logs.insert("end", mensaje + "\n") + self.text_logs.see("end") + + self.after(0, _escribir) + + def _on_close(self): + if self._agent: + self._agent.stop() + self.destroy() diff --git a/bots/bot_descarga/bot_descarga/ui/progress_tracker.py b/bots/bot_descarga/bot_descarga/ui/progress_tracker.py new file mode 100644 index 0000000..f30e975 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/ui/progress_tracker.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from bot_descarga.audit.events import BotEvent, EventType + + +class ProgressTracker: + """Traduce BotEvent a contadores de progreso para la UI.""" + + def __init__(self): + self.total = 0 + self.procesados = 0 + self.exitosos = 0 + self.fallidos = 0 + self.skipped = 0 + self.nit_actual = "—" + self.nit_estado = "Esperando inicio del proceso..." + + def reset(self) -> None: + self.total = 0 + self.procesados = 0 + self.exitosos = 0 + self.fallidos = 0 + self.skipped = 0 + self.nit_actual = "—" + self.nit_estado = "Esperando inicio del proceso..." + + def handle(self, event: BotEvent) -> None: + if event.index is not None and event.total is not None: + self.procesados = event.index + self.total = event.total + + if event.event_type == EventType.RUN_STARTED: + if "total" in event.metadata: + self.total = int(event.metadata["total"]) + elif event.total is not None: + self.total = event.total + + elif event.event_type == EventType.NIT_STARTED: + self.nit_actual = event.nit or "—" + self.nit_estado = "Procesando radicados de este NIT..." + + elif event.event_type == EventType.NIT_FINISHED: + nit = event.nit or "" + meta = event.metadata + self.nit_estado = ( + f"Finalizó NIT {nit} | " + f"OK: {meta.get('ok', 0)} | " + f"Fallidos: {meta.get('fallidos', 0)} | " + f"Ya existían: {meta.get('omitidos', 0)}" + ) + + elif event.event_type == EventType.RADICADO_SAVED: + self.exitosos += 1 + + elif event.event_type == EventType.RADICADO_SKIPPED: + self.skipped += 1 + + elif event.event_type == EventType.RADICADO_FAILED: + if "PROCESO TERMINADO" not in event.message: + self.fallidos += 1 + + @property + def percent(self) -> float: + total = self.total if self.total > 0 else 1 + return min(self.procesados / total, 1.0) diff --git a/bots/bot_descarga/bot_descarga/ui/settings_store.py b/bots/bot_descarga/bot_descarga/ui/settings_store.py new file mode 100644 index 0000000..01372db --- /dev/null +++ b/bots/bot_descarga/bot_descarga/ui/settings_store.py @@ -0,0 +1,59 @@ +"""Persistencia simple de configuración de la GUI del worker.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any, Dict + +DEFAULT_API_URL = os.getenv( + "BOT_API_URL", + "https://pys1.builker.xyz/bot-descarga", +) +DEFAULT_TOKEN = os.getenv("BOT_API_TOKEN", "dev-token-change-me") + +_SETTINGS_PATH = Path("data") / "gui_settings.json" + + +def load_gui_settings() -> Dict[str, Any]: + defaults = { + "api_url": DEFAULT_API_URL, + "worker_id": "", + "token": DEFAULT_TOKEN, + } + if not _SETTINGS_PATH.exists(): + return defaults + try: + data = json.loads(_SETTINGS_PATH.read_text(encoding="utf-8")) + if not isinstance(data, dict): + return defaults + return {**defaults, **data} + except Exception: + return defaults + + +def save_gui_settings(api_url: str, worker_id: str, token: str) -> None: + _SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True) + payload = { + "api_url": api_url.strip(), + "worker_id": worker_id.strip(), + "token": token.strip(), + } + _SETTINGS_PATH.write_text( + json.dumps(payload, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + +def normalize_api_url(url: str) -> str: + """Normaliza la URL base del API bot-descarga.""" + u = (url or "").strip().rstrip("/") + if not u: + return DEFAULT_API_URL.rstrip("/") + # Si el usuario pone solo el host del gateway, añade el prefijo + if u.endswith("/bot-descarga"): + return u + if "builker.xyz" in u or u.endswith(":8189"): + return f"{u}/bot-descarga" + return u diff --git a/bots/bot_descarga/bot_descarga/worker/__init__.py b/bots/bot_descarga/bot_descarga/worker/__init__.py new file mode 100644 index 0000000..7f1e43e --- /dev/null +++ b/bots/bot_descarga/bot_descarga/worker/__init__.py @@ -0,0 +1,3 @@ +from bot_descarga.worker.agent import WorkerAgent + +__all__ = ["WorkerAgent"] diff --git a/bots/bot_descarga/bot_descarga/worker/__main__.py b/bots/bot_descarga/bot_descarga/worker/__main__.py new file mode 100644 index 0000000..0f35c38 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/worker/__main__.py @@ -0,0 +1,6 @@ +"""Permite: python -m bot_descarga.worker.run""" + +from bot_descarga.worker.run import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/bots/bot_descarga/bot_descarga/worker/agent.py b/bots/bot_descarga/bot_descarga/worker/agent.py new file mode 100644 index 0000000..7edc9a0 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/worker/agent.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +import socket +import tempfile +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +import httpx + +from bot_descarga.core.download_service import DownloadService +from bot_descarga.worker.minio_upload import MinioSettings, WorkerMinioUploader + +JOB_COMPLETED = "completed" +JOB_FAILED = "failed" +JOB_CANCELLED = "cancelled" + + +class RemoteApiClient: + """Cliente HTTP hacia la API en IPSExtractor (vía gateway). Sin puertos locales.""" + + def __init__(self, base_url: str, token: str): + self.base_url = base_url.rstrip("/") + self.token = token + self._client = httpx.Client( + timeout=httpx.Timeout(60.0, connect=10.0), + headers={"Authorization": f"Bearer {token}"}, + ) + + def close(self) -> None: + self._client.close() + + def _url(self, path: str) -> str: + return f"{self.base_url}{path}" + + def register(self, worker_id: str, hostname: str = "", max_concurrent: int = 1) -> dict: + r = self._client.post( + self._url("/workers/register"), + json={ + "worker_id": worker_id, + "hostname": hostname or socket.gethostname(), + "max_concurrent": max_concurrent, + }, + ) + r.raise_for_status() + return r.json() + + def heartbeat(self, worker_id: str) -> dict: + r = self._client.post( + self._url("/workers/heartbeat"), + json={"worker_id": worker_id}, + ) + r.raise_for_status() + return r.json() + + def claim_pending(self, worker_id: str) -> Optional[dict]: + r = self._client.get( + self._url("/workers/me/pending"), + params={"worker_id": worker_id}, + ) + r.raise_for_status() + data = r.json() + return data.get("job") + + def download_excel(self, job_id: str, dest: Path) -> Path: + r = self._client.get(self._url(f"/jobs/{job_id}/file")) + r.raise_for_status() + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(r.content) + return dest + + def post_event(self, job_id: str, event: Dict[str, Any]) -> None: + r = self._client.post(self._url(f"/jobs/{job_id}/events"), json=event) + r.raise_for_status() + + def cancel_flag(self, job_id: str) -> bool: + r = self._client.get(self._url(f"/jobs/{job_id}/cancel-flag")) + r.raise_for_status() + return bool(r.json().get("cancel_requested")) + + def patch_job(self, job_id: str, **body) -> dict: + r = self._client.patch(self._url(f"/jobs/{job_id}"), json=body) + r.raise_for_status() + return r.json() + + +class WorkerAgent: + """Agent idle: heartbeat + claim solo jobs asignados por el usuario.""" + + def __init__( + self, + api_url: str, + worker_id: str, + token: str, + poll_seconds: float = 8.0, + staging_root: Optional[Path] = None, + log_func=None, + ): + self.api = RemoteApiClient(api_url, token) + self.worker_id = worker_id + self.poll_seconds = poll_seconds + self.staging_root = staging_root or Path(tempfile.gettempdir()) / "bot_descarga" + self.minio = WorkerMinioUploader(MinioSettings.from_env()) + self._running = True + self._uploaded_keys: List[str] = [] + self._log = log_func or print + + def stop(self) -> None: + self._running = False + + def run_forever(self) -> None: + self.api.register(self.worker_id) + self._log(f"[worker {self.worker_id}] registrado — idle (esperando jobs manuales)") + try: + while self._running: + try: + self.api.heartbeat(self.worker_id) + job = self.api.claim_pending(self.worker_id) + if job: + self._run_job(job) + else: + time.sleep(self.poll_seconds) + except KeyboardInterrupt: + raise + except Exception as e: + self._log(f"[worker {self.worker_id}] error en loop: {e}") + time.sleep(self.poll_seconds) + finally: + self.api.close() + self._log(f"[worker {self.worker_id}] detenido") + + def _run_job(self, job: dict) -> None: + job_id = job["id"] + self._uploaded_keys = [] + work_dir = self.staging_root / job_id + out_dir = work_dir / "out" + out_dir.mkdir(parents=True, exist_ok=True) + excel_path = work_dir / (Path(job.get("excel_key") or "input.xlsx").name) + + self._log(f"[worker {self.worker_id}] job {job_id} iniciado") + + try: + self.api.download_excel(job_id, excel_path) + + def should_cancel() -> bool: + try: + return self.api.cancel_flag(job_id) + except Exception: + return False + + def on_event(event) -> None: + self._log(event.display_message()) + payload = { + "event_type": event.event_type.value + if hasattr(event.event_type, "value") + else str(event.event_type), + "message": event.message, + "radicado": event.radicado, + "nit": event.nit, + "index": event.index, + "total": event.total, + "metadata": event.metadata, + "timestamp": event.timestamp, + } + try: + self.api.post_event(job_id, payload) + except Exception as e: + self._log(f"[worker] no se pudo enviar evento: {e}") + + def on_pdf_saved(nit: str, radicado: str, local_path: str) -> None: + key = self.minio.upload_pdf(job_id, nit, radicado, local_path) + self._uploaded_keys.append(key) + + service = DownloadService( + ruta_excel=str(excel_path), + carpeta_descarga=str(out_dir), + event_callback=on_event, + should_cancel=should_cancel, + on_pdf_saved=on_pdf_saved, + ) + summary = service.run() + + try: + self.minio.write_manifest(job_id, self._uploaded_keys) + except Exception as e: + self._log(f"[worker] manifest no escrito: {e}") + + fallidos_count = len(summary.fallidos) + if summary.fatal_error and fallidos_count == 0 and summary.total: + fallidos_count = summary.total - summary.exitosos + + result = { + "total": summary.total, + "exitosos": summary.exitosos, + "fallidos": fallidos_count, + "cancelled": summary.cancelled, + "downloads_prefix": job.get("downloads_prefix"), + "uploaded_count": len(self._uploaded_keys), + } + + if summary.cancelled: + status = JOB_CANCELLED + error_msg = None + elif summary.fatal_error: + status = JOB_FAILED + error_msg = summary.fatal_error + elif summary.fallidos and summary.exitosos == 0: + status = JOB_FAILED + error_msg = "Todos los radicados fallaron" + else: + status = JOB_COMPLETED + error_msg = None + + if error_msg: + try: + self.api.post_event( + job_id, + { + "event_type": "warning", + "message": error_msg, + "metadata": {"fatal": bool(summary.fatal_error)}, + }, + ) + except Exception: + pass + + self.api.patch_job( + job_id, + status=status, + result=result, + error=error_msg, + ) + self._log(f"[worker {self.worker_id}] job {job_id} → {status}") + + except Exception as e: + self._log(f"[worker {self.worker_id}] job {job_id} error: {e}") + try: + self.api.post_event( + job_id, + { + "event_type": "warning", + "message": f"Error fatal: {e}", + "metadata": {"fatal": True}, + }, + ) + except Exception: + pass + try: + self.api.patch_job( + job_id, + status=JOB_FAILED, + error=str(e), + ) + except Exception: + pass diff --git a/bots/bot_descarga/bot_descarga/worker/minio_upload.py b/bots/bot_descarga/bot_descarga/worker/minio_upload.py new file mode 100644 index 0000000..9dc5fe6 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/worker/minio_upload.py @@ -0,0 +1,113 @@ +"""Cliente MinIO mínimo para subir PDFs desde el worker (solo salida).""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from pathlib import Path +from typing import List, Optional + + +@dataclass(frozen=True) +class MinioSettings: + endpoint: str = "https://s3.builker.xyz" + bucket: str = "pys-cuentas-medicas" + access_key: str = "" + secret_key: str = "" + region: str = "us-east-1" + anonymous: bool = True + + @classmethod + def from_env(cls) -> "MinioSettings": + access_key = (os.getenv("MINIO_ACCESS_KEY") or "").strip() + secret_key = (os.getenv("MINIO_SECRET_KEY") or "").strip() + # pys-cuentas-medicas suele ser anónimo; solo firma si hay ambas keys. + anon_flag = (os.getenv("MINIO_ANONYMOUS") or "").strip().lower() + if anon_flag in ("0", "false", "no"): + anonymous = False + elif anon_flag in ("1", "true", "yes"): + anonymous = True + else: + anonymous = not (access_key and secret_key) + return cls( + endpoint=os.getenv("MINIO_ENDPOINT", "https://s3.builker.xyz").rstrip("/"), + bucket=os.getenv("MINIO_BUCKET", "pys-cuentas-medicas"), + access_key=access_key, + secret_key=secret_key, + region=os.getenv("MINIO_REGION", "us-east-1"), + anonymous=anonymous, + ) + + +def pdf_key(job_id: str, nit: str, radicado: str) -> str: + return f"downloads/{job_id}/{nit}/{radicado}.pdf" + + +def manifest_key(job_id: str) -> str: + return f"downloads/{job_id}/_manifest.json" + + +def _s3_client(settings: MinioSettings): + import boto3 + from botocore import UNSIGNED + from botocore.client import Config + + s3_cfg = { + "addressing_style": "path", + # Evita aws-chunked / checksums que rompen PutObject anónimo en MinIO. + "request_checksum_calculation": "when_required", + "response_checksum_validation": "when_required", + } + if settings.anonymous: + return boto3.client( + "s3", + endpoint_url=settings.endpoint, + region_name=settings.region, + config=Config(signature_version=UNSIGNED, s3=s3_cfg), + ) + return boto3.client( + "s3", + endpoint_url=settings.endpoint, + aws_access_key_id=settings.access_key, + aws_secret_access_key=settings.secret_key, + region_name=settings.region, + config=Config(signature_version="s3v4", s3=s3_cfg), + ) + + +class WorkerMinioUploader: + """Solo PutObject de PDFs / manifest (sin API ni SQLite).""" + + def __init__(self, settings: Optional[MinioSettings] = None): + self.settings = settings or MinioSettings.from_env() + self._client = None + + @property + def client(self): + if self._client is None: + self._client = _s3_client(self.settings) + return self._client + + def upload_pdf(self, job_id: str, nit: str, radicado: str, local_path: str | Path) -> str: + key = pdf_key(job_id, nit, radicado) + path = Path(local_path) + # put_object (no multipart) es más fiable en buckets anónimos. + self.client.put_object( + Bucket=self.settings.bucket, + Key=key, + Body=path.read_bytes(), + ContentType="application/pdf", + ) + return key + + def write_manifest(self, job_id: str, keys: List[str]) -> str: + key = manifest_key(job_id) + payload = {"job_id": job_id, "files": keys, "count": len(keys)} + self.client.put_object( + Bucket=self.settings.bucket, + Key=key, + Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + ContentType="application/json", + ) + return key diff --git a/bots/bot_descarga/bot_descarga/worker/run.py b/bots/bot_descarga/bot_descarga/worker/run.py new file mode 100644 index 0000000..699b9c5 --- /dev/null +++ b/bots/bot_descarga/bot_descarga/worker/run.py @@ -0,0 +1,57 @@ +"""CLI del agent worker: idle hasta que el usuario asigne un job.""" + +from __future__ import annotations + +import argparse +import os +import sys + +from bot_descarga.worker.agent import WorkerAgent + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + description="Agent BotDescarga: registra el PC y espera jobs asignados manualmente." + ) + p.add_argument( + "--api-url", + default=os.getenv("BOT_API_URL", "https://pys1.builker.xyz/bot-descarga"), + help="URL del API vía gateway (default: https://pys1.builker.xyz/bot-descarga)", + ) + p.add_argument( + "--worker-id", + required=True, + help="Identificador único de este PC (ej. PC-CONTABILIDAD-01)", + ) + p.add_argument( + "--token", + default=os.getenv("BOT_API_TOKEN", "dev-token-change-me"), + help="Bearer token (BOT_API_TOKEN)", + ) + p.add_argument( + "--poll", + type=float, + default=8.0, + help="Segundos entre heartbeat/claim", + ) + return p + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + agent = WorkerAgent( + api_url=args.api_url, + worker_id=args.worker_id, + token=args.token, + poll_seconds=args.poll, + ) + try: + agent.run_forever() + except KeyboardInterrupt: + print("\nDetenido por usuario") + agent.stop() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bots/bot_descarga/main.py b/bots/bot_descarga/main.py new file mode 100644 index 0000000..946d77f --- /dev/null +++ b/bots/bot_descarga/main.py @@ -0,0 +1,58 @@ +"""CLI para el Bot de Descargas PDF.""" + +from __future__ import annotations + +import argparse +import sys + +from bot_descarga.audit.events import BotEvent +from bot_descarga.config.settings import DEFAULT_SETTINGS +from bot_descarga.core.download_service import DownloadService + + +def _print_event(event: BotEvent) -> None: + print(event.display_message(), flush=True) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Descarga facturas PDF desde Imaging y las organiza por NIT." + ) + parser.add_argument( + "--excel", + required=True, + help="Ruta al archivo Excel (col A=radicado, col B=NIT)", + ) + parser.add_argument( + "--carpeta", + required=True, + help="Carpeta destino de las descargas", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + + service = DownloadService( + ruta_excel=args.excel, + carpeta_descarga=args.carpeta, + settings=DEFAULT_SETTINGS, + event_callback=_print_event, + ) + summary = service.run() + + audit_path = getattr(service.bus.jsonl_logger, "path", None) + if audit_path: + print(f"Auditoría JSONL: {audit_path}", flush=True) + + print( + f"Resumen: {summary.exitosos}/{summary.total} exitosos, " + f"{len(summary.fallidos)} fallidos", + flush=True, + ) + return 0 if not summary.fallidos else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bots/bot_descarga/requirements.txt b/bots/bot_descarga/requirements.txt new file mode 100644 index 0000000..a0d49c6 --- /dev/null +++ b/bots/bot_descarga/requirements.txt @@ -0,0 +1,7 @@ +selenium>=4.15.0 +openpyxl>=3.1.0 +pikepdf>=8.0.0 +httpx>=0.27.0 +boto3>=1.34.0 +customtkinter>=5.2.0 +pytest>=7.0.0 diff --git a/bots/bot_descarga/run_gui.py b/bots/bot_descarga/run_gui.py new file mode 100644 index 0000000..e748f30 --- /dev/null +++ b/bots/bot_descarga/run_gui.py @@ -0,0 +1,12 @@ +"""Punto de entrada de la interfaz gráfica.""" + +from bot_descarga.ui.dashboard import BotDashboard + + +def main(): + app = BotDashboard() + app.mainloop() + + +if __name__ == "__main__": + main() diff --git a/bots/bot_descarga/tests/__init__.py b/bots/bot_descarga/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bots/bot_descarga/tests/test_browser_fatal.py b/bots/bot_descarga/tests/test_browser_fatal.py new file mode 100644 index 0000000..74605f6 --- /dev/null +++ b/bots/bot_descarga/tests/test_browser_fatal.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from bot_descarga.config.settings import Settings +from bot_descarga.core.download_service import DownloadService +from bot_descarga.models.resultado import RunSummary + + +def test_friendly_dns_error_message(): + svc = DownloadService(ruta_excel="x.xlsx", carpeta_descarga=".", event_callback=lambda e: None) + msg = svc._friendly_browser_error( + Exception("unknown error: net::ERR_NAME_NOT_RESOLVED\n (Session info: chrome=1)") + ) + assert "ERR_NAME_NOT_RESOLVED" in msg + assert "Imaging" in msg + assert svc.settings.url_sistema in msg + + +def test_browser_start_failure_sets_fatal_error(monkeypatch): + events = [] + + def fake_read(_self, _path): + from bot_descarga.models.radicado import RadicadoRecord + + return [RadicadoRecord("1", "NIT1")] + + class BoomBrowser: + def start(self, *a, **k): + raise Exception("unknown error: net::ERR_NAME_NOT_RESOLVED") + + def close(self): + pass + + monkeypatch.setattr( + "bot_descarga.core.download_service.ExcelReader.read", + fake_read, + ) + monkeypatch.setattr( + "bot_descarga.core.download_service.BrowserManager", + lambda settings=None: BoomBrowser(), + ) + monkeypatch.setattr( + "bot_descarga.core.download_service.JsonlLogger", + lambda *a, **k: type( + "L", + (), + {"path": "x", "write": lambda *x: None, "close": lambda *x: None}, + )(), + ) + + svc = DownloadService( + ruta_excel="dummy.xlsx", + carpeta_descarga=".", + settings=Settings(url_sistema="http://serviciosimaging:8180/"), + event_callback=lambda e: events.append(e), + ) + summary = svc.run() + assert isinstance(summary, RunSummary) + assert summary.fatal_error + assert "ERR_NAME_NOT_RESOLVED" in summary.fatal_error + assert any("FALLIDO" in e.message or "ERR_NAME_NOT_RESOLVED" in e.message for e in events) diff --git a/bots/bot_descarga/tests/test_download_cancel.py b/bots/bot_descarga/tests/test_download_cancel.py new file mode 100644 index 0000000..8ad2adc --- /dev/null +++ b/bots/bot_descarga/tests/test_download_cancel.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from bot_descarga.models.resultado import RunSummary +from bot_descarga.core.download_service import DownloadService + + +def test_download_service_accepts_should_cancel(monkeypatch): + """Verifica que should_cancel se honra antes de procesar radicados.""" + calls = {"cancel_checks": 0} + + def fake_read(_self, _path): + from bot_descarga.models.radicado import RadicadoRecord + return [ + RadicadoRecord("1", "NIT1"), + RadicadoRecord("2", "NIT1"), + ] + + class FakeBrowser: + def start(self, *a, **k): + return object(), object() + + def is_alive(self): + return True + + def close(self): + pass + + def reconnect(self, *a, **k): + return object(), object() + + monkeypatch.setattr( + "bot_descarga.core.download_service.ExcelReader.read", + fake_read, + ) + monkeypatch.setattr( + "bot_descarga.core.download_service.BrowserManager", + lambda settings=None: FakeBrowser(), + ) + monkeypatch.setattr( + "bot_descarga.core.download_service.ImagingClient", + lambda *a, **k: object(), + ) + monkeypatch.setattr( + "bot_descarga.core.download_service.JsonlLogger", + lambda *a, **k: type("L", (), {"path": "x", "write": lambda *x: None, "close": lambda *x: None})(), + ) + + def should_cancel(): + calls["cancel_checks"] += 1 + return True + + svc = DownloadService( + ruta_excel="dummy.xlsx", + carpeta_descarga=".", + should_cancel=should_cancel, + event_callback=lambda e: None, + ) + # Evitar crear carpeta real problemática: FileManager.mkdirs ok + summary = svc.run() + assert isinstance(summary, RunSummary) + assert summary.cancelled is True + assert calls["cancel_checks"] >= 1 diff --git a/bots/bot_descarga/tests/test_excel_reader.py b/bots/bot_descarga/tests/test_excel_reader.py new file mode 100644 index 0000000..b361511 --- /dev/null +++ b/bots/bot_descarga/tests/test_excel_reader.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import tempfile +from pathlib import Path + +from openpyxl import Workbook + +from bot_descarga.infrastructure.excel_reader import ExcelReader, limpiar_nombre_carpeta + + +def _make_excel(rows, path: Path) -> Path: + wb = Workbook() + ws = wb.active + for row in rows: + ws.append(row) + wb.save(path) + return path + + +def test_limpiar_nombre_carpeta_caracteres_invalidos(): + assert limpiar_nombre_carpeta(r"900/123:456") == "900_123_456" + assert limpiar_nombre_carpeta(" ") == "SIN_NIT" + assert limpiar_nombre_carpeta("") == "SIN_NIT" + + +def test_excel_reader_lee_radicados_y_nit(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "datos.xlsx" + _make_excel( + [ + ["Radicado", "NIT"], + ["100001", "900111222"], + ["100002", "900111222"], + ["100003", None], + ["", "800"], + ], + path, + ) + records = ExcelReader().read(str(path)) + + assert len(records) == 3 + assert records[0].radicado == "100001" + assert records[0].nit == "900111222" + assert records[2].nit == "SIN_NIT" + + +def test_excel_reader_omite_encabezados_variantes(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "datos.xlsx" + _make_excel( + [ + ["proceso", "nit"], + ["número", "nit"], + ["555", "1"], + ], + path, + ) + records = ExcelReader().read(str(path)) + + assert len(records) == 1 + assert records[0].radicado == "555" diff --git a/bots/bot_descarga/tests/test_file_manager.py b/bots/bot_descarga/tests/test_file_manager.py new file mode 100644 index 0000000..e92519a --- /dev/null +++ b/bots/bot_descarga/tests/test_file_manager.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +import tempfile +from pathlib import Path + +from bot_descarga.config.settings import Settings +from bot_descarga.infrastructure.file_manager import FileManager +from bot_descarga.infrastructure.pdf_validator import PdfValidator + + +class AlwaysValid: + def validate(self, ruta: str): + return True, "OK" + + +class AlwaysInvalid: + def validate(self, ruta: str): + return False, "corrupto" + + +def test_ruta_final_crea_carpeta_nit(): + with tempfile.TemporaryDirectory() as tmp: + fm = FileManager(tmp, AlwaysValid()) + ruta = fm.ruta_final("900111", "12345") + assert Path(ruta).parent.exists() + assert Path(ruta).name == "12345.pdf" + assert "900111" in ruta + + +def test_exists_valid_cuando_no_existe(): + with tempfile.TemporaryDirectory() as tmp: + fm = FileManager(tmp, AlwaysValid()) + ok, ruta, motivo = fm.exists_valid("900", "1") + assert ok is False + assert motivo == "no existe" + assert ruta.endswith("1.pdf") + + +def test_exists_valid_cuando_valido(): + with tempfile.TemporaryDirectory() as tmp: + fm = FileManager(tmp, AlwaysValid()) + ruta = fm.ruta_final("900", "1") + Path(ruta).write_bytes(b"%PDF-dummy") + ok, _, motivo = fm.exists_valid("900", "1") + assert ok is True + assert motivo == "OK" + assert Path(ruta).exists() + + +def test_exists_valid_elimina_corrupto(): + with tempfile.TemporaryDirectory() as tmp: + fm = FileManager(tmp, AlwaysInvalid()) + ruta = fm.ruta_final("900", "1") + Path(ruta).write_bytes(b"basura") + ok, _, motivo = fm.exists_valid("900", "1") + assert ok is False + assert motivo == "corrupto" + assert not Path(ruta).exists() + + +def test_move_to_final(): + with tempfile.TemporaryDirectory() as tmp: + fm = FileManager(tmp, PdfValidator(Settings(peso_minimo_bytes=1))) + src = Path(tmp) / "temp.pdf" + src.write_bytes(b"%PDF-1.4") + dest = fm.ruta_final("NIT1", "R1") + fm.move_to_final(str(src), dest) + assert Path(dest).exists() + assert not src.exists() diff --git a/bots/bot_descarga/tests/test_gui_settings.py b/bots/bot_descarga/tests/test_gui_settings.py new file mode 100644 index 0000000..4c4abf9 --- /dev/null +++ b/bots/bot_descarga/tests/test_gui_settings.py @@ -0,0 +1,15 @@ +from bot_descarga.ui.settings_store import DEFAULT_API_URL, normalize_api_url + + +def test_default_api_is_pys1(): + assert "pys1.builker.xyz" in DEFAULT_API_URL + assert DEFAULT_API_URL.rstrip("/").endswith("bot-descarga") or "/bot-descarga" in DEFAULT_API_URL + + +def test_normalize_adds_prefix(): + assert normalize_api_url("https://pys1.builker.xyz") == "https://pys1.builker.xyz/bot-descarga" + assert ( + normalize_api_url("https://pys1.builker.xyz/bot-descarga/") + == "https://pys1.builker.xyz/bot-descarga" + ) + assert normalize_api_url("http://127.0.0.1:8189") == "http://127.0.0.1:8189/bot-descarga" diff --git a/bots/bot_descarga/tests/test_pdf_validator.py b/bots/bot_descarga/tests/test_pdf_validator.py new file mode 100644 index 0000000..98a74e3 --- /dev/null +++ b/bots/bot_descarga/tests/test_pdf_validator.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import tempfile +from pathlib import Path + +from bot_descarga.config.settings import Settings +from bot_descarga.infrastructure.pdf_validator import PdfValidator + + +def test_validate_rechaza_archivo_pequeno(): + settings = Settings(peso_minimo_bytes=10_000) + validator = PdfValidator(settings) + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "tiny.pdf" + path.write_bytes(b"%PDF-1.4\n%%EOF\n") + ok, motivo = validator.validate(str(path)) + + assert ok is False + assert "Peso insuficiente" in motivo + + +def test_validate_rechaza_sin_firma_pdf(): + settings = Settings(peso_minimo_bytes=10) + validator = PdfValidator(settings) + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "fake.pdf" + path.write_bytes(b"NOTA PDF" + b"x" * 100) + ok, motivo = validator.validate(str(path)) + + assert ok is False + assert "firma" in motivo.lower() + + +def test_validate_acepta_pdf_minimo_valido(): + """PDF mínimo con firma, EOF y estructura pikepdf-compatible.""" + import pikepdf + + settings = Settings(peso_minimo_bytes=50) + validator = PdfValidator(settings) + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "ok.pdf" + pdf = pikepdf.Pdf.new() + # Varias páginas para superar el peso mínimo sin corromper el archivo + for _ in range(5): + pdf.add_blank_page(page_size=(612, 792)) + pdf.save(path) + + ok, motivo = validator.validate(str(path)) + + assert ok is True, motivo + assert motivo == "OK" diff --git a/bots/bot_descarga/tests/test_worker_minio_keys.py b/bots/bot_descarga/tests/test_worker_minio_keys.py new file mode 100644 index 0000000..94ecc04 --- /dev/null +++ b/bots/bot_descarga/tests/test_worker_minio_keys.py @@ -0,0 +1,26 @@ +import os + +from bot_descarga.worker.minio_upload import MinioSettings, manifest_key, pdf_key + + +def test_pdf_and_manifest_keys(): + assert pdf_key("abc", "900", "123") == "downloads/abc/900/123.pdf" + assert manifest_key("abc") == "downloads/abc/_manifest.json" + + +def test_minio_defaults_to_anonymous(monkeypatch): + monkeypatch.delenv("MINIO_ACCESS_KEY", raising=False) + monkeypatch.delenv("MINIO_SECRET_KEY", raising=False) + monkeypatch.delenv("MINIO_ANONYMOUS", raising=False) + monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False) + monkeypatch.delenv("AWS_SECRET_ACCESS_KEY", raising=False) + settings = MinioSettings.from_env() + assert settings.anonymous is True + + +def test_minio_signed_only_with_both_keys(monkeypatch): + monkeypatch.setenv("MINIO_ACCESS_KEY", "user") + monkeypatch.setenv("MINIO_SECRET_KEY", "secret") + monkeypatch.delenv("MINIO_ANONYMOUS", raising=False) + settings = MinioSettings.from_env() + assert settings.anonymous is False diff --git a/docker-compose.yml b/docker-compose.yml index 829374e..31f331f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -151,6 +151,33 @@ services: limits: memory: ${COUNT_API_MEMORY:-4g} + bot-descarga: + build: + context: ./services/bot_descarga + dockerfile: Dockerfile + container_name: ipsextractor-bot-descarga + restart: unless-stopped + depends_on: + database: + condition: service_healthy + environment: + BOT_API_TOKEN: ${BOT_API_TOKEN:-dev-token-change-me} + BOT_API_HOST: 0.0.0.0 + BOT_API_PORT: "8200" + DATABASE_URL: postgresql+psycopg://ipsextractor:ipsextractor@database:5432/ipsextractor + MINIO_ENDPOINT: ${MINIO_ENDPOINT:-https://s3.builker.xyz} + MINIO_BUCKET: ${MINIO_BUCKET:-pys-cuentas-medicas} + MINIO_ANONYMOUS: ${MINIO_ANONYMOUS:-1} + MINIO_ACCESS_KEY: ${MINIO_ACCESS_KEY:-} + MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:-} + MINIO_REGION: ${MINIO_REGION:-us-east-1} + healthcheck: + test: ["CMD", "curl", "-f", "http://127.0.0.1:8200/health"] + interval: 10s + timeout: 5s + retries: 5 + start_period: 20s + api_gateway: build: context: ./services/api_gateway @@ -163,10 +190,13 @@ services: condition: service_started count-api: condition: service_healthy + bot-descarga: + condition: service_healthy environment: EXTRACTION_URL: http://extraction:8001 CLEANING_URL: http://cleaning:8002 COUNT_URL: http://count-api:8080 + BOT_DESCARGA_URL: http://bot-descarga:8200 ports: # Único puerto expuesto al host. Los demás servicios solo son accesibles en la red Docker. - "0.0.0.0:8189:8000" diff --git a/docs/database.dbml b/docs/database.dbml index 58929a0..263d3ba 100644 --- a/docs/database.dbml +++ b/docs/database.dbml @@ -9,8 +9,8 @@ Project ipsextractor { database_type: 'PostgreSQL' Note: ''' Base de datos compartida entre todos los microservicios (extraction, cleaning, - count-facturas). Cada servicio accede directamente a las tablas que necesita. - Las migraciones se gestionan con Alembic (services/extraction/alembic/). + count-facturas, bot-descarga). Cada servicio accede directamente a las tablas + que necesita. Las migraciones se gestionan con Alembic (services/db/alembic/). Cada servicio es desacoplado; comparten solo la DB como fuente de verdad. ''' } @@ -35,6 +35,12 @@ TableGroup counting { count_job_files } +TableGroup bot_descarga { + download_workers + download_jobs + download_job_events +} + // --------------------------------------------------------------------------- // Orchestration — trabajo multimodular del usuario // --------------------------------------------------------------------------- @@ -61,7 +67,7 @@ Table jobs { Table tasks { id varchar(36) [pk, note: 'UUID v4'] job_id varchar(36) [ref: > jobs.id, note: 'FK nullable — compat. con tasks anteriores'] - type varchar(64) [not null, note: 'counting | extraction | cleaning'] + type varchar(64) [not null, note: 'counting | extraction | cleaning | downloads'] status varchar(32) [not null, default: 'pending', note: 'pending | running | completed | failed'] title text request jsonb [not null, default: '{}', note: 'Payload de la petición que originó la tarea'] @@ -82,11 +88,77 @@ Table tasks { Note: ''' Módulo/resultado dentro de un Job. Una task representa la ejecución - de un servicio específico (conteo, extracción, limpieza). + de un servicio específico (conteo, extracción, limpieza, bot de descarga). job_id es nullable para compatibilidad con tasks creadas antes de jobs. ''' } +// --------------------------------------------------------------------------- +// Bot de descarga — workers remotos + corridas PDF → MinIO +// --------------------------------------------------------------------------- + +Table download_workers { + id varchar(36) [pk, note: 'ID del agent (ej. PC-01 / zer0)'] + hostname text [not null, default: ''] + max_concurrent integer [not null, default: 1] + status varchar(32) [not null, note: 'online | offline | busy'] + last_heartbeat timestamptz [not null] + current_job_id varchar(36) [note: 'download_jobs.id en ejecución'] + + indexes { + status + } + + Note: 'PC remota registrada por el agent BotDescarga (heartbeat).' +} + +Table download_jobs { + id varchar(36) [pk, note: 'UUID v4 de la corrida'] + job_id varchar(36) [null, ref: > jobs.id, note: 'Job Desktop contenedor'] + task_id varchar(36) [null, ref: > tasks.id, note: 'Task type=downloads'] + worker_id varchar(36) [not null, note: 'download_workers.id asignado'] + status varchar(32) [not null, default: 'pending', note: 'pending | running | completed | failed | cancelled'] + title text + excel_key text [not null, default: '', note: 'Clave MinIO del Excel'] + downloads_prefix text [not null, default: '', note: 'downloads/{id}/'] + original_filename text [not null, default: ''] + cancel_requested boolean [not null, default: false] + result jsonb [not null, default: '{}'] + error text + created_at timestamptz [not null] + started_at timestamptz + finished_at timestamptz + + indexes { + job_id + task_id + worker_id + status + (worker_id, status) [name: 'ix_download_jobs_worker_status'] + } + + Note: 'Corrida de descarga remota. PDFs en MinIO bajo downloads/{id}/.' +} + +Table download_job_events { + id integer [pk, increment] + download_job_id varchar(36) [not null, ref: > download_jobs.id] + event_type varchar(64) [not null] + message text [not null] + timestamp timestamptz [not null] + radicado varchar(255) + nit varchar(64) + idx integer + total integer + metadata jsonb [not null, default: '{}'] + + indexes { + (download_job_id, id) [name: 'ix_download_job_events_job_id'] + } + + Note: 'Log incremental de progreso / errores de una corrida.' +} + // --------------------------------------------------------------------------- // Counting — resultados del servicio de conteo de facturas médicas // --------------------------------------------------------------------------- diff --git a/services/api_gateway/api/main.py b/services/api_gateway/api/main.py index b5960d5..3bbcce0 100644 --- a/services/api_gateway/api/main.py +++ b/services/api_gateway/api/main.py @@ -18,6 +18,7 @@ logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name EXTRACTION_URL = os.getenv("EXTRACTION_URL", "http://extraction:8001").rstrip("/") CLEANING_URL = os.getenv("CLEANING_URL", "http://cleaning:8002").rstrip("/") COUNT_URL = os.getenv("COUNT_URL", "http://count-api:8080").rstrip("/") +BOT_DESCARGA_URL = os.getenv("BOT_DESCARGA_URL", "http://bot-descarga:8200").rstrip("/") # Timeouts: JSON ligero vs uploads OCR/conteo _TIMEOUT_JSON = httpx.Timeout(connect=5.0, read=30.0, write=30.0, pool=5.0) @@ -33,7 +34,13 @@ async def lifespan(app: FastAPI): timeout=_TIMEOUT_JSON, limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), ) - _log.info("Gateway iniciado — extraction=%s cleaning=%s count=%s", EXTRACTION_URL, CLEANING_URL, COUNT_URL) + _log.info( + "Gateway iniciado — extraction=%s cleaning=%s count=%s bot_descarga=%s", + EXTRACTION_URL, + CLEANING_URL, + COUNT_URL, + BOT_DESCARGA_URL, + ) yield await _http_client.aclose() _http_client = None @@ -72,6 +79,7 @@ def _is_upload_path(path: str) -> bool: or "/upload" in path or path.startswith("/cleaning/analyze-csv") or path.startswith("/cleaning/jobs/from-csv") + or path.startswith("/bot-descarga/jobs") ) @@ -91,12 +99,14 @@ async def health() -> dict[str, Any]: extraction = await _safe_health(client, EXTRACTION_URL) cleaning = await _safe_health(client, CLEANING_URL) counting = await _safe_health(client, COUNT_URL) + bot_descarga = await _safe_health(client, BOT_DESCARGA_URL) return { "status": "ok" if extraction["ok"] and cleaning["ok"] else "degraded", "services": { "extraction": extraction, "cleaning": cleaning, "counting": counting, + "bot_descarga": bot_descarga, }, } @@ -152,6 +162,11 @@ async def count_jobs_proxy(path: str, request: Request) -> Response: return await _proxy_request(request, f"{EXTRACTION_URL}/count-jobs/{path}") +@app.api_route("/bot-descarga/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"]) +async def bot_descarga_proxy(path: str, request: Request) -> Response: + return await _proxy_request(request, f"{BOT_DESCARGA_URL}/{path}") + + @app.api_route("/jobs", methods=["GET", "POST"]) async def jobs_root_proxy(request: Request) -> Response: return await _proxy_request(request, f"{EXTRACTION_URL}/jobs") diff --git a/services/bot_descarga/Dockerfile b/services/bot_descarga/Dockerfile new file mode 100644 index 0000000..4394423 --- /dev/null +++ b/services/bot_descarga/Dockerfile @@ -0,0 +1,28 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PYTHONPATH=/app + +WORKDIR /app + +RUN apt-get update \ + && apt-get install -y --no-install-recommends curl \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +COPY domain /app/domain +COPY infrastructure /app/infrastructure +COPY api /app/api + +ENV BOT_API_HOST=0.0.0.0 +ENV BOT_API_PORT=8200 + +EXPOSE 8200 + +HEALTHCHECK --interval=10s --timeout=5s --retries=5 --start-period=15s \ + CMD curl -f http://127.0.0.1:8200/health || exit 1 + +CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8200"] diff --git a/services/bot_descarga/README.md b/services/bot_descarga/README.md new file mode 100644 index 0000000..3346689 --- /dev/null +++ b/services/bot_descarga/README.md @@ -0,0 +1,51 @@ +# Bot Descarga API (infraestructura IPSExtractor) + +Microservicio FastAPI de control remoto de bots. Corre en Docker; +las PCs de bot **no** exponen puertos — solo el agent en [`bots/bot_descarga`](../../bots/bot_descarga). + +Persistencia: **PostgreSQL compartido** (`DATABASE_URL`), tablas +`download_workers`, `download_jobs`, `download_job_events` + `tasks` (`type=downloads`). + +## Layout MinIO (`pys-cuentas-medicas`) + +``` +uploads/{download_job_id}/input.xlsx +downloads/{download_job_id}/{nit}/{radicado}.pdf +``` + +## Arranque (compose) + +Servicio `bot-descarga` en el `docker-compose.yml` raíz. +Gateway: `/bot-descarga/*` → este servicio `:8200`. + +## Variables + +``` +BOT_API_TOKEN=dev-token-change-me +DATABASE_URL=postgresql+psycopg://ipsextractor:ipsextractor@database:5432/ipsextractor +MINIO_ENDPOINT=https://s3.builker.xyz +MINIO_BUCKET=pys-cuentas-medicas +MINIO_ANONYMOUS=1 +``` + +## Worker en PCs (`bots/bot_descarga`) + +```bash +cd bots/bot_descarga +python -m bot_descarga.worker.run \ + --api-url http://GATEWAY:8189/bot-descarga \ + --worker-id PC-01 \ + --token $BOT_API_TOKEN +``` + +O con GUI: `python run_gui.py`. Ver [README del agent](../../bots/bot_descarga/README.md). + +El agent queda idle hasta que el usuario cree un job desde el desktop +(enviando `job_id` del Job IPSExtractor). + +## Tests (API) + +```bash +cd services/bot_descarga +pytest -q +``` diff --git a/services/bot_descarga/api/__init__.py b/services/bot_descarga/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/bot_descarga/api/main.py b/services/bot_descarga/api/main.py new file mode 100644 index 0000000..53a212f --- /dev/null +++ b/services/bot_descarga/api/main.py @@ -0,0 +1,293 @@ +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Dict, List, Optional + +from fastapi import Depends, FastAPI, File, Form, HTTPException, Query, UploadFile +from fastapi.responses import Response +from pydantic import BaseModel, Field + +from domain.models import JobStatus +from infrastructure.auth import require_bearer +from infrastructure.storage_minio import MinioSettings, MinioStorage +from infrastructure.store import JobStore + +store = JobStore() +minio = MinioStorage(MinioSettings.from_env()) + +app = FastAPI(title="BotDescarga Remote API", version="1.0.0") + + +class RegisterWorkerBody(BaseModel): + worker_id: str + hostname: str = "" + max_concurrent: int = 1 + + +class HeartbeatBody(BaseModel): + worker_id: str + + +class EventBody(BaseModel): + event_type: str + message: str + radicado: Optional[str] = None + nit: Optional[str] = None + index: Optional[int] = None + total: Optional[int] = None + metadata: Dict[str, Any] = Field(default_factory=dict) + timestamp: Optional[str] = None + + +class PatchJobBody(BaseModel): + status: Optional[str] = None + result: Optional[Dict[str, Any]] = None + error: Optional[str] = None + + +@app.get("/health") +def health() -> dict: + return {"status": "ok", "service": "bot-descarga"} + + +@app.post("/workers/register") +def register_worker(body: RegisterWorkerBody, _: str = Depends(require_bearer)) -> dict: + worker = store.register_worker( + body.worker_id, + hostname=body.hostname, + max_concurrent=body.max_concurrent, + ) + return worker.to_dict() + + +@app.post("/workers/heartbeat") +def heartbeat(body: HeartbeatBody, _: str = Depends(require_bearer)) -> dict: + worker = store.heartbeat(body.worker_id) + if not worker: + raise HTTPException(status_code=404, detail="Worker no registrado") + return worker.to_dict() + + +@app.get("/workers") +def list_workers(_: str = Depends(require_bearer)) -> List[dict]: + return [w.to_dict() for w in store.list_workers()] + + +@app.get("/workers/me/pending") +def claim_pending( + worker_id: str = Query(...), + _: str = Depends(require_bearer), +) -> dict: + job = store.claim_pending(worker_id) + if not job: + return {"job": None} + return {"job": job.to_dict()} + + +@app.post("/jobs") +async def create_job( + worker_id: str = Form(...), + job_id: str = Form(...), + title: str = Form(""), + file: UploadFile = File(...), + _: str = Depends(require_bearer), +) -> dict: + if not worker_id.strip(): + raise HTTPException(status_code=400, detail="worker_id obligatorio") + parent_job_id = job_id.strip() + if not parent_job_id: + raise HTTPException(status_code=400, detail="job_id (Job Desktop) obligatorio") + if not store.parent_job_exists(parent_job_id): + raise HTTPException(status_code=404, detail="Trabajo (job) no encontrado") + + if not store.is_worker_online(worker_id): + raise HTTPException( + status_code=409, + detail=f"Worker '{worker_id}' no está online. Enciende el agent primero.", + ) + + data = await file.read() + if not data: + raise HTTPException(status_code=400, detail="Archivo vacío") + + original = file.filename or "input.xlsx" + content_type = file.content_type or "application/octet-stream" + title_clean = title.strip() + + job = store.create_job( + worker_id=worker_id.strip(), + excel_key="", + original_filename=original, + title=title_clean, + job_id=parent_job_id, + ) + task_id = store.create_downloads_task( + parent_job_id=parent_job_id, + download_job_id=job.id, + worker_id=worker_id.strip(), + title=title_clean or f"Bot descarga ({original})", + original_filename=original, + ) + job = store.get_job(job.id) + assert job is not None + + try: + key = minio.upload_excel(job.id, data, original, content_type=content_type) + except Exception as e: + store.patch_job(job.id, status=JobStatus.FAILED.value, error=f"MinIO upload failed: {e}") + raise HTTPException(status_code=502, detail=f"Error subiendo a MinIO: {e}") from e + + job = store.update_excel_key(job.id, key) + assert job is not None + store.add_event( + job.id, + "run_started", + f"Job creado y asignado a {worker_id}", + metadata={ + "excel_key": key, + "worker_id": worker_id, + "job_id": parent_job_id, + "task_id": task_id, + }, + ) + return job.to_dict() + + +@app.get("/jobs") +def list_jobs( + limit: int = Query(100, ge=1, le=500), + job_id: Optional[str] = Query(None, description="Filtrar por Job Desktop"), + _: str = Depends(require_bearer), +) -> List[dict]: + return [j.to_dict() for j in store.list_jobs(limit=limit, job_id=job_id)] + + +@app.get("/jobs/{download_job_id}") +def get_job(download_job_id: str, _: str = Depends(require_bearer)) -> dict: + job = store.get_job(download_job_id) + if not job: + raise HTTPException(status_code=404, detail="Job no encontrado") + return job.to_dict() + + +@app.get("/jobs/{download_job_id}/files") +def list_job_files(download_job_id: str, _: str = Depends(require_bearer)) -> dict: + job = store.get_job(download_job_id) + if not job: + raise HTTPException(status_code=404, detail="Job no encontrado") + try: + return minio.list_download_files(download_job_id, prefix=job.downloads_prefix) + except Exception as e: + raise HTTPException(status_code=502, detail=f"Error listando MinIO: {e}") from e + + +@app.get("/jobs/{download_job_id}/events") +def get_events( + download_job_id: str, + since: int = Query(0, ge=0), + _: str = Depends(require_bearer), +) -> dict: + if not store.get_job(download_job_id): + raise HTTPException(status_code=404, detail="Job no encontrado") + events = store.list_events(download_job_id, since_id=since) + return {"events": [e.to_dict() for e in events]} + + +@app.post("/jobs/{download_job_id}/cancel") +def cancel_job(download_job_id: str, _: str = Depends(require_bearer)) -> dict: + job = store.request_cancel(download_job_id) + if not job: + raise HTTPException(status_code=404, detail="Job no encontrado") + store.add_event(download_job_id, "warning", "Cancelación solicitada por el usuario") + return job.to_dict() + + +@app.get("/jobs/{download_job_id}/file") +def get_job_file(download_job_id: str, _: str = Depends(require_bearer)): + job = store.get_job(download_job_id) + if not job: + raise HTTPException(status_code=404, detail="Job no encontrado") + if not job.excel_key: + raise HTTPException(status_code=404, detail="Excel no disponible") + + try: + data = minio.get_bytes(job.excel_key) + except Exception as e: + raise HTTPException(status_code=502, detail=f"Error leyendo MinIO: {e}") from e + + filename = Path(job.excel_key).name + return Response( + content=data, + media_type="application/octet-stream", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + + +@app.get("/jobs/{download_job_id}/file-url") +def get_job_file_url(download_job_id: str, _: str = Depends(require_bearer)) -> dict: + job = store.get_job(download_job_id) + if not job or not job.excel_key: + raise HTTPException(status_code=404, detail="Job/archivo no encontrado") + try: + url = minio.presigned_get(job.excel_key) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) from e + return {"url": url, "excel_key": job.excel_key} + + +@app.post("/jobs/{download_job_id}/events") +def post_event( + download_job_id: str, + body: EventBody, + _: str = Depends(require_bearer), +) -> dict: + if not store.get_job(download_job_id): + raise HTTPException(status_code=404, detail="Job no encontrado") + event = store.add_event( + download_job_id, + body.event_type, + body.message, + radicado=body.radicado, + nit=body.nit, + index=body.index, + total=body.total, + metadata=body.metadata, + timestamp=body.timestamp, + ) + job = store.get_job(download_job_id) + if job and body.index is not None and body.total is not None: + result = dict(job.result or {}) + result["processed"] = body.index + result["total"] = body.total + if body.nit: + result["nit_actual"] = body.nit + store.patch_job(download_job_id, result=result) + return event.to_dict() + + +@app.patch("/jobs/{download_job_id}") +def patch_job( + download_job_id: str, + body: PatchJobBody, + _: str = Depends(require_bearer), +) -> dict: + if body.status and body.status not in {s.value for s in JobStatus}: + raise HTTPException(status_code=400, detail="status inválido") + job = store.patch_job( + download_job_id, + status=body.status, + result=body.result, + error=body.error, + ) + if not job: + raise HTTPException(status_code=404, detail="Job no encontrado") + return job.to_dict() + + +@app.get("/jobs/{download_job_id}/cancel-flag") +def cancel_flag(download_job_id: str, _: str = Depends(require_bearer)) -> dict: + job = store.get_job(download_job_id) + if not job: + raise HTTPException(status_code=404, detail="Job no encontrado") + return {"cancel_requested": job.cancel_requested, "status": job.status} diff --git a/services/bot_descarga/domain/__init__.py b/services/bot_descarga/domain/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/bot_descarga/domain/models.py b/services/bot_descarga/domain/models.py new file mode 100644 index 0000000..5953ef6 --- /dev/null +++ b/services/bot_descarga/domain/models.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from enum import Enum +from typing import Any, Dict, List, Optional + + +def utc_now() -> str: + return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") + + +class WorkerStatus(str, Enum): + ONLINE = "online" + OFFLINE = "offline" + BUSY = "busy" + + +class JobStatus(str, Enum): + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass +class Worker: + id: str + hostname: str = "" + max_concurrent: int = 1 + status: str = WorkerStatus.ONLINE.value + last_heartbeat: str = field(default_factory=utc_now) + current_job_id: Optional[str] = None + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass +class Job: + id: str + worker_id: str + status: str = JobStatus.PENDING.value + title: str = "" + excel_key: str = "" + downloads_prefix: str = "" + original_filename: str = "" + cancel_requested: bool = False + result: Dict[str, Any] = field(default_factory=dict) + error: Optional[str] = None + created_at: str = field(default_factory=utc_now) + started_at: Optional[str] = None + finished_at: Optional[str] = None + job_id: Optional[str] = None + task_id: Optional[str] = None + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass +class JobEventRecord: + id: int + job_id: str + event_type: str + message: str + timestamp: str + radicado: Optional[str] = None + nit: Optional[str] = None + index: Optional[int] = None + total: Optional[int] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict: + return asdict(self) diff --git a/services/bot_descarga/infrastructure/__init__.py b/services/bot_descarga/infrastructure/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/bot_descarga/infrastructure/auth.py b/services/bot_descarga/infrastructure/auth.py new file mode 100644 index 0000000..968cbc6 --- /dev/null +++ b/services/bot_descarga/infrastructure/auth.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +import os +from typing import Optional + +from fastapi import Header, HTTPException + + +def get_api_token() -> str: + return os.getenv("BOT_API_TOKEN", "dev-token-change-me") + + +def require_bearer(authorization: Optional[str] = Header(default=None)) -> str: + expected = get_api_token() + if not authorization or not authorization.lower().startswith("bearer "): + raise HTTPException(status_code=401, detail="Authorization Bearer requerido") + token = authorization.split(" ", 1)[1].strip() + if token != expected: + raise HTTPException(status_code=401, detail="Token inválido") + return token diff --git a/services/bot_descarga/infrastructure/database.py b/services/bot_descarga/infrastructure/database.py new file mode 100644 index 0000000..281c986 --- /dev/null +++ b/services/bot_descarga/infrastructure/database.py @@ -0,0 +1,159 @@ +"""Persistencia Postgres del servicio bot-descarga (DB compartida IPSExtractor).""" +from __future__ import annotations + +import os +import uuid +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Any, Iterator + +from sqlalchemy import ( + JSON, + Boolean, + DateTime, + ForeignKey, + Integer, + String, + Text, + create_engine, + text, +) +from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker + +DATABASE_URL = os.getenv( + "DATABASE_URL", + "postgresql+psycopg://ipsextractor:ipsextractor@database:5432/ipsextractor", +) + +engine = create_engine(DATABASE_URL, pool_pre_ping=True) +SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) + + +class Base(DeclarativeBase): + pass + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def new_id() -> str: + return str(uuid.uuid4()) + + +def _iso(dt: datetime | None) -> str | None: + if dt is None: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone().isoformat(timespec="seconds") + + +class Job(Base): + """Contenedor Desktop (lectura / FK).""" + + __tablename__ = "jobs" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_id) + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + title: Mapped[str | None] = mapped_column(Text, nullable=True) + job_metadata: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + + +class Task(Base): + __tablename__ = "tasks" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_id) + job_id: Mapped[str | None] = mapped_column( + String(36), ForeignKey("jobs.id"), nullable=True, index=True + ) + type: Mapped[str] = mapped_column(String(64), index=True) + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + title: Mapped[str | None] = mapped_column(Text, nullable=True) + request: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict) + result: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + owner_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class DownloadWorkerRow(Base): + __tablename__ = "download_workers" + + id: Mapped[str] = mapped_column(String(36), primary_key=True) + hostname: Mapped[str] = mapped_column(Text, default="") + max_concurrent: Mapped[int] = mapped_column(Integer, default=1) + status: Mapped[str] = mapped_column(String(32), index=True) + last_heartbeat: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + current_job_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + + +class DownloadJobRow(Base): + __tablename__ = "download_jobs" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_id) + job_id: Mapped[str | None] = mapped_column( + String(36), ForeignKey("jobs.id"), nullable=True, index=True + ) + task_id: Mapped[str | None] = mapped_column( + String(36), ForeignKey("tasks.id"), nullable=True, index=True + ) + worker_id: Mapped[str] = mapped_column(String(36), index=True) + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + title: Mapped[str | None] = mapped_column(Text, nullable=True) + excel_key: Mapped[str] = mapped_column(Text, default="") + downloads_prefix: Mapped[str] = mapped_column(Text, default="") + original_filename: Mapped[str] = mapped_column(Text, default="") + cancel_requested: Mapped[bool] = mapped_column(Boolean, default=False) + result: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class DownloadJobEventRow(Base): + __tablename__ = "download_job_events" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + download_job_id: Mapped[str] = mapped_column( + String(36), ForeignKey("download_jobs.id"), index=True + ) + event_type: Mapped[str] = mapped_column(String(64)) + message: Mapped[str] = mapped_column(Text) + timestamp: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + radicado: Mapped[str | None] = mapped_column(String(255), nullable=True) + nit: Mapped[str | None] = mapped_column(String(64), nullable=True) + idx: Mapped[int | None] = mapped_column(Integer, nullable=True) + total: Mapped[int | None] = mapped_column(Integer, nullable=True) + metadata_json: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict) + + +def init_db() -> None: + if DATABASE_URL.startswith("postgresql"): + with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as connection: + connection.execute(text("SELECT pg_advisory_lock(74512004)")) + try: + Base.metadata.create_all(bind=connection) + finally: + connection.execute(text("SELECT pg_advisory_unlock(74512004)")) + return + Base.metadata.create_all(bind=engine) + + +@contextmanager +def session_scope() -> Iterator[Session]: + session = SessionLocal() + try: + yield session + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() diff --git a/services/bot_descarga/infrastructure/storage_minio.py b/services/bot_descarga/infrastructure/storage_minio.py new file mode 100644 index 0000000..3cd2720 --- /dev/null +++ b/services/bot_descarga/infrastructure/storage_minio.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from pathlib import Path +from typing import BinaryIO, List, Optional + + +@dataclass(frozen=True) +class MinioSettings: + endpoint: str = "https://s3.builker.xyz" + bucket: str = "pys-cuentas-medicas" + access_key: str = "" + secret_key: str = "" + region: str = "us-east-1" + anonymous: bool = True + + @classmethod + def from_env(cls) -> "MinioSettings": + access_key = (os.getenv("MINIO_ACCESS_KEY") or "").strip() + secret_key = (os.getenv("MINIO_SECRET_KEY") or "").strip() + anon_flag = (os.getenv("MINIO_ANONYMOUS") or "").strip().lower() + if anon_flag in ("0", "false", "no"): + anonymous = False + elif anon_flag in ("1", "true", "yes"): + anonymous = True + else: + anonymous = not (access_key and secret_key) + return cls( + endpoint=os.getenv("MINIO_ENDPOINT", "https://s3.builker.xyz").rstrip("/"), + bucket=os.getenv("MINIO_BUCKET", "pys-cuentas-medicas"), + access_key=access_key, + secret_key=secret_key, + region=os.getenv("MINIO_REGION", "us-east-1"), + anonymous=anonymous, + ) + + +def excel_key(job_id: str, filename: str = "input.xlsx") -> str: + return f"uploads/{job_id}/{filename}" + + +def downloads_prefix(job_id: str) -> str: + return f"downloads/{job_id}/" + + +def pdf_key(job_id: str, nit: str, radicado: str) -> str: + return f"downloads/{job_id}/{nit}/{radicado}.pdf" + + +def meta_key(job_id: str) -> str: + return f"uploads/{job_id}/meta.json" + + +def manifest_key(job_id: str) -> str: + return f"downloads/{job_id}/_manifest.json" + + +def _s3_client(settings: MinioSettings): + import boto3 + from botocore import UNSIGNED + from botocore.client import Config + + s3_cfg = { + "addressing_style": "path", + "request_checksum_calculation": "when_required", + "response_checksum_validation": "when_required", + } + if settings.anonymous: + return boto3.client( + "s3", + endpoint_url=settings.endpoint, + region_name=settings.region, + config=Config(signature_version=UNSIGNED, s3=s3_cfg), + ) + return boto3.client( + "s3", + endpoint_url=settings.endpoint, + aws_access_key_id=settings.access_key, + aws_secret_access_key=settings.secret_key, + region_name=settings.region, + config=Config(signature_version="s3v4", s3=s3_cfg), + ) + + +class MinioStorage: + """Cliente S3-compatible para MinIO (uploads/downloads por job_id).""" + + def __init__(self, settings: Optional[MinioSettings] = None): + self.settings = settings or MinioSettings.from_env() + self._client = None + + @property + def client(self): + if self._client is None: + self._client = _s3_client(self.settings) + return self._client + + @property + def bucket(self) -> str: + return self.settings.bucket + + def put_bytes(self, key: str, data: bytes, content_type: str = "application/octet-stream") -> str: + self.client.put_object( + Bucket=self.bucket, + Key=key, + Body=data, + ContentType=content_type, + ) + return key + + def put_file(self, key: str, path: str | Path, content_type: str = "application/octet-stream") -> str: + path = Path(path) + return self.put_bytes(key, path.read_bytes(), content_type=content_type) + + def put_stream(self, key: str, stream: BinaryIO, content_type: str = "application/octet-stream") -> str: + data = stream.read() + return self.put_bytes(key, data, content_type=content_type) + + def get_bytes(self, key: str) -> bytes: + obj = self.client.get_object(Bucket=self.bucket, Key=key) + return obj["Body"].read() + + def download_file(self, key: str, dest: str | Path) -> Path: + dest = Path(dest) + dest.parent.mkdir(parents=True, exist_ok=True) + self.client.download_file(self.bucket, key, str(dest)) + return dest + + def exists(self, key: str) -> bool: + try: + self.client.head_object(Bucket=self.bucket, Key=key) + return True + except Exception: + return False + + def list_keys(self, prefix: str) -> List[str]: + keys: List[str] = [] + token = None + while True: + kwargs = {"Bucket": self.bucket, "Prefix": prefix} + if token: + kwargs["ContinuationToken"] = token + resp = self.client.list_objects_v2(**kwargs) + for item in resp.get("Contents") or []: + keys.append(item["Key"]) + if not resp.get("IsTruncated"): + break + token = resp.get("NextContinuationToken") + return keys + + def presigned_get(self, key: str, expires_in: int = 3600) -> str: + if self.settings.anonymous: + return f"{self.settings.endpoint}/{self.bucket}/{key}" + return self.client.generate_presigned_url( + "get_object", + Params={"Bucket": self.bucket, "Key": key}, + ExpiresIn=expires_in, + ) + + def upload_excel( + self, + job_id: str, + data: bytes, + original_filename: str, + content_type: str = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ) -> str: + ext = Path(original_filename).suffix.lower() or ".xlsx" + canonical = f"input{ext}" if ext in (".xlsx", ".xls", ".csv") else "input.xlsx" + key = excel_key(job_id, canonical) + self.put_bytes(key, data, content_type=content_type) + meta = { + "original_filename": original_filename, + "size": len(data), + "canonical": canonical, + } + self.put_bytes(meta_key(job_id), json.dumps(meta).encode("utf-8"), "application/json") + return key + + def upload_pdf(self, job_id: str, nit: str, radicado: str, local_path: str | Path) -> str: + key = pdf_key(job_id, nit, radicado) + self.put_file(key, local_path, content_type="application/pdf") + return key + + def write_manifest(self, job_id: str, keys: List[str]) -> str: + key = manifest_key(job_id) + payload = {"job_id": job_id, "files": keys, "count": len(keys)} + self.put_bytes(key, json.dumps(payload, ensure_ascii=False).encode("utf-8"), "application/json") + return key + + def list_download_files(self, job_id: str, prefix: str | None = None) -> dict: + """Lista PDFs de una corrida (manifest o list_keys).""" + prefix = prefix or downloads_prefix(job_id) + files: List[dict] = [] + try: + raw = self.get_bytes(manifest_key(job_id)) + payload = json.loads(raw.decode("utf-8")) + for key in payload.get("files") or []: + files.append(self._file_entry(str(key))) + except Exception: + for key in self.list_keys(prefix): + if key.endswith("/_manifest.json") or key.endswith("_manifest.json"): + continue + files.append(self._file_entry(key)) + return {"job_id": job_id, "prefix": prefix, "files": files, "count": len(files)} + + @staticmethod + def _file_entry(key: str) -> dict: + # downloads/{job_id}/{nit}/{radicado}.pdf + parts = key.strip("/").split("/") + nit = parts[-2] if len(parts) >= 3 else None + name = parts[-1] if parts else key + radicado = name[:-4] if name.lower().endswith(".pdf") else name + return {"key": key, "nit": nit, "radicado": radicado} diff --git a/services/bot_descarga/infrastructure/store.py b/services/bot_descarga/infrastructure/store.py new file mode 100644 index 0000000..c0a0899 --- /dev/null +++ b/services/bot_descarga/infrastructure/store.py @@ -0,0 +1,466 @@ +"""Repositorio sobre PostgreSQL (tablas download_* + sync de tasks).""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional + +from sqlalchemy import select + +from domain.models import ( + Job, + JobEventRecord, + JobStatus, + Worker, + WorkerStatus, + utc_now, +) +from infrastructure.database import ( + DownloadJobEventRow, + DownloadJobRow, + DownloadWorkerRow, + Task, + _iso, + init_db, + session_scope, + utcnow, +) +from infrastructure.storage_minio import downloads_prefix + +HEARTBEAT_OFFLINE_SECONDS = 45 + + +def _parse_iso(value: str | None) -> datetime | None: + if not value: + return None + dt = datetime.fromisoformat(value) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +class JobStore: + """API de persistencia compatible con el store SQLite anterior.""" + + def __init__(self, db_path: str | None = None): + # db_path ignorado (compat); usa DATABASE_URL. + init_db() + + def _row_worker(self, row: DownloadWorkerRow) -> Worker: + worker = Worker( + id=row.id, + hostname=row.hostname or "", + max_concurrent=int(row.max_concurrent or 1), + status=row.status, + last_heartbeat=_iso(row.last_heartbeat) or utc_now(), + current_job_id=row.current_job_id, + ) + return self._refresh_worker_status(worker) + + def _row_job(self, row: DownloadJobRow) -> Job: + return Job( + id=row.id, + worker_id=row.worker_id, + status=row.status, + title=row.title or "", + excel_key=row.excel_key or "", + downloads_prefix=row.downloads_prefix or "", + original_filename=row.original_filename or "", + cancel_requested=bool(row.cancel_requested), + result=dict(row.result or {}), + error=row.error, + created_at=_iso(row.created_at) or utc_now(), + started_at=_iso(row.started_at), + finished_at=_iso(row.finished_at), + job_id=row.job_id, + task_id=row.task_id, + ) + + def _refresh_worker_status(self, worker: Worker) -> Worker: + try: + hb = datetime.fromisoformat(worker.last_heartbeat) + if hb.tzinfo is None: + hb = hb.replace(tzinfo=timezone.utc) + age = datetime.now(timezone.utc) - hb.astimezone(timezone.utc) + if age > timedelta(seconds=HEARTBEAT_OFFLINE_SECONDS): + worker.status = WorkerStatus.OFFLINE.value + elif worker.current_job_id: + worker.status = WorkerStatus.BUSY.value + else: + worker.status = WorkerStatus.ONLINE.value + except Exception: + worker.status = WorkerStatus.OFFLINE.value + return worker + + def register_worker( + self, + worker_id: str, + hostname: str = "", + max_concurrent: int = 1, + ) -> Worker: + with session_scope() as session: + row = session.get(DownloadWorkerRow, worker_id) + now = utcnow() + if row: + row.hostname = hostname + row.max_concurrent = max_concurrent + row.status = WorkerStatus.ONLINE.value + row.last_heartbeat = now + row.current_job_id = None + else: + row = DownloadWorkerRow( + id=worker_id, + hostname=hostname, + max_concurrent=max_concurrent, + status=WorkerStatus.ONLINE.value, + last_heartbeat=now, + current_job_id=None, + ) + session.add(row) + session.flush() + return self._row_worker(row) + + def heartbeat(self, worker_id: str) -> Optional[Worker]: + with session_scope() as session: + row = session.get(DownloadWorkerRow, worker_id) + if not row: + return None + row.last_heartbeat = utcnow() + row.status = ( + WorkerStatus.BUSY.value if row.current_job_id else WorkerStatus.ONLINE.value + ) + session.flush() + return self._row_worker(row) + + def get_worker(self, worker_id: str) -> Optional[Worker]: + with session_scope() as session: + row = session.get(DownloadWorkerRow, worker_id) + if not row: + return None + return self._row_worker(row) + + def list_workers(self) -> List[Worker]: + with session_scope() as session: + rows = session.scalars( + select(DownloadWorkerRow).order_by(DownloadWorkerRow.id) + ).all() + return [self._row_worker(r) for r in rows] + + def is_worker_online(self, worker_id: str) -> bool: + worker = self.get_worker(worker_id) + if not worker: + return False + return worker.status in (WorkerStatus.ONLINE.value, WorkerStatus.BUSY.value) + + def update_excel_key(self, job_id: str, excel_key: str) -> Optional[Job]: + with session_scope() as session: + row = session.get(DownloadJobRow, job_id) + if not row: + return None + row.excel_key = excel_key + session.flush() + return self._row_job(row) + + def create_job( + self, + worker_id: str, + excel_key: str, + original_filename: str = "", + title: str = "", + *, + job_id: Optional[str] = None, + task_id: Optional[str] = None, + ) -> Job: + import uuid as uuid_mod + + download_id = str(uuid_mod.uuid4()) + prefix = downloads_prefix(download_id) + with session_scope() as session: + row = DownloadJobRow( + id=download_id, + job_id=job_id, + task_id=task_id, + worker_id=worker_id, + status=JobStatus.PENDING.value, + title=title or f"Descarga {original_filename}", + excel_key=excel_key, + downloads_prefix=prefix, + original_filename=original_filename, + cancel_requested=False, + result={}, + created_at=utcnow(), + ) + session.add(row) + session.flush() + return self._row_job(row) + + def get_job(self, job_id: str) -> Optional[Job]: + with session_scope() as session: + row = session.get(DownloadJobRow, job_id) + if not row: + return None + return self._row_job(row) + + def list_jobs( + self, + limit: int = 100, + *, + job_id: Optional[str] = None, + ) -> List[Job]: + with session_scope() as session: + stmt = select(DownloadJobRow).order_by(DownloadJobRow.created_at.desc()).limit(limit) + if job_id: + stmt = ( + select(DownloadJobRow) + .where(DownloadJobRow.job_id == job_id) + .order_by(DownloadJobRow.created_at.desc()) + .limit(limit) + ) + rows = session.scalars(stmt).all() + return [self._row_job(r) for r in rows] + + def claim_pending(self, worker_id: str) -> Optional[Job]: + with session_scope() as session: + worker = session.get(DownloadWorkerRow, worker_id) + if not worker or worker.current_job_id: + return None + + row = session.scalars( + select(DownloadJobRow) + .where( + DownloadJobRow.worker_id == worker_id, + DownloadJobRow.status == JobStatus.PENDING.value, + ) + .order_by(DownloadJobRow.created_at.asc()) + .limit(1) + ).first() + if not row: + return None + + now = utcnow() + row.status = JobStatus.RUNNING.value + row.started_at = now + worker.current_job_id = row.id + worker.status = WorkerStatus.BUSY.value + worker.last_heartbeat = now + + if row.task_id: + task = session.get(Task, row.task_id) + if task: + task.status = JobStatus.RUNNING.value + task.started_at = now + task.updated_at = now + + session.flush() + return self._row_job(row) + + def request_cancel(self, job_id: str) -> Optional[Job]: + with session_scope() as session: + row = session.get(DownloadJobRow, job_id) + if not row: + return None + if row.status in ( + JobStatus.COMPLETED.value, + JobStatus.FAILED.value, + JobStatus.CANCELLED.value, + ): + return self._row_job(row) + + now = utcnow() + row.cancel_requested = True + if row.status == JobStatus.PENDING.value: + row.status = JobStatus.CANCELLED.value + row.finished_at = now + self._sync_task(session, row, status=JobStatus.CANCELLED.value) + session.flush() + return self._row_job(row) + + def is_cancel_requested(self, job_id: str) -> bool: + job = self.get_job(job_id) + return bool(job and job.cancel_requested) + + def _sync_task( + self, + session, + row: DownloadJobRow, + *, + status: Optional[str] = None, + result: Optional[Dict[str, Any]] = None, + error: Optional[str] = None, + ) -> None: + if not row.task_id: + return + task = session.get(Task, row.task_id) + if not task: + return + now = utcnow() + if status: + task.status = status + if result is not None: + task.result = { + **dict(task.result or {}), + "download_job_id": row.id, + **result, + } + if error is not None: + task.error = error + task.updated_at = now + if status in ( + JobStatus.COMPLETED.value, + JobStatus.FAILED.value, + JobStatus.CANCELLED.value, + ): + task.finished_at = now + + def patch_job( + self, + job_id: str, + *, + status: Optional[str] = None, + result: Optional[Dict[str, Any]] = None, + error: Optional[str] = None, + ) -> Optional[Job]: + with session_scope() as session: + row = session.get(DownloadJobRow, job_id) + if not row: + return None + + new_status = status or row.status + if result is not None: + row.result = result + if error is not None: + row.error = error + row.status = new_status + + finished = False + if new_status in ( + JobStatus.COMPLETED.value, + JobStatus.FAILED.value, + JobStatus.CANCELLED.value, + ): + row.finished_at = row.finished_at or utcnow() + finished = True + + self._sync_task( + session, + row, + status=new_status if status else None, + result=result, + error=error, + ) + + if finished: + worker = session.get(DownloadWorkerRow, row.worker_id) + if worker and worker.current_job_id == job_id: + worker.current_job_id = None + worker.status = WorkerStatus.ONLINE.value + + session.flush() + return self._row_job(row) + + def add_event( + self, + job_id: str, + event_type: str, + message: str, + *, + radicado: Optional[str] = None, + nit: Optional[str] = None, + index: Optional[int] = None, + total: Optional[int] = None, + metadata: Optional[Dict[str, Any]] = None, + timestamp: Optional[str] = None, + ) -> JobEventRecord: + with session_scope() as session: + ts = _parse_iso(timestamp) or utcnow() + ev = DownloadJobEventRow( + download_job_id=job_id, + event_type=event_type, + message=message, + timestamp=ts, + radicado=radicado, + nit=nit, + idx=index, + total=total, + metadata_json=metadata or {}, + ) + session.add(ev) + session.flush() + return JobEventRecord( + id=int(ev.id), + job_id=job_id, + event_type=event_type, + message=message, + timestamp=_iso(ts) or utc_now(), + radicado=radicado, + nit=nit, + index=index, + total=total, + metadata=metadata or {}, + ) + + def list_events(self, job_id: str, since_id: int = 0) -> List[JobEventRecord]: + with session_scope() as session: + rows = session.scalars( + select(DownloadJobEventRow) + .where( + DownloadJobEventRow.download_job_id == job_id, + DownloadJobEventRow.id > since_id, + ) + .order_by(DownloadJobEventRow.id.asc()) + ).all() + return [ + JobEventRecord( + id=int(r.id), + job_id=r.download_job_id, + event_type=r.event_type, + message=r.message, + timestamp=_iso(r.timestamp) or utc_now(), + radicado=r.radicado, + nit=r.nit, + index=r.idx, + total=r.total, + metadata=dict(r.metadata_json or {}), + ) + for r in rows + ] + + def create_downloads_task( + self, + *, + parent_job_id: str, + download_job_id: str, + worker_id: str, + title: str, + original_filename: str, + ) -> str: + """Crea Task(type=downloads) y la asocia al download_job. Devuelve task_id.""" + with session_scope() as session: + now = utcnow() + task = Task( + job_id=parent_job_id, + type="downloads", + status=JobStatus.PENDING.value, + title=title or f"Bot descarga ({original_filename})", + request={ + "worker_id": worker_id, + "original_filename": original_filename, + "download_job_id": download_job_id, + }, + result={"download_job_id": download_job_id}, + created_at=now, + updated_at=now, + ) + session.add(task) + session.flush() + row = session.get(DownloadJobRow, download_job_id) + if row: + row.task_id = task.id + row.job_id = parent_job_id + session.flush() + return task.id + + def parent_job_exists(self, parent_job_id: str) -> bool: + from infrastructure.database import Job as IpsJob + + with session_scope() as session: + return session.get(IpsJob, parent_job_id) is not None diff --git a/services/bot_descarga/requirements.txt b/services/bot_descarga/requirements.txt new file mode 100644 index 0000000..4a41418 --- /dev/null +++ b/services/bot_descarga/requirements.txt @@ -0,0 +1,9 @@ +fastapi>=0.110.0 +uvicorn[standard]>=0.27.0 +python-multipart>=0.0.9 +boto3>=1.34.0 +httpx>=0.27.0 +pydantic>=2.0.0 +sqlalchemy>=2.0.0 +psycopg[binary]>=3.1.0 +pytest>=7.0.0 diff --git a/services/bot_descarga/tests/test_store.py b/services/bot_descarga/tests/test_store.py new file mode 100644 index 0000000..b133ded --- /dev/null +++ b/services/bot_descarga/tests/test_store.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import importlib +import sys +from pathlib import Path + +import pytest + +from infrastructure.storage_minio import downloads_prefix, excel_key, pdf_key + + +def test_minio_key_helpers(): + assert excel_key("abc", "input.xlsx") == "uploads/abc/input.xlsx" + assert downloads_prefix("abc") == "downloads/abc/" + assert pdf_key("abc", "900", "123") == "downloads/abc/900/123.pdf" + + +@pytest.fixture() +def store(tmp_path, monkeypatch): + db_url = f"sqlite:///{(tmp_path / 'bot.db').as_posix()}" + monkeypatch.setenv("DATABASE_URL", db_url) + + # Recargar módulos para que el engine use SQLITE + for name in [ + "infrastructure.database", + "infrastructure.store", + ]: + if name in sys.modules: + importlib.reload(sys.modules[name]) + else: + importlib.import_module(name) + + from domain.models import JobStatus, WorkerStatus + from infrastructure import database as db + from infrastructure.store import JobStore + + importlib.reload(db) + import infrastructure.store as store_mod + + importlib.reload(store_mod) + return store_mod.JobStore(), JobStatus, WorkerStatus + + +def test_register_heartbeat_and_offline(store): + job_store, _JobStatus, WorkerStatus = store + w = job_store.register_worker("PC-01", hostname="host1") + assert w.status == WorkerStatus.ONLINE.value + assert job_store.is_worker_online("PC-01") + + job_store.heartbeat("PC-01") + workers = job_store.list_workers() + assert len(workers) == 1 + assert workers[0].id == "PC-01" + + +def test_create_job_requires_worker_and_claim_is_atomic(store): + job_store, JobStatus, _WorkerStatus = store + job_store.register_worker("PC-01") + + job = job_store.create_job( + worker_id="PC-01", + excel_key="uploads/x/input.xlsx", + original_filename="datos.xlsx", + title="Lote 1", + ) + assert job.status == JobStatus.PENDING.value + assert job.worker_id == "PC-01" + assert job.downloads_prefix == downloads_prefix(job.id) + + claimed = job_store.claim_pending("PC-01") + assert claimed is not None + assert claimed.id == job.id + assert claimed.status == JobStatus.RUNNING.value + + again = job_store.claim_pending("PC-01") + assert again is None + + job_store.register_worker("PC-02") + other = job_store.claim_pending("PC-02") + assert other is None + + +def test_cancel_pending_marks_cancelled(store): + job_store, JobStatus, _WorkerStatus = store + job_store.register_worker("PC-01") + job = job_store.create_job("PC-01", "uploads/j/input.xlsx", "a.xlsx") + cancelled = job_store.request_cancel(job.id) + assert cancelled is not None + assert cancelled.status == JobStatus.CANCELLED.value + assert cancelled.cancel_requested is True + + +def test_cancel_running_sets_flag(store): + job_store, JobStatus, _WorkerStatus = store + job_store.register_worker("PC-01") + job = job_store.create_job("PC-01", "uploads/j/input.xlsx", "a.xlsx") + job_store.claim_pending("PC-01") + flagged = job_store.request_cancel(job.id) + assert flagged is not None + assert flagged.cancel_requested is True + assert flagged.status == JobStatus.RUNNING.value + assert job_store.is_cancel_requested(job.id) is True + + +def test_events_and_patch_complete(store): + job_store, JobStatus, _WorkerStatus = store + job_store.register_worker("PC-01") + job = job_store.create_job("PC-01", "uploads/j/input.xlsx", "a.xlsx") + job_store.claim_pending("PC-01") + e1 = job_store.add_event( + job.id, "radicado_saved", "ok", radicado="1", nit="9", index=1, total=2 + ) + assert e1.id >= 1 + events = job_store.list_events(job.id, since_id=0) + assert len(events) == 1 + + done = job_store.patch_job( + job.id, + status=JobStatus.COMPLETED.value, + result={"exitosos": 1}, + ) + assert done is not None + assert done.status == JobStatus.COMPLETED.value + worker = job_store.get_worker("PC-01") + assert worker is not None + assert worker.current_job_id is None + + +def test_list_jobs_by_parent(store): + job_store, _JobStatus, _WorkerStatus = store + from infrastructure.database import Job as IpsJob + from infrastructure.database import session_scope, utcnow + + job_store.register_worker("PC-01") + with session_scope() as session: + session.add( + IpsJob( + id="parent-1", + status="pending", + title="Trabajo", + job_metadata={}, + created_at=utcnow(), + updated_at=utcnow(), + ) + ) + + a = job_store.create_job("PC-01", "uploads/a/input.xlsx", "a.xlsx", job_id="parent-1") + job_store.create_job("PC-01", "uploads/b/input.xlsx", "b.xlsx", job_id="other") + listed = job_store.list_jobs(limit=50, job_id="parent-1") + assert len(listed) == 1 + assert listed[0].id == a.id diff --git a/services/db/alembic/versions/0004_bot_descarga_tables.py b/services/db/alembic/versions/0004_bot_descarga_tables.py new file mode 100644 index 0000000..013db9e --- /dev/null +++ b/services/db/alembic/versions/0004_bot_descarga_tables.py @@ -0,0 +1,120 @@ +"""Bot descarga: download_workers, download_jobs, download_job_events. + +Revision ID: 0004 +Revises: 0003 +Create Date: 2026-07-12 + +""" +from __future__ import annotations + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0004" +down_revision: Union[str, None] = "0003" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _table_exists(conn: sa.engine.Connection, table: str) -> bool: + return sa.inspect(conn).has_table(table) + + +def upgrade() -> None: + conn = op.get_bind() + json_type = ( + postgresql.JSONB() + if conn.dialect.name == "postgresql" + else sa.JSON() + ) + + if not _table_exists(conn, "download_workers"): + op.create_table( + "download_workers", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("hostname", sa.Text(), nullable=False, server_default=""), + sa.Column("max_concurrent", sa.Integer(), nullable=False, server_default="1"), + sa.Column("status", sa.String(32), nullable=False), + sa.Column("last_heartbeat", sa.DateTime(timezone=True), nullable=False), + sa.Column("current_job_id", sa.String(36), nullable=True), + ) + op.create_index("ix_download_workers_status", "download_workers", ["status"]) + + if not _table_exists(conn, "download_jobs"): + op.create_table( + "download_jobs", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("job_id", sa.String(36), sa.ForeignKey("jobs.id"), nullable=True), + sa.Column("task_id", sa.String(36), sa.ForeignKey("tasks.id"), nullable=True), + sa.Column("worker_id", sa.String(36), nullable=False), + sa.Column("status", sa.String(32), nullable=False, server_default="pending"), + sa.Column("title", sa.Text(), nullable=True), + sa.Column("excel_key", sa.Text(), nullable=False, server_default=""), + sa.Column("downloads_prefix", sa.Text(), nullable=False, server_default=""), + sa.Column("original_filename", sa.Text(), nullable=False, server_default=""), + sa.Column("cancel_requested", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("result", json_type, nullable=False, server_default=sa.text("'{}'::jsonb") if conn.dialect.name == "postgresql" else sa.text("'{}'")), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_download_jobs_job_id", "download_jobs", ["job_id"]) + op.create_index("ix_download_jobs_task_id", "download_jobs", ["task_id"]) + op.create_index("ix_download_jobs_worker_id", "download_jobs", ["worker_id"]) + op.create_index("ix_download_jobs_status", "download_jobs", ["status"]) + op.create_index( + "ix_download_jobs_worker_status", + "download_jobs", + ["worker_id", "status"], + ) + + if not _table_exists(conn, "download_job_events"): + op.create_table( + "download_job_events", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column( + "download_job_id", + sa.String(36), + sa.ForeignKey("download_jobs.id"), + nullable=False, + ), + sa.Column("event_type", sa.String(64), nullable=False), + sa.Column("message", sa.Text(), nullable=False), + sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False), + sa.Column("radicado", sa.String(255), nullable=True), + sa.Column("nit", sa.String(64), nullable=True), + sa.Column("idx", sa.Integer(), nullable=True), + sa.Column("total", sa.Integer(), nullable=True), + sa.Column( + "metadata", + json_type, + nullable=False, + server_default=sa.text("'{}'::jsonb") if conn.dialect.name == "postgresql" else sa.text("'{}'"), + ), + ) + op.create_index( + "ix_download_job_events_job_id", + "download_job_events", + ["download_job_id", "id"], + ) + + +def downgrade() -> None: + conn = op.get_bind() + if _table_exists(conn, "download_job_events"): + op.drop_index("ix_download_job_events_job_id", table_name="download_job_events") + op.drop_table("download_job_events") + if _table_exists(conn, "download_jobs"): + op.drop_index("ix_download_jobs_worker_status", table_name="download_jobs") + op.drop_index("ix_download_jobs_status", table_name="download_jobs") + op.drop_index("ix_download_jobs_worker_id", table_name="download_jobs") + op.drop_index("ix_download_jobs_task_id", table_name="download_jobs") + op.drop_index("ix_download_jobs_job_id", table_name="download_jobs") + op.drop_table("download_jobs") + if _table_exists(conn, "download_workers"): + op.drop_index("ix_download_workers_status", table_name="download_workers") + op.drop_table("download_workers") diff --git a/services/desktop_app/package.json b/services/desktop_app/package.json index deb583a..6974a07 100644 --- a/services/desktop_app/package.json +++ b/services/desktop_app/package.json @@ -1,6 +1,6 @@ { "name": "ipsextractor-desktop", - "version": "1.0.3", + "version": "1.1.0", "private": true, "description": "Desktop extraction client for IPSExtractor microservices.", "main": "electron/main.cjs", diff --git a/services/desktop_app/src/renderer/App.tsx b/services/desktop_app/src/renderer/App.tsx index 0e3315f..69ab238 100644 --- a/services/desktop_app/src/renderer/App.tsx +++ b/services/desktop_app/src/renderer/App.tsx @@ -29,6 +29,7 @@ import { UpdatePrompt, CheckUpdatesButton, useAutoUpdateCheck } from "./componen import "./modules/counting/CountingModule"; import "./modules/extraction/ExtractionModule"; import "./modules/cleaning/CleaningModule"; +import "./modules/downloads/DownloadsModule"; type Status = "idle" | "ok" | "error" | "busy"; type View = "home" | "job" | "settings"; @@ -39,7 +40,7 @@ function App() { const [settingsReturnView, setSettingsReturnView] = useState<"home" | "job">("home"); const [settingsSection, setSettingsSection] = useState("connection"); const [activeJob, setActiveJob] = useState(null); - const [jobSection, setJobSection] = useState("counting"); + const [jobSection, setJobSection] = useState("downloads"); const [jobSectionContext, setJobSectionContext] = useState>({}); const [apiBaseUrl, setApiBaseUrl] = useState(loadApiBaseUrl); const [connectionStatus, setConnectionStatus] = useState("idle"); @@ -109,7 +110,7 @@ function App() { function openJob(job: Job) { setActiveJob(job); - setJobSection("counting"); + setJobSection("downloads"); setJobSectionContext({}); setView("job"); } @@ -201,6 +202,14 @@ function App() {
); diff --git a/services/desktop_app/src/renderer/views/JobView.tsx b/services/desktop_app/src/renderer/views/JobView.tsx index 75b4f26..d0ee822 100644 --- a/services/desktop_app/src/renderer/views/JobView.tsx +++ b/services/desktop_app/src/renderer/views/JobView.tsx @@ -7,6 +7,7 @@ import { BackgroundProgressBanner } from "../components/BackgroundProgressBanner import { CountingWizardModule } from "../modules/counting/CountingWizardModule"; import { CleaningWizardModule } from "../modules/cleaning/CleaningWizardModule"; import { JobExtractionWizard } from "../modules/extraction/JobExtractionWizard"; +import { DownloadsWizardModule } from "../modules/downloads/DownloadsWizardModule"; import { JobResultsView } from "./JobResultsView"; interface Props { @@ -120,6 +121,7 @@ export function JobView({ client, job, section, sectionContext, onActivateSectio {section === "counting" && "Conteo de servicios"} {section === "extraction" && "Extracción de datos"} {section === "cleaning" && "Limpieza de datos"} + {section === "downloads" && "Bot de descarga"} {section === "results" && "Resultados del trabajo"} @@ -179,6 +181,10 @@ export function JobView({ client, job, section, sectionContext, onActivateSectio /> +
+ +
+
onActivateSection("counting")} onStartExtraction={(ctx?: Record) => onActivateSection("extraction", ctx)} onStartCleaning={(ctx?: Record) => onActivateSection("cleaning", ctx)} + onStartDownload={() => onActivateSection("downloads")} />
diff --git a/services/desktop_app/src/shared/types.ts b/services/desktop_app/src/shared/types.ts index cf9ad41..e3922f8 100644 --- a/services/desktop_app/src/shared/types.ts +++ b/services/desktop_app/src/shared/types.ts @@ -303,6 +303,62 @@ export interface UpdateProgress { percent?: number; } +/** Worker del bot de descargas PDF (control remoto). */ +export interface DownloadWorker { + id: string; + hostname: string; + max_concurrent: number; + status: "online" | "offline" | "busy" | string; + last_heartbeat: string; + current_job_id: string | null; +} + +export interface DownloadJob { + id: string; + worker_id: string; + status: "pending" | "running" | "completed" | "failed" | "cancelled" | string; + title: string; + excel_key: string; + downloads_prefix: string; + original_filename: string; + cancel_requested: boolean; + result: Record; + error: string | null; + created_at: string; + started_at: string | null; + finished_at: string | null; + /** Job Desktop contenedor (IPSExtractor jobs.id) */ + job_id?: string | null; + /** Task type=downloads */ + task_id?: string | null; +} + +export interface DownloadJobFile { + key: string; + nit?: string | null; + radicado?: string | null; +} + +export interface DownloadJobFilesResponse { + job_id: string; + prefix: string; + files: DownloadJobFile[]; + count: number; +} + +export interface DownloadJobEvent { + id: number; + job_id: string; + event_type: string; + message: string; + timestamp: string; + radicado?: string | null; + nit?: string | null; + index?: number | null; + total?: number | null; + metadata?: Record; +} + export interface DesktopBridge { getRuntimeConfig(): Promise; selectDirectory(): Promise;