Refactor digitador bot and add runtime monitoring

This commit is contained in:
erik
2026-08-30 14:02:58 -05:00
parent 3b01456875
commit d9e2012e22
46 changed files with 1720 additions and 248 deletions
+10
View File
@@ -0,0 +1,10 @@
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
@@ -0,0 +1,37 @@
from __future__ import annotations
from types import SimpleNamespace
from bot_digitador.core.application import select_document_type
class FakeWrapper:
def __init__(self):
self.keys: list[str] = []
def type_keys(self, value: str, **_kwargs) -> None:
self.keys.append(value)
def test_select_document_type_uses_home_and_exact_c_sequence():
wrapper = FakeWrapper()
window = SimpleNamespace(wrapper_object=lambda: wrapper)
clicks: list[tuple[int, int]] = []
logs: list[str] = []
ok = select_document_type(
window,
"CC",
log=logs.append,
sleep=lambda _seconds: None,
click=lambda _window, x, y: clicks.append((x, y)),
)
assert ok is True
assert clicks == [(729, 279)]
assert wrapper.keys == ["{HOME}", "c", "c", "c", "{ENTER}"]
def test_select_document_type_rejects_unknown_type():
assert select_document_type(SimpleNamespace(), "XX", log=lambda _message: None) is False
+17
View File
@@ -0,0 +1,17 @@
from __future__ import annotations
from pathlib import Path
from bot_digitador.core.coords import DEFAULT_APP_PATH, resolve_app_path
def test_default_app_path_uses_current_user_home():
assert Path(DEFAULT_APP_PATH) == Path.home() / "Documents" / "IT SaludTotal.lnk"
def test_app_path_expands_worker_user_profile(monkeypatch):
monkeypatch.setenv("USERPROFILE", r"C:\Users\worker-02")
resolved = resolve_app_path(r"%USERPROFILE%\Documents\IT SaludTotal.lnk")
assert resolved == r"C:\Users\worker-02\Documents\IT SaludTotal.lnk"
@@ -0,0 +1,33 @@
from __future__ import annotations
from bot_digitador.core.csv_rules import (
convert_date,
detect_csv_delimiter,
format_money_for_input,
parse_money,
valid_document_types,
)
HEADERS = "radicado|documento|ingreso|total_cargos|autorizacion|tipo_documento|tipo_servicio|copago|cuota_moderadora\n"
def test_detects_pipe_csv(tmp_path):
source = tmp_path / "input.csv"
source.write_text(HEADERS + "R1|123|2026-08-30|100|A|CC|ambulatorio|0|0\n", encoding="utf-8")
assert detect_csv_delimiter(str(source), preferred=",") == "|"
def test_dates_and_money_match_original_rules():
assert convert_date("30/08/2026", "dd/mm/yyyy") == ("08302026", True)
assert convert_date("2026-08-30", "dd/mm/yyyy") == ("08302026", True)
assert parse_money("$ 12.500") == 12500.0
assert parse_money("12,50") == 12.5
assert format_money_for_input("12,50") == "12,5"
def test_document_candidates_preserve_strategy_rules():
candidates = valid_document_types("1234567890")
assert {"CC", "TI", "RC", "CN", "AS", "ME", "PS"}.issubset(candidates)
assert valid_document_types("") == []
@@ -0,0 +1,35 @@
from __future__ import annotations
import importlib.util
from pathlib import Path
from zipfile import ZipFile
from bot_digitador.core.config import REQUIRED_TEMPLATE_NAMES
SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "package_bot.py"
SPEC = importlib.util.spec_from_file_location("digitador_package_bot", SCRIPT)
assert SPEC and SPEC.loader
package_bot = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(package_bot)
def test_package_is_clean_and_contains_required_templates(tmp_path):
root = tmp_path / "source"
(root / "bot_digitador" / "core").mkdir(parents=True)
(root / "template").mkdir()
for name in ("README.md", "requirements.txt", "run_gui.py", "main.py", ".env.example"):
(root / name).write_text(name, encoding="utf-8")
(root / "bot_digitador" / "core" / "service.py").write_text("pass", encoding="utf-8")
(root / "bot_digitador" / "core" / "cached.pyc").write_bytes(b"cache")
(root / "data").mkdir()
(root / "data" / "gui_settings.json").write_text("secret", encoding="utf-8")
for name in REQUIRED_TEMPLATE_NAMES:
(root / "template" / name).write_bytes(b"png")
target = package_bot.build_package(root=root, output=tmp_path / "bot.zip")
with ZipFile(target) as archive:
names = archive.namelist()
assert "bot_digitador/template/radicado_existente.png" in names
assert not any("__pycache__" in name or name.endswith(".pyc") or "/data/" in name for name in names)
@@ -0,0 +1,7 @@
from bot_digitador.ui.progress_overlay import format_duration
def test_format_duration_for_local_viewer():
assert format_duration(0) == "00:00:00"
assert format_duration(65) == "00:01:05"
assert format_duration(3661) == "01:01:01"
@@ -0,0 +1,33 @@
from __future__ import annotations
import csv
import pytest
from bot_digitador.core.config import REQUIRED_TEMPLATE_NAMES
from bot_digitador.core.reports import write_reports
from bot_digitador.core.vision import missing_templates, validate_templates
def test_template_inventory_reports_every_missing_asset(tmp_path):
assert missing_templates(str(tmp_path)) == list(REQUIRED_TEMPLATE_NAMES)
with pytest.raises(RuntimeError, match="Faltan 9 plantillas"):
validate_templates(str(tmp_path))
for name in REQUIRED_TEMPLATE_NAMES:
(tmp_path / name).write_bytes(b"png")
validate_templates(str(tmp_path))
def test_report_writer_creates_all_outputs(tmp_path):
omitted = [{"radicado": "R2", "estado": "OMITIDO", "tipo_error": "ERROR"}]
results = [{
"radicado": "R1", "estado": "OK", "tipo_error": "", "motivo": "Cuenta registrada",
"detalle": "BIEN", "fecha_hora": "2026-08-30 12:00:00", "duracion_segundos": 10,
"accion_bot": "Continuó",
}]
write_reports(str(tmp_path), log_entries=["inicio"], omitted_records=omitted, results=results)
assert (tmp_path / "ejecucion_bot.log").read_text(encoding="utf-8") == "inicio\n"
with (tmp_path / "RESULTADOS.csv").open(encoding="utf-8-sig") as source:
assert list(csv.DictReader(source))[0]["radicado"] == "R1"
assert (tmp_path / "OMITIDOS.csv").is_file()
@@ -0,0 +1,29 @@
from __future__ import annotations
from bot_digitador.core.digitador_service import DigitadorService
def test_events_include_remote_overlay_metrics(tmp_path):
events = []
service = DigitadorService(
csv_path=str(tmp_path / "input.csv"),
output_dir=str(tmp_path / "out"),
event_callback=events.append,
)
service.inicio_proceso = None
service.total_registros = 10
service.procesados_ok = 2
service.errores = 1
service.existentes = 1
service.reinicios = 3
service.registro_actual = "RAD-4"
service.estado_actual = "Procesando"
service.ultimo_evento = "Registro 4/10"
service._emit("radicado_processing", "Procesando RAD-4")
metadata = events[0].metadata
assert metadata["estado"] == "Procesando"
assert metadata["processed"] == 4
assert metadata["reinicios"] == 3
assert metadata["radicado_actual"] == "RAD-4"