fix: parallelize WA sync and surface scheduler errors in historial
- whatsapp_sync: send patients concurrently (asyncio.gather + semaphore=10) instead of sequentially — eliminates long waits with many patients - envios_erp: auto-refresh historial every 60s when scheduler tab is active - main: wrap scheduler job to catch exceptions and log them to sync_wa_log so failures are visible in historial Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
5ca97159b4
commit
cffeae7e94
@@ -5,6 +5,7 @@ Mapea campos Firebird al formato esperado por /api/lab/ingest_paciente.php.
|
|||||||
|
|
||||||
import re
|
import re
|
||||||
import json
|
import json
|
||||||
|
import asyncio
|
||||||
import httpx
|
import httpx
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
@@ -175,25 +176,36 @@ async def sync_paciente(
|
|||||||
|
|
||||||
|
|
||||||
async def sync_todos(rows: list, url: str, api_key: str, timeout: int = 30, modo: str = "upsert",
|
async def sync_todos(rows: list, url: str, api_key: str, timeout: int = 30, modo: str = "upsert",
|
||||||
examenes_map: Optional[dict] = None) -> dict:
|
examenes_map: Optional[dict] = None, concurrencia: int = 10) -> dict:
|
||||||
"""Envía una lista de filas de pacientes. Retorna resumen {total, created, skipped, updated, errores}."""
|
"""Envía una lista de filas de pacientes en paralelo. Retorna resumen {total, created, skipped, updated, errores}."""
|
||||||
resultado = {"total": len(rows), "created": 0, "skipped": 0, "updated": 0, "errores": 0, "detalle": []}
|
resultado = {"total": len(rows), "created": 0, "skipped": 0, "updated": 0, "errores": 0, "detalle": []}
|
||||||
|
sem = asyncio.Semaphore(concurrencia)
|
||||||
|
|
||||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
async def _enviar(row: dict, client: httpx.AsyncClient) -> dict:
|
||||||
for row in rows:
|
async with sem:
|
||||||
doc = str(row.get("DOCIDENT") or "").strip()
|
doc = str(row.get("DOCIDENT") or "").strip()
|
||||||
examenes = examenes_map.get(doc) if examenes_map else None
|
examenes = examenes_map.get(doc) if examenes_map else None
|
||||||
r = await sync_paciente(row, url, api_key, client, modo, examenes)
|
return await sync_paciente(row, url, api_key, client, modo, examenes)
|
||||||
if r["ok"]:
|
|
||||||
action = r["action"]
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||||
if action == "created":
|
tasks = [_enviar(row, client) for row in rows]
|
||||||
resultado["created"] += 1
|
resultados = await asyncio.gather(*tasks, return_exceptions=True)
|
||||||
elif action == "skipped":
|
|
||||||
resultado["skipped"] += 1
|
for r in resultados:
|
||||||
else:
|
if isinstance(r, Exception):
|
||||||
resultado["updated"] += 1
|
resultado["errores"] += 1
|
||||||
|
resultado["detalle"].append({"ok": False, "action": "error", "message": str(r), "doc": "", "nombre": ""})
|
||||||
|
continue
|
||||||
|
if r["ok"]:
|
||||||
|
action = r["action"]
|
||||||
|
if action == "created":
|
||||||
|
resultado["created"] += 1
|
||||||
|
elif action == "skipped":
|
||||||
|
resultado["skipped"] += 1
|
||||||
else:
|
else:
|
||||||
resultado["errores"] += 1
|
resultado["updated"] += 1
|
||||||
resultado["detalle"].append(r)
|
else:
|
||||||
|
resultado["errores"] += 1
|
||||||
|
resultado["detalle"].append(r)
|
||||||
|
|
||||||
return resultado
|
return resultado
|
||||||
|
|||||||
@@ -151,6 +151,7 @@ const _TAB_ACTIVE = 'border-green-500 text-green-700 bg-green-50';
|
|||||||
const _TAB_INACTIVE = 'border-transparent text-gray-500 hover:text-gray-700 hover:border-gray-300';
|
const _TAB_INACTIVE = 'border-transparent text-gray-500 hover:text-gray-700 hover:border-gray-300';
|
||||||
|
|
||||||
let _schedulerLoaded = false;
|
let _schedulerLoaded = false;
|
||||||
|
let _histRefreshTimer = null;
|
||||||
|
|
||||||
function switchTab(name) {
|
function switchTab(name) {
|
||||||
document.querySelectorAll('.tab-panel').forEach(p => p.classList.add('hidden'));
|
document.querySelectorAll('.tab-panel').forEach(p => p.classList.add('hidden'));
|
||||||
@@ -168,10 +169,17 @@ function switchTab(name) {
|
|||||||
const iframe = document.getElementById('iframe-pacientes');
|
const iframe = document.getElementById('iframe-pacientes');
|
||||||
if (!iframe.src) iframe.src = '/pacientes?embed=1';
|
if (!iframe.src) iframe.src = '/pacientes?embed=1';
|
||||||
}
|
}
|
||||||
if (name === 'scheduler' && !_schedulerLoaded) {
|
if (name === 'scheduler') {
|
||||||
_schedulerLoaded = true;
|
if (!_schedulerLoaded) {
|
||||||
cargarHistorialScheduler();
|
_schedulerLoaded = true;
|
||||||
pollSchedulerStatus();
|
cargarHistorialScheduler();
|
||||||
|
pollSchedulerStatus();
|
||||||
|
}
|
||||||
|
// Auto-refresh historial cada 60s mientras el tab está activo
|
||||||
|
clearInterval(_histRefreshTimer);
|
||||||
|
_histRefreshTimer = setInterval(cargarHistorialScheduler, 60000);
|
||||||
|
} else {
|
||||||
|
clearInterval(_histRefreshTimer);
|
||||||
}
|
}
|
||||||
|
|
||||||
sessionStorage.setItem('erp-tab', name);
|
sessionStorage.setItem('erp-tab', name);
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|||||||
from app.database import init_db
|
from app.database import init_db
|
||||||
from app.auth import decode_token
|
from app.auth import decode_token
|
||||||
from app.services.scheduler import sync_recientes
|
from app.services.scheduler import sync_recientes
|
||||||
|
from app.services.whatsapp_sync import guardar_sync_log
|
||||||
|
|
||||||
app = FastAPI(title="RIPS Manager", version="1.0.0")
|
app = FastAPI(title="RIPS Manager", version="1.0.0")
|
||||||
_scheduler = AsyncIOScheduler()
|
_scheduler = AsyncIOScheduler()
|
||||||
@@ -63,8 +64,27 @@ async def startup():
|
|||||||
config_defaults()
|
config_defaults()
|
||||||
query_defaults()
|
query_defaults()
|
||||||
contratos_defaults()
|
contratos_defaults()
|
||||||
_scheduler.add_job(sync_recientes, "interval", minutes=1, id="wa_sync_recientes",
|
async def _job_sync():
|
||||||
kwargs={"ventana_min": 10}, max_instances=1, coalesce=True)
|
try:
|
||||||
|
resultado = await sync_recientes(ventana_min=10)
|
||||||
|
if not resultado.get("ok") and resultado.get("error"):
|
||||||
|
guardar_sync_log(
|
||||||
|
{"total": 0, "created": 0, "skipped": 0, "updated": 0, "errores": 1, "detalle": [
|
||||||
|
{"ok": False, "action": "error", "message": resultado["error"], "doc": "", "nombre": "scheduler-error"}
|
||||||
|
]},
|
||||||
|
None, origen="scheduler", modo="upsert"
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
import traceback
|
||||||
|
guardar_sync_log(
|
||||||
|
{"total": 0, "created": 0, "skipped": 0, "updated": 0, "errores": 1, "detalle": [
|
||||||
|
{"ok": False, "action": "error", "message": str(exc), "doc": "", "nombre": "scheduler-exception"}
|
||||||
|
]},
|
||||||
|
None, origen="scheduler", modo="upsert"
|
||||||
|
)
|
||||||
|
|
||||||
|
_scheduler.add_job(_job_sync, "interval", minutes=1, id="wa_sync_recientes",
|
||||||
|
max_instances=1, coalesce=True)
|
||||||
_scheduler.start()
|
_scheduler.start()
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user