diff --git a/bots/bot_descarga/bot_descarga/ui/dashboard.py b/bots/bot_descarga/bot_descarga/ui/dashboard.py index 880508a..5502f8f 100644 --- a/bots/bot_descarga/bot_descarga/ui/dashboard.py +++ b/bots/bot_descarga/bot_descarga/ui/dashboard.py @@ -281,6 +281,7 @@ class BotDashboard(ctk.CTk): self.entry_api.delete(0, "end") self.entry_api.insert(0, api) self._persist_flags() + self._limpiar_logs() self._agent = WorkerAgent( api_url=api, diff --git a/bots/bot_digitador/bot_digitador/ui/dashboard.py b/bots/bot_digitador/bot_digitador/ui/dashboard.py index 1ce595c..ca1b7bd 100644 --- a/bots/bot_digitador/bot_digitador/ui/dashboard.py +++ b/bots/bot_digitador/bot_digitador/ui/dashboard.py @@ -291,6 +291,7 @@ class BotDashboard(ctk.CTk): self.entry_api.delete(0, "end") self.entry_api.insert(0, api) self._persist_flags() + self._limpiar_logs() self._agent = WorkerAgent( api_url=api, diff --git a/services/bot_descarga/infrastructure/store.py b/services/bot_descarga/infrastructure/store.py index a18ff3a..3991fd9 100644 --- a/services/bot_descarga/infrastructure/store.py +++ b/services/bot_descarga/infrastructure/store.py @@ -27,6 +27,8 @@ from infrastructure.database import ( from infrastructure.storage_minio import downloads_prefix HEARTBEAT_OFFLINE_SECONDS = 45 +# Un agent recién conectado no recibe jobs reclamados de otros workers. +_FRESH_WORKER_GRACE_SECONDS = 45 def _parse_iso(value: str | None) -> datetime | None: @@ -44,6 +46,7 @@ class JobStore: def __init__(self, db_path: str | None = None): # db_path ignorado (compat); usa DATABASE_URL. init_db() + self._fresh_until: dict[str, datetime] = {} def _row_worker(self, row: DownloadWorkerRow) -> Worker: worker = Worker( @@ -120,7 +123,67 @@ class JobStore: ) session.add(row) session.flush() - return self._row_worker(row) + parent_ids = self._abandon_assigned_jobs( + session, + worker_id, + now, + "Agent reconectado: job anterior abandonado. Lanza una corrida nueva desde el desktop.", + ) + session.flush() + worker = self._row_worker(row) + self._fresh_until[worker_id] = now + timedelta(seconds=_FRESH_WORKER_GRACE_SECONDS) + for parent_id in parent_ids: + try: + self.refresh_parent_from_shards(parent_id) + except Exception: + pass + return worker + + def _abandon_assigned_jobs( + self, session, worker_id: str, now: datetime, message: str + ) -> list[str]: + """Cancela pending/running de este worker para que un agent nuevo arranque idle.""" + rows = session.scalars( + select(DownloadJobRow).where( + DownloadJobRow.worker_id == worker_id, + DownloadJobRow.status.in_((JobStatus.PENDING.value, JobStatus.RUNNING.value)), + ) + ).all() + parent_ids: set[str] = set() + for row in rows: + is_parent = ( + row.parent_download_job_id is None + and row.shard_count is not None + and int(row.shard_count) > 1 + ) + if is_parent: + continue + row.status = JobStatus.CANCELLED.value + row.cancel_requested = True + row.finished_at = now + row.error = message + self._sync_task(session, row, status=JobStatus.CANCELLED.value, error=message) + session.add( + DownloadJobEventRow( + download_job_id=row.id, + event_type="warning", + message=message, + timestamp=now, + metadata_json={"abandoned_on_register": True, "worker_id": worker_id}, + ) + ) + if row.parent_download_job_id: + parent_ids.add(row.parent_download_job_id) + session.add( + DownloadJobEventRow( + download_job_id=row.parent_download_job_id, + event_type="warning", + message=f"Shard {row.shard_index} abandonado (agent reconectado)", + timestamp=now, + metadata_json={"shard_id": row.id, "worker_id": worker_id}, + ) + ) + return list(parent_ids) def heartbeat(self, worker_id: str) -> Optional[Worker]: with session_scope() as session: @@ -609,6 +672,9 @@ class JobStore: w.current_job_id = None w.status = WorkerStatus.OFFLINE.value elif not w.current_job_id: + fresh_until = self._fresh_until.get(w.id) + if fresh_until is not None and now < fresh_until: + continue online_free.append(w) if not offline_ids: diff --git a/services/bot_descarga/tests/test_store.py b/services/bot_descarga/tests/test_store.py index b133ded..0353ce7 100644 --- a/services/bot_descarga/tests/test_store.py +++ b/services/bot_descarga/tests/test_store.py @@ -80,6 +80,32 @@ def test_create_job_requires_worker_and_claim_is_atomic(store): assert other is None +def test_reregister_abandons_leftover_job(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", + ) + job_store.claim_pending("PC-01") + + job_store.register_worker("PC-01") + leftover = job_store.get_job(job.id) + assert leftover is not None + assert leftover.status == JobStatus.CANCELLED.value + assert job_store.claim_pending("PC-01") is None + + fresh = job_store.create_job( + worker_id="PC-01", + excel_key="uploads/y/input.xlsx", + original_filename="nuevo.xlsx", + ) + claimed = job_store.claim_pending("PC-01") + assert claimed is not None + assert claimed.id == fresh.id + + def test_cancel_pending_marks_cancelled(store): job_store, JobStatus, _WorkerStatus = store job_store.register_worker("PC-01") diff --git a/services/bot_digitador/infrastructure/store.py b/services/bot_digitador/infrastructure/store.py index 3d7dd74..5894bd7 100644 --- a/services/bot_digitador/infrastructure/store.py +++ b/services/bot_digitador/infrastructure/store.py @@ -27,6 +27,7 @@ from infrastructure.database import ( from infrastructure.storage_minio import reports_prefix HEARTBEAT_OFFLINE_SECONDS = 45 +_FRESH_WORKER_GRACE_SECONDS = 45 def _parse_iso(value: str | None) -> datetime | None: @@ -44,6 +45,7 @@ class JobStore: def __init__(self, db_path: str | None = None): # db_path ignorado (compat); usa DATABASE_URL. init_db() + self._fresh_until: dict[str, datetime] = {} def _row_worker(self, row: DigitadorWorkerRow) -> Worker: worker = Worker( @@ -120,7 +122,67 @@ class JobStore: ) session.add(row) session.flush() - return self._row_worker(row) + parent_ids = self._abandon_assigned_jobs( + session, + worker_id, + now, + "Agent reconectado: job anterior abandonado. Lanza una corrida nueva desde el desktop.", + ) + session.flush() + worker = self._row_worker(row) + self._fresh_until[worker_id] = now + timedelta(seconds=_FRESH_WORKER_GRACE_SECONDS) + for parent_id in parent_ids: + try: + self.refresh_parent_from_shards(parent_id) + except Exception: + pass + return worker + + def _abandon_assigned_jobs( + self, session, worker_id: str, now: datetime, message: str + ) -> list[str]: + """Cancela pending/running de este worker para que un agent nuevo arranque idle.""" + rows = session.scalars( + select(DigitadorJobRow).where( + DigitadorJobRow.worker_id == worker_id, + DigitadorJobRow.status.in_((JobStatus.PENDING.value, JobStatus.RUNNING.value)), + ) + ).all() + parent_ids: set[str] = set() + for row in rows: + is_parent = ( + row.parent_digitador_job_id is None + and row.shard_count is not None + and int(row.shard_count) > 1 + ) + if is_parent: + continue + row.status = JobStatus.CANCELLED.value + row.cancel_requested = True + row.finished_at = now + row.error = message + self._sync_task(session, row, status=JobStatus.CANCELLED.value, error=message) + session.add( + DigitadorJobEventRow( + digitador_job_id=row.id, + event_type="warning", + message=message, + timestamp=now, + metadata_json={"abandoned_on_register": True, "worker_id": worker_id}, + ) + ) + if row.parent_digitador_job_id: + parent_ids.add(row.parent_digitador_job_id) + session.add( + DigitadorJobEventRow( + digitador_job_id=row.parent_digitador_job_id, + event_type="warning", + message=f"Shard {row.shard_index} abandonado (agent reconectado)", + timestamp=now, + metadata_json={"shard_id": row.id, "worker_id": worker_id}, + ) + ) + return list(parent_ids) def heartbeat(self, worker_id: str) -> Optional[Worker]: with session_scope() as session: @@ -609,6 +671,9 @@ class JobStore: w.current_job_id = None w.status = WorkerStatus.OFFLINE.value elif not w.current_job_id: + fresh_until = self._fresh_until.get(w.id) + if fresh_until is not None and now < fresh_until: + continue online_free.append(w) if not offline_ids: diff --git a/services/desktop_app/electron/autoUpdate.cjs b/services/desktop_app/electron/autoUpdate.cjs index c123249..3d0bceb 100644 --- a/services/desktop_app/electron/autoUpdate.cjs +++ b/services/desktop_app/electron/autoUpdate.cjs @@ -210,7 +210,7 @@ async function downloadAndInstall(manifest, onProgress) { setTimeout(() => { app.quit(); - }, 400); + }, 2500); return { ok: true, destPath }; } diff --git a/services/desktop_app/package.json b/services/desktop_app/package.json index 19b8d87..1a8e877 100644 --- a/services/desktop_app/package.json +++ b/services/desktop_app/package.json @@ -1,6 +1,6 @@ { "name": "ipsextractor-desktop", - "version": "1.2.0", + "version": "1.3.0", "private": true, "description": "Desktop extraction client for IPSExtractor microservices.", "main": "electron/main.cjs", diff --git a/services/desktop_app/src/renderer/components/UpdatePrompt.tsx b/services/desktop_app/src/renderer/components/UpdatePrompt.tsx index 6d39ed1..8a63cba 100644 --- a/services/desktop_app/src/renderer/components/UpdatePrompt.tsx +++ b/services/desktop_app/src/renderer/components/UpdatePrompt.tsx @@ -244,7 +244,13 @@ export function useAutoUpdateCheck() { try { await window.desktopApi.installUpdate(manifest); } catch (e) { - setError(e instanceof Error ? e.message : String(e)); + const msg = e instanceof Error ? e.message : String(e); + if (msg.toLowerCase().includes("aborted")) { + setProgress({ phase: "installing", percent: 100 }); + setCheckMessage("El instalador se inició. Si la app no se cierra sola, vuelve a abrirla después de unos segundos."); + return; + } + setError(msg); setInstalling(false); } } diff --git a/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx b/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx index 6364be8..355667c 100644 --- a/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx +++ b/services/desktop_app/src/renderer/modules/digitador/DigitadorWizardModule.tsx @@ -13,6 +13,7 @@ import { } from "lucide-react"; import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import type { ApiClient } from "../../lib/apiClient"; +import { formatDateTime } from "../../lib/time"; import type { DigitadorJob, DigitadorJobEvent, @@ -305,21 +306,107 @@ export function DigitadorWizardModule({ client, job, initialSource = null }: Pro
{step === 1 && (
-
-

- - Bot digitador -

-

- Digita en Integración Total solo los aptos (PASO_BOT), o un CSV - propio con el mismo formato. +

+
+

+ Bots disponibles + {onlineWorkers.length > 0 && ( + {onlineWorkers.length} online + )} +

+ +
+

+ Con varios bots online el CSV se reparte automáticamente; los reportes RESULTADOS / + OMITIDOS quedan en un solo resultado.

+ + {onlineWorkers.length >= 2 && ( + + )} + + {workers.length === 0 ? ( +

Ningún worker registrado. Enciende el agent en el PC del bot.

+ ) : ( +
+ {workers.map((w) => { + const online = w.status === "online" || w.status === "busy"; + const selected = effectiveTargets.includes(w.id); + return ( + + ); + })} +
+ )} + {onlineWorkers.length === 0 && workers.length > 0 && ( +

No hay bots online. Inicia el agent en la PC correspondiente.

+ )} + {effectiveTargets.length > 1 && ( +

+ Se repartirá entre {effectiveTargets.length} bots. +

+ )}
- {/* 1. Origen */}
-

1. Origen

-
+

Origen del CSV

+

+ Solo aptos (PASO_BOT) de población apta, o un CSV propio con el mismo formato. +

+
- Solo se envían filas PASO_BOT. Los no aptos no se digitan. + Solo se envían filas PASO_BOT.
@@ -462,115 +550,6 @@ export function DigitadorWizardModule({ client, job, initialSource = null }: Pro )}
- - {/* 2. Bots */} -
-
-

- 2. Bots - {onlineWorkers.length > 0 && ( - {onlineWorkers.length} online - )} -

- -
-

- El agent debe estar encendido en la PC con Integración Total. Con varios bots, las - filas se reparte automáticamente. -

- - {onlineWorkers.length >= 2 && ( - - )} - - {workers.length === 0 ? ( -
-
- -
-
Ningún bot conectado
-

- En la PC del digitador ejecuta run_gui.py del agent y conéctalo - al gateway. -

-
-
-
- ) : ( -
- {workers.map((w) => { - const online = w.status === "online" || w.status === "busy"; - const selected = effectiveTargets.includes(w.id); - return ( - - ); - })} -
- )} - {onlineWorkers.length === 0 && workers.length > 0 && ( -

- Hay workers registrados pero ninguno online. Revisa el agent. -

- )} - {effectiveTargets.length > 1 && ( -

- Se repartirá entre {effectiveTargets.length} bots. -

- )} -
)}