arreglo de la sincronizacion de WhatsApp: paso de 74 s a 0.15 s (insercion por lotes, sincronizacion incremental por marcas de tiempo, contactos solo los referenciados, poda unicamente en el ciclo completo) + fin del bucle infinito de peticiones en la bandeja y refresco cada 30 s con indicador de hora + aviso inmediato desde el puente de WhatsApp
This commit is contained in:
@@ -6,6 +6,7 @@ import Layout from '../components/Layout';
|
|||||||
import { Card, Button, Input, EmptyState, Skeleton, toast, Badge, Modal } from '../components/ui';
|
import { Card, Button, Input, EmptyState, Skeleton, toast, Badge, Modal } from '../components/ui';
|
||||||
import { odooApi, type WacrmConversation, type WacrmMessage, type Patient, type WacrmTag } from '../services/odoo';
|
import { odooApi, type WacrmConversation, type WacrmMessage, type Patient, type WacrmTag } from '../services/odoo';
|
||||||
import { useAuth } from '../lib/auth';
|
import { useAuth } from '../lib/auth';
|
||||||
|
import { useAutoRefresh, AutoRefreshBadge } from '../lib/autoRefresh';
|
||||||
import WacrmMetrics from '../components/wacrm/WacrmMetrics';
|
import WacrmMetrics from '../components/wacrm/WacrmMetrics';
|
||||||
import ConversationNotas from '../components/wacrm/ConversationNotas';
|
import ConversationNotas from '../components/wacrm/ConversationNotas';
|
||||||
import ConversationEtiquetas from '../components/wacrm/ConversationEtiquetas';
|
import ConversationEtiquetas from '../components/wacrm/ConversationEtiquetas';
|
||||||
@@ -18,9 +19,35 @@ interface Member {
|
|||||||
role: string;
|
role: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// El backend devuelve marcas ISO con microsegundos: se recortan a 3 dígitos
|
||||||
|
// para que `new Date()` las entienda en todos los navegadores.
|
||||||
|
const parseMarca = (iso?: string | null): Date | null => {
|
||||||
|
if (!iso) return null;
|
||||||
|
const limpio = iso.replace(/(\.\d{3})\d+/, '$1');
|
||||||
|
const fecha = new Date(limpio);
|
||||||
|
return Number.isNaN(fecha.getTime()) ? null : fecha;
|
||||||
|
};
|
||||||
|
|
||||||
|
const hhmmss = (fecha: Date | null) =>
|
||||||
|
fecha ? fecha.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit', second: '2-digit' }) : null;
|
||||||
|
|
||||||
|
// Huella de los campos que la bandeja pinta: sirve para no re-renderizar
|
||||||
|
// cuando el refresco trae exactamente lo mismo.
|
||||||
|
const firmaConversacion = (c: WacrmConversation) => JSON.stringify([
|
||||||
|
c.last_message_at, c.last_message_text, c.unread_count, c.status,
|
||||||
|
c.assigned_agent_id, c.assigned_agent, (c.tags || []).map((t) => t.id),
|
||||||
|
]);
|
||||||
|
|
||||||
|
// Cada cuánto se sincroniza el caché del servidor. Cada ciclo es incremental
|
||||||
|
// (milisegundos), así que 30 s es holgado y deja la bandeja viva.
|
||||||
|
const SYNC_INTERVAL_MS = 30000;
|
||||||
|
|
||||||
const WacrmMessages: FC = () => {
|
const WacrmMessages: FC = () => {
|
||||||
const [conversations, setConversations] = useState<WacrmConversation[]>([]);
|
const [conversations, setConversations] = useState<WacrmConversation[]>([]);
|
||||||
const [messages, setMessages] = useState<WacrmMessage[]>([]);
|
const [messages, setMessages] = useState<WacrmMessage[]>([]);
|
||||||
|
// Espejo de `messages` para que `loadMessages` no dependa del estado (ver
|
||||||
|
// comentario en loadMessages).
|
||||||
|
const messagesRef = useRef<WacrmMessage[]>([]);
|
||||||
const [msgPage, setMsgPage] = useState(1);
|
const [msgPage, setMsgPage] = useState(1);
|
||||||
const [msgHasMore, setMsgHasMore] = useState(false);
|
const [msgHasMore, setMsgHasMore] = useState(false);
|
||||||
const [loadingOlder, setLoadingOlder] = useState(false);
|
const [loadingOlder, setLoadingOlder] = useState(false);
|
||||||
@@ -31,7 +58,8 @@ const WacrmMessages: FC = () => {
|
|||||||
const cercaDelFondoRef = useRef(true);
|
const cercaDelFondoRef = useRef(true);
|
||||||
const [members, setMembers] = useState<Member[]>([]);
|
const [members, setMembers] = useState<Member[]>([]);
|
||||||
const [loading, setLoading] = useState(true);
|
const [loading, setLoading] = useState(true);
|
||||||
const [syncing, setSyncing] = useState(false);
|
// Hora (del servidor) de la última sincronización del caché con WACRM.
|
||||||
|
const [lastSyncAt, setLastSyncAt] = useState<string | null>(null);
|
||||||
const [search, setSearch] = useState('');
|
const [search, setSearch] = useState('');
|
||||||
const [selected, setSelected] = useState<WacrmConversation | null>(null);
|
const [selected, setSelected] = useState<WacrmConversation | null>(null);
|
||||||
const [assignOpen, setAssignOpen] = useState(false);
|
const [assignOpen, setAssignOpen] = useState(false);
|
||||||
@@ -77,6 +105,10 @@ const WacrmMessages: FC = () => {
|
|||||||
searchRef.current = search;
|
searchRef.current = search;
|
||||||
}, [search]);
|
}, [search]);
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
messagesRef.current = messages;
|
||||||
|
}, [messages]);
|
||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
if (currentAgentId && typeof window !== 'undefined') {
|
if (currentAgentId && typeof window !== 'undefined') {
|
||||||
localStorage.setItem('skeen_current_agent_id', currentAgentId);
|
localStorage.setItem('skeen_current_agent_id', currentAgentId);
|
||||||
@@ -88,11 +120,13 @@ const WacrmMessages: FC = () => {
|
|||||||
const res = await odooApi.getWacrmConversations(searchRef.current);
|
const res = await odooApi.getWacrmConversations(searchRef.current);
|
||||||
if (res.status === 'success') {
|
if (res.status === 'success') {
|
||||||
setConversations(res.conversations);
|
setConversations(res.conversations);
|
||||||
|
if (res.last_sync_at) setLastSyncAt(res.last_sync_at);
|
||||||
const currentSelected = selectedRef.current;
|
const currentSelected = selectedRef.current;
|
||||||
if (currentSelected) {
|
if (currentSelected) {
|
||||||
const updated = res.conversations.find((c) => c.external_id === currentSelected.external_id);
|
const updated = res.conversations.find((c) => c.external_id === currentSelected.external_id);
|
||||||
if (updated && updated.external_id !== currentSelected.external_id) {
|
// La conversación abierta también se refresca (último mensaje, no
|
||||||
// Solo actualizar si realmente cambió la referencia (datos nuevos)
|
// leídos, asignación...), pero solo si de verdad cambió algo.
|
||||||
|
if (updated && firmaConversacion(updated) !== firmaConversacion(currentSelected)) {
|
||||||
setSelected(updated);
|
setSelected(updated);
|
||||||
}
|
}
|
||||||
} else if (!userSelectedRef.current && res.conversations.length > 0) {
|
} else if (!userSelectedRef.current && res.conversations.length > 0) {
|
||||||
@@ -138,9 +172,10 @@ const WacrmMessages: FC = () => {
|
|||||||
setMsgHasMore(res.has_more);
|
setMsgHasMore(res.has_more);
|
||||||
if (page === 1) {
|
if (page === 1) {
|
||||||
// Merge: conserva el historial antiguo ya cargado (scroll manual)
|
// Merge: conserva el historial antiguo ya cargado (scroll manual)
|
||||||
|
const previos = messagesRef.current;
|
||||||
const known = new Set(res.messages.map((m) => m.external_id));
|
const known = new Set(res.messages.map((m) => m.external_id));
|
||||||
const minTs = res.messages.length ? res.messages[0].created_at : '';
|
const minTs = res.messages.length ? res.messages[0].created_at : '';
|
||||||
const older = messages.filter((m) => !known.has(m.external_id) && (m.created_at ?? '') < (minTs ?? ''));
|
const older = previos.filter((m) => !known.has(m.external_id) && (m.created_at ?? '') < (minTs ?? ''));
|
||||||
setMessages([...older, ...res.messages]);
|
setMessages([...older, ...res.messages]);
|
||||||
// Solo baja al final si cambió de conversación o el usuario ya estaba
|
// Solo baja al final si cambió de conversación o el usuario ya estaba
|
||||||
// abajo; si estaba leyendo mensajes anteriores, se respeta su posición.
|
// abajo; si estaba leyendo mensajes anteriores, se respeta su posición.
|
||||||
@@ -148,7 +183,7 @@ const WacrmMessages: FC = () => {
|
|||||||
scrollThreadToBottom();
|
scrollThreadToBottom();
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
const prev = messages;
|
const prev = messagesRef.current;
|
||||||
const known = new Set(prev.map((m) => m.external_id));
|
const known = new Set(prev.map((m) => m.external_id));
|
||||||
const older = res.messages.filter((m) => !known.has(m.external_id));
|
const older = res.messages.filter((m) => !known.has(m.external_id));
|
||||||
setMessages([...older, ...prev]);
|
setMessages([...older, ...prev]);
|
||||||
@@ -163,7 +198,10 @@ const WacrmMessages: FC = () => {
|
|||||||
toast.error('Error al cargar mensajes');
|
toast.error('Error al cargar mensajes');
|
||||||
console.error(err);
|
console.error(err);
|
||||||
}
|
}
|
||||||
}, [messages]);
|
// Sin dependencia de `messages` a propósito: si la tuviera, cada mensaje
|
||||||
|
// nuevo la recrearía y el efecto que la llama volvería a pedir el hilo en
|
||||||
|
// bucle (una petición por render).
|
||||||
|
}, []);
|
||||||
|
|
||||||
const loadMembers = useCallback(async () => {
|
const loadMembers = useCallback(async () => {
|
||||||
try {
|
try {
|
||||||
@@ -176,25 +214,35 @@ const WacrmMessages: FC = () => {
|
|||||||
}
|
}
|
||||||
}, []);
|
}, []);
|
||||||
|
|
||||||
const sync = async () => {
|
// Un ciclo = sincronizar el caché del servidor y volver a leer la bandeja y
|
||||||
|
// el hilo abierto. El hook de auto-refresco se encarga de que nunca haya dos
|
||||||
|
// ciclos a la vez y de que no corra con la pestaña oculta.
|
||||||
|
const refreshAll = useCallback(async () => {
|
||||||
try {
|
try {
|
||||||
setSyncing(true);
|
|
||||||
const res = await odooApi.syncWacrm();
|
const res = await odooApi.syncWacrm();
|
||||||
if (res.status === 'success') {
|
if (res.status === 'success') {
|
||||||
toast.success(`Sincronizado: ${JSON.stringify(res.result)}`);
|
if (res.result?.last_sync_at) setLastSyncAt(String(res.result.last_sync_at));
|
||||||
await loadConversations();
|
|
||||||
} else if (res.status === 'busy') {
|
} else if (res.status === 'busy') {
|
||||||
// La auto-sincronización cada 15 s ya está corriendo
|
// Otro ciclo (esta u otra pestaña) está corriendo: no es un error.
|
||||||
toast.info('Ya hay una sincronización en curso');
|
console.info('Sincronización WACRM en curso; se reintenta al siguiente ciclo');
|
||||||
await loadConversations();
|
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
toast.error('Error al sincronizar con WACRM');
|
// Un fallo de sincronización no debe romper la bandeja: se sigue viendo
|
||||||
console.error(err);
|
// el caché y se reintenta en el siguiente ciclo.
|
||||||
} finally {
|
console.error('Error al sincronizar con WACRM', err);
|
||||||
setSyncing(false);
|
|
||||||
}
|
}
|
||||||
};
|
await loadConversations();
|
||||||
|
const current = selectedRef.current;
|
||||||
|
if (current) await loadMessages(current.external_id);
|
||||||
|
}, [loadConversations, loadMessages]);
|
||||||
|
|
||||||
|
const auto = useAutoRefresh({
|
||||||
|
onRefresh: refreshAll,
|
||||||
|
intervalMs: SYNC_INTERVAL_MS,
|
||||||
|
storageKey: 'skeen_wacrm_autorefresh_paused',
|
||||||
|
});
|
||||||
|
|
||||||
|
const sync = useCallback(() => auto.refreshNow(), [auto]);
|
||||||
|
|
||||||
const loadTags = useCallback(async () => {
|
const loadTags = useCallback(async () => {
|
||||||
try {
|
try {
|
||||||
@@ -289,17 +337,9 @@ const WacrmMessages: FC = () => {
|
|||||||
}
|
}
|
||||||
}, [selected, loadMessages]);
|
}, [selected, loadMessages]);
|
||||||
|
|
||||||
// Auto-sync cada 15 segundos
|
// El ciclo de sincronización + refresco lo lleva `useAutoRefresh` (ver
|
||||||
useEffect(() => {
|
// `auto` arriba): espera a que termine el anterior, no corre con la pestaña
|
||||||
const interval = setInterval(() => {
|
// oculta y se puede pausar desde el indicador.
|
||||||
loadConversations();
|
|
||||||
const currentSelected = selectedRef.current;
|
|
||||||
if (currentSelected) {
|
|
||||||
loadMessages(currentSelected.external_id);
|
|
||||||
}
|
|
||||||
}, 15000);
|
|
||||||
return () => clearInterval(interval);
|
|
||||||
}, [loadConversations, loadMessages]);
|
|
||||||
|
|
||||||
const handleSelect = useCallback((c: WacrmConversation) => {
|
const handleSelect = useCallback((c: WacrmConversation) => {
|
||||||
userSelectedRef.current = true;
|
userSelectedRef.current = true;
|
||||||
@@ -402,7 +442,7 @@ const WacrmMessages: FC = () => {
|
|||||||
className="pl-10"
|
className="pl-10"
|
||||||
/>
|
/>
|
||||||
</div>
|
</div>
|
||||||
<div className="flex items-center gap-3">
|
<div className="flex flex-wrap items-center gap-3">
|
||||||
<div className="relative">
|
<div className="relative">
|
||||||
<select
|
<select
|
||||||
value={currentAgentId}
|
value={currentAgentId}
|
||||||
@@ -416,7 +456,18 @@ const WacrmMessages: FC = () => {
|
|||||||
</select>
|
</select>
|
||||||
<ChevronDown size={14} className="absolute right-2.5 top-1/2 -translate-y-1/2 text-theme-muted pointer-events-none" />
|
<ChevronDown size={14} className="absolute right-2.5 top-1/2 -translate-y-1/2 text-theme-muted pointer-events-none" />
|
||||||
</div>
|
</div>
|
||||||
<Button onClick={sync} loading={syncing}>
|
{/* Indicador discreto: cuándo se sincronizó con WACRM y cuándo se
|
||||||
|
recargó esta bandeja. */}
|
||||||
|
<div className="flex flex-col items-start gap-0.5">
|
||||||
|
<AutoRefreshBadge
|
||||||
|
state={auto}
|
||||||
|
hint={`La bandeja se sincroniza sola cada ${SYNC_INTERVAL_MS / 1000} segundos`}
|
||||||
|
/>
|
||||||
|
<span className="text-[10px] text-theme-muted">
|
||||||
|
Última sincronización: {hhmmss(parseMarca(lastSyncAt)) || '—'}
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
|
<Button onClick={sync} loading={auto.refreshing}>
|
||||||
<RefreshCw size={16} className="mr-2" />
|
<RefreshCw size={16} className="mr-2" />
|
||||||
Sincronizar
|
Sincronizar
|
||||||
</Button>
|
</Button>
|
||||||
|
|||||||
@@ -5403,7 +5403,7 @@ export const odooApi = {
|
|||||||
// ============================================================
|
// ============================================================
|
||||||
// WACRM Proxy
|
// WACRM Proxy
|
||||||
// ============================================================
|
// ============================================================
|
||||||
async syncWacrm(): Promise<{ status: string; result: Record<string, number | string | null> }> {
|
async syncWacrm(): Promise<{ status: string; result: Record<string, unknown> }> {
|
||||||
const { data } = await axios.post(`${ODOO_BASE}/skeen/frontend/v1/wacrm/sync`);
|
const { data } = await axios.post(`${ODOO_BASE}/skeen/frontend/v1/wacrm/sync`);
|
||||||
return data;
|
return data;
|
||||||
},
|
},
|
||||||
@@ -5420,7 +5420,7 @@ export const odooApi = {
|
|||||||
return data;
|
return data;
|
||||||
},
|
},
|
||||||
|
|
||||||
async getWacrmConversations(search?: string): Promise<{ status: string; conversations: WacrmConversation[] }> {
|
async getWacrmConversations(search?: string): Promise<{ status: string; conversations: WacrmConversation[]; last_sync_at?: string | null }> {
|
||||||
const { data } = await axios.get(`${ODOO_BASE}/skeen/frontend/v1/wacrm/conversations`, { params: search ? { search } : {} });
|
const { data } = await axios.get(`${ODOO_BASE}/skeen/frontend/v1/wacrm/conversations`, { params: search ? { search } : {} });
|
||||||
return data;
|
return data;
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ from odoo.http import request
|
|||||||
from .frontend import json_response, _parse_json_body
|
from .frontend import json_response, _parse_json_body
|
||||||
from .auth import require_role, require_perm
|
from .auth import require_role, require_perm
|
||||||
import os
|
import os
|
||||||
|
import hmac
|
||||||
import logging
|
import logging
|
||||||
import requests
|
import requests
|
||||||
import re
|
import re
|
||||||
@@ -18,6 +19,32 @@ from datetime import datetime, timedelta
|
|||||||
|
|
||||||
_logger = logging.getLogger(__name__)
|
_logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# ============================================================
|
||||||
|
# Sincronización incremental
|
||||||
|
#
|
||||||
|
# El caché local se llena desde Supabase (WACRM). Recorrer las 9.750 filas
|
||||||
|
# de `contacts` y todos los mensajes en cada ciclo costaba ~75 s, así que la
|
||||||
|
# pantalla nunca alcanzaba a sincronizar. Ahora:
|
||||||
|
# - los contactos se traen SOLO si están referenciados por una conversación
|
||||||
|
# o un deal, o si cambiaron desde la última marca;
|
||||||
|
# - conversaciones/mensajes/deals se traen por fecha de actualización;
|
||||||
|
# - las escrituras van en lote (una búsqueda + un create múltiple);
|
||||||
|
# - la poda de lo que ya no existe en el origen SOLO corre en un ciclo
|
||||||
|
# completo (cada FULL_SYNC_INTERVAL segundos), nunca en un parcial.
|
||||||
|
# Las marcas viven en ir.config_parameter y solo avanzan, nunca retroceden.
|
||||||
|
# ============================================================
|
||||||
|
PARAM_CONV_MARK = 'skeen.wacrm.last_sync' # marca de conversations
|
||||||
|
PARAM_MSG_MARK = 'skeen.wacrm.last_msg_sync' # marca de messages
|
||||||
|
PARAM_CONTACT_MARK = 'skeen.wacrm.last_contact_sync'
|
||||||
|
PARAM_DEAL_MARK = 'skeen.wacrm.last_deal_sync'
|
||||||
|
PARAM_LAST_SYNC_AT = 'skeen.wacrm.last_sync_at' # hora local del último ciclo ok
|
||||||
|
PARAM_LAST_FULL_AT = 'skeen.wacrm.last_full_sync_at'
|
||||||
|
PARAM_NOTIFY_KEY = 'skeen.wacrm.notify_key' # llave del aviso puntual
|
||||||
|
FULL_SYNC_INTERVAL = 300 # segundos entre ciclos completos (poda)
|
||||||
|
INCREMENTAL_CONTACT_PAGE = 1000 # tope de contactos cambiados por ciclo
|
||||||
|
SUPABASE_PAGE_SIZE = 1000
|
||||||
|
EPOCH_SUPABASE = '1970-01-01T00:00:00.000000+00:00'
|
||||||
|
|
||||||
|
|
||||||
def _get_supabase_config():
|
def _get_supabase_config():
|
||||||
"""Lee configuración de Supabase desde variables de entorno o ir.config_parameter."""
|
"""Lee configuración de Supabase desde variables de entorno o ir.config_parameter."""
|
||||||
@@ -84,6 +111,26 @@ def _hermes_bridge_request(path, json=None):
|
|||||||
return response.json()
|
return response.json()
|
||||||
|
|
||||||
|
|
||||||
|
def _supabase_server_time():
|
||||||
|
"""Hora del propio Supabase (cabecera Date), en UTC naive.
|
||||||
|
|
||||||
|
Se usa para arrancar las marcas con el reloj del origen y no con el de
|
||||||
|
Odoo, que puede estar desfasado.
|
||||||
|
"""
|
||||||
|
url, key = _get_supabase_config()
|
||||||
|
try:
|
||||||
|
from email.utils import parsedate_to_datetime
|
||||||
|
from datetime import timezone
|
||||||
|
response = requests.get(
|
||||||
|
f'{url}/rest/v1/',
|
||||||
|
headers={'apikey': key, 'Authorization': f'Bearer {key}'},
|
||||||
|
timeout=10)
|
||||||
|
return parsedate_to_datetime(response.headers['Date']).astimezone(timezone.utc).replace(tzinfo=None)
|
||||||
|
except Exception as e:
|
||||||
|
_logger.warning('No se pudo leer la hora de Supabase: %s', e)
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _supabase_request_paginated(table, params=None, page_size=1000):
|
def _supabase_request_paginated(table, params=None, page_size=1000):
|
||||||
"""Itera sobre todas las páginas de resultados de PostgREST.
|
"""Itera sobre todas las páginas de resultados de PostgREST.
|
||||||
|
|
||||||
@@ -120,6 +167,141 @@ def _parse_datetime(value):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _to_supabase_dt(dt):
|
||||||
|
"""Datetime naive UTC → texto ISO 8601 que PostgREST acepta en filtros."""
|
||||||
|
return dt.strftime('%Y-%m-%dT%H:%M:%S.%f') + '+00:00'
|
||||||
|
|
||||||
|
|
||||||
|
def _max_dt(rows, field):
|
||||||
|
"""Máximo (naive UTC) del campo en las filas, o None si no hay ninguno."""
|
||||||
|
mayor = None
|
||||||
|
for row in rows:
|
||||||
|
dt = _parse_datetime(row.get(field))
|
||||||
|
if dt and (mayor is None or dt > mayor):
|
||||||
|
mayor = dt
|
||||||
|
return mayor
|
||||||
|
|
||||||
|
|
||||||
|
def _chunks(items, size):
|
||||||
|
for i in range(0, len(items), size):
|
||||||
|
yield items[i:i + size]
|
||||||
|
|
||||||
|
|
||||||
|
def _changed_vals(record, vals):
|
||||||
|
"""Subconjunto de `vals` que difiere de lo que ya tiene el registro."""
|
||||||
|
cambios = {}
|
||||||
|
for field, valor in vals.items():
|
||||||
|
actual = record[field]
|
||||||
|
if hasattr(actual, 'ids'):
|
||||||
|
actual = actual.id or False
|
||||||
|
valor = valor or False
|
||||||
|
# Un campo vacío en Odoo se lee como False; el origen manda None. Se
|
||||||
|
# igualan para no reescribir la fila en cada ciclo.
|
||||||
|
if valor is None:
|
||||||
|
valor = False
|
||||||
|
if actual is None:
|
||||||
|
actual = False
|
||||||
|
if isinstance(actual, datetime) and isinstance(valor, datetime):
|
||||||
|
if abs((actual - valor).total_seconds()) < 0.001:
|
||||||
|
continue
|
||||||
|
elif actual == valor:
|
||||||
|
continue
|
||||||
|
cambios[field] = valor
|
||||||
|
return cambios
|
||||||
|
|
||||||
|
|
||||||
|
def _bulk_upsert(model, rows, key='external_id'):
|
||||||
|
"""Crea/actualiza en lote. Devuelve {clave: registro}.
|
||||||
|
|
||||||
|
Una sola búsqueda por lote, un `create` múltiple y escrituras agrupadas
|
||||||
|
por valores idénticos: nada de una consulta por fila.
|
||||||
|
"""
|
||||||
|
if not rows:
|
||||||
|
return {}
|
||||||
|
by_key = {rec[key]: rec for rec in model.search([(key, 'in', [r[key] for r in rows])])}
|
||||||
|
nuevos = [r for r in rows if r[key] not in by_key]
|
||||||
|
nuevos_ids = set()
|
||||||
|
if nuevos:
|
||||||
|
for rec in model.create(nuevos):
|
||||||
|
by_key[rec[key]] = rec
|
||||||
|
nuevos_ids.add(rec.id)
|
||||||
|
grupos = {}
|
||||||
|
for row in rows:
|
||||||
|
rec = by_key.get(row[key])
|
||||||
|
if rec is None or rec.id in nuevos_ids:
|
||||||
|
continue
|
||||||
|
cambios = _changed_vals(rec, {k: v for k, v in row.items() if k != key})
|
||||||
|
if not cambios:
|
||||||
|
continue
|
||||||
|
firma = tuple(sorted((k, str(v)) for k, v in cambios.items()))
|
||||||
|
previos = grupos.get(firma)
|
||||||
|
grupos[firma] = (cambios, (previos[1] | rec) if previos else rec)
|
||||||
|
for cambios, recs in grupos.values():
|
||||||
|
recs.write(cambios)
|
||||||
|
return by_key
|
||||||
|
|
||||||
|
|
||||||
|
def _contacto_a_vals(item):
|
||||||
|
return {
|
||||||
|
'external_id': item['id'],
|
||||||
|
'account_id': item.get('account_id'),
|
||||||
|
'phone': item.get('phone'),
|
||||||
|
'name': item.get('name') or item.get('phone'),
|
||||||
|
'email': item.get('email'),
|
||||||
|
'company': item.get('company'),
|
||||||
|
'avatar_url': item.get('avatar_url'),
|
||||||
|
'created_at': _parse_datetime(item.get('created_at')),
|
||||||
|
'updated_at': _parse_datetime(item.get('updated_at')),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _conversacion_a_vals(item, contacts_map):
|
||||||
|
vals = {
|
||||||
|
'external_id': item['id'],
|
||||||
|
'status': item.get('status') or 'open',
|
||||||
|
'assigned_agent_id': item.get('assigned_agent_id') or None,
|
||||||
|
'assigned_agent': str(item.get('assigned_agent_id')) if item.get('assigned_agent_id') else None,
|
||||||
|
'last_message_text': item.get('last_message_text'),
|
||||||
|
'last_message_at': _parse_datetime(item.get('last_message_at')),
|
||||||
|
'unread_count': item.get('unread_count', 0),
|
||||||
|
'created_at': _parse_datetime(item.get('created_at')),
|
||||||
|
'updated_at': _parse_datetime(item.get('updated_at')),
|
||||||
|
}
|
||||||
|
contacto = contacts_map.get(item.get('contact_id'))
|
||||||
|
if contacto:
|
||||||
|
vals['contact_id'] = contacto.id
|
||||||
|
elif not item.get('contact_id'):
|
||||||
|
vals['contact_id'] = False
|
||||||
|
# Si el contacto no vino en el lote no se toca el vínculo ya guardado.
|
||||||
|
return vals
|
||||||
|
|
||||||
|
|
||||||
|
def _traer_contactos(env, contact_ids):
|
||||||
|
"""Upserta los contactos indicados trayéndolos del origen. Devuelve el mapa."""
|
||||||
|
if not contact_ids:
|
||||||
|
return {}
|
||||||
|
items = []
|
||||||
|
for trozo in _chunks(sorted(contact_ids), 150):
|
||||||
|
items.extend(_supabase_request('GET', 'contacts', {
|
||||||
|
'select': '*', 'id': 'in.(%s)' % ','.join(trozo)}))
|
||||||
|
return _bulk_upsert(env['skeen.wacrm.contact'].sudo(),
|
||||||
|
[_contacto_a_vals(i) for i in items])
|
||||||
|
|
||||||
|
|
||||||
|
def _traer_conversaciones(env, conv_ids, contacts_map):
|
||||||
|
"""Trae del origen las conversaciones indicadas (y sus contactos)."""
|
||||||
|
if not conv_ids:
|
||||||
|
return {}
|
||||||
|
items = []
|
||||||
|
for trozo in _chunks(sorted(conv_ids), 100):
|
||||||
|
items.extend(_supabase_request('GET', 'conversations', {
|
||||||
|
'select': '*', 'id': 'in.(%s)' % ','.join(trozo)}))
|
||||||
|
faltan = {i['contact_id'] for i in items if i.get('contact_id')} - set(contacts_map)
|
||||||
|
contacts_map.update(_traer_contactos(env, faltan))
|
||||||
|
return _bulk_upsert(env['skeen.wacrm.conversation'].sudo(),
|
||||||
|
[_conversacion_a_vals(i, contacts_map) for i in items])
|
||||||
|
|
||||||
|
|
||||||
class SkeenWacrmProxyController(http.Controller):
|
class SkeenWacrmProxyController(http.Controller):
|
||||||
"""Endpoints proxy WACRM para el frontend."""
|
"""Endpoints proxy WACRM para el frontend."""
|
||||||
|
|
||||||
@@ -128,32 +310,41 @@ class SkeenWacrmProxyController(http.Controller):
|
|||||||
# ============================================================
|
# ============================================================
|
||||||
@http.route('/skeen/frontend/v1/wacrm/sync', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False)
|
@http.route('/skeen/frontend/v1/wacrm/sync', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False)
|
||||||
def sync_wacrm(self, **kw):
|
def sync_wacrm(self, **kw):
|
||||||
"""Sincroniza contactos, conversaciones, mensajes, pipelines y deals desde Supabase hacia Odoo.
|
"""Sincroniza el caché local con Supabase (WACRM).
|
||||||
|
|
||||||
El frontend tiene auto-sync cada 15 s y un botón manual: si ambos
|
Por defecto corre en modo incremental: solo trae lo que cambió desde
|
||||||
coinciden, dos transacciones escriben las mismas filas del caché y
|
la última marca. Cada FULL_SYNC_INTERVAL segundos —o con `full=1`, o
|
||||||
Postgres aborta una con "could not serialize access" (500). Se
|
la primera vez— corre un ciclo completo, que además poda del caché lo
|
||||||
reintenta con la transacción limpia, que es lo que hace Odoo en sus
|
que ya no existe en el origen.
|
||||||
propios reintentos.
|
|
||||||
|
Un solo sync a la vez: dos transacciones escribiendo las mismas filas
|
||||||
|
se abortan mutuamente ("could not serialize access"), así que el
|
||||||
|
segundo responde 'busy' y ante un conflicto real se reintenta.
|
||||||
"""
|
"""
|
||||||
# Un solo sync a la vez: el frontend auto-sincroniza cada 15 s y tiene
|
# El lock de transacción se libera solo al terminar la petición.
|
||||||
# botón manual; dos syncs simultáneos chocan en las mismas filas del
|
|
||||||
# caché y Postgres aborta uno. El lock de transacción se libera solo.
|
|
||||||
try:
|
try:
|
||||||
request.env.cr.execute(
|
request.env.cr.execute(
|
||||||
"SELECT pg_try_advisory_xact_lock(hashtext('skeen_wacrm_sync'))")
|
"SELECT pg_try_advisory_xact_lock(hashtext('skeen_wacrm_sync'))")
|
||||||
libre = request.env.cr.fetchone()[0]
|
libre = request.env.cr.fetchone()[0]
|
||||||
except Exception:
|
except Exception:
|
||||||
libre = True
|
libre = True
|
||||||
|
ICP = request.env['ir.config_parameter'].sudo()
|
||||||
if not libre:
|
if not libre:
|
||||||
return json_response({
|
return json_response({
|
||||||
'status': 'busy',
|
'status': 'busy',
|
||||||
'message': 'Ya hay una sincronización en curso; se actualizará en unos segundos',
|
'message': 'Ya hay una sincronización en curso; se actualizará en unos segundos',
|
||||||
|
'last_sync_at': ICP.get_param(PARAM_LAST_SYNC_AT),
|
||||||
})
|
})
|
||||||
|
pedir_full = str(kw.get('full', '')).lower() in ('1', 'true', 'yes', 'si')
|
||||||
|
if not pedir_full:
|
||||||
|
try:
|
||||||
|
pedir_full = bool((_parse_json_body() or {}).get('full'))
|
||||||
|
except Exception:
|
||||||
|
pedir_full = False
|
||||||
ultimo = None
|
ultimo = None
|
||||||
for intento in range(3):
|
for intento in range(3):
|
||||||
try:
|
try:
|
||||||
result = self._do_sync()
|
result = self._do_sync(full=pedir_full)
|
||||||
return json_response({'status': 'success', 'result': result})
|
return json_response({'status': 'success', 'result': result})
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
ultimo = e
|
ultimo = e
|
||||||
@@ -165,168 +356,164 @@ class SkeenWacrmProxyController(http.Controller):
|
|||||||
_logger.exception('Error sincronizando WACRM')
|
_logger.exception('Error sincronizando WACRM')
|
||||||
return json_response({'status': 'error', 'message': str(ultimo)}, 500)
|
return json_response({'status': 'error', 'message': str(ultimo)}, 500)
|
||||||
|
|
||||||
def _do_sync(self):
|
def _do_sync(self, full=False):
|
||||||
|
inicio = time.monotonic()
|
||||||
|
ICP = request.env['ir.config_parameter'].sudo()
|
||||||
|
Contact = request.env['skeen.wacrm.contact'].sudo()
|
||||||
|
Conv = request.env['skeen.wacrm.conversation'].sudo()
|
||||||
|
Msg = request.env['skeen.wacrm.message'].sudo()
|
||||||
|
Deal = request.env['skeen.wacrm.deal'].sudo()
|
||||||
|
|
||||||
|
marca_conv = ICP.get_param(PARAM_CONV_MARK)
|
||||||
|
marca_msg = ICP.get_param(PARAM_MSG_MARK)
|
||||||
|
marca_contacto = ICP.get_param(PARAM_CONTACT_MARK)
|
||||||
|
marca_deal = ICP.get_param(PARAM_DEAL_MARK)
|
||||||
|
ultimo_full = _parse_datetime(ICP.get_param(PARAM_LAST_FULL_AT))
|
||||||
|
|
||||||
|
# Sin marcas previas (primer arranque) o con el ciclo completo vencido,
|
||||||
|
# se hace una pasada completa: es la única que poda lo que ya no existe.
|
||||||
|
if not marca_conv or not marca_msg:
|
||||||
|
full = True
|
||||||
|
elif not ultimo_full or (fields.Datetime.now() - ultimo_full).total_seconds() >= FULL_SYNC_INTERVAL:
|
||||||
|
full = True
|
||||||
|
modo = 'full' if full else 'incremental'
|
||||||
account_id = None
|
account_id = None
|
||||||
|
|
||||||
# 1. Contacts (paginado por si PostgREST limita a 1000 filas)
|
# 1. Conversaciones (completas la primera vez y en cada ciclo completo;
|
||||||
contacts_data = _supabase_request_paginated('contacts', {'select': '*'})
|
# después solo las que cambiaron desde la marca).
|
||||||
contacts_map = {}
|
params_conv = {'select': '*', 'order': 'id.asc' if full else 'updated_at.asc,id.asc'}
|
||||||
|
if not full:
|
||||||
|
params_conv['updated_at'] = 'gte.%s' % marca_conv
|
||||||
|
conversations_data = _supabase_request_paginated('conversations', params_conv)
|
||||||
|
|
||||||
|
# 2. Deals (igual criterio)
|
||||||
|
params_deal = {'select': '*', 'order': 'id.asc' if full else 'updated_at.asc,id.asc'}
|
||||||
|
if not full:
|
||||||
|
params_deal['updated_at'] = 'gte.%s' % (marca_deal or EPOCH_SUPABASE)
|
||||||
|
deals_data = _supabase_request_paginated('deals', params_deal)
|
||||||
|
|
||||||
|
# 3. Contactos: solo los que hacen falta. Los referenciados por lo que
|
||||||
|
# se acaba de traer, más los que cambiaron desde la última marca.
|
||||||
|
# La marca arranca en la hora DEL ORIGEN en el primer ciclo: el
|
||||||
|
# histórico de pacientes importados no se recorre, solo se siguen
|
||||||
|
# los cambios posteriores (acotado a una página por ciclo, para que
|
||||||
|
# un reimport masivo no arrastre las 9.750 filas de golpe).
|
||||||
|
if not marca_contacto:
|
||||||
|
servidor = _supabase_server_time()
|
||||||
|
if servidor:
|
||||||
|
marca_contacto = _to_supabase_dt(servidor)
|
||||||
|
ICP.set_param(PARAM_CONTACT_MARK, marca_contacto)
|
||||||
|
referenciados = [i['contact_id'] for i in conversations_data + deals_data if i.get('contact_id')]
|
||||||
|
referenciados = list(dict.fromkeys(referenciados))
|
||||||
|
contacts_data = []
|
||||||
|
for trozo in _chunks(referenciados, 150):
|
||||||
|
contacts_data.extend(_supabase_request('GET', 'contacts', {
|
||||||
|
'select': '*', 'id': 'in.(%s)' % ','.join(trozo)}))
|
||||||
|
contactos_barridos = []
|
||||||
|
if not full and marca_contacto:
|
||||||
|
contactos_barridos = _supabase_request('GET', 'contacts', {
|
||||||
|
'select': '*', 'updated_at': 'gte.%s' % marca_contacto,
|
||||||
|
'order': 'updated_at.asc,id.asc', 'limit': INCREMENTAL_CONTACT_PAGE})
|
||||||
|
contacts_data.extend(contactos_barridos)
|
||||||
|
contacts_data = list({i['id']: i for i in contacts_data}.values())
|
||||||
for item in contacts_data:
|
for item in contacts_data:
|
||||||
if not account_id and item.get('account_id'):
|
if not account_id and item.get('account_id'):
|
||||||
account_id = item['account_id']
|
account_id = item['account_id']
|
||||||
vals = {
|
if not account_id:
|
||||||
'external_id': item['id'],
|
fila = _supabase_request('GET', 'contacts', {'select': 'account_id', 'limit': 1})
|
||||||
'account_id': item.get('account_id'),
|
account_id = fila[0].get('account_id') if fila else None
|
||||||
'phone': item.get('phone'),
|
contacts_map = _bulk_upsert(Contact, [_contacto_a_vals(i) for i in contacts_data])
|
||||||
'name': item.get('name') or item.get('phone'),
|
|
||||||
'email': item.get('email'),
|
|
||||||
'company': item.get('company'),
|
|
||||||
'avatar_url': item.get('avatar_url'),
|
|
||||||
'created_at': _parse_datetime(item.get('created_at')),
|
|
||||||
'updated_at': _parse_datetime(item.get('updated_at')),
|
|
||||||
}
|
|
||||||
existing = request.env['skeen.wacrm.contact'].sudo().search([('external_id', '=', item['id'])], limit=1)
|
|
||||||
if existing:
|
|
||||||
existing.write(vals)
|
|
||||||
contacts_map[item['id']] = existing
|
|
||||||
else:
|
|
||||||
contacts_map[item['id']] = request.env['skeen.wacrm.contact'].sudo().create(vals)
|
|
||||||
|
|
||||||
# 2. Conversations
|
# 4. Conversaciones: el vínculo con el contacto se resuelve con el mapa
|
||||||
# Además del upsert, se REFLEJAN ELIMINACIONES: lo que ya no exista
|
# anterior. Si un contacto no vino en el lote, no se pisa el vínculo.
|
||||||
# en Supabase se borra del caché local (antes el caché acumulaba
|
conversations_map = _bulk_upsert(
|
||||||
# conversaciones borradas y la bandeja de SKEEN mostraba chats muertos).
|
Conv, [_conversacion_a_vals(i, contacts_map) for i in conversations_data])
|
||||||
conversations_data = _supabase_request_paginated('conversations', {'select': '*'})
|
|
||||||
conversations_map = {}
|
|
||||||
for item in conversations_data:
|
|
||||||
vals = {
|
|
||||||
'external_id': item['id'],
|
|
||||||
'contact_id': contacts_map.get(item.get('contact_id'), request.env['skeen.wacrm.contact']).id if contacts_map.get(item.get('contact_id')) else False,
|
|
||||||
'status': item.get('status', 'open'),
|
|
||||||
'assigned_agent_id': item.get('assigned_agent_id') or None,
|
|
||||||
'assigned_agent': str(item.get('assigned_agent_id')) if item.get('assigned_agent_id') else None,
|
|
||||||
'last_message_text': item.get('last_message_text'),
|
|
||||||
'last_message_at': _parse_datetime(item.get('last_message_at')),
|
|
||||||
'unread_count': item.get('unread_count', 0),
|
|
||||||
'created_at': _parse_datetime(item.get('created_at')),
|
|
||||||
'updated_at': _parse_datetime(item.get('updated_at')),
|
|
||||||
}
|
|
||||||
existing = request.env['skeen.wacrm.conversation'].sudo().search([('external_id', '=', item['id'])], limit=1)
|
|
||||||
if existing:
|
|
||||||
existing.write(vals)
|
|
||||||
conversations_map[item['id']] = existing
|
|
||||||
else:
|
|
||||||
conversations_map[item['id']] = request.env['skeen.wacrm.conversation'].sudo().create(vals)
|
|
||||||
|
|
||||||
remote_conv_ids = set(conversations_map.keys())
|
# 5. Mensajes: completos la primera vez y en cada ciclo completo; luego
|
||||||
stale_convs = request.env['skeen.wacrm.conversation'].sudo().search([
|
# solo los nuevos, en orden ascendente (con descendente + offset los
|
||||||
('external_id', 'not in', list(remote_conv_ids))]) if remote_conv_ids else \
|
# mensajes que van entrando desplazan las páginas y se duplican).
|
||||||
request.env['skeen.wacrm.conversation'].sudo().search([])
|
params_msg = {'select': '*', 'order': 'created_at.asc,id.asc'}
|
||||||
if stale_convs:
|
if not full:
|
||||||
stale_convs.unlink()
|
params_msg['created_at'] = 'gte.%s' % marca_msg
|
||||||
conversations_map = {k: v for k, v in conversations_map.items()}
|
messages_data = _supabase_request_paginated('messages', params_msg, page_size=SUPABASE_PAGE_SIZE)
|
||||||
|
|
||||||
# 3. Messages (todos los del origen, sin recorte)
|
# Una conversación que no vino en el lote puede tener mensajes nuevos
|
||||||
messages_data = _supabase_request_paginated(
|
# (lo normal en incremental): se resuelve contra el caché y, si tampoco
|
||||||
'messages',
|
# está, se trae del origen junto con su contacto.
|
||||||
{'select': '*', 'order': 'created_at.desc'},
|
pendientes = {m['conversation_id'] for m in messages_data if m.get('conversation_id')} - set(conversations_map)
|
||||||
page_size=1000
|
if pendientes:
|
||||||
)
|
locales = Conv.search([('external_id', 'in', list(pendientes))])
|
||||||
remote_msg_ids = {item['id'] for item in messages_data}
|
for conv in locales:
|
||||||
stale_msgs = request.env['skeen.wacrm.message'].sudo().search([
|
conversations_map[conv.external_id] = conv
|
||||||
('external_id', 'not in', list(remote_msg_ids))]) if remote_msg_ids else \
|
pendientes -= set(locales.mapped('external_id'))
|
||||||
request.env['skeen.wacrm.message'].sudo().search([])
|
if pendientes:
|
||||||
stale_msgs.unlink()
|
conversations_map.update(_traer_conversaciones(request.env, pendientes, contacts_map))
|
||||||
|
|
||||||
|
msg_vals = []
|
||||||
for item in messages_data:
|
for item in messages_data:
|
||||||
conversation = conversations_map.get(item.get('conversation_id'))
|
conversation = conversations_map.get(item.get('conversation_id'))
|
||||||
if not conversation:
|
if not conversation:
|
||||||
continue
|
continue
|
||||||
vals = {
|
msg_vals.append({
|
||||||
'external_id': item['id'],
|
'external_id': item['id'],
|
||||||
'conversation_id': conversation.id,
|
'conversation_id': conversation.id,
|
||||||
'sender_type': {'contact': 'customer', 'user': 'agent'}.get(item.get('sender_type', 'customer'), item.get('sender_type', 'customer')),
|
'sender_type': {'contact': 'customer', 'user': 'agent'}.get(
|
||||||
'content_type': item.get('content_type', 'text'),
|
item.get('sender_type', 'customer'), item.get('sender_type', 'customer')),
|
||||||
|
'content_type': item.get('content_type') or 'text',
|
||||||
'content_text': item.get('content_text') or item.get('content'),
|
'content_text': item.get('content_text') or item.get('content'),
|
||||||
'media_url': item.get('media_url'),
|
'media_url': item.get('media_url'),
|
||||||
'template_name': item.get('template_name'),
|
'template_name': item.get('template_name'),
|
||||||
'status': item.get('status', 'sent'),
|
'status': item.get('status', 'sent'),
|
||||||
'created_at': _parse_datetime(item.get('created_at')),
|
'created_at': _parse_datetime(item.get('created_at')),
|
||||||
}
|
})
|
||||||
existing = request.env['skeen.wacrm.message'].sudo().search([('external_id', '=', item['id'])], limit=1)
|
_bulk_upsert(Msg, msg_vals)
|
||||||
if existing:
|
|
||||||
existing.write(vals)
|
|
||||||
else:
|
|
||||||
request.env['skeen.wacrm.message'].sudo().create(vals)
|
|
||||||
|
|
||||||
# 4. Pipelines & Stages
|
# 6. Pipelines & Stages (catálogo diminuto: siempre completo)
|
||||||
pipelines_data = _supabase_request_paginated('pipelines', {'select': '*'})
|
pipelines_data = _supabase_request_paginated('pipelines', {'select': '*', 'order': 'id.asc'})
|
||||||
pipelines_map = {}
|
pipelines_map = _bulk_upsert(request.env['skeen.wacrm.pipeline'].sudo(), [{
|
||||||
stages_map = {}
|
|
||||||
for item in pipelines_data:
|
|
||||||
vals = {
|
|
||||||
'external_id': item['id'],
|
'external_id': item['id'],
|
||||||
'name': item.get('name'),
|
'name': item.get('name'),
|
||||||
'created_at': _parse_datetime(item.get('created_at')),
|
'created_at': _parse_datetime(item.get('created_at')),
|
||||||
}
|
} for item in pipelines_data])
|
||||||
existing = request.env['skeen.wacrm.pipeline'].sudo().search([('external_id', '=', item['id'])], limit=1)
|
|
||||||
if existing:
|
|
||||||
existing.write(vals)
|
|
||||||
pipelines_map[item['id']] = existing
|
|
||||||
else:
|
|
||||||
pipelines_map[item['id']] = request.env['skeen.wacrm.pipeline'].sudo().create(vals)
|
|
||||||
|
|
||||||
stages_data = _supabase_request_paginated('pipeline_stages', {'select': '*'})
|
stages_data = _supabase_request_paginated('pipeline_stages', {'select': '*', 'order': 'id.asc'})
|
||||||
for item in stages_data:
|
stages_map = _bulk_upsert(request.env['skeen.wacrm.stage'].sudo(), [{
|
||||||
vals = {
|
|
||||||
'external_id': item['id'],
|
'external_id': item['id'],
|
||||||
'pipeline_id': pipelines_map.get(item.get('pipeline_id'), request.env['skeen.wacrm.pipeline']).id if pipelines_map.get(item.get('pipeline_id')) else False,
|
'pipeline_id': pipelines_map[item['pipeline_id']].id if pipelines_map.get(item.get('pipeline_id')) else False,
|
||||||
'name': item.get('name'),
|
'name': item.get('name'),
|
||||||
'position': item.get('position', 0),
|
'position': item.get('position', 0),
|
||||||
'color': item.get('color', '#3b82f6'),
|
'color': item.get('color') or '#3b82f6',
|
||||||
}
|
} for item in stages_data])
|
||||||
existing = request.env['skeen.wacrm.stage'].sudo().search([('external_id', '=', item['id'])], limit=1)
|
|
||||||
if existing:
|
|
||||||
existing.write(vals)
|
|
||||||
stages_map[item['id']] = existing
|
|
||||||
else:
|
|
||||||
stages_map[item['id']] = request.env['skeen.wacrm.stage'].sudo().create(vals)
|
|
||||||
|
|
||||||
# 5. Deals
|
# 7. Deals — mapa profiles.id → user_id (Supabase guarda assigned_to como
|
||||||
# Mapa profiles.id → user_id (Supabase guarda assigned_to como profiles.id;
|
# profiles.id; localmente usamos user_id, como assigned_agent_id).
|
||||||
# localmente usamos user_id, consistente con conversations.assigned_agent_id)
|
|
||||||
profiles_usermap = {}
|
profiles_usermap = {}
|
||||||
try:
|
try:
|
||||||
for p in _supabase_request('GET', 'profiles', params={'select': 'id,user_id'}):
|
for p in _supabase_request('GET', 'profiles', params={'select': 'id,user_id'}):
|
||||||
profiles_usermap[p['id']] = p['user_id']
|
profiles_usermap[p['id']] = p['user_id']
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
deals_data = _supabase_request_paginated('deals', {'select': '*'})
|
_bulk_upsert(Deal, [{
|
||||||
for item in deals_data:
|
|
||||||
vals = {
|
|
||||||
'external_id': item['id'],
|
'external_id': item['id'],
|
||||||
'title': item.get('title') or 'Sin título',
|
'title': item.get('title') or 'Sin título',
|
||||||
'contact_id': contacts_map.get(item.get('contact_id'), request.env['skeen.wacrm.contact']).id if contacts_map.get(item.get('contact_id')) else False,
|
'contact_id': contacts_map[item['contact_id']].id if contacts_map.get(item.get('contact_id')) else False,
|
||||||
'conversation_id': item.get('conversation_id'),
|
'conversation_id': item.get('conversation_id'),
|
||||||
'pipeline_id': pipelines_map.get(item.get('pipeline_id'), request.env['skeen.wacrm.pipeline']).id if pipelines_map.get(item.get('pipeline_id')) else False,
|
'pipeline_id': pipelines_map[item['pipeline_id']].id if pipelines_map.get(item.get('pipeline_id')) else False,
|
||||||
'stage_id': stages_map.get(item.get('stage_id'), request.env['skeen.wacrm.stage']).id if stages_map.get(item.get('stage_id')) else False,
|
'stage_id': stages_map[item['stage_id']].id if stages_map.get(item.get('stage_id')) else False,
|
||||||
'value': float(item.get('value', 0) or 0),
|
'value': float(item.get('value', 0) or 0),
|
||||||
'currency': item.get('currency', 'USD'),
|
'currency': item.get('currency', 'USD'),
|
||||||
'status': item.get('status', 'open'),
|
'status': item.get('status', 'open'),
|
||||||
'expected_close_date': item.get('expected_close_date'),
|
'expected_close_date': fields.Date.to_date(item.get('expected_close_date')),
|
||||||
'notes': item.get('notes'),
|
'notes': item.get('notes'),
|
||||||
'assigned_to': profiles_usermap.get(item.get('assigned_to')) if item.get('assigned_to') else None,
|
'assigned_to': profiles_usermap.get(item.get('assigned_to')) if item.get('assigned_to') else None,
|
||||||
'created_at': _parse_datetime(item.get('created_at')),
|
'created_at': _parse_datetime(item.get('created_at')),
|
||||||
'updated_at': _parse_datetime(item.get('updated_at')),
|
'updated_at': _parse_datetime(item.get('updated_at')),
|
||||||
}
|
} for item in deals_data])
|
||||||
existing = request.env['skeen.wacrm.deal'].sudo().search([('external_id', '=', item['id'])], limit=1)
|
|
||||||
if existing:
|
|
||||||
existing.write(vals)
|
|
||||||
else:
|
|
||||||
request.env['skeen.wacrm.deal'].sudo().create(vals)
|
|
||||||
|
|
||||||
# 6. Auto-crear deals: contactos con conversación sin deal abierto.
|
# 8. Auto-crear deals: contactos con conversación sin deal abierto.
|
||||||
# Se crean EN SUPABASE (la app WACRM nativa los ve) y se espejan local.
|
# Se crean EN SUPABASE (la app WACRM nativa los ve) y se espejan local.
|
||||||
# Idempotente: un deal abierto por contacto.
|
# Idempotente: un deal por contacto. En incremental `deals_data` es
|
||||||
|
# parcial, así que la lista de contactos con deal se pide al origen.
|
||||||
deals_created = 0
|
deals_created = 0
|
||||||
try:
|
try:
|
||||||
if conversations_map and pipelines_map and account_id:
|
if conversations_map and pipelines_map and account_id:
|
||||||
@@ -338,9 +525,15 @@ class SkeenWacrmProxyController(http.Controller):
|
|||||||
'select': 'user_id', 'order': 'created_at.asc', 'limit': 1})
|
'select': 'user_id', 'order': 'created_at.asc', 'limit': 1})
|
||||||
owner_id = owner[0]['user_id'] if owner else None
|
owner_id = owner[0]['user_id'] if owner else None
|
||||||
if stage and owner_id:
|
if stage and owner_id:
|
||||||
Deal = request.env['skeen.wacrm.deal'].sudo()
|
# Un deal por contacto: no crear si ya tiene cualquiera (open o ganado)
|
||||||
# Un deal por contacto: no crear si ya tiene cualquier deal (open o ganado)
|
|
||||||
contacts_with_deals = {str(d.get('contact_id')) for d in deals_data if d.get('contact_id')}
|
contacts_with_deals = {str(d.get('contact_id')) for d in deals_data if d.get('contact_id')}
|
||||||
|
if not full:
|
||||||
|
con_conv = [c.external_id for c in conversations_map.values() if c.contact_id]
|
||||||
|
for trozo in _chunks(con_conv, 100):
|
||||||
|
remotos = _supabase_request('GET', 'deals', {
|
||||||
|
'select': 'contact_id',
|
||||||
|
'contact_id': 'in.(%s)' % ','.join(trozo)})
|
||||||
|
contacts_with_deals |= {str(d['contact_id']) for d in remotos if d.get('contact_id')}
|
||||||
for conv in conversations_map.values():
|
for conv in conversations_map.values():
|
||||||
contact = conv.contact_id
|
contact = conv.contact_id
|
||||||
if not contact:
|
if not contact:
|
||||||
@@ -382,7 +575,46 @@ class SkeenWacrmProxyController(http.Controller):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
_logger.warning('Auto-deals omitidos: %s', e)
|
_logger.warning('Auto-deals omitidos: %s', e)
|
||||||
|
|
||||||
|
# 9. Poda: SOLO en el ciclo completo. Comparar contra un lote parcial
|
||||||
|
# (incremental o una página suelta) borraría filas válidas; por eso
|
||||||
|
# el borrado queda reservado a la pasada que trajo TODO el origen.
|
||||||
|
# Los mensajes de una conversación borrada caen por cascade.
|
||||||
|
borrados = {'conversations': 0, 'messages': 0}
|
||||||
|
if full:
|
||||||
|
remotos_conv = set(conversations_map)
|
||||||
|
sobran_conv = Conv.search([]).filtered(lambda c: c.external_id not in remotos_conv)
|
||||||
|
if sobran_conv:
|
||||||
|
borrados['conversations'] = len(sobran_conv)
|
||||||
|
sobran_conv.unlink()
|
||||||
|
remotos_msg = {m['id'] for m in messages_data}
|
||||||
|
sobran_msg = Msg.search([]).filtered(lambda m: m.external_id not in remotos_msg)
|
||||||
|
if sobran_msg:
|
||||||
|
borrados['messages'] = len(sobran_msg)
|
||||||
|
sobran_msg.unlink()
|
||||||
|
|
||||||
|
# 10. Marcas: avanzan al máximo visto en cada tabla, nunca retroceden.
|
||||||
|
ahora = fields.Datetime.now()
|
||||||
|
ICP.set_param(PARAM_LAST_SYNC_AT, _to_supabase_dt(ahora))
|
||||||
|
if full:
|
||||||
|
ICP.set_param(PARAM_LAST_FULL_AT, _to_supabase_dt(ahora))
|
||||||
|
|
||||||
|
def _avanzar(param, nuevo):
|
||||||
|
if not nuevo:
|
||||||
|
return
|
||||||
|
previo = _parse_datetime(ICP.get_param(param))
|
||||||
|
if previo is None or nuevo > previo:
|
||||||
|
ICP.set_param(param, _to_supabase_dt(nuevo))
|
||||||
|
|
||||||
|
_avanzar(PARAM_CONV_MARK, _max_dt(conversations_data, 'updated_at'))
|
||||||
|
_avanzar(PARAM_MSG_MARK, _max_dt(messages_data, 'created_at'))
|
||||||
|
# Solo la página del barrido mueve la marca de contactos: los
|
||||||
|
# referenciados se traen siempre aparte, y usarlos aquí saltaría
|
||||||
|
# contactos intermedios del barrido.
|
||||||
|
_avanzar(PARAM_CONTACT_MARK, _max_dt(contactos_barridos, 'updated_at'))
|
||||||
|
_avanzar(PARAM_DEAL_MARK, _max_dt(deals_data, 'updated_at'))
|
||||||
|
|
||||||
return {
|
return {
|
||||||
|
'mode': modo,
|
||||||
'contacts': len(contacts_data),
|
'contacts': len(contacts_data),
|
||||||
'conversations': len(conversations_data),
|
'conversations': len(conversations_data),
|
||||||
'messages': len(messages_data),
|
'messages': len(messages_data),
|
||||||
@@ -390,9 +622,75 @@ class SkeenWacrmProxyController(http.Controller):
|
|||||||
'stages': len(stages_data),
|
'stages': len(stages_data),
|
||||||
'deals': len(deals_data),
|
'deals': len(deals_data),
|
||||||
'deals_created': deals_created,
|
'deals_created': deals_created,
|
||||||
|
'deleted': borrados,
|
||||||
'account_id': account_id,
|
'account_id': account_id,
|
||||||
|
'duration_ms': int((time.monotonic() - inicio) * 1000),
|
||||||
|
'last_sync_at': ICP.get_param(PARAM_LAST_SYNC_AT),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def _sync_conversacion(self, external_id):
|
||||||
|
"""Alta puntual de UNA conversación, su contacto y sus mensajes."""
|
||||||
|
Conv = request.env['skeen.wacrm.conversation'].sudo()
|
||||||
|
convs = _supabase_request('GET', 'conversations', {
|
||||||
|
'select': '*', 'id': 'eq.%s' % external_id})
|
||||||
|
if convs:
|
||||||
|
contacts_map = _traer_contactos(
|
||||||
|
request.env, {c['contact_id'] for c in convs if c.get('contact_id')})
|
||||||
|
_bulk_upsert(Conv, [_conversacion_a_vals(c, contacts_map) for c in convs])
|
||||||
|
conv = Conv.search([('external_id', '=', external_id)], limit=1)
|
||||||
|
if not conv:
|
||||||
|
return {'conversation': 0, 'messages': 0}
|
||||||
|
msgs = _supabase_request_paginated('messages', {
|
||||||
|
'select': '*', 'conversation_id': 'eq.%s' % external_id,
|
||||||
|
'order': 'created_at.asc,id.asc'})
|
||||||
|
_bulk_upsert(request.env['skeen.wacrm.message'].sudo(), [{
|
||||||
|
'external_id': m['id'],
|
||||||
|
'conversation_id': conv.id,
|
||||||
|
'sender_type': {'contact': 'customer', 'user': 'agent'}.get(
|
||||||
|
m.get('sender_type', 'customer'), m.get('sender_type', 'customer')),
|
||||||
|
'content_type': m.get('content_type') or 'text',
|
||||||
|
'content_text': m.get('content_text') or m.get('content'),
|
||||||
|
'media_url': m.get('media_url'),
|
||||||
|
'template_name': m.get('template_name'),
|
||||||
|
'status': m.get('status', 'sent'),
|
||||||
|
'created_at': _parse_datetime(m.get('created_at')),
|
||||||
|
} for m in msgs])
|
||||||
|
return {'conversation': len(convs), 'messages': len(msgs)}
|
||||||
|
|
||||||
|
@http.route('/skeen/frontend/v1/wacrm/notify', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False)
|
||||||
|
def wacrm_notify(self, **kw):
|
||||||
|
"""Aviso puntual del puente de Hermes: "llegó un mensaje a esta conversación".
|
||||||
|
|
||||||
|
Da de alta solo esa conversación, su contacto y sus mensajes, para que
|
||||||
|
la bandeja se actualice en segundos sin esperar el ciclo. Es aditivo y
|
||||||
|
a prueba de fallos: si el aviso falla o se pierde, la sincronización
|
||||||
|
periódica cubre el mensaje igual. Se protege con llave compartida.
|
||||||
|
"""
|
||||||
|
if request.httprequest.method == 'OPTIONS':
|
||||||
|
return json_response({}, 200)
|
||||||
|
try:
|
||||||
|
esperada = os.environ.get('SKEEN_WACRM_NOTIFY_KEY') or \
|
||||||
|
request.env['ir.config_parameter'].sudo().get_param(PARAM_NOTIFY_KEY)
|
||||||
|
if not esperada:
|
||||||
|
return json_response({'status': 'error', 'message': 'Aviso puntual deshabilitado'}, 503)
|
||||||
|
if not hmac.compare_digest(request.httprequest.headers.get('X-Skeen-Key') or '', esperada):
|
||||||
|
return json_response({'status': 'error', 'message': 'Llave inválida'}, 401)
|
||||||
|
external_id = (_parse_json_body().get('conversation_id') or '').strip()
|
||||||
|
if not external_id:
|
||||||
|
return json_response({'status': 'error', 'message': 'Falta conversation_id'}, 400)
|
||||||
|
try:
|
||||||
|
request.env.cr.execute(
|
||||||
|
"SELECT pg_try_advisory_xact_lock(hashtext('skeen_wacrm_sync'))")
|
||||||
|
libre = request.env.cr.fetchone()[0]
|
||||||
|
except Exception:
|
||||||
|
libre = True
|
||||||
|
if not libre:
|
||||||
|
return json_response({'status': 'busy'})
|
||||||
|
return json_response({'status': 'success', 'result': self._sync_conversacion(external_id)})
|
||||||
|
except Exception as e:
|
||||||
|
_logger.exception('Error en el aviso puntual de WACRM')
|
||||||
|
return json_response({'status': 'error', 'message': str(e)}, 500)
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# Mensajes
|
# Mensajes
|
||||||
# ============================================================
|
# ============================================================
|
||||||
@@ -522,6 +820,7 @@ class SkeenWacrmProxyController(http.Controller):
|
|||||||
return json_response({
|
return json_response({
|
||||||
'status': 'success',
|
'status': 'success',
|
||||||
'conversations': [self._conversation_to_dict(c) for c in conversations],
|
'conversations': [self._conversation_to_dict(c) for c in conversations],
|
||||||
|
'last_sync_at': request.env['ir.config_parameter'].sudo().get_param(PARAM_LAST_SYNC_AT),
|
||||||
})
|
})
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
_logger.exception('Error listando conversaciones WACRM')
|
_logger.exception('Error listando conversaciones WACRM')
|
||||||
|
|||||||
Reference in New Issue
Block a user