API Reference
En esta sección se detalla la documentación del código fuente de la aplicación, extraída directamente de los docstrings. La documentación está organizada siguiendo el flujo de ejecución del procesamiento de los Bills of Lading (BOLs).
Scheduler
El proceso puede ser iniciado automáticamente por un planificador de tareas (scheduler) que se ejecuta a intervalos regulares.
app.scheduler.jobs
Este módulo se encarga de configurar y ejecutar la tarea programada que llama al endpoint de procesamiento de BOLs.
app.scheduler.jobs
Functions
call_process_bol()
async
Realiza una llamada HTTP POST al endpoint de procesamiento de BOLs.
Esta función asíncrona construye la URL y la fecha de inicio para la solicitud,
y luego envía una petición POST al servicio process_bol.
Returns:
| Name | Type | Description |
|---|---|---|
None |
La función no retorna un valor explícito, pero registra el resultado de la llamada HTTP. |
Source code in app/scheduler/jobs.py
async def call_process_bol():
"""
Realiza una llamada HTTP POST al endpoint de procesamiento de BOLs.
Esta función asíncrona construye la URL y la fecha de inicio para la solicitud,
y luego envía una petición POST al servicio `process_bol`.
Args:
No recibe argumentos.
Returns:
None: La función no retorna un valor explícito, pero registra el resultado de la llamada HTTP.
"""
url = settings.process_bol_url
timeout = settings.http_client_timeout_seconds
start_date = datetime.now(ZoneInfo(settings.timezone)).isoformat()
async with httpx.AsyncClient(timeout=timeout) as client:
r = await client.post(url, json={"start_date": start_date})
logger.info(
"[Scheduler] Called process_bol with start_date=%s -> %s: %s",
start_date,
r.status_code,
r.text,
)
start_scheduler()
Inicia el planificador de tareas (scheduler) para ejecutar call_process_bol a intervalos regulares.
Configura un AsyncIOScheduler que ejecutará la función call_process_bol
cada cierto número de minutos, definido en las configuraciones de la aplicación.
Returns:
| Name | Type | Description |
|---|---|---|
AsyncIOScheduler |
AsyncIOScheduler
|
Una instancia del planificador de tareas iniciado. |
Source code in app/scheduler/jobs.py
def start_scheduler() -> AsyncIOScheduler:
"""
Inicia el planificador de tareas (scheduler) para ejecutar `call_process_bol` a intervalos regulares.
Configura un `AsyncIOScheduler` que ejecutará la función `call_process_bol`
cada cierto número de minutos, definido en las configuraciones de la aplicación.
Args:
No recibe argumentos.
Returns:
AsyncIOScheduler: Una instancia del planificador de tareas iniciado.
"""
scheduler = AsyncIOScheduler(timezone=settings.timezone)
scheduler.add_job(
call_process_bol, IntervalTrigger(minutes=settings.internal_trigger_minutes)
)
scheduler.start()
logger.info("Scheduler started")
return schedulerAPI Endpoint
El endpoint principal que orquesta todo el flujo de procesamiento de BOLs.
app.api.process
Este módulo contiene el endpoint /fetch-dataload que recibe la solicitud de procesamiento.
app.api.process
Functions
fetch_dataload(request, body)
async
Procesa los Bills of Lading (BOLs) para crear y actualizar cargas (loads).
Este endpoint orquesta un flujo de varios pasos que comienza con la recuperación de
archivos adjuntos (BOLs en formato PDF) desde una cuenta de correo electrónico a partir
de una fecha determinada. Posteriormente, delega el procesamiento completo a
_process_bol_attachments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Request
|
El objeto de solicitud de FastAPI, que contiene el estado de la aplicación (clientes de servicios, etc.). |
required |
body
|
StartDateRequest
|
El cuerpo de la solicitud, que debe contener la fecha de inicio para filtrar los correos electrónicos. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
JSONResponse |
Una respuesta JSON que contiene los IDs de las cargas (loads) creadas en el CRM, con un código de estado 200 en caso de éxito. Si no se encuentran archivos adjuntos, retorna una lista vacía. |
Source code in app/api/process.py
@router.post(
"/fetch-dataload",
response_class=JSONResponse,
response_model=List[FinalLoadV3],
status_code=200,
responses={200: {"description": "Loads fetched successfully"}},
)
async def fetch_dataload(request: Request, body: StartDateRequest):
"""
Procesa los Bills of Lading (BOLs) para crear y actualizar cargas (loads).
Este endpoint orquesta un flujo de varios pasos que comienza con la recuperación de
archivos adjuntos (BOLs en formato PDF) desde una cuenta de correo electrónico a partir
de una fecha determinada. Posteriormente, delega el procesamiento completo a
`_process_bol_attachments`.
Args:
request (Request): El objeto de solicitud de FastAPI, que contiene el estado de la
aplicación (clientes de servicios, etc.).
body (StartDateRequest): El cuerpo de la solicitud, que debe contener la fecha
de inicio para filtrar los correos electrónicos.
Returns:
JSONResponse: Una respuesta JSON que contiene los IDs de las cargas (loads) creadas
en el CRM, con un código de estado 200 en caso de éxito. Si no se
encuentran archivos adjuntos, retorna una lista vacía.
"""
t0 = time.time()
state: AppState = request.app.state
start_dt = datetime.fromisoformat(body.start_date)
# 1. Traer adjuntos desde correo por fecha
attachments = await state.mail_client.fetch_bol_attachments(start_dt, limit=10)
if not attachments:
logger.warning("No attachments found for start_date=%s", start_dt)
return JSONResponse(status_code=200, content=jsonable_encoder([]))
loads_id = await _process_bol_attachments(attachments, state)
elapsed = time.time() - t0
logger.info(f"Processed BOLs in {elapsed:.2f} seconds")
return JSONResponse(status_code=200, content=loads_id)
reprocess_bols(request, body)
async
Reprocesa un conjunto de Bills of Lading (BOLs) específicos buscándolos por asunto en el correo.
Este endpoint permite forzar el reprocesamiento de BOLs concretos sin depender de
la fecha de llegada del correo. Busca los correos cuyo asunto coincida con los subjects
indicados a través del endpoint /api/graph/messages-by-subject (sin filtro de fecha),
descarga sus adjuntos y delega el procesamiento completo a _process_bol_attachments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Request
|
El objeto de solicitud de FastAPI, que contiene el estado de la aplicación (clientes de servicios, etc.). |
required |
body
|
ReprocessBolRequest
|
El cuerpo de la solicitud, con la lista de subjects de correo a buscar (máximo 5 por llamada al servicio de graph). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
JSONResponse |
Una respuesta JSON con los IDs de las cargas (loads) creadas o actualizadas en el CRM, con código 200. Si no se encuentran adjuntos para los subjects indicados, retorna una lista vacía. |
Source code in app/api/process.py
@router.post(
"/reprocess-bols",
response_class=JSONResponse,
response_model=List[FinalLoadV3],
status_code=200,
responses={200: {"description": "BOLs reprocessed successfully"}},
)
async def reprocess_bols(
request: Request,
body: ReprocessBolRequest,
):
"""
Reprocesa un conjunto de Bills of Lading (BOLs) específicos buscándolos por asunto en el correo.
Este endpoint permite forzar el reprocesamiento de BOLs concretos sin depender de
la fecha de llegada del correo. Busca los correos cuyo asunto coincida con los subjects
indicados a través del endpoint `/api/graph/messages-by-subject` (sin filtro de fecha),
descarga sus adjuntos y delega el procesamiento completo a `_process_bol_attachments`.
Args:
request (Request): El objeto de solicitud de FastAPI, que contiene el estado de la
aplicación (clientes de servicios, etc.).
body (ReprocessBolRequest): El cuerpo de la solicitud, con la lista de subjects de
correo a buscar (máximo 5 por llamada al servicio de graph).
Returns:
JSONResponse: Una respuesta JSON con los IDs de las cargas (loads) creadas o
actualizadas en el CRM, con código 200. Si no se encuentran adjuntos
para los subjects indicados, retorna una lista vacía.
"""
t0 = time.time()
state: AppState = request.app.state
# 1. Buscar adjuntos por asunto completo en el servicio de correo
attachments = await state.mail_client.fetch_attachments_by_subject(body.subjects)
if not attachments:
logger.warning("No attachments found for subjects: %s", body.subjects)
return JSONResponse(status_code=200, content=jsonable_encoder([]))
loads_id = await _process_bol_attachments(attachments, state)
elapsed = time.time() - t0
logger.info(f"Reprocessed BOLs in {elapsed:.2f} seconds")
return JSONResponse(status_code=200, content=loads_id)Servicios
A continuación se detallan los diferentes servicios y clientes que son utilizados por el endpoint principal para llevar a cabo el procesamiento.
Mail Client
Este servicio se conecta a la API de correo para obtener los correos electrónicos y extraer los archivos adjuntos (PDFs de los BOLs).
app.services.mail_client
app.services.mail_client
Classes
MailClient
Cliente para interactuar con la API interna de correo.
Esta clase gestiona la comunicación con un servicio de correo para obtener correos electrónicos y sus archivos adjuntos, aplicando filtros específicos para los Bills of Lading (BOLs). También implementa un mecanismo para evitar el procesamiento duplicado de correos ya vistos.
Source code in app/services/mail_client.py
class MailClient:
"""
Cliente para interactuar con la API interna de correo.
Esta clase gestiona la comunicación con un servicio de correo para obtener
correos electrónicos y sus archivos adjuntos, aplicando filtros específicos
para los Bills of Lading (BOLs). También implementa un mecanismo para evitar
el procesamiento duplicado de correos ya vistos.
"""
def __init__(self):
"""
Inicializa el cliente de correo.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.mail_api_url
self.subject_filter = settings.mail_filter_subject
self.from_filter = settings.mail_filter_from
self.local_tz = ZoneInfo(settings.timezone)
self.lookback_minutes = settings.lookback_minutes
self.timeout = settings.http_client_timeout_seconds
# Guardamos {graph_id: timestamp_visto}
self._seen_ids: dict[str, datetime] = {}
self.ttl_hours = 24 # caducidad de 24h
# Retry logic
self.max_retries = 2
self.retry_delays = [2, 4] # segundos: 2s, 4s
def _cleanup_seen_ids(self):
"""
Limpia los IDs de correos que han superado su tiempo de vida (TTL).
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
cutoff = datetime.now(timezone.utc) - timedelta(hours=self.ttl_hours)
expired = [gid for gid, ts in self._seen_ids.items() if ts < cutoff]
for gid in expired:
del self._seen_ids[gid]
def _mark_seen(self, graph_id: str):
"""
Registra un ID de correo como "visto".
Args:
graph_id (str): El ID único del correo a registrar.
Returns:
No retorna ningún valor.
"""
self._seen_ids[graph_id] = datetime.now(timezone.utc)
def _is_seen(self, graph_id: str) -> bool:
"""
Verifica si un ID de correo ya ha sido procesado recientemente.
Args:
graph_id (str): El ID único del correo a verificar.
Returns:
bool: True si el ID ya ha sido visto, False en caso contrario.
"""
return graph_id in self._seen_ids
def extract_bol_codes(self, attachments: list[dict]):
"""
Extrae los códigos de BOL de una lista de archivos adjuntos.
Utiliza una expresión regular para encontrar archivos adjuntos que coincidan
con el patrón "BOL_XXXX.pdf" y extrae el código numérico.
Args:
attachments (list[dict]): Una lista de diccionarios de archivos adjuntos.
Returns:
tuple[list[str], list[dict]]: Una tupla que contiene una lista de códigos
de BOL extraídos y una lista de los archivos
adjuntos que coincidieron con el patrón.
"""
pattern = re.compile(r"BOL[_-](\d+)\.pdf$", re.IGNORECASE)
codes = []
filtered_attachments = []
for att in attachments:
name = att.get("name", "")
match = pattern.search(name)
if match:
codes.append(match.group(1))
filtered_attachments.append(att)
return codes, filtered_attachments
async def fetch_bol_attachments(
self, start_date: datetime, limit: int = 10
) -> list[dict]:
"""
Recupera adjuntos de correos que contienen Bills of Lading (BOL).
Consulta el endpoint `/filtered-messages` de la API de correo para obtener
mensajes que coincidan con los filtros configurados (asunto y remitente).
Luego, para cada correo, obtiene sus archivos adjuntos.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Args:
start_date (datetime): La fecha y hora de inicio para filtrar los correos.
limit (int, optional): El número máximo de correos a procesar.
Defaults to 10.
Returns:
list[dict]: Una lista de diccionarios de archivos adjuntos. Cada diccionario
contiene el nombre ('name') y el contenido ('content') del adjunto.
Retorna una lista vacía si falla después de los reintentos.
"""
# limpiar viejos ids antes de cada consulta
self._cleanup_seen_ids()
# asegurar que start_date tiene zona horaria local
start_date = start_date.replace(tzinfo=self.local_tz)
# convertir a UTC después de ajustar el lookback
utc_start_date = (
start_date - timedelta(minutes=self.lookback_minutes)
).astimezone(timezone.utc)
logger.info("[Graph] Called GRAPH start_date=%s", utc_start_date.isoformat())
for attempt in range(self.max_retries):
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
logger.info(
f"Consultando Mail API (intento {attempt + 1}/{self.max_retries})"
)
resp = await client.get(
f"{self.base_url}/api/graph/filtered-messages",
params={
"start_date": utc_start_date.isoformat(), # Graph espera UTC
"limit": limit,
"subject_filter": self.subject_filter,
"from_filter": self.from_filter,
},
)
resp.raise_for_status()
mails = resp.json()
att_data = []
for mail in mails:
graph_id = mail.get("graph_id")
if not graph_id:
continue
# ignorar si ya fue procesado
if self._is_seen(graph_id):
continue
try:
att_resp = await client.get(
f"{self.base_url}/api/graph/v2/attachments/{graph_id}"
)
att_resp.raise_for_status()
attachments = att_resp.json()
if attachments:
att_data.extend(attachments)
self._mark_seen(graph_id)
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión obteniendo adjuntos para graph_id={graph_id}: {str(e)}"
)
continue
except httpx.TimeoutException as e:
logger.warning(
f"Timeout obteniendo adjuntos para graph_id={graph_id}: {str(e)}"
)
continue
except Exception as e:
logger.warning(
f"Error inesperado obteniendo adjuntos para graph_id={graph_id}: {str(e)}"
)
continue
logger.info(
f"Consulta a Mail API exitosa, {len(att_data)} adjuntos encontrados"
)
return att_data
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión TCP en Mail API (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
logger.warning(
f"Verifica: 1) URL de Mail API correcta ({self.base_url}), 2) Conexión de red, 3) Firewall"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
logger.error(
f"Fallo conexión a Mail API después de {self.max_retries} intentos. Error: {str(e)}"
)
return []
except httpx.TimeoutException as e:
logger.warning(
f"Timeout en Mail API (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
logger.warning(
f"Considera aumentar http_client_timeout_seconds (actual: {self.timeout}s)"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
logger.error(
f"Timeout en Mail API después de {self.max_retries} intentos. Error: {str(e)}"
)
return []
except Exception as e:
logger.error(
f"Error inesperado en Mail API (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
logger.error(
f"Error en Mail API después de {self.max_retries} intentos. Error: {str(e)}"
)
return []
return []
async def fetch_attachments_by_subject(self, subjects: list[str]) -> list[dict]:
"""
Recupera adjuntos de correos buscando por asunto completo.
Consulta el endpoint `/messages-by-subject` de la API de correo usando los subjects
proporcionados directamente. A diferencia de `fetch_bol_attachments`, este método no
aplica filtro de fecha, por lo que es adecuado para reprocesar correos históricos.
Para cada mensaje que tenga adjuntos, descarga los archivos asociados.
Args:
subjects (list[str]): Lista de asuntos (completos o parciales) a buscar en los correos.
La búsqueda es case-insensitive y por coincidencia parcial.
El servicio de graph acepta un máximo de 5 subjects por llamada.
Returns:
list[dict]: Una lista de diccionarios de archivos adjuntos. Cada diccionario
contiene el nombre ('name') y el contenido ('content') del adjunto.
Levanta una excepción si la solicitud HTTP falla.
"""
logger.info("[Graph] Fetching messages by subject: %s", subjects)
async with httpx.AsyncClient(timeout=self.timeout) as client:
resp = await client.get(
f"{self.base_url}/api/graph/messages-by-subject",
params=[("subjects", s) for s in subjects],
)
resp.raise_for_status()
messages: list[dict] = resp.json()
att_data = []
for message in messages:
if not message.get("has_attachments"):
continue
graph_id = message.get("graph_id")
if not graph_id:
continue
att_resp = await client.get(
f"{self.base_url}/api/graph/v2/attachments/{graph_id}"
)
att_resp.raise_for_status()
attachments = att_resp.json()
if attachments:
att_data.extend(attachments)
return att_dataFunctions
__init__()
Inicializa el cliente de correo.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor. |
Source code in app/services/mail_client.py
def __init__(self):
"""
Inicializa el cliente de correo.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.mail_api_url
self.subject_filter = settings.mail_filter_subject
self.from_filter = settings.mail_filter_from
self.local_tz = ZoneInfo(settings.timezone)
self.lookback_minutes = settings.lookback_minutes
self.timeout = settings.http_client_timeout_seconds
# Guardamos {graph_id: timestamp_visto}
self._seen_ids: dict[str, datetime] = {}
self.ttl_hours = 24 # caducidad de 24h
# Retry logic
self.max_retries = 2
self.retry_delays = [2, 4] # segundos: 2s, 4s
extract_bol_codes(attachments)
Extrae los códigos de BOL de una lista de archivos adjuntos.
Utiliza una expresión regular para encontrar archivos adjuntos que coincidan con el patrón "BOL_XXXX.pdf" y extrae el código numérico.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
attachments
|
list[dict]
|
Una lista de diccionarios de archivos adjuntos. |
required |
Returns:
| Type | Description |
|---|---|
|
tuple[list[str], list[dict]]: Una tupla que contiene una lista de códigos de BOL extraídos y una lista de los archivos adjuntos que coincidieron con el patrón. |
Source code in app/services/mail_client.py
def extract_bol_codes(self, attachments: list[dict]):
"""
Extrae los códigos de BOL de una lista de archivos adjuntos.
Utiliza una expresión regular para encontrar archivos adjuntos que coincidan
con el patrón "BOL_XXXX.pdf" y extrae el código numérico.
Args:
attachments (list[dict]): Una lista de diccionarios de archivos adjuntos.
Returns:
tuple[list[str], list[dict]]: Una tupla que contiene una lista de códigos
de BOL extraídos y una lista de los archivos
adjuntos que coincidieron con el patrón.
"""
pattern = re.compile(r"BOL[_-](\d+)\.pdf$", re.IGNORECASE)
codes = []
filtered_attachments = []
for att in attachments:
name = att.get("name", "")
match = pattern.search(name)
if match:
codes.append(match.group(1))
filtered_attachments.append(att)
return codes, filtered_attachments
fetch_attachments_by_subject(subjects)
async
Recupera adjuntos de correos buscando por asunto completo.
Consulta el endpoint /messages-by-subject de la API de correo usando los subjects
proporcionados directamente. A diferencia de fetch_bol_attachments, este método no
aplica filtro de fecha, por lo que es adecuado para reprocesar correos históricos.
Para cada mensaje que tenga adjuntos, descarga los archivos asociados.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
subjects
|
list[str]
|
Lista de asuntos (completos o parciales) a buscar en los correos. La búsqueda es case-insensitive y por coincidencia parcial. El servicio de graph acepta un máximo de 5 subjects por llamada. |
required |
Returns:
| Type | Description |
|---|---|
list[dict]
|
list[dict]: Una lista de diccionarios de archivos adjuntos. Cada diccionario contiene el nombre ('name') y el contenido ('content') del adjunto. Levanta una excepción si la solicitud HTTP falla. |
Source code in app/services/mail_client.py
async def fetch_attachments_by_subject(self, subjects: list[str]) -> list[dict]:
"""
Recupera adjuntos de correos buscando por asunto completo.
Consulta el endpoint `/messages-by-subject` de la API de correo usando los subjects
proporcionados directamente. A diferencia de `fetch_bol_attachments`, este método no
aplica filtro de fecha, por lo que es adecuado para reprocesar correos históricos.
Para cada mensaje que tenga adjuntos, descarga los archivos asociados.
Args:
subjects (list[str]): Lista de asuntos (completos o parciales) a buscar en los correos.
La búsqueda es case-insensitive y por coincidencia parcial.
El servicio de graph acepta un máximo de 5 subjects por llamada.
Returns:
list[dict]: Una lista de diccionarios de archivos adjuntos. Cada diccionario
contiene el nombre ('name') y el contenido ('content') del adjunto.
Levanta una excepción si la solicitud HTTP falla.
"""
logger.info("[Graph] Fetching messages by subject: %s", subjects)
async with httpx.AsyncClient(timeout=self.timeout) as client:
resp = await client.get(
f"{self.base_url}/api/graph/messages-by-subject",
params=[("subjects", s) for s in subjects],
)
resp.raise_for_status()
messages: list[dict] = resp.json()
att_data = []
for message in messages:
if not message.get("has_attachments"):
continue
graph_id = message.get("graph_id")
if not graph_id:
continue
att_resp = await client.get(
f"{self.base_url}/api/graph/v2/attachments/{graph_id}"
)
att_resp.raise_for_status()
attachments = att_resp.json()
if attachments:
att_data.extend(attachments)
return att_data
fetch_bol_attachments(start_date, limit=10)
async
Recupera adjuntos de correos que contienen Bills of Lading (BOL).
Consulta el endpoint /filtered-messages de la API de correo para obtener
mensajes que coincidan con los filtros configurados (asunto y remitente).
Luego, para cada correo, obtiene sus archivos adjuntos.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start_date
|
datetime
|
La fecha y hora de inicio para filtrar los correos. |
required |
limit
|
int
|
El número máximo de correos a procesar. Defaults to 10. |
10
|
Returns:
| Type | Description |
|---|---|
list[dict]
|
list[dict]: Una lista de diccionarios de archivos adjuntos. Cada diccionario contiene el nombre ('name') y el contenido ('content') del adjunto. Retorna una lista vacía si falla después de los reintentos. |
Source code in app/services/mail_client.py
async def fetch_bol_attachments(
self, start_date: datetime, limit: int = 10
) -> list[dict]:
"""
Recupera adjuntos de correos que contienen Bills of Lading (BOL).
Consulta el endpoint `/filtered-messages` de la API de correo para obtener
mensajes que coincidan con los filtros configurados (asunto y remitente).
Luego, para cada correo, obtiene sus archivos adjuntos.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Args:
start_date (datetime): La fecha y hora de inicio para filtrar los correos.
limit (int, optional): El número máximo de correos a procesar.
Defaults to 10.
Returns:
list[dict]: Una lista de diccionarios de archivos adjuntos. Cada diccionario
contiene el nombre ('name') y el contenido ('content') del adjunto.
Retorna una lista vacía si falla después de los reintentos.
"""
# limpiar viejos ids antes de cada consulta
self._cleanup_seen_ids()
# asegurar que start_date tiene zona horaria local
start_date = start_date.replace(tzinfo=self.local_tz)
# convertir a UTC después de ajustar el lookback
utc_start_date = (
start_date - timedelta(minutes=self.lookback_minutes)
).astimezone(timezone.utc)
logger.info("[Graph] Called GRAPH start_date=%s", utc_start_date.isoformat())
for attempt in range(self.max_retries):
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
logger.info(
f"Consultando Mail API (intento {attempt + 1}/{self.max_retries})"
)
resp = await client.get(
f"{self.base_url}/api/graph/filtered-messages",
params={
"start_date": utc_start_date.isoformat(), # Graph espera UTC
"limit": limit,
"subject_filter": self.subject_filter,
"from_filter": self.from_filter,
},
)
resp.raise_for_status()
mails = resp.json()
att_data = []
for mail in mails:
graph_id = mail.get("graph_id")
if not graph_id:
continue
# ignorar si ya fue procesado
if self._is_seen(graph_id):
continue
try:
att_resp = await client.get(
f"{self.base_url}/api/graph/v2/attachments/{graph_id}"
)
att_resp.raise_for_status()
attachments = att_resp.json()
if attachments:
att_data.extend(attachments)
self._mark_seen(graph_id)
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión obteniendo adjuntos para graph_id={graph_id}: {str(e)}"
)
continue
except httpx.TimeoutException as e:
logger.warning(
f"Timeout obteniendo adjuntos para graph_id={graph_id}: {str(e)}"
)
continue
except Exception as e:
logger.warning(
f"Error inesperado obteniendo adjuntos para graph_id={graph_id}: {str(e)}"
)
continue
logger.info(
f"Consulta a Mail API exitosa, {len(att_data)} adjuntos encontrados"
)
return att_data
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión TCP en Mail API (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
logger.warning(
f"Verifica: 1) URL de Mail API correcta ({self.base_url}), 2) Conexión de red, 3) Firewall"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
logger.error(
f"Fallo conexión a Mail API después de {self.max_retries} intentos. Error: {str(e)}"
)
return []
except httpx.TimeoutException as e:
logger.warning(
f"Timeout en Mail API (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
logger.warning(
f"Considera aumentar http_client_timeout_seconds (actual: {self.timeout}s)"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
logger.error(
f"Timeout en Mail API después de {self.max_retries} intentos. Error: {str(e)}"
)
return []
except Exception as e:
logger.error(
f"Error inesperado en Mail API (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
logger.error(
f"Error en Mail API después de {self.max_retries} intentos. Error: {str(e)}"
)
return []
return []Primus Client
Cliente utilizado para conectarse a la API de Primus y obtener la información de las reservas (bookings) asociada a los BOLs.
app.services.primus_client
app.services.primus_client
Classes
PrimusClient
Cliente para interactuar con la API de Primus.
Esta clase maneja la autenticación y las solicitudes a la API de Primus para obtener datos de reservas (bookings), con lógica de reintento y mejor manejo de errores de conexión.
Source code in app/services/primus_client.py
class PrimusClient:
"""
Cliente para interactuar con la API de Primus.
Esta clase maneja la autenticación y las solicitudes a la API de Primus
para obtener datos de reservas (bookings), con lógica de reintento
y mejor manejo de errores de conexión.
"""
def __init__(self):
"""
Inicializa el cliente de Primus.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.primus_base_url
self.user = settings.primus_user
self.password = settings.primus_password
self.timeout = settings.http_client_timeout_seconds
self.token = None
self.max_retries = 2
self.retry_delays = [2, 4] # segundos: 2s, 4s
async def login(self):
"""
Autentica con la API de Primus y almacena el token de acceso.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor, pero almacena el token de acceso en self.token.
Levanta una excepción si la autenticación falla después de los reintentos.
"""
for attempt in range(self.max_retries):
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
logger.info(
f"Intentando login en Primus (intento {attempt + 1}/{self.max_retries})"
)
resp = await client.post(
f"{self.base_url}/login",
json={"username": self.user, "password": self.password},
)
resp.raise_for_status()
data = resp.json()
self.token = data.get("data", {}).get("accessToken")
logger.info("Login en Primus exitoso, token obtenido")
return
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión TCP en login (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Fallo conexión a Primus después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": "N/A",
"error_message": f"[PRIMUS LOGIN] {error_msg}",
}
)
raise
except httpx.TimeoutException as e:
logger.warning(
f"Timeout en login (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Timeout en Primus login después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": "N/A",
"error_message": f"[PRIMUS LOGIN TIMEOUT] {error_msg}",
}
)
raise
except Exception as e:
logger.error(
f"Error inesperado en login (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Error en Primus login después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": "N/A",
"error_message": f"[PRIMUS LOGIN ERROR] {error_msg}",
}
)
raise
async def get_booking_data(self, bol_numbers: list[str]) -> dict:
"""
Recupera la información de la reserva para una lista de números de BOL de Primus.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Args:
bol_numbers (list[str]): Una lista de números de BOL (Bill of Lading) para consultar.
Returns:
dict: Un diccionario con los datos de la reserva obtenidos de la API.
Si el token ha expirado, intenta autenticarse de nuevo y reintentar la solicitud.
Levanta una excepción si la solicitud a la API falla.
"""
if not self.token:
await self.login()
headers = {"Authorization": f"Bearer {self.token}"}
for attempt in range(self.max_retries):
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
logger.info(
f"Consultando Primus para BOLs: {bol_numbers} (intento {attempt + 1}/{self.max_retries})"
)
resp = await client.get(
f"{self.base_url}/book",
headers=headers,
params={"BOLNumber[]": bol_numbers},
)
if resp.status_code == 401: # Token expirado
logger.warning("Token expirado, realizando re-login")
await self.login()
return await self.get_booking_data(bol_numbers)
resp.raise_for_status()
logger.info(f"Consulta a Primus exitosa para BOLs: {bol_numbers}")
return resp.json()
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión TCP (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Fallo conexión a Primus para BOLs {bol_numbers} después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": ",".join(bol_numbers),
"error_message": f"[PRIMUS CONNECT ERROR] {error_msg}",
}
)
raise
except httpx.TimeoutException as e:
logger.warning(
f"Timeout en consulta (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Timeout en Primus para BOLs {bol_numbers} después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": ",".join(bol_numbers),
"error_message": f"[PRIMUS TIMEOUT] {error_msg}",
}
)
raise
except Exception as e:
logger.error(
f"Error inesperado (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Error en Primus para BOLs {bol_numbers} después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": ",".join(bol_numbers),
"error_message": f"[PRIMUS ERROR] {error_msg}",
}
)
raise
return {}Functions
__init__()
Inicializa el cliente de Primus.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor. |
Source code in app/services/primus_client.py
def __init__(self):
"""
Inicializa el cliente de Primus.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.primus_base_url
self.user = settings.primus_user
self.password = settings.primus_password
self.timeout = settings.http_client_timeout_seconds
self.token = None
self.max_retries = 2
self.retry_delays = [2, 4] # segundos: 2s, 4s
get_booking_data(bol_numbers)
async
Recupera la información de la reserva para una lista de números de BOL de Primus.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bol_numbers
|
list[str]
|
Una lista de números de BOL (Bill of Lading) para consultar. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict
|
Un diccionario con los datos de la reserva obtenidos de la API. Si el token ha expirado, intenta autenticarse de nuevo y reintentar la solicitud. |
dict
|
Levanta una excepción si la solicitud a la API falla. |
Source code in app/services/primus_client.py
async def get_booking_data(self, bol_numbers: list[str]) -> dict:
"""
Recupera la información de la reserva para una lista de números de BOL de Primus.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Args:
bol_numbers (list[str]): Una lista de números de BOL (Bill of Lading) para consultar.
Returns:
dict: Un diccionario con los datos de la reserva obtenidos de la API.
Si el token ha expirado, intenta autenticarse de nuevo y reintentar la solicitud.
Levanta una excepción si la solicitud a la API falla.
"""
if not self.token:
await self.login()
headers = {"Authorization": f"Bearer {self.token}"}
for attempt in range(self.max_retries):
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
logger.info(
f"Consultando Primus para BOLs: {bol_numbers} (intento {attempt + 1}/{self.max_retries})"
)
resp = await client.get(
f"{self.base_url}/book",
headers=headers,
params={"BOLNumber[]": bol_numbers},
)
if resp.status_code == 401: # Token expirado
logger.warning("Token expirado, realizando re-login")
await self.login()
return await self.get_booking_data(bol_numbers)
resp.raise_for_status()
logger.info(f"Consulta a Primus exitosa para BOLs: {bol_numbers}")
return resp.json()
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión TCP (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Fallo conexión a Primus para BOLs {bol_numbers} después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": ",".join(bol_numbers),
"error_message": f"[PRIMUS CONNECT ERROR] {error_msg}",
}
)
raise
except httpx.TimeoutException as e:
logger.warning(
f"Timeout en consulta (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Timeout en Primus para BOLs {bol_numbers} después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": ",".join(bol_numbers),
"error_message": f"[PRIMUS TIMEOUT] {error_msg}",
}
)
raise
except Exception as e:
logger.error(
f"Error inesperado (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Error en Primus para BOLs {bol_numbers} después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": ",".join(bol_numbers),
"error_message": f"[PRIMUS ERROR] {error_msg}",
}
)
raise
return {}
login()
async
Autentica con la API de Primus y almacena el token de acceso.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor, pero almacena el token de acceso en self.token. |
|
|
Levanta una excepción si la autenticación falla después de los reintentos. |
Source code in app/services/primus_client.py
async def login(self):
"""
Autentica con la API de Primus y almacena el token de acceso.
Incluye lógica de reintento para manejar fallos transitorios de conexión.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor, pero almacena el token de acceso en self.token.
Levanta una excepción si la autenticación falla después de los reintentos.
"""
for attempt in range(self.max_retries):
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
logger.info(
f"Intentando login en Primus (intento {attempt + 1}/{self.max_retries})"
)
resp = await client.post(
f"{self.base_url}/login",
json={"username": self.user, "password": self.password},
)
resp.raise_for_status()
data = resp.json()
self.token = data.get("data", {}).get("accessToken")
logger.info("Login en Primus exitoso, token obtenido")
return
except httpx.ConnectError as e:
logger.warning(
f"Error de conexión TCP en login (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Fallo conexión a Primus después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": "N/A",
"error_message": f"[PRIMUS LOGIN] {error_msg}",
}
)
raise
except httpx.TimeoutException as e:
logger.warning(
f"Timeout en login (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Timeout en Primus login después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": "N/A",
"error_message": f"[PRIMUS LOGIN TIMEOUT] {error_msg}",
}
)
raise
except Exception as e:
logger.error(
f"Error inesperado en login (intento {attempt + 1}/{self.max_retries}): {str(e)}"
)
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delays[attempt])
else:
error_msg = f"Error en Primus login después de {self.max_retries} intentos. Error: {str(e)}"
logger.error(error_msg)
email_service.send_email(
dynamic_data={
"bol_number": "N/A",
"error_message": f"[PRIMUS LOGIN ERROR] {error_msg}",
}
)
raiseAgente de IA (PDF Processing)
Este conjunto de módulos se encarga de procesar los archivos PDF de los BOLs utilizando un modelo de IA generativa (Gemini) para extraer la información estructurada.
app.services.agent.process_pdf
Este módulo orquesta el procesamiento en paralelo de los PDFs, gestionando la concurrencia y los reintentos.
app.services.agent.process_pdf
Classes
Functions
process_pdfs(agent, prompt, attachments, response_schema)
async
Procesa una lista de archivos PDF en paralelo utilizando el LLM.
Crea y ejecuta tareas asíncronas para cada archivo adjunto, gestionando la concurrencia. Recopila los resultados y maneja las excepciones de forma individual para cada tarea, asegurando que el fallo de un PDF no detenga el procesamiento de los demás.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent
|
GenaiProcessor
|
La instancia del procesador Genai a utilizar. |
required |
prompt
|
str
|
El prompt a enviar al LLM para cada PDF. |
required |
attachments
|
list[dict]
|
Una lista de diccionarios, donde cada uno representa un archivo PDF a procesar. |
required |
response_schema
|
dict
|
El esquema JSON que el LLM debe seguir para las respuestas. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, dict]
|
dict[str, dict]: Un diccionario consolidado con los resultados exitosos, donde cada clave es un número de BOL. Los procesamientos fallidos se registran como errores y no se incluyen en el retorno. |
Source code in app/services/agent/process_pdf.py
async def process_pdfs(
agent: GenaiProcessor,
prompt: str,
attachments: list[dict],
response_schema: dict,
) -> dict[str, dict]:
"""
Procesa una lista de archivos PDF en paralelo utilizando el LLM.
Crea y ejecuta tareas asíncronas para cada archivo adjunto, gestionando la
concurrencia. Recopila los resultados y maneja las excepciones de forma
individual para cada tarea, asegurando que el fallo de un PDF no detenga
el procesamiento de los demás.
Args:
agent (GenaiProcessor): La instancia del procesador Genai a utilizar.
prompt (str): El prompt a enviar al LLM para cada PDF.
attachments (list[dict]): Una lista de diccionarios, donde cada uno
representa un archivo PDF a procesar.
response_schema (dict): El esquema JSON que el LLM debe seguir para
las respuestas.
Returns:
dict[str, dict]: Un diccionario consolidado con los resultados exitosos,
donde cada clave es un número de BOL. Los procesamientos fallidos
se registran como errores y no se incluyen en el retorno.
"""
tasks = [
_call_llm_threadsafe(agent, prompt, att, response_schema) for att in attachments
]
results = await asyncio.gather(*tasks, return_exceptions=True)
final_results: dict[str, dict] = {}
for i, r in enumerate(results):
if isinstance(r, Exception):
logger.error("Falló el procesamiento del PDF %d: %s", i, r)
continue
# r tiene la forma {"BOL123": {"parsed": ..., "usage": ...}}
final_results.update(cast(dict, r))
return final_results
retry_policy()
Define y retorna una política de reintentos para manejar errores transitorios.
Utiliza una estrategia de backoff exponencial, reintentando hasta 5 veces
cuando se producen errores específicos como RateLimitError, APIError de Google,
o HTTPStatusError.
Returns:
| Name | Type | Description |
|---|---|---|
Callable |
Callable
|
Un decorador de reintentos de la librería |
Source code in app/services/agent/process_pdf.py
def retry_policy() -> Callable:
"""
Define y retorna una política de reintentos para manejar errores transitorios.
Utiliza una estrategia de backoff exponencial, reintentando hasta 5 veces
cuando se producen errores específicos como `RateLimitError`, `APIError` de Google,
o `HTTPStatusError`.
Returns:
Callable: Un decorador de reintentos de la librería `tenacity`.
"""
return retry(
wait=wait_exponential(multiplier=1, min=2, max=30), # 2s, 4s, 8s... hasta 30s
stop=stop_after_attempt(5), # máximo 5 intentos
retry=retry_if_exception_type((RateLimitError, APIError, HTTPStatusError)),
reraise=True,
)
safe_json_load(response, bol_number)
Parsea una cadena JSON de forma segura, manejando posibles errores.
Si la cadena no es un JSON válido, registra un error, envía un correo de notificación y retorna una estructura de respuesta por defecto.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
str
|
La cadena de texto que se espera que sea un JSON. |
required |
bol_number
|
str
|
El número de BOL asociado a la respuesta, para incluirlo en notificaciones de error. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict
|
El objeto JSON parseado, o un diccionario por defecto si ocurre un error. |
Source code in app/services/agent/process_pdf.py
def safe_json_load(response: str, bol_number: str) -> dict:
"""
Parsea una cadena JSON de forma segura, manejando posibles errores.
Si la cadena no es un JSON válido, registra un error, envía un correo
de notificación y retorna una estructura de respuesta por defecto.
Args:
response (str): La cadena de texto que se espera que sea un JSON.
bol_number (str): El número de BOL asociado a la respuesta, para
incluirlo en notificaciones de error.
Returns:
dict: El objeto JSON parseado, o un diccionario por defecto si ocurre un error.
"""
try:
return json.loads(response)
except (json.JSONDecodeError, TypeError):
logger.error("Error parsing JSON response from LLM")
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": f"[ERROR] Respuesta invalida LLM para BOL: {bol_number}",
}
)
return DEFAULT_LLM_RESPONSE.model_dump()app.services.agent.genai_client
Este cliente encapsula la comunicación con la API del modelo de IA generativa.
app.services.agent.genai_client
Classes
GenaiProcessor
Clase para interactuar con los modelos de IA generativa de Google a través del SDK de Genai.
Esta clase encapsula la configuración y la comunicación con los modelos de Vertex AI, proporcionando métodos para generar contenido a partir de URIs, bytes de archivos, y para manejar sesiones de chat.
Source code in app/services/agent/genai_client.py
class GenaiProcessor:
"""
Clase para interactuar con los modelos de IA generativa de Google a través del SDK de Genai.
Esta clase encapsula la configuración y la comunicación con los modelos de Vertex AI,
proporcionando métodos para generar contenido a partir de URIs, bytes de archivos,
y para manejar sesiones de chat.
"""
def __init__(
self,
model_name: str,
system_instruction: str | None = None,
) -> None:
"""
Inicializa el cliente de GenaiProcessor.
Args:
model_name (str): El nombre del modelo a utilizar (ej. "gemini-2.5-pro").
system_instruction (str | None, optional): Instrucción de sistema para
guiar el comportamiento del modelo. Defaults to None.
"""
self.client = genai.Client(
vertexai=True,
project=os.environ["VERTEX_PROJECT_ID"],
location=os.environ["VERTEX_LOCATION"],
)
self.model_name = model_name
self.system_instruction = system_instruction
self._thinking_models = {"gemini-2.5-flash", "gemini-2.5-pro"}
def _get_safety_settings(
self, threshold: HarmBlockThreshold
) -> list[SafetySetting]:
"""
Construye la configuración de seguridad para las solicitudes a la API.
Args:
threshold (HarmBlockThreshold): El nivel de umbral para bloquear contenido
dañino.
Returns:
list[SafetySetting]: Una lista de objetos SafetySetting para cada categoría
de contenido dañino.
"""
return [
SafetySetting(category=category, threshold=threshold)
for category in [
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
HarmCategory.HARM_CATEGORY_HARASSMENT,
HarmCategory.HARM_CATEGORY_HATE_SPEECH,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
]
]
def _normalize_files(
self,
files: list[dict[str, str]] | None,
bucket_path: str | None,
mime_type: str | None,
) -> list[dict[str, str]]:
"""
Normaliza la lista de archivos para la solicitud a la API.
Valida y formatea la lista de archivos. Si `files` es None, la inicializa.
Si `bucket_path` se proporciona y la lista de archivos está vacía, crea
una entrada de archivo para el bucket.
Args:
files (list[dict[str, str]] | None): La lista de archivos a normalizar.
Cada archivo es un dict con 'file_uri' y 'mime_type'.
bucket_path (str | None): La ruta GCS a un archivo.
mime_type (str | None): El tipo MIME del archivo en `bucket_path`.
Returns:
list[dict[str, str]]: La lista de archivos normalizada.
Raises:
ValueError: Si `files` no es una lista de diccionarios o si a un
archivo le falta 'file_uri' or 'mime_type'.
"""
if files is None:
files = []
if not isinstance(files, list) or not all(isinstance(f, dict) for f in files):
raise ValueError("`files` must be a list of dictionaries.")
if not files and bucket_path:
files = [
{
"file_uri": bucket_path,
"mime_type": mime_type or "application/pdf",
},
]
for f in files:
if not f.get("file_uri") or not f.get("mime_type"):
raise ValueError("Each file must include 'file_uri' and 'mime_type'.")
return files
def _build_generation_config(
self,
*,
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> GenerateContentConfig:
"""
Construye el objeto de configuración para la generación de contenido.
Args:
response_schema (dict | None, optional): El esquema para la respuesta. Defaults to None.
response_mime_type (str, optional): El tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): La temperatura de la generación. Defaults to 0.0.
max_output_tokens (int, optional): Máximo de tokens en la respuesta. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Top-p para el muestreo. Defaults to 1.0.
seed (int, optional): Semilla para la reproducibilidad. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Umbral de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
GenerateContentConfig: El objeto de configuración de generación.
"""
config = {
"system_instruction": self.system_instruction,
"response_schema": response_schema,
"response_mime_type": response_mime_type,
"temperature": temperature,
"max_output_tokens": max_output_tokens,
"presence_penalty": presence_penalty,
"frequency_penalty": frequency_penalty,
"top_p": top_p,
"seed": seed,
"safety_settings": self._get_safety_settings(safety_threshold),
}
if self.model_name in self._thinking_models:
config["thinking_config"] = ThinkingConfig(
thinking_budget=thinking_budget,
)
return GenerateContentConfig(**config)
def extract_usage_metadata(self, response: Any) -> dict:
"""
Extrae los metadatos de uso de la respuesta de la API.
Args:
response (object): El objeto de respuesta de la API de Genai.
Returns:
dict: Un diccionario con el recuento de tokens (prompt, respuesta, total, y thoughts si aplica).
"""
usage = {
"prompt_tokens": getattr(
response.usage_metadata, "prompt_token_count", None
),
"response_tokens": getattr(
response.usage_metadata, "candidates_token_count", None
),
"total_tokens": getattr(response.usage_metadata, "total_token_count", None),
}
if self.model_name in self._thinking_models:
usage["thoughts_tokens"] = getattr(
response.usage_metadata,
"thoughts_token_count",
None,
)
return usage
def generate_from_uri(
self,
prompt: str,
*,
files: list[dict[str, str]] | None = None,
bucket_path: str | None = None,
mime_type: str | None = None,
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> dict[str, object]:
"""
Genera contenido a partir de un prompt y archivos especificados por URI (GCS).
Args:
prompt (str): El prompt de texto para el modelo.
files (list[dict[str, str]] | None, optional): Lista de archivos en GCS. Defaults to None.
bucket_path (str | None, optional): Ruta a un único archivo en GCS. Defaults to None.
mime_type (str | None, optional): Tipo MIME para `bucket_path`. Defaults to None.
response_schema (dict | None, optional): Esquema de respuesta esperado. Defaults to None.
response_mime_type (str, optional): Tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
dict[str, object]: Un diccionario con la respuesta completa de la API,
los metadatos de uso y la respuesta parseada.
Raises:
VertexAIExceptionError: Si ocurre un error en la llamada a la API de Genai.
"""
files = self._normalize_files(files, bucket_path, mime_type)
contents = [
Part.from_uri(
file_uri=f["file_uri"],
mime_type=f["mime_type"],
)
for f in files
] + [
Part.from_text(text=prompt),
]
config = self._build_generation_config(
response_schema=response_schema,
response_mime_type=response_mime_type,
temperature=temperature,
max_output_tokens=max_output_tokens,
presence_penalty=presence_penalty,
frequency_penalty=frequency_penalty,
top_p=top_p,
seed=seed,
safety_threshold=safety_threshold,
thinking_budget=thinking_budget,
)
try:
response = self.client.models.generate_content(
model=self.model_name,
contents=contents,
config=config,
)
except APIError as e:
raise VertexAIExceptionError(
original_exception=e,
) from e
usage = self.extract_usage_metadata(response)
return {
"response": response,
"usage": usage,
"parsed": self.parse_response(response),
}
def generate_from_bytes(
self,
prompt: str,
*,
pdf_bytes: bytes,
mime_type: str = "application/pdf",
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> dict[str, object]:
"""
Genera contenido a partir de un prompt y los bytes de un archivo PDF.
Args:
prompt (str): El prompt de texto para el modelo.
pdf_bytes (bytes): Los bytes del archivo PDF a procesar.
mime_type (str, optional): El tipo MIME del archivo. Defaults to "application/pdf".
response_schema (dict | None, optional): Esquema de respuesta esperado. Defaults to None.
response_mime_type (str, optional): Tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
dict[str, object]: Un diccionario con la respuesta completa de la API,
los metadatos de uso y la respuesta parseada.
"""
contents = [
Part.from_bytes(
data=pdf_bytes,
mime_type=mime_type,
),
Part.from_text(
text=prompt,
),
]
config = {
"system_instruction": self.system_instruction,
"response_mime_type": response_mime_type,
"response_schema": response_schema,
"temperature": temperature,
"max_output_tokens": max_output_tokens,
"presence_penalty": presence_penalty,
"frequency_penalty": frequency_penalty,
"top_p": top_p,
"seed": seed,
"safety_settings": self._get_safety_settings(
safety_threshold,
),
}
if self.model_name in self._thinking_models:
config["thinking_config"] = ThinkingConfig(
thinking_budget=thinking_budget,
)
response = self.client.models.generate_content(
model=self.model_name,
contents=contents,
config=GenerateContentConfig(**config),
)
usage = self.extract_usage_metadata(response)
return {
"response": response,
"usage": usage,
"parsed": self.parse_response(response),
}
def get_chat_session(
self,
*,
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> object:
"""
Crea y retorna una nueva sesión de chat con la configuración especificada.
Args:
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
object: Un objeto de sesión de chat del SDK de Genai.
"""
config = self._build_generation_config(
temperature=temperature,
max_output_tokens=max_output_tokens,
presence_penalty=presence_penalty,
frequency_penalty=frequency_penalty,
top_p=top_p,
seed=seed,
safety_threshold=safety_threshold,
thinking_budget=thinking_budget,
)
return self.client.chats.create(
model=self.model_name,
config=config,
)
def generate_chat(
self,
chat: Any,
prompt: str,
*,
files: list[dict[str, str]] | None = None,
bucket_path: str | None = None,
mime_type: str | None = None,
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> dict[str, object]:
"""
Envía un mensaje a una sesión de chat existente y obtiene una respuesta.
Args:
chat (object): La sesión de chat activa.
prompt (str): El prompt de texto para el modelo.
files (list[dict[str, str]] | None, optional): Lista de archivos en GCS. Defaults to None.
bucket_path (str | None, optional): Ruta a un único archivo en GCS. Defaults to None.
mime_type (str | None, optional): Tipo MIME para `bucket_path`. Defaults to None.
response_schema (dict | None, optional): Esquema de respuesta esperado. Defaults to None.
response_mime_type (str, optional): Tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
dict[str, object]: Un diccionario con la respuesta completa de la API,
los metadatos de uso y la respuesta parseada.
Raises:
VertexAIExceptionError: Si ocurre un error en la llamada a la API de Genai.
"""
files = self._normalize_files(files, bucket_path, mime_type)
parts = [
Part.from_text(text=prompt),
] + [
Part.from_uri(
file_uri=f["file_uri"],
mime_type=f["mime_type"],
)
for f in files
]
config = self._build_generation_config(
response_schema=response_schema,
response_mime_type=response_mime_type,
temperature=temperature,
max_output_tokens=max_output_tokens,
presence_penalty=presence_penalty,
frequency_penalty=frequency_penalty,
top_p=top_p,
seed=seed,
safety_threshold=safety_threshold,
thinking_budget=thinking_budget,
)
try:
response = chat.send_message(message=parts, config=config)
except APIError as e:
raise VertexAIExceptionError(
original_exception=e,
) from e
usage = self.extract_usage_metadata(response)
return {
"response": response,
"usage": usage,
"parsed": self.parse_response(response),
}
def parse_response(self, response: Any) -> dict:
"""
Parsea la respuesta de la API para separar los 'thoughts' de las 'answers'.
Args:
response (object): El objeto de respuesta de la API de Genai.
Returns:
dict: Un diccionario con dos claves, 'thoughts' y 'answers', cada una
conteniendo una lista de las partes correspondientes de la respuesta.
"""
parsed: dict[str, list] = {
"thoughts": [],
"answers": [],
}
if not response.candidates or not response.candidates[0].content.parts:
return parsed
for part in response.candidates[0].content.parts:
if part.thought:
parsed["thoughts"].append(part.text)
else:
parsed["answers"].append(part.text)
return parsedFunctions
__init__(model_name, system_instruction=None)
Inicializa el cliente de GenaiProcessor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name
|
str
|
El nombre del modelo a utilizar (ej. "gemini-2.5-pro"). |
required |
system_instruction
|
str | None
|
Instrucción de sistema para guiar el comportamiento del modelo. Defaults to None. |
None
|
Source code in app/services/agent/genai_client.py
def __init__(
self,
model_name: str,
system_instruction: str | None = None,
) -> None:
"""
Inicializa el cliente de GenaiProcessor.
Args:
model_name (str): El nombre del modelo a utilizar (ej. "gemini-2.5-pro").
system_instruction (str | None, optional): Instrucción de sistema para
guiar el comportamiento del modelo. Defaults to None.
"""
self.client = genai.Client(
vertexai=True,
project=os.environ["VERTEX_PROJECT_ID"],
location=os.environ["VERTEX_LOCATION"],
)
self.model_name = model_name
self.system_instruction = system_instruction
self._thinking_models = {"gemini-2.5-flash", "gemini-2.5-pro"}
extract_usage_metadata(response)
Extrae los metadatos de uso de la respuesta de la API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
object
|
El objeto de respuesta de la API de Genai. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict
|
Un diccionario con el recuento de tokens (prompt, respuesta, total, y thoughts si aplica). |
Source code in app/services/agent/genai_client.py
def extract_usage_metadata(self, response: Any) -> dict:
"""
Extrae los metadatos de uso de la respuesta de la API.
Args:
response (object): El objeto de respuesta de la API de Genai.
Returns:
dict: Un diccionario con el recuento de tokens (prompt, respuesta, total, y thoughts si aplica).
"""
usage = {
"prompt_tokens": getattr(
response.usage_metadata, "prompt_token_count", None
),
"response_tokens": getattr(
response.usage_metadata, "candidates_token_count", None
),
"total_tokens": getattr(response.usage_metadata, "total_token_count", None),
}
if self.model_name in self._thinking_models:
usage["thoughts_tokens"] = getattr(
response.usage_metadata,
"thoughts_token_count",
None,
)
return usage
generate_chat(chat, prompt, *, files=None, bucket_path=None, mime_type=None, response_schema=None, response_mime_type='application/json', temperature=0.0, max_output_tokens=8192, presence_penalty=0.0, frequency_penalty=0.0, top_p=1.0, seed=20180507, safety_threshold=HarmBlockThreshold.BLOCK_NONE, thinking_budget=0)
Envía un mensaje a una sesión de chat existente y obtiene una respuesta.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
chat
|
object
|
La sesión de chat activa. |
required |
prompt
|
str
|
El prompt de texto para el modelo. |
required |
files
|
list[dict[str, str]] | None
|
Lista de archivos en GCS. Defaults to None. |
None
|
bucket_path
|
str | None
|
Ruta a un único archivo en GCS. Defaults to None. |
None
|
mime_type
|
str | None
|
Tipo MIME para |
None
|
response_schema
|
dict | None
|
Esquema de respuesta esperado. Defaults to None. |
None
|
response_mime_type
|
str
|
Tipo MIME de la respuesta. Defaults to "application/json". |
'application/json'
|
temperature
|
float
|
Controla la aleatoriedad. Defaults to 0.0. |
0.0
|
max_output_tokens
|
int
|
Límite de tokens de salida. Defaults to 8192. |
8192
|
presence_penalty
|
float
|
Penalización por presencia. Defaults to 0.0. |
0.0
|
frequency_penalty
|
float
|
Penalización por frecuencia. Defaults to 0.0. |
0.0
|
top_p
|
float
|
Muestreo nucleus. Defaults to 1.0. |
1.0
|
seed
|
int
|
Semilla para resultados deterministas. Defaults to 20180507. |
20180507
|
safety_threshold
|
HarmBlockThreshold
|
Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE. |
BLOCK_NONE
|
thinking_budget
|
int
|
Presupuesto para 'thinking'. Defaults to 0. |
0
|
Returns:
| Type | Description |
|---|---|
dict[str, object]
|
dict[str, object]: Un diccionario con la respuesta completa de la API, los metadatos de uso y la respuesta parseada. |
Raises:
| Type | Description |
|---|---|
VertexAIExceptionError
|
Si ocurre un error en la llamada a la API de Genai. |
Source code in app/services/agent/genai_client.py
def generate_chat(
self,
chat: Any,
prompt: str,
*,
files: list[dict[str, str]] | None = None,
bucket_path: str | None = None,
mime_type: str | None = None,
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> dict[str, object]:
"""
Envía un mensaje a una sesión de chat existente y obtiene una respuesta.
Args:
chat (object): La sesión de chat activa.
prompt (str): El prompt de texto para el modelo.
files (list[dict[str, str]] | None, optional): Lista de archivos en GCS. Defaults to None.
bucket_path (str | None, optional): Ruta a un único archivo en GCS. Defaults to None.
mime_type (str | None, optional): Tipo MIME para `bucket_path`. Defaults to None.
response_schema (dict | None, optional): Esquema de respuesta esperado. Defaults to None.
response_mime_type (str, optional): Tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
dict[str, object]: Un diccionario con la respuesta completa de la API,
los metadatos de uso y la respuesta parseada.
Raises:
VertexAIExceptionError: Si ocurre un error en la llamada a la API de Genai.
"""
files = self._normalize_files(files, bucket_path, mime_type)
parts = [
Part.from_text(text=prompt),
] + [
Part.from_uri(
file_uri=f["file_uri"],
mime_type=f["mime_type"],
)
for f in files
]
config = self._build_generation_config(
response_schema=response_schema,
response_mime_type=response_mime_type,
temperature=temperature,
max_output_tokens=max_output_tokens,
presence_penalty=presence_penalty,
frequency_penalty=frequency_penalty,
top_p=top_p,
seed=seed,
safety_threshold=safety_threshold,
thinking_budget=thinking_budget,
)
try:
response = chat.send_message(message=parts, config=config)
except APIError as e:
raise VertexAIExceptionError(
original_exception=e,
) from e
usage = self.extract_usage_metadata(response)
return {
"response": response,
"usage": usage,
"parsed": self.parse_response(response),
}
generate_from_bytes(prompt, *, pdf_bytes, mime_type='application/pdf', response_schema=None, response_mime_type='application/json', temperature=0.0, max_output_tokens=8192, presence_penalty=0.0, frequency_penalty=0.0, top_p=1.0, seed=20180507, safety_threshold=HarmBlockThreshold.BLOCK_NONE, thinking_budget=0)
Genera contenido a partir de un prompt y los bytes de un archivo PDF.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt
|
str
|
El prompt de texto para el modelo. |
required |
pdf_bytes
|
bytes
|
Los bytes del archivo PDF a procesar. |
required |
mime_type
|
str
|
El tipo MIME del archivo. Defaults to "application/pdf". |
'application/pdf'
|
response_schema
|
dict | None
|
Esquema de respuesta esperado. Defaults to None. |
None
|
response_mime_type
|
str
|
Tipo MIME de la respuesta. Defaults to "application/json". |
'application/json'
|
temperature
|
float
|
Controla la aleatoriedad. Defaults to 0.0. |
0.0
|
max_output_tokens
|
int
|
Límite de tokens de salida. Defaults to 8192. |
8192
|
presence_penalty
|
float
|
Penalización por presencia. Defaults to 0.0. |
0.0
|
frequency_penalty
|
float
|
Penalización por frecuencia. Defaults to 0.0. |
0.0
|
top_p
|
float
|
Muestreo nucleus. Defaults to 1.0. |
1.0
|
seed
|
int
|
Semilla para resultados deterministas. Defaults to 20180507. |
20180507
|
safety_threshold
|
HarmBlockThreshold
|
Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE. |
BLOCK_NONE
|
thinking_budget
|
int
|
Presupuesto para 'thinking'. Defaults to 0. |
0
|
Returns:
| Type | Description |
|---|---|
dict[str, object]
|
dict[str, object]: Un diccionario con la respuesta completa de la API, los metadatos de uso y la respuesta parseada. |
Source code in app/services/agent/genai_client.py
def generate_from_bytes(
self,
prompt: str,
*,
pdf_bytes: bytes,
mime_type: str = "application/pdf",
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> dict[str, object]:
"""
Genera contenido a partir de un prompt y los bytes de un archivo PDF.
Args:
prompt (str): El prompt de texto para el modelo.
pdf_bytes (bytes): Los bytes del archivo PDF a procesar.
mime_type (str, optional): El tipo MIME del archivo. Defaults to "application/pdf".
response_schema (dict | None, optional): Esquema de respuesta esperado. Defaults to None.
response_mime_type (str, optional): Tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
dict[str, object]: Un diccionario con la respuesta completa de la API,
los metadatos de uso y la respuesta parseada.
"""
contents = [
Part.from_bytes(
data=pdf_bytes,
mime_type=mime_type,
),
Part.from_text(
text=prompt,
),
]
config = {
"system_instruction": self.system_instruction,
"response_mime_type": response_mime_type,
"response_schema": response_schema,
"temperature": temperature,
"max_output_tokens": max_output_tokens,
"presence_penalty": presence_penalty,
"frequency_penalty": frequency_penalty,
"top_p": top_p,
"seed": seed,
"safety_settings": self._get_safety_settings(
safety_threshold,
),
}
if self.model_name in self._thinking_models:
config["thinking_config"] = ThinkingConfig(
thinking_budget=thinking_budget,
)
response = self.client.models.generate_content(
model=self.model_name,
contents=contents,
config=GenerateContentConfig(**config),
)
usage = self.extract_usage_metadata(response)
return {
"response": response,
"usage": usage,
"parsed": self.parse_response(response),
}
generate_from_uri(prompt, *, files=None, bucket_path=None, mime_type=None, response_schema=None, response_mime_type='application/json', temperature=0.0, max_output_tokens=8192, presence_penalty=0.0, frequency_penalty=0.0, top_p=1.0, seed=20180507, safety_threshold=HarmBlockThreshold.BLOCK_NONE, thinking_budget=0)
Genera contenido a partir de un prompt y archivos especificados por URI (GCS).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt
|
str
|
El prompt de texto para el modelo. |
required |
files
|
list[dict[str, str]] | None
|
Lista de archivos en GCS. Defaults to None. |
None
|
bucket_path
|
str | None
|
Ruta a un único archivo en GCS. Defaults to None. |
None
|
mime_type
|
str | None
|
Tipo MIME para |
None
|
response_schema
|
dict | None
|
Esquema de respuesta esperado. Defaults to None. |
None
|
response_mime_type
|
str
|
Tipo MIME de la respuesta. Defaults to "application/json". |
'application/json'
|
temperature
|
float
|
Controla la aleatoriedad. Defaults to 0.0. |
0.0
|
max_output_tokens
|
int
|
Límite de tokens de salida. Defaults to 8192. |
8192
|
presence_penalty
|
float
|
Penalización por presencia. Defaults to 0.0. |
0.0
|
frequency_penalty
|
float
|
Penalización por frecuencia. Defaults to 0.0. |
0.0
|
top_p
|
float
|
Muestreo nucleus. Defaults to 1.0. |
1.0
|
seed
|
int
|
Semilla para resultados deterministas. Defaults to 20180507. |
20180507
|
safety_threshold
|
HarmBlockThreshold
|
Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE. |
BLOCK_NONE
|
thinking_budget
|
int
|
Presupuesto para 'thinking'. Defaults to 0. |
0
|
Returns:
| Type | Description |
|---|---|
dict[str, object]
|
dict[str, object]: Un diccionario con la respuesta completa de la API, los metadatos de uso y la respuesta parseada. |
Raises:
| Type | Description |
|---|---|
VertexAIExceptionError
|
Si ocurre un error en la llamada a la API de Genai. |
Source code in app/services/agent/genai_client.py
def generate_from_uri(
self,
prompt: str,
*,
files: list[dict[str, str]] | None = None,
bucket_path: str | None = None,
mime_type: str | None = None,
response_schema: dict | None = None,
response_mime_type: str = "application/json",
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> dict[str, object]:
"""
Genera contenido a partir de un prompt y archivos especificados por URI (GCS).
Args:
prompt (str): El prompt de texto para el modelo.
files (list[dict[str, str]] | None, optional): Lista de archivos en GCS. Defaults to None.
bucket_path (str | None, optional): Ruta a un único archivo en GCS. Defaults to None.
mime_type (str | None, optional): Tipo MIME para `bucket_path`. Defaults to None.
response_schema (dict | None, optional): Esquema de respuesta esperado. Defaults to None.
response_mime_type (str, optional): Tipo MIME de la respuesta. Defaults to "application/json".
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
dict[str, object]: Un diccionario con la respuesta completa de la API,
los metadatos de uso y la respuesta parseada.
Raises:
VertexAIExceptionError: Si ocurre un error en la llamada a la API de Genai.
"""
files = self._normalize_files(files, bucket_path, mime_type)
contents = [
Part.from_uri(
file_uri=f["file_uri"],
mime_type=f["mime_type"],
)
for f in files
] + [
Part.from_text(text=prompt),
]
config = self._build_generation_config(
response_schema=response_schema,
response_mime_type=response_mime_type,
temperature=temperature,
max_output_tokens=max_output_tokens,
presence_penalty=presence_penalty,
frequency_penalty=frequency_penalty,
top_p=top_p,
seed=seed,
safety_threshold=safety_threshold,
thinking_budget=thinking_budget,
)
try:
response = self.client.models.generate_content(
model=self.model_name,
contents=contents,
config=config,
)
except APIError as e:
raise VertexAIExceptionError(
original_exception=e,
) from e
usage = self.extract_usage_metadata(response)
return {
"response": response,
"usage": usage,
"parsed": self.parse_response(response),
}
get_chat_session(*, temperature=0.0, max_output_tokens=8192, presence_penalty=0.0, frequency_penalty=0.0, top_p=1.0, seed=20180507, safety_threshold=HarmBlockThreshold.BLOCK_NONE, thinking_budget=0)
Crea y retorna una nueva sesión de chat con la configuración especificada.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
temperature
|
float
|
Controla la aleatoriedad. Defaults to 0.0. |
0.0
|
max_output_tokens
|
int
|
Límite de tokens de salida. Defaults to 8192. |
8192
|
presence_penalty
|
float
|
Penalización por presencia. Defaults to 0.0. |
0.0
|
frequency_penalty
|
float
|
Penalización por frecuencia. Defaults to 0.0. |
0.0
|
top_p
|
float
|
Muestreo nucleus. Defaults to 1.0. |
1.0
|
seed
|
int
|
Semilla para resultados deterministas. Defaults to 20180507. |
20180507
|
safety_threshold
|
HarmBlockThreshold
|
Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE. |
BLOCK_NONE
|
thinking_budget
|
int
|
Presupuesto para 'thinking'. Defaults to 0. |
0
|
Returns:
| Name | Type | Description |
|---|---|---|
object |
object
|
Un objeto de sesión de chat del SDK de Genai. |
Source code in app/services/agent/genai_client.py
def get_chat_session(
self,
*,
temperature: float = 0.0,
max_output_tokens: int = 8192,
presence_penalty: float = 0.0,
frequency_penalty: float = 0.0,
top_p: float = 1.0,
seed: int = 20180507,
safety_threshold: HarmBlockThreshold = HarmBlockThreshold.BLOCK_NONE,
thinking_budget: int = 0,
) -> object:
"""
Crea y retorna una nueva sesión de chat con la configuración especificada.
Args:
temperature (float, optional): Controla la aleatoriedad. Defaults to 0.0.
max_output_tokens (int, optional): Límite de tokens de salida. Defaults to 8192.
presence_penalty (float, optional): Penalización por presencia. Defaults to 0.0.
frequency_penalty (float, optional): Penalización por frecuencia. Defaults to 0.0.
top_p (float, optional): Muestreo nucleus. Defaults to 1.0.
seed (int, optional): Semilla para resultados deterministas. Defaults to 20180507.
safety_threshold (HarmBlockThreshold, optional): Nivel de seguridad. Defaults to HarmBlockThreshold.BLOCK_NONE.
thinking_budget (int, optional): Presupuesto para 'thinking'. Defaults to 0.
Returns:
object: Un objeto de sesión de chat del SDK de Genai.
"""
config = self._build_generation_config(
temperature=temperature,
max_output_tokens=max_output_tokens,
presence_penalty=presence_penalty,
frequency_penalty=frequency_penalty,
top_p=top_p,
seed=seed,
safety_threshold=safety_threshold,
thinking_budget=thinking_budget,
)
return self.client.chats.create(
model=self.model_name,
config=config,
)
parse_response(response)
Parsea la respuesta de la API para separar los 'thoughts' de las 'answers'.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response
|
object
|
El objeto de respuesta de la API de Genai. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict
|
Un diccionario con dos claves, 'thoughts' y 'answers', cada una conteniendo una lista de las partes correspondientes de la respuesta. |
Source code in app/services/agent/genai_client.py
def parse_response(self, response: Any) -> dict:
"""
Parsea la respuesta de la API para separar los 'thoughts' de las 'answers'.
Args:
response (object): El objeto de respuesta de la API de Genai.
Returns:
dict: Un diccionario con dos claves, 'thoughts' y 'answers', cada una
conteniendo una lista de las partes correspondientes de la respuesta.
"""
parsed: dict[str, list] = {
"thoughts": [],
"answers": [],
}
if not response.candidates or not response.candidates[0].content.parts:
return parsed
for part in response.candidates[0].content.parts:
if part.thought:
parsed["thoughts"].append(part.text)
else:
parsed["answers"].append(part.text)
return parsedMapper
Este módulo contiene funciones de utilidad para mapear y transformar datos entre los diferentes formatos y sistemas (Primus, LLM, CRM).
app.services.mapper.mapper
app.services.mapper.mapper
Classes
Functions
build_account_obj(primus_result, account_type)
Construye un objeto Accounts para un shipper o consignee.
A partir de un resultado de Primus, esta función crea un objeto Accounts
que representa a un shipper (remitente) o a un consignee (destinatario),
truncando los campos de nombre y calle a 40 caracteres si es necesario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_result
|
dict
|
Un diccionario que representa un único booking de Primus. |
required |
account_type
|
str
|
El tipo de cuenta a construir, debe ser 'shipper' o 'consignee'. |
required |
Returns:
| Type | Description |
|---|---|
Optional[Accounts]
|
Optional[Accounts]: Un objeto |
Source code in app/services/mapper/mapper.py
def build_account_obj(primus_result: dict, account_type: str) -> Optional[Accounts]:
"""
Construye un objeto Accounts para un shipper o consignee.
A partir de un resultado de Primus, esta función crea un objeto `Accounts`
que representa a un shipper (remitente) o a un consignee (destinatario),
truncando los campos de nombre y calle a 40 caracteres si es necesario.
Args:
primus_result (dict): Un diccionario que representa un único booking de Primus.
account_type (str): El tipo de cuenta a construir, debe ser 'shipper' o 'consignee'.
Returns:
Optional[Accounts]: Un objeto `Accounts` con la información de la cuenta.
Retorna `None` si el `account_type` no se encuentra en `primus_result`.
"""
if account_type not in primus_result:
return None
account_data = primus_result[account_type]
country_code_2letter = ""
country_code_3letter = account_data.get("country", "")
account_name = account_data.get("name", "")
street = account_data.get("address1", "")
if country_code_3letter == "USA":
country_code_2letter = "US"
elif country_code_3letter == "CAN":
country_code_2letter = "CA"
elif country_code_3letter == "MEX":
country_code_2letter = "MX"
if len(account_name) > 40:
account_name = account_name[:40] # Truncar a 40 caracteres si es necesario
if len(street) > 40:
street = street[:40]
return Accounts(
name=account_name,
street=street,
city=account_data.get("city", ""),
postal_code=account_data.get("zipCode", ""),
country_code=country_code_2letter,
state_code=account_data.get("state", ""),
phone=account_data.get("contactPhone", ""),
)
build_carrier_quote_object(primus_result, llm_entry, load_id, carrier_service_id, carrier_id)
Construye carrier_quote con datos de Primus, LLM y CRM.
Source code in app/services/mapper/mapper.py
def build_carrier_quote_object(
primus_result: dict,
llm_entry: dict,
load_id: str,
carrier_service_id: Optional[str],
carrier_id: Optional[str],
) -> CarrierQuoteV3:
"""
Construye carrier_quote con datos de Primus, LLM y CRM.
"""
vendor = primus_result.get("vendor", {})
bill_to_third_party = (
llm_entry.get("parsed", {}).get("load", {}).get("bill_to_third_party")
)
if not bill_to_third_party:
contract = ""
elif "BASINGEN" in str(bill_to_third_party).upper():
contract = "BA"
else:
contract = "GLT"
raw_cost = vendor.get("cost")
try:
line_haul = float(raw_cost) if raw_cost is not None else 0.0
except (TypeError, ValueError):
line_haul = 0.0
raw_transit_days = vendor.get("transitDays")
try:
transit_days = int(raw_transit_days) if raw_transit_days is not None else 0
except (TypeError, ValueError):
transit_days = 0
return CarrierQuoteV3(
carrier_service=carrier_service_id,
quote_contract_id=f"{vendor.get('SCAC')}_{contract}:{vendor.get('quoteNumber')}",
service_class=str(vendor.get("serviceLevel") or "").strip(),
line_haul=line_haul,
net_line_haul=line_haul,
carrier_total=line_haul,
load_id=load_id,
vendor_id=carrier_id,
estimated_transit_time_days=transit_days,
account_group=contract,
fuel_surcharge=0.0,
accessorial_charges=0.0,
accepted=True,
currency="USD",
)
build_customer_quote_object(primus_result, load_id)
Construye un objeto CustomerQuote a partir de los datos de Primus.
Crea un objeto CustomerQuote que representa la cotización para el cliente,
extrayendo la información de contabilidad y la fecha de la cotización.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_result
|
dict
|
Un diccionario que representa un único booking de Primus. |
required |
load_id
|
str
|
El ID de la carga (load) a la que se asocia la cotización. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
CustomerQuote |
CustomerQuoteV3
|
Un objeto |
Source code in app/services/mapper/mapper.py
def build_customer_quote_object(primus_result: dict, load_id: str) -> CustomerQuoteV3:
"""
Construye un objeto CustomerQuote a partir de los datos de Primus.
Crea un objeto `CustomerQuote` que representa la cotización para el cliente,
extrayendo la información de contabilidad y la fecha de la cotización.
Args:
primus_result (dict): Un diccionario que representa un único booking de Primus.
load_id (str): El ID de la carga (load) a la que se asocia la cotización.
Returns:
CustomerQuote: Un objeto `CustomerQuote` con los detalles de la cotización
del cliente.
"""
accounting = primus_result.get("accountingInformation", {})
tracking = primus_result.get("trackingInformation", {})
quote_date = tracking.get("bookDate")
try:
dt = datetime.strptime(quote_date, "%Y-%m-%d %H:%M:%S")
except (TypeError, ValueError):
logger.warning(
f"bookDate inválido o ausente para BOL: {quote_date}, usando fecha actual"
)
dt = datetime.now()
return CustomerQuoteV3(
status=PanelAdminStatus.ACCEPTED,
currency_iso_code="USD",
load_id=load_id,
net_line_haul=_to_float(accounting.get("customerQuoteAmount"), 0.0),
pricing="contract",
quote_date=str(dt.date()),
)
build_final_loads_v3(primus_data, llm_results, customer_mode_ids, accessorials_results, linear_feet_results, updates_data)
Construye una lista de objetos FinalLoadV3 a partir de datos procesados.
Versión V3 que incluye objetos Accounts y ContactInput anidados en lugar de IDs, y cotizaciones requeridas para la creación en un solo paso.
Source code in app/services/mapper/mapper.py
def build_final_loads_v3(
primus_data: list[dict],
llm_results: dict[str, dict],
customer_mode_ids: dict[str, dict[str, str | None]],
accessorials_results: dict[str, Any],
linear_feet_results: dict[str, float],
updates_data: dict[str, dict],
) -> List[FinalLoadV3]:
"""
Construye una lista de objetos FinalLoadV3 a partir de datos procesados.
Versión V3 que incluye objetos Accounts y ContactInput anidados en lugar de IDs,
y cotizaciones requeridas para la creación en un solo paso.
"""
final_loads = []
for primus_entry in primus_data:
bol_number = str(primus_entry.get("BOLNumber"))
llm_entry = llm_results.get(bol_number, {})
parsed_llm = llm_entry.get("parsed", {})
updates = updates_data.get(bol_number, {})
# Extraer IDs base
customer_id = customer_mode_ids.get(bol_number, {}).get("customer_id")
# Mode ID puede venir de updates (override) o de customer_mode_ids
mode_id = updates.get("mode_id") or customer_mode_ids.get(bol_number, {}).get(
"mode_id"
)
if not customer_id:
logger.warning(
f"No se encontró customer_id para BOL {bol_number}, omitiendo"
)
customer_id = None
if not mode_id:
logger.warning(f"No se encontró mode_id para BOL {bol_number}, omitiendo")
mode_id = None
# Construir Accounts (Shipper/Consignee)
shipper_account = build_account_obj(primus_entry, "shipper")
consignee_account = build_account_obj(primus_entry, "consignee")
# Construir Stops
stops = []
tracking_info = primus_entry.get("trackingInformation", {})
vendor_data = primus_entry.get("vendor") or {}
# Stop 1: Pickup
stops.append(
StopV3(
name="",
number=1,
location_id=None,
contact_info_id=None,
shipping_receiving_hours=primus_entry.get("shipper", {}).get(
"shippingHours"
),
expected_date=(
tracking_info.get("pickupDateEstimated")
or primus_entry.get("pickupDate")
),
references=vendor_data.get("pickupNumber", ""),
instructions=primus_entry.get("BOLInstructions"),
pickup_delivery_number=primus_entry.get("shipper", {}).get(
"referenceNumber"
),
is_dropoff=False,
is_pickup=True,
city=primus_entry.get("shipper", {}).get("city"),
state_province=primus_entry.get("shipper", {}).get("state"),
postal_code=primus_entry.get("shipper", {}).get("zipCode"),
country=primus_entry.get("shipper", {}).get("country"),
)
)
# Stop 2: Delivery
stops.append(
StopV3(
name="",
number=2,
location_id=None,
contact_info_id=None,
shipping_receiving_hours=primus_entry.get("consignee", {}).get(
"shippingHours"
),
expected_date=(
tracking_info.get("deliveryDateEstimated")
or primus_entry.get("deliveryDate")
),
references=vendor_data.get("deliveryNumber", ""),
instructions=primus_entry.get("BOLInstructions"),
pickup_delivery_number=primus_entry.get("consignee", {}).get(
"referenceNumber"
),
is_dropoff=True,
is_pickup=False,
city=primus_entry.get("consignee", {}).get("city"),
state_province=primus_entry.get("consignee", {}).get("state"),
postal_code=primus_entry.get("consignee", {}).get("zipCode"),
country=primus_entry.get("consignee", {}).get("country"),
)
)
# Line Items: Primus (freightInfo) es la fuente principal de datos de carga.
# El LLM (parsed_llm["line_items"]) enriquece con hazmat details, packaging y dimension_units.
line_items = []
line_items_llm = parsed_llm.get("line_items") or []
dimension_units = "IN"
if line_items_llm and isinstance(line_items_llm[0], dict):
dimension_units = line_items_llm[0].get("dimension_units") or "IN"
w_units = None # Let CRM inherit weight_units from the load
freight_rows = primus_entry.get("freightInfo", []) or []
# Linear Feet
linear_feet = linear_feet_results.get(bol_number, 0.0)
for idx, fi in enumerate(freight_rows):
llm_li = get_llm_line_item(parsed_llm, idx)
dim_u = llm_li.get("dimension_units") or dimension_units
qty = int(fi.get("qty") or 1)
hu_display = (
MAPPER_HU_DATA.get(fi.get("dimType")) or fi.get("dimType") or "Pallets"
)
length = _to_float(fi.get("length") or 0)
width = _to_float(fi.get("width") or 0)
height = _to_float(fi.get("height") or 0)
total_weight = _to_float(fi.get("weight") or 0) * qty
un_number = fi.get("UN") or llm_li.get("hazmat_number") or ""
hazmat_number_type = str(llm_li.get("hazmat_number_type") or "")
is_hazmat = bool(fi.get("hazmat", False))
if is_hazmat and un_number:
llm_description = str(fi.get("commodity") or "").strip() or "Shipping"
hazmat_description = str(llm_li.get("description") or "")
line_description = f"{llm_description} {hazmat_description}"
else:
line_description = str(fi.get("commodity") or "").strip() or "Shipping"
line_items.append(
LineItemV3(
name=f"Line Item {idx + 1}",
description=line_description,
dimension_units=dim_u,
handling_unit_count=qty,
handling_units=hu_display,
height=height,
length=length,
linear_feet=linear_feet,
nmfc_class=str(fi.get("class") or ""),
weight_units=w_units,
weight=total_weight,
width=width,
hazardous_materials=is_hazmat,
hazmat_class_division=str(
llm_li.get("hazmat_class_division") or ""
),
hazmat_number_type=hazmat_number_type,
hazmat_number=un_number,
hazmat_packaging_group=str(
llm_li.get("hazmat_packaging_group") or ""
),
nmfc_number=str(fi.get("nmfc") or "") or None,
packaging_unit_count=int(llm_li.get("packaging_unit_count") or 1),
packaging_units=str(llm_li.get("packaging_units") or ""),
)
)
# Accessorials — map_accesorials_to_glt retorna mezcla de dicts (del JSON) y Accessorial (stackable)
raw_accessorials = accessorials_results.get(bol_number, {})
if isinstance(raw_accessorials, dict):
raw_accessorials = raw_accessorials.get("accessorials", [])
accessorials: list[Accessorial] = []
for a in raw_accessorials or []:
if isinstance(a, Accessorial):
accessorials.append(a)
elif isinstance(a, dict):
acc_id = a.get("accessorial_id")
name = a.get("name")
if acc_id is not None or name is not None:
accessorials.append(
Accessorial(accessorial_id=str(acc_id or ""), name=name)
)
# Carrier ID must come from CRM carrier service mapping (updates_data).
# Send it only when it looks like a valid CRM/Salesforce ID.
raw_carrier_id = updates.get("carrier_id")
carrier_id = None
if raw_carrier_id is not None:
candidate_carrier_id = str(raw_carrier_id).strip()
if len(candidate_carrier_id) in (15, 18) and candidate_carrier_id.isalnum():
carrier_id = candidate_carrier_id
elif candidate_carrier_id:
logger.warning(
"Carrier ID inválido para BOL %s: %s. Se enviará como null.",
bol_number,
candidate_carrier_id,
)
else:
logger.warning(
"Carrier ID vacío para BOL %s. El load será rechazado por Salesforce si rtms__Vendor__c es requerido.",
bol_number,
)
raw_carrier_service_id = updates.get("carrier_service")
carrier_service_id = None
if raw_carrier_service_id is not None:
candidate_carrier_service_id = str(raw_carrier_service_id).strip()
if (
len(candidate_carrier_service_id) in (15, 18)
and candidate_carrier_service_id.isalnum()
):
carrier_service_id = candidate_carrier_service_id
elif candidate_carrier_service_id:
logger.warning(
"Carrier service ID inválido para BOL %s: %s. Se enviará como null.",
bol_number,
candidate_carrier_service_id,
)
# Construir FinalLoadV3
final_load = FinalLoadV3(
load=LoadV3(
total_weight=float(primus_entry.get("totalWeight") or 0.0),
mode_id=mode_id or "",
carrier_id=carrier_id,
carrier_bill_id=updates.get("carrier_bill_id") or "",
document_options_id=updates.get("document_options_id") or "",
customer_id=customer_id,
bol_number=bol_number,
weight_units=parsed_llm.get("load", {}).get("weight_units", "LB"),
linear_feet=linear_feet,
pro_number=updates.get("pro_number"),
po_number=updates.get("po_number") or None,
billing_reference=bol_number,
external_tms_integration=updates.get("external_tms_integration__c"),
),
stops=stops,
line_items=line_items,
accessorials=accessorials,
shipper_account=shipper_account,
shipper_contact=Contact(
first_name=(
(primus_entry.get("shipper", {}).get("contact") or "").split(" ")[0]
)[:40]
or None,
last_name=(
" ".join(
(primus_entry.get("shipper", {}).get("contact") or "").split(
" "
)[1:]
)
)[:40]
or "Shipping",
phone=primus_entry.get("shipper", {}).get("contactPhone") or None,
email=primus_entry.get("shipper", {}).get("email") or None,
),
consignee_account=consignee_account,
consignee_contact=Contact(
first_name=(
(primus_entry.get("consignee", {}).get("contact") or "").split(" ")[
0
]
)[:40]
or None,
last_name=(
" ".join(
(primus_entry.get("consignee", {}).get("contact") or "").split(
" "
)[1:]
)
)[:40]
or "Consignee",
phone=primus_entry.get("consignee", {}).get("contactPhone") or None,
email=primus_entry.get("consignee", {}).get("email") or None,
),
carrier_quote=build_carrier_quote_object(
primus_result=primus_entry,
llm_entry=llm_entry,
load_id=bol_number,
carrier_service_id=carrier_service_id,
carrier_id=carrier_id,
)
if carrier_service_id
else None,
customer_quote=build_customer_quote_object(
primus_result=primus_entry,
load_id=bol_number,
),
hazmat_contact=Contact(
first_name="Hazmat",
last_name=primus_entry.get("emergencyContact"),
phone=primus_entry.get("emergencyPhone") or None,
email="",
)
if any(
fi.get("hazmat", False) for fi in primus_entry.get("freightInfo", [])
)
else None,
)
final_loads.append(final_load)
return final_loads
build_linear_feet_obj(primus_result, llm_results)
Construye un objeto LinearFeet a partir de los datos de Primus y del LLM.
Esta función extrae la información de los items de un booking de Primus y los
resultados del LLM para determinar si la carga es apilable (stackable) y
construir un objeto LinearFeet con los detalles de la carga.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_result
|
dict
|
Un diccionario que representa un único booking de Primus. |
required |
llm_results
|
dict[str, dict]
|
Un diccionario con los resultados del procesamiento del LLM, usando el BOLNumber como clave. |
required |
Returns:
| Type | Description |
|---|---|
Optional[LinearFeet]
|
Optional[LinearFeet]: Un objeto |
Source code in app/services/mapper/mapper.py
def build_linear_feet_obj(
primus_result: dict, llm_results: dict[str, dict]
) -> Optional[LinearFeet]:
"""
Construye un objeto LinearFeet a partir de los datos de Primus y del LLM.
Esta función extrae la información de los items de un booking de Primus y los
resultados del LLM para determinar si la carga es apilable (stackable) y
construir un objeto `LinearFeet` con los detalles de la carga.
Args:
primus_result (dict): Un diccionario que representa un único booking de Primus.
llm_results (dict[str, dict]): Un diccionario con los resultados del
procesamiento del LLM, usando el BOLNumber como clave.
Returns:
Optional[LinearFeet]: Un objeto `LinearFeet` si se encuentra una coincidencia
de BOL en los resultados del LLM, de lo contrario, retorna `None`.
"""
bol_number = str(primus_result["BOLNumber"])
# Buscar el matching en LLM
llm_entry = llm_results.get(bol_number, {})
if not llm_entry:
return None # no hay matching LLM para este BOL
# Determinar stackable:
# puede ser a nivel global (si todos los ítems son stackables)
# Ejemplo: basta que al menos uno sea stackable para marcar True
llm_line_items = llm_entry.get("parsed", {}).get("line_items", [])
stackable = any(li.get("stackable") for li in llm_line_items)
# Mapear freightInfo de Primus a LineItembyLinearFeet
freight_info = primus_result.get("freightInfo", [])
line_items = [
LineItembyLinearFeet(
lenght=item["length"],
# length=item["length"] - Cambio despues de fixear megatron
width=item["width"],
height=item["height"],
weight=item["weight"],
hu_count=item["qty"],
)
for item in freight_info
]
return LinearFeet(line_items=line_items, stackable=stackable)
get_llm_line_item(parsed_llm, idx)
Obtiene de forma segura un 'line item' de los resultados del LLM.
Intenta acceder a un 'line item' específico por su índice dentro de la lista de 'line_items' en los resultados procesados del LLM.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parsed_llm
|
dict
|
El diccionario de resultados procesados del LLM. |
required |
idx
|
int
|
El índice del 'line item' a obtener. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
El 'line item' encontrado como un diccionario. Retorna un diccionario vacío si el índice está fuera de rango o si ocurre un error de tipo. |
Source code in app/services/mapper/mapper.py
def get_llm_line_item(parsed_llm, idx):
"""
Obtiene de forma segura un 'line item' de los resultados del LLM.
Intenta acceder a un 'line item' específico por su índice dentro de la lista
de 'line_items' en los resultados procesados del LLM.
Args:
parsed_llm (dict): El diccionario de resultados procesados del LLM.
idx (int): El índice del 'line item' a obtener.
Returns:
dict: El 'line item' encontrado como un diccionario. Retorna un
diccionario vacío si el índice está fuera de rango o si ocurre
un error de tipo.
"""
try:
return parsed_llm.get("line_items", [])[idx] or {}
except (IndexError, TypeError):
return {}
load_mapper(jsonfilename)
Carga un archivo de configuración JSON desde el directorio actual.
Esta función construye la ruta a un archivo JSON basándose en el nombre de archivo proporcionado y lo carga en un diccionario de Python.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
jsonfilename
|
str
|
El nombre del archivo JSON a cargar. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict
|
Un diccionario con el contenido del archivo JSON. |
Source code in app/services/mapper/mapper.py
def load_mapper(jsonfilename: str) -> dict:
"""
Carga un archivo de configuración JSON desde el directorio actual.
Esta función construye la ruta a un archivo JSON basándose en el nombre de archivo
proporcionado y lo carga en un diccionario de Python.
Args:
jsonfilename (str): El nombre del archivo JSON a cargar.
Returns:
dict: Un diccionario con el contenido del archivo JSON.
"""
base_dir = os.path.dirname(__file__) # carpeta actual: app/services/mapper
filepath = os.path.join(
base_dir,
jsonfilename,
)
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
map_accesorials_to_glt(all_primus_data, llm_results, mapper_data)
Mapea los 'accessorials' de Primus a los de GLT y los consolida.
Esta función procesa una lista de bookings de Primus, mapea sus 'accessorials' utilizando un diccionario de mapeo, agrega 'accessorials' adicionales basados en los resultados del LLM (ej. 'Stackable'), y elimina duplicados.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
all_primus_data
|
list[dict]
|
Lista de bookings de Primus. |
required |
llm_results
|
dict[str, dict]
|
Resultados del procesamiento del LLM, usando el BOLNumber como clave. |
required |
mapper_data
|
dict
|
Diccionario para mapear los 'accessorials' de Primus a GLT. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, dict]
|
dict[str, dict]: Un diccionario donde cada clave es un BOLNumber y el valor es un diccionario que contiene una lista de 'accessorials' únicos para ese BOL. |
Source code in app/services/mapper/mapper.py
def map_accesorials_to_glt(
all_primus_data: list[dict], llm_results: dict[str, dict], mapper_data: dict
) -> dict[str, dict]:
"""
Mapea los 'accessorials' de Primus a los de GLT y los consolida.
Esta función procesa una lista de bookings de Primus, mapea sus 'accessorials'
utilizando un diccionario de mapeo, agrega 'accessorials' adicionales basados
en los resultados del LLM (ej. 'Stackable'), y elimina duplicados.
Args:
all_primus_data (list[dict]): Lista de bookings de Primus.
llm_results (dict[str, dict]): Resultados del procesamiento del LLM,
usando el BOLNumber como clave.
mapper_data (dict): Diccionario para mapear los 'accessorials' de Primus a GLT.
Returns:
dict[str, dict]: Un diccionario donde cada clave es un BOLNumber y el valor
es un diccionario que contiene una lista de 'accessorials' únicos para ese BOL.
"""
results = {}
for booking in all_primus_data:
bol_number = str(booking["BOLNumber"])
parsed_llm = llm_results.get(bol_number, {}).get("parsed", {})
accessorials_by_booking = booking["accessorials"]
accessorials = []
for accessorial in accessorials_by_booking: # list[str]
mapped = mapper_data.get(accessorial)
if not mapped:
logger.warning(
f"BOL {bol_number}: accessorial code '{accessorial}' "
f"not in primus_glt_mapping.json"
)
continue
accessorials.extend(mapped) # list[dict]
for item in parsed_llm.get("line_items", []):
if item["stackable"]:
accessorials.append(
Accessorial(
accessorial_id="a01Rc000002Fw8z",
name="Stackable",
)
)
else:
accessorials.append(
Accessorial(
accessorial_id="a013s00000PKXDc",
name="Non Stackable - Fragile",
)
)
unique_accessorials = list(
{
frozenset(
(d.model_dump() if hasattr(d, "model_dump") else d).items()
): d
for d in accessorials
}.values()
)
results[bol_number] = {"accessorials": unique_accessorials}
return results
map_units(primus_data, llm_results)
Convierte las unidades de peso y dimensión de los datos de Primus.
Esta función itera sobre los datos de Primus y, basándose en la información extraída por el LLM, convierte el peso de KG a LBS y las dimensiones de CM a IN cuando es necesario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_data
|
list[dict]
|
Una lista de diccionarios, donde cada diccionario representa un booking de Primus. |
required |
llm_results
|
dict[str, dict]
|
Un diccionario con los resultados del procesamiento del LLM, usando el BOLNumber como clave. |
required |
Returns:
| Type | Description |
|---|---|
|
list[dict]: La lista de bookings de Primus con las unidades de peso y dimensión convertidas. |
Source code in app/services/mapper/mapper.py
def map_units(primus_data: list[dict], llm_results: dict[str, dict]):
"""
Convierte las unidades de peso y dimensión de los datos de Primus.
Esta función itera sobre los datos de Primus y, basándose en la información
extraída por el LLM, convierte el peso de KG a LBS y las dimensiones de CM a IN
cuando es necesario.
Args:
primus_data (list[dict]): Una lista de diccionarios, donde cada diccionario
representa un booking de Primus.
llm_results (dict[str, dict]): Un diccionario con los resultados del
procesamiento del LLM, usando el BOLNumber como clave.
Returns:
list[dict]: La lista de bookings de Primus con las unidades de peso y
dimensión convertidas.
"""
for booking in primus_data:
bol_number = str(booking["BOLNumber"])
parsed_llm = llm_results.get(bol_number, {}).get("parsed", {})
load_weight_units = parsed_llm.get("load", {}).get("weight_units", "LB")
line_item_dimension_units = any(
li.get("dimension_units") == "CM" for li in parsed_llm.get("line_items", [])
)
if load_weight_units == "KG":
for fi in booking.get("freightInfo", []):
weight_in_lb = float(fi.get("weight", 0)) * 2.205 # Convertir KG a LBS
fi["weight"] = math.ceil(weight_in_lb)
total_weight_in_lb = (
float(booking.get("totalWeight", 0)) * 2.205
) # Convertir KG a LBS
booking["totalWeight"] = math.ceil(total_weight_in_lb)
if line_item_dimension_units:
for fi in booking.get("freightInfo", []):
length_in = float(fi.get("length", 0)) * 0.393701 # Convertir CM a IN
width_in = float(fi.get("width", 0)) * 0.393701
height_in = float(fi.get("height", 0)) * 0.393701
fi["length"] = math.ceil(length_in)
fi["width"] = math.ceil(width_in)
fi["height"] = math.ceil(height_in)
return primus_dataCRM Client
Cliente para interactuar con la API del CRM (Salesforce). Se encarga de crear y actualizar cuentas, contactos, cargas (loads) y cotizaciones.
app.services.crm_client
app.services.crm_client
Classes
CRMClient
Cliente para interactuar con la API de CRM.
Esta clase maneja la autenticación y las solicitudes a la API de CRM para crear y actualizar cuentas, contactos, cargas y cotizaciones.
Source code in app/services/crm_client.py
class CRMClient:
"""
Cliente para interactuar con la API de CRM.
Esta clase maneja la autenticación y las solicitudes a la API de CRM
para crear y actualizar cuentas, contactos, cargas y cotizaciones.
"""
def __init__(self):
"""
Inicializa el cliente de CRM.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.crm_base_url
self.user = settings.crm_user
self.password = settings.crm_password
self.timeout = settings.http_client_timeout_seconds
self.semaphore = asyncio.Semaphore(settings.concurrency)
self.token = None
self._token_expires_at: float = 0.0
self._login_lock = asyncio.Lock()
self._TOKEN_BUFFER_SECONDS = 300 # Refresh 5 min before actual expiry
def _token_is_valid(self) -> bool:
return bool(self.token) and time.monotonic() < self._token_expires_at
async def login(self):
"""
Autentica con el CRM y guarda el token de acceso.
Usa un Lock para evitar logins concurrentes (race condition cuando
múltiples BOLs arrancan en paralelo con token=None). El patrón
double-check dentro del lock evita logins redundantes.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor, pero almacena el token de acceso en self.token.
Levanta una excepción si la autenticación falla.
"""
async with self._login_lock:
# Double-check: another coroutine may have already refreshed the token
if self._token_is_valid():
return
async with httpx.AsyncClient(timeout=self.timeout) as client:
resp = await client.post(
f"{self.base_url}/auth/access-token-v2",
data={
"grant_type": "password",
"username": self.user,
"password": self.password,
"scope": "",
"client_id": "",
"client_secret": "",
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
resp.raise_for_status()
data = resp.json()
self.token = data.get("access_token")
expires_in = data.get("expires_in", 3600)
self._token_expires_at = (
time.monotonic() + expires_in - self._TOKEN_BUFFER_SECONDS
)
logger.info("CRM login OK, nuevo token obtenido")
def _headers(self) -> dict:
"""
Genera los encabezados de autorización para las solicitudes al CRM.
Args:
No recibe argumentos.
Returns:
dict: Un diccionario con los encabezados de autorización y tipo de contenido.
"""
return {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json",
}
async def _request_with_reauth(
self, method: str, path: str, **kwargs
) -> httpx.Response:
"""
Envía una solicitud HTTP al CRM, con reintento automático si el token está expirado.
Args:
method (str): El método HTTP a utilizar (GET, POST, PATCH, etc.).
path (str): La ruta del endpoint del CRM.
**kwargs: Argumentos adicionales para la solicitud httpx.request.
Returns:
httpx.Response: La respuesta de la solicitud HTTP.
Levanta una excepción si la solicitud falla después de los reintentos.
"""
if not self._token_is_valid():
await self.login()
url = f"{self.base_url}/{path.lstrip('/')}"
async with httpx.AsyncClient(timeout=self.timeout) as client:
resp = await client.request(method, url, headers=self._headers(), **kwargs)
if resp.status_code in (401, 403):
logger.warning("Token expirado o inválido, reintentando login...")
await self.login()
resp = await client.request(
method, url, headers=self._headers(), **kwargs
)
resp.raise_for_status()
return resp
# -------------------------------
# Métodos usando _request_with_reauth
# -------------------------------
async def _post_load_v3(self, bol_number: str, final_load: FinalLoadV3):
"""
Crea una carga (load) en el CRM usando la versión V3 del endpoint.
Args:
bol_number (str): El número de BOL asociado a la carga.
final_load (FinalLoadV3): El objeto Pydantic con los datos finales de la carga (incluyendo cuentas y contactos).
Returns:
tuple: Una tupla que contiene el número de BOL y un diccionario con el ID de la carga creada.
Si falla, retorna un diccionario con un ID None.
"""
async with self.semaphore:
try:
resp = await self._request_with_reauth(
"POST",
"load/v3",
json=final_load.model_dump(exclude_none=True, mode="json"),
)
resp_json = resp.json()
load_id = resp_json.get("id")
if not load_id:
logger.warning(
f"Load v3 creado con ID nulo para BOL {bol_number}. Respuesta: {resp_json}"
)
logger.warning(f"Payload enviado para BOL {bol_number}")
return bol_number, {"id": load_id, "full_response": resp_json}
except Exception as e:
error_detail = ""
if hasattr(e, "response") and e.response is not None:
try:
error_json = e.response.json()
error_detail = error_json.get("detail", str(error_json))
except Exception:
error_detail = e.response.text
else:
error_detail = str(e)
error_detail = clean_error(error_detail)
logger.exception(
f"Error creando load v3 para BOL {bol_number}: {e} - {error_detail}"
)
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": error_detail,
}
)
return bol_number, {"id": None, "error": f"{e} - {error_detail}"}
async def create_loads_v3(self, final_loads: list[FinalLoadV3]):
"""
Crea múltiples cargas en el CRM usando V3.
Args:
final_loads (list[FinalLoadV3]): Una lista de objetos Pydantic con los datos finales de las cargas.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
con el ID de la carga creada.
"""
tasks = [self._post_load_v3(fl.load.bol_number or "", fl) for fl in final_loads]
results = await asyncio.gather(*tasks)
return {bol: load_out for bol, load_out in results}
async def _get_carrier_service_by_scacs(self, bol_number: str, scac: str):
"""
Obtiene el servicio de transportista (carrier service) del CRM por su código SCAC.
Args:
bol_number (str): El número de BOL asociado.
scac (str): El código SCAC del transportista.
Returns:
tuple: Una tupla que contiene el número de BOL y un diccionario con la información del servicio de transportista.
Si falla, retorna un diccionario vacío.
"""
async with self.semaphore:
try:
resp = await self._request_with_reauth(
"GET", "carrier-service/primus/scacs", params={"scacs": [scac]}
)
carriers = resp.json()
carrier_service: dict | None
if scac == "FDEG":
carrier_service = next(
(
item
for item in carriers
if item["name"] == "FEDEX PACKAGING"
),
{},
)
else:
carrier_service = next(
(item for item in carriers if item["name"].endswith("(LTL)")),
None,
)
if not carrier_service:
project44_carriers = [
item
for item in carriers
if item.get("api_provider") == "Project44"
]
if len(project44_carriers) == 1:
carrier_service = project44_carriers[0]
elif len(project44_carriers) > 1:
carrier_service = next(
(
item
for item in project44_carriers
if "LTL" in item["name"]
),
project44_carriers[0],
)
else:
carrier_service = {}
return bol_number, carrier_service
except Exception as e:
error_detail = ""
if hasattr(e, "response") and e.response is not None:
try:
error_json = e.response.json()
error_detail = error_json.get("detail", str(error_json))
except Exception:
error_detail = e.response.text
else:
error_detail = str(e)
logger.exception(
f"Error consultando carrier SCAC={scac}: {e} - {error_detail}"
)
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": f"Error consultando carrier SCAC={scac}: {e} - {error_detail}",
}
)
return bol_number, {}
async def carrier_service(self, primus_results: list[dict]):
"""
Obtiene los servicios de transportista para múltiples resultados de Primus.
Args:
primus_results (list[dict]): Una lista de diccionarios con los resultados de Primus.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
con la información del servicio de transportista.
"""
results = await asyncio.gather(
*[
self._get_carrier_service_by_scacs(
str(p["BOLNumber"]),
SCACS_MAPPING.get(
p.get("vendor", {}).get("SCAC"), p.get("vendor", {}).get("SCAC")
),
)
for p in primus_results
]
)
return {bol: cs for bol, cs in results}
async def _get_account_by_name(self, bol_number: str, account_name: str):
"""
Obtiene el ID de una cuenta del CRM por su nombre.
Args:
bol_number (str): El número de BOL asociado.
account_name (str): El nombre de la cuenta a buscar.
Returns:
tuple: Una tupla que contiene el número de BOL y un diccionario con el ID de la cuenta.
Si falla, retorna un diccionario con un ID None.
"""
async with self.semaphore:
try:
resp = await self._request_with_reauth(
"GET",
"account/name",
params={"account_name": account_name, "billing_account": True},
)
id = next(
(
item["id"]
for item in resp.json()
if item["name"].lower() == account_name.lower()
),
None,
)
return bol_number, {"id": id}
except Exception as e:
error_detail = ""
if hasattr(e, "response") and e.response is not None:
try:
error_json = e.response.json()
error_detail = error_json.get("detail", str(error_json))
except Exception:
error_detail = e.response.text
else:
error_detail = str(e)
logger.exception(
f"Error consultando account={account_name}: {e} - {error_detail}"
)
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": f"Error consultando account={account_name}: {e} - {error_detail}",
}
)
return bol_number, {"id": None}
async def create_carrier_bill_id(self, llm_results: dict[str, dict]):
"""
Crea los IDs de facturación del transportista (carrier bill ID) basándose en los resultados del LLM.
Args:
llm_results (dict[str, dict]): Un diccionario con los resultados del LLM, indexados por número de BOL.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
con el ID de la cuenta de facturación del transportista.
"""
results = []
for bol_number, bol_data in llm_results.items():
bill_to_third_party = (
bol_data.get("parsed", {}).get("load", {}).get("bill_to_third_party")
)
logger.info(f"BOL {bol_number}, bill_to_third_party: {bill_to_third_party}")
if bill_to_third_party:
results.append(
await self._get_account_by_name(bol_number, bill_to_third_party)
)
return {bol: out for bol, out in results}
async def _get_mode_id(self, bol_number: str, mode: str):
"""
Obtiene el ID de un modo de transporte (mode) del CRM.
Args:
bol_number (str): El número de BOL asociado.
mode (str): El nombre del modo de transporte.
Returns:
tuple: Una tupla que contiene el número de BOL y un diccionario con el ID del modo.
Si falla, retorna un diccionario con un ID vacío.
"""
async with self.semaphore:
try:
resp = await self._request_with_reauth(
"GET", "mode", params={"department": mode}
)
response = resp.json()
return bol_number, {
"mode_id": next(
(item["id"] for item in response if item["name"] == mode), ""
)
}
except Exception as e:
error_detail = ""
if hasattr(e, "response") and e.response is not None:
try:
error_json = e.response.json()
error_detail = error_json.get("detail", str(error_json))
except Exception:
error_detail = e.response.text
else:
error_detail = str(e)
logger.exception(
f"Error obteniendo mode_id para BOL {bol_number}: {e} - {error_detail}"
)
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": f"Error obteniendo mode_id para BOL {bol_number}: {e} - {error_detail}",
}
)
return bol_number, {"mode_id": ""}
async def _get_customer_id(self, bol_number: str, third_party: str):
"""
Obtiene el ID de un cliente (customer) del CRM basado en el third_party.
Args:
bol_number (str): El número de BOL asociado.
third_party (str): El identificador del tercer partido.
Returns:
tuple: Una tupla que contiene el número de BOL y un diccionario con el ID del cliente.
Si falla, retorna un diccionario con un ID vacío.
"""
async with self.semaphore:
try:
resp = await self._request_with_reauth(
"GET", f"account/tai/{third_party}"
)
response = resp.json()
return bol_number, {"customer_id": response[0].get("id")}
except Exception as e:
error_detail = ""
if hasattr(e, "response") and e.response is not None:
try:
error_json = e.response.json()
error_detail = error_json.get("detail", str(error_json))
except Exception:
error_detail = e.response.text
else:
error_detail = str(e)
logger.exception(
f"Error obteniendo customer_id para BOL {bol_number}: {e} - {error_detail}"
)
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": f"Error obteniendo customer_id para BOL {bol_number}: {e} - {error_detail}",
}
)
return bol_number, {"customer_id": ""}
async def get_customer_and_mode_id(self, primus_results: list):
"""
Obtiene los IDs de cliente y modo de transporte para cada resultado de Primus.
Args:
primus_results (list): Una lista de diccionarios con los resultados de Primus.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
que contiene "customer_id" y/o "mode_id".
"""
tasks = []
for primus_result in primus_results:
bol_number = str(primus_result["BOLNumber"])
third_party = primus_result.get("thirdParty", {}).get("id")
mode = primus_result.get("vendor", {}).get("mode")
if not third_party and not mode:
logger.warning(
f"No se encontró thirdParty ni mode en BOL {bol_number}, se omite"
)
continue
if third_party:
tasks.append(self._get_customer_id(bol_number, third_party))
if mode:
tasks.append(self._get_mode_id(bol_number, mode))
results = await asyncio.gather(*tasks)
output: dict[str, dict[str, str | None]] = {}
for bol_number, result in results:
if bol_number not in output:
output[bol_number] = {}
output[bol_number].update(result)
return outputFunctions
__init__()
Inicializa el cliente de CRM.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor. |
Source code in app/services/crm_client.py
def __init__(self):
"""
Inicializa el cliente de CRM.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.crm_base_url
self.user = settings.crm_user
self.password = settings.crm_password
self.timeout = settings.http_client_timeout_seconds
self.semaphore = asyncio.Semaphore(settings.concurrency)
self.token = None
self._token_expires_at: float = 0.0
self._login_lock = asyncio.Lock()
self._TOKEN_BUFFER_SECONDS = 300 # Refresh 5 min before actual expiry
carrier_service(primus_results)
async
Obtiene los servicios de transportista para múltiples resultados de Primus.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_results
|
list[dict]
|
Una lista de diccionarios con los resultados de Primus. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
Un diccionario donde la clave es el número de BOL y el valor es un diccionario con la información del servicio de transportista. |
Source code in app/services/crm_client.py
async def carrier_service(self, primus_results: list[dict]):
"""
Obtiene los servicios de transportista para múltiples resultados de Primus.
Args:
primus_results (list[dict]): Una lista de diccionarios con los resultados de Primus.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
con la información del servicio de transportista.
"""
results = await asyncio.gather(
*[
self._get_carrier_service_by_scacs(
str(p["BOLNumber"]),
SCACS_MAPPING.get(
p.get("vendor", {}).get("SCAC"), p.get("vendor", {}).get("SCAC")
),
)
for p in primus_results
]
)
return {bol: cs for bol, cs in results}
create_carrier_bill_id(llm_results)
async
Crea los IDs de facturación del transportista (carrier bill ID) basándose en los resultados del LLM.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
llm_results
|
dict[str, dict]
|
Un diccionario con los resultados del LLM, indexados por número de BOL. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
Un diccionario donde la clave es el número de BOL y el valor es un diccionario con el ID de la cuenta de facturación del transportista. |
Source code in app/services/crm_client.py
async def create_carrier_bill_id(self, llm_results: dict[str, dict]):
"""
Crea los IDs de facturación del transportista (carrier bill ID) basándose en los resultados del LLM.
Args:
llm_results (dict[str, dict]): Un diccionario con los resultados del LLM, indexados por número de BOL.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
con el ID de la cuenta de facturación del transportista.
"""
results = []
for bol_number, bol_data in llm_results.items():
bill_to_third_party = (
bol_data.get("parsed", {}).get("load", {}).get("bill_to_third_party")
)
logger.info(f"BOL {bol_number}, bill_to_third_party: {bill_to_third_party}")
if bill_to_third_party:
results.append(
await self._get_account_by_name(bol_number, bill_to_third_party)
)
return {bol: out for bol, out in results}
create_loads_v3(final_loads)
async
Crea múltiples cargas en el CRM usando V3.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
final_loads
|
list[FinalLoadV3]
|
Una lista de objetos Pydantic con los datos finales de las cargas. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
Un diccionario donde la clave es el número de BOL y el valor es un diccionario con el ID de la carga creada. |
Source code in app/services/crm_client.py
async def create_loads_v3(self, final_loads: list[FinalLoadV3]):
"""
Crea múltiples cargas en el CRM usando V3.
Args:
final_loads (list[FinalLoadV3]): Una lista de objetos Pydantic con los datos finales de las cargas.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
con el ID de la carga creada.
"""
tasks = [self._post_load_v3(fl.load.bol_number or "", fl) for fl in final_loads]
results = await asyncio.gather(*tasks)
return {bol: load_out for bol, load_out in results}
get_customer_and_mode_id(primus_results)
async
Obtiene los IDs de cliente y modo de transporte para cada resultado de Primus.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_results
|
list
|
Una lista de diccionarios con los resultados de Primus. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
Un diccionario donde la clave es el número de BOL y el valor es un diccionario que contiene "customer_id" y/o "mode_id". |
Source code in app/services/crm_client.py
async def get_customer_and_mode_id(self, primus_results: list):
"""
Obtiene los IDs de cliente y modo de transporte para cada resultado de Primus.
Args:
primus_results (list): Una lista de diccionarios con los resultados de Primus.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
que contiene "customer_id" y/o "mode_id".
"""
tasks = []
for primus_result in primus_results:
bol_number = str(primus_result["BOLNumber"])
third_party = primus_result.get("thirdParty", {}).get("id")
mode = primus_result.get("vendor", {}).get("mode")
if not third_party and not mode:
logger.warning(
f"No se encontró thirdParty ni mode en BOL {bol_number}, se omite"
)
continue
if third_party:
tasks.append(self._get_customer_id(bol_number, third_party))
if mode:
tasks.append(self._get_mode_id(bol_number, mode))
results = await asyncio.gather(*tasks)
output: dict[str, dict[str, str | None]] = {}
for bol_number, result in results:
if bol_number not in output:
output[bol_number] = {}
output[bol_number].update(result)
return output
login()
async
Autentica con el CRM y guarda el token de acceso.
Usa un Lock para evitar logins concurrentes (race condition cuando múltiples BOLs arrancan en paralelo con token=None). El patrón double-check dentro del lock evita logins redundantes.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor, pero almacena el token de acceso en self.token. |
|
|
Levanta una excepción si la autenticación falla. |
Source code in app/services/crm_client.py
async def login(self):
"""
Autentica con el CRM y guarda el token de acceso.
Usa un Lock para evitar logins concurrentes (race condition cuando
múltiples BOLs arrancan en paralelo con token=None). El patrón
double-check dentro del lock evita logins redundantes.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor, pero almacena el token de acceso en self.token.
Levanta una excepción si la autenticación falla.
"""
async with self._login_lock:
# Double-check: another coroutine may have already refreshed the token
if self._token_is_valid():
return
async with httpx.AsyncClient(timeout=self.timeout) as client:
resp = await client.post(
f"{self.base_url}/auth/access-token-v2",
data={
"grant_type": "password",
"username": self.user,
"password": self.password,
"scope": "",
"client_id": "",
"client_secret": "",
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
resp.raise_for_status()
data = resp.json()
self.token = data.get("access_token")
expires_in = data.get("expires_in", 3600)
self._token_expires_at = (
time.monotonic() + expires_in - self._TOKEN_BUFFER_SECONDS
)
logger.info("CRM login OK, nuevo token obtenido")Functions
evaluate_if_need_hazmat_contact(primus_result)
Evalúa si se necesita un contacto Hazmat basado en el contenido del primus_result.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_result
|
dict
|
El resultado del procesamiento de Primus que contiene la información de la carga. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True si se encuentra algún artículo peligroso (hazmat), False en caso contrario. |
Source code in app/services/crm_client.py
def evaluate_if_need_hazmat_contact(primus_result: dict) -> bool:
"""
Evalúa si se necesita un contacto Hazmat basado en el contenido del primus_result.
Args:
primus_result (dict): El resultado del procesamiento de Primus que contiene la información de la carga.
Returns:
bool: True si se encuentra algún artículo peligroso (hazmat), False en caso contrario.
"""
lineItems = primus_result.get("freightInfo", [])
for lineItem in lineItems:
hazmat = lineItem.get("hazmat", False)
if hazmat:
return True
return FalseMegatron Client
Cliente que se conecta a la API de Megatron, utilizada para realizar cálculos específicos como los pies lineales (linear feet).
app.services.megatron_client
app.services.megatron_client
Classes
MegatronClient
Cliente para interactuar con la API de Megatron.
Esta clase se encarga de calcular los pies lineales (linear feet) de los Bills of Lading (BOLs) utilizando la API de Megatron.
Source code in app/services/megatron_client.py
class MegatronClient:
"""
Cliente para interactuar con la API de Megatron.
Esta clase se encarga de calcular los pies lineales (linear feet)
de los Bills of Lading (BOLs) utilizando la API de Megatron.
"""
def __init__(self):
"""
Inicializa el cliente de Megatron.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.megatron_base_url
self.token = settings.megatron_token
async def _post_linear_feet(
self,
client: httpx.AsyncClient,
bol_number: str,
lf_in: LinearFeet,
headers: dict,
) -> tuple[str, dict]:
"""
Envía una solicitud POST a la API de Megatron para calcular los pies lineales de un solo BOL.
Args:
client (httpx.AsyncClient): Cliente HTTP asíncrono para realizar la solicitud.
bol_number (str): El número de BOL para el cual se calcularán los pies lineales.
lf_in (LinearFeet): Objeto Pydantic con los datos de entrada para el cálculo de pies lineales.
headers (dict): Diccionario de encabezados HTTP para la solicitud.
Returns:
tuple[str, dict]: Una tupla que contiene el número de BOL y un diccionario con el total de pies lineales calculados.
Si ocurre un error, retorna el número de BOL y un diccionario con 'total_linear_feet' en 0.0.
"""
try:
resp = await client.post(
f"{self.base_url}/linear-feet/calculate",
headers=headers,
json=lf_in.model_dump(),
)
resp.raise_for_status()
return bol_number, {
"total_linear_feet": resp.json().get("total_linear_feet", 0.0)
}
except Exception as e:
error_detail = ""
if hasattr(e, "response") and e.response is not None:
try:
error_json = e.response.json()
error_detail = error_json.get("detail", str(error_json))
except Exception:
error_detail = e.response.text
else:
error_detail = str(e)
logger.exception(
f"Error calculating linear feet for BOL {bol_number}: {e} - {error_detail}"
)
email_service.send_email(
dynamic_data={
"bol_number": bol_number,
"error_message": f"Error calculating linear feet for BOL {bol_number}: {e} - {error_detail}",
}
)
return bol_number, {"total_linear_feet": 0.0}
async def calculate_linear_feet_per_bol(
self, primus_results: list, llm_results: dict[str, dict]
):
"""
Calcula los pies lineales para múltiples Bills of Lading (BOLs) en paralelo.
Para cada BOL, construye el objeto LinearFeetIn, realiza una llamada a la API de Megatron
y devuelve los resultados.
Args:
primus_results (list): Una lista de diccionarios con los resultados de Primus.
llm_results (dict[str, dict]): Un diccionario con los resultados del LLM, indexados por número de BOL.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
que contiene el total de pies lineales calculados.
"""
headers = {"access-token": self.token, "Content-Type": "application/json"}
tasks = []
async with httpx.AsyncClient() as client:
for primus_result in primus_results:
bol_number = str(primus_result["BOLNumber"])
lf_in = build_linear_feet_obj(primus_result, llm_results)
if not lf_in:
logger.warning(f"No hay data LLM para BOL {bol_number}, se omite")
continue
tasks.append(self._post_linear_feet(client, bol_number, lf_in, headers))
results = await asyncio.gather(*tasks, return_exceptions=False)
return {bol: lf_out for bol, lf_out in results}Functions
__init__()
Inicializa el cliente de Megatron.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor. |
Source code in app/services/megatron_client.py
def __init__(self):
"""
Inicializa el cliente de Megatron.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.base_url = settings.megatron_base_url
self.token = settings.megatron_token
calculate_linear_feet_per_bol(primus_results, llm_results)
async
Calcula los pies lineales para múltiples Bills of Lading (BOLs) en paralelo.
Para cada BOL, construye el objeto LinearFeetIn, realiza una llamada a la API de Megatron y devuelve los resultados.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
primus_results
|
list
|
Una lista de diccionarios con los resultados de Primus. |
required |
llm_results
|
dict[str, dict]
|
Un diccionario con los resultados del LLM, indexados por número de BOL. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
dict |
Un diccionario donde la clave es el número de BOL y el valor es un diccionario que contiene el total de pies lineales calculados. |
Source code in app/services/megatron_client.py
async def calculate_linear_feet_per_bol(
self, primus_results: list, llm_results: dict[str, dict]
):
"""
Calcula los pies lineales para múltiples Bills of Lading (BOLs) en paralelo.
Para cada BOL, construye el objeto LinearFeetIn, realiza una llamada a la API de Megatron
y devuelve los resultados.
Args:
primus_results (list): Una lista de diccionarios con los resultados de Primus.
llm_results (dict[str, dict]): Un diccionario con los resultados del LLM, indexados por número de BOL.
Returns:
dict: Un diccionario donde la clave es el número de BOL y el valor es un diccionario
que contiene el total de pies lineales calculados.
"""
headers = {"access-token": self.token, "Content-Type": "application/json"}
tasks = []
async with httpx.AsyncClient() as client:
for primus_result in primus_results:
bol_number = str(primus_result["BOLNumber"])
lf_in = build_linear_feet_obj(primus_result, llm_results)
if not lf_in:
logger.warning(f"No hay data LLM para BOL {bol_number}, se omite")
continue
tasks.append(self._post_linear_feet(client, bol_number, lf_in, headers))
results = await asyncio.gather(*tasks, return_exceptions=False)
return {bol: lf_out for bol, lf_out in results}Functions
Email Service
Servicio de utilidad para el envío de notificaciones por correo electrónico, principalmente en caso de errores durante el proceso.
app.services.send_email
app.services.send_email
Classes
EmailService
Servicio para el envío de correos electrónicos utilizando SendGrid.
Esta clase encapsula la lógica para enviar correos electrónicos, especialmente para notificaciones de errores, utilizando una plantilla dinámica.
Source code in app/services/send_email.py
class EmailService:
"""
Servicio para el envío de correos electrónicos utilizando SendGrid.
Esta clase encapsula la lógica para enviar correos electrónicos,
especialmente para notificaciones de errores, utilizando una plantilla dinámica.
"""
def __init__(self):
"""
Inicializa el cliente de SendGrid.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.sg_client = SendGridAPIClient(settings.sendgrid_api_key)
def send_email(self, *, dynamic_data: dict):
"""
Envía un correo electrónico utilizando una plantilla dinámica de SendGrid.
Args:
dynamic_data (dict): Un diccionario con los datos dinámicos que se inyectarán en la plantilla del correo.
Debe contener al menos una clave "subject" si se desea un asunto personalizado,
de lo contrario, se usará un asunto predeterminado.
Returns:
None: Este método no retorna ningún valor explícito. Registra un error si el envío falla.
"""
sg_message = Mail(
from_email=settings.sendgrid_from,
to_emails=settings.sendgrid_to,
)
sg_message.template_id = settings.sendgrid_id_template
sg_message.dynamic_template_data = {
"subject": "Primus load creation failed",
**dynamic_data,
}
try:
self.sg_client.send(sg_message)
except Exception as e:
logger.error(f"[ERROR] enviando email de notificación: {str(e)}")
return NoneFunctions
__init__()
Inicializa el cliente de SendGrid.
Returns:
| Type | Description |
|---|---|
|
No retorna ningún valor. |
Source code in app/services/send_email.py
def __init__(self):
"""
Inicializa el cliente de SendGrid.
Args:
No recibe argumentos.
Returns:
No retorna ningún valor.
"""
self.sg_client = SendGridAPIClient(settings.sendgrid_api_key)
send_email(*, dynamic_data)
Envía un correo electrónico utilizando una plantilla dinámica de SendGrid.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dynamic_data
|
dict
|
Un diccionario con los datos dinámicos que se inyectarán en la plantilla del correo. Debe contener al menos una clave "subject" si se desea un asunto personalizado, de lo contrario, se usará un asunto predeterminado. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
None |
Este método no retorna ningún valor explícito. Registra un error si el envío falla. |
Source code in app/services/send_email.py
def send_email(self, *, dynamic_data: dict):
"""
Envía un correo electrónico utilizando una plantilla dinámica de SendGrid.
Args:
dynamic_data (dict): Un diccionario con los datos dinámicos que se inyectarán en la plantilla del correo.
Debe contener al menos una clave "subject" si se desea un asunto personalizado,
de lo contrario, se usará un asunto predeterminado.
Returns:
None: Este método no retorna ningún valor explícito. Registra un error si el envío falla.
"""
sg_message = Mail(
from_email=settings.sendgrid_from,
to_emails=settings.sendgrid_to,
)
sg_message.template_id = settings.sendgrid_id_template
sg_message.dynamic_template_data = {
"subject": "Primus load creation failed",
**dynamic_data,
}
try:
self.sg_client.send(sg_message)
except Exception as e:
logger.error(f"[ERROR] enviando email de notificación: {str(e)}")
return None