va/int-layer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix lint errors

Browse Source
This commit is contained in:
Anibal Angulo
2026-02-20 04:59:56 +00:00
parent bcdc41ecd5
commit 58393a538e
19 changed files with 442 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
src/capa_de_integracion/services/conversation_manager.py
View File

@@ -1,5 +1,8 @@
"""Conversation manager service for orchestrating user conversations."""
import logging
from datetime import datetime, timedelta
import re
from datetime import UTC, datetime, timedelta
from uuid import uuid4
from capa_de_integracion.config import Settings
@@ -10,6 +13,7 @@ from capa_de_integracion.models import (
DetectIntentResponse,
QueryResult,
)
from capa_de_integracion.models.notification import NotificationSession
from .dlp_service import DLPService
from .firestore_service import FirestoreService
@@ -46,10 +50,10 @@ class ConversationManagerService:
logger.info("ConversationManagerService initialized successfully")
async def manage_conversation(
async def manage_conversation( # noqa: PLR0915
self, request: ConversationRequest,
) -> DetectIntentResponse:
"""Main entry point for managing conversations.
"""Manage conversation flow and return response.
Args:
request: External conversation request from client
@@ -82,10 +86,11 @@ class ConversationManagerService:
# Step 2: Check for pantallaContexto in existing session
if session.pantallaContexto:
# Check if pantallaContexto is stale (10 minutes)
if self._is_pantalla_context_valid(session.lastModified):
if self._is_pantalla_context_valid(session.last_modified):
logger.info(
f"Detected 'pantallaContexto' in session: {session.pantallaContexto}. "
f"Delegating to QuickReplies flow.",
"Detected 'pantallaContexto' in session: %s. "
"Delegating to QuickReplies flow.",
session.pantallaContexto,
)
response = await self._manage_quick_reply_conversation(
request, session.pantallaContexto,
@@ -95,62 +100,64 @@ class ConversationManagerService:
user_entry = ConversationEntry(
entity="user",
type="CONVERSACION",
timestamp=datetime.now(),
timestamp=datetime.now(UTC),
text=request.mensaje,
parameters=None,
canal=getattr(request, "canal", None),
)
await self.firestore_service.save_entry(
session.sessionId, user_entry,
session.session_id, user_entry,
)
# Save quick reply response to Firestore
response_text = (
response.queryResult.responseText
if response.queryResult
response.query_result.response_text
if response.query_result
else ""
) or ""
assistant_entry = ConversationEntry(
entity="assistant",
type="CONVERSACION",
timestamp=datetime.now(),
timestamp=datetime.now(UTC),
text=response_text,
parameters=None,
canal=getattr(request, "canal", None),
)
await self.firestore_service.save_entry(
session.sessionId, assistant_entry,
session.session_id, assistant_entry,
)
# Update session with last message and timestamp
session.lastMessage = response_text
session.lastModified = datetime.now()
session.last_message = response_text
session.last_modified = datetime.now(UTC)
await self.firestore_service.save_session(session)
await self.redis_service.save_session(session)
return response
else:
logger.info(
"Detected STALE 'pantallaContexto'. Ignoring and proceeding with normal flow.",
"Detected STALE 'pantallaContexto'. "
"Ignoring and proceeding with normal flow.",
)
# Step 3: Continue with standard conversation flow
nickname = request.usuario.nickname
logger.info(
f"Primary Check (Redis): Looking up session for phone: {telefono}",
"Primary Check (Redis): Looking up session for phone: %s",
telefono,
)
# Step 3a: Load conversation history from Firestore
entries = await self.firestore_service.get_entries(
session.sessionId,
session.session_id,
limit=self.settings.conversation_context_message_limit,
)
logger.info(f"Loaded {len(entries)} conversation entries from Firestore")
logger.info("Loaded %s conversation entries from Firestore", len(entries))
# Step 3b: Retrieve active notifications for this user
notifications = await self._get_active_notifications(telefono)
logger.info(f"Retrieved {len(notifications)} active notifications")
logger.info("Retrieved %s active notifications", len(notifications))
# Step 3c: Prepare context for RAG service
messages = await self._prepare_rag_messages(
@@ -165,36 +172,37 @@ class ConversationManagerService:
logger.info("Sending query to RAG service")
assistant_response = await self.rag_service.query(messages)
logger.info(
f"Received response from RAG service: {assistant_response[:100]}...",
"Received response from RAG service: %s...",
assistant_response[:100],
)
# Step 3e: Save user message to Firestore
user_entry = ConversationEntry(
entity="user",
type="CONVERSACION",
timestamp=datetime.now(),
timestamp=datetime.now(UTC),
text=request.mensaje,
parameters=None,
canal=getattr(request, "canal", None),
)
await self.firestore_service.save_entry(session.sessionId, user_entry)
await self.firestore_service.save_entry(session.session_id, user_entry)
logger.info("Saved user message to Firestore")
# Step 3f: Save assistant response to Firestore
assistant_entry = ConversationEntry(
entity="assistant",
type="LLM",
timestamp=datetime.now(),
timestamp=datetime.now(UTC),
text=assistant_response,
parameters=None,
canal=getattr(request, "canal", None),
)
await self.firestore_service.save_entry(session.sessionId, assistant_entry)
await self.firestore_service.save_entry(session.session_id, assistant_entry)
logger.info("Saved assistant response to Firestore")
# Step 3g: Update session with last message and timestamp
session.lastMessage = assistant_response
session.lastModified = datetime.now()
session.last_message = assistant_response
session.last_modified = datetime.now(UTC)
await self.firestore_service.save_session(session)
await self.redis_service.save_session(session)
logger.info("Updated session in Firestore and Redis")
@@ -202,26 +210,26 @@ class ConversationManagerService:
# Step 3h: Mark notifications as processed if any were included
if notifications:
await self._mark_notifications_as_processed(telefono)
logger.info(f"Marked {len(notifications)} notifications as processed")
logger.info("Marked %s notifications as processed", len(notifications))
# Step 3i: Return response object
return DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(
responseText=assistant_response,
response_id=str(uuid4()),
query_result=QueryResult(
response_text=assistant_response,
parameters=None,
),
quick_replies=None,
)
except Exception as e:
logger.error(f"Error managing conversation: {e!s}", exc_info=True)
except Exception:
logger.exception("Error managing conversation")
raise
def _is_pantalla_context_valid(self, last_modified: datetime) -> bool:
"""Check if pantallaContexto is still valid (not stale)."""
time_diff = datetime.now() - last_modified
time_diff = datetime.now(UTC) - last_modified
return time_diff < timedelta(minutes=self.SCREEN_CONTEXT_TIMEOUT_MINUTES)
async def _manage_quick_reply_conversation(
@@ -234,7 +242,7 @@ class ConversationManagerService:
# If no questions available, delegate to normal conversation flow
if not quick_reply_screen.preguntas:
logger.warning(f"No quick replies found for screen: {screen_id}.")
logger.warning("No quick replies found for screen: %s.", screen_id)
return None
# Match user message to a quick reply question
@@ -245,19 +253,20 @@ class ConversationManagerService:
# Simple matching: check if question title matches user message
if pregunta.titulo.lower().strip() == user_message_lower:
matched_answer = pregunta.respuesta
logger.info(f"Matched quick reply: {pregunta.titulo}")
logger.info("Matched quick reply: %s", pregunta.titulo)
break
# If no match, use first question as default or delegate to normal flow
if not matched_answer:
logger.warning(
f"No matching quick reply found for message: '{request.mensaje}'.",
"No matching quick reply found for message: '%s'.",
request.mensaje,
)
# Create response with the matched quick reply answer
return DetectIntentResponse(
responseId=str(uuid4()),
queryResult=QueryResult(responseText=matched_answer, parameters=None),
response_id=str(uuid4()),
query_result=QueryResult(response_text=matched_answer, parameters=None),
quick_replies=quick_reply_screen,
)
@@ -281,8 +290,6 @@ class ConversationManagerService:
# If not in Redis, try Firestore
if not notification_session:
# Firestore uses phone as document ID for notifications
from capa_de_integracion.models.notification import NotificationSession
doc_ref = self.firestore_service.db.collection(
self.firestore_service.notifications_collection,
).document(telefono)
@@ -294,20 +301,19 @@ class ConversationManagerService:
# Filter for active notifications only
if notification_session and notification_session.notificaciones:
return [
active_notifications = [
notif
for notif in notification_session.notificaciones
if notif.status == "active"
]
else:
active_notifications = []
except Exception:
logger.exception("Error retrieving notifications for %s", telefono)
return []
except Exception as e:
logger.error(
f"Error retrieving notifications for {telefono}: {e!s}",
exc_info=True,
)
return []
else:
return active_notifications
async def _prepare_rag_messages(
self,
@@ -339,7 +345,10 @@ class ConversationManagerService:
messages.append(
{
"role": "system",
"content": f"Historial de conversación:\n{conversation_context}",
"content": (
f"Historial de conversación:\n"
f"{conversation_context}"
),
},
)
@@ -353,7 +362,10 @@ class ConversationManagerService:
messages.append(
{
"role": "system",
"content": f"Notificaciones pendientes para el usuario:\n{notifications_text}",
"content": (
f"Notificaciones pendientes para el usuario:\n"
f"{notifications_text}"
),
},
)
@@ -382,17 +394,17 @@ class ConversationManagerService:
# Update or delete from Redis
await self.redis_service.delete_notification_session(telefono)
logger.info(f"Marked notifications as processed for {telefono}")
logger.info("Marked notifications as processed for %s", telefono)
except Exception as e:
logger.error(
f"Error marking notifications as processed for {telefono}: {e!s}",
exc_info=True,
except Exception:
logger.exception(
"Error marking notifications as processed for %s",
telefono,
)
def _format_conversation_history(
self,
session: ConversationSession,
session: ConversationSession, # noqa: ARG002
entries: list[ConversationEntry],
) -> str:
"""Format conversation history with business rule limits.
@@ -414,7 +426,7 @@ class ConversationManagerService:
return ""
# Filter by date (30 days)
cutoff_date = datetime.now() - timedelta(
cutoff_date = datetime.now(UTC) - timedelta(
days=self.settings.conversation_context_days_limit,
)
recent_entries = [
@@ -445,7 +457,7 @@ class ConversationManagerService:
if not entries:
return ""
MAX_BYTES = 50 * 1024 # 50KB
max_bytes = 50 * 1024 # 50KB
formatted_messages = [self._format_entry(entry) for entry in entries]
# Build from newest to oldest
@@ -456,7 +468,7 @@ class ConversationManagerService:
message_line = message + "\n"
message_bytes = len(message_line.encode("utf-8"))
if current_size + message_bytes > MAX_BYTES:
if current_size + message_bytes > max_bytes:
break
text_block.insert(0, message_line)
@@ -481,8 +493,6 @@ class ConversationManagerService:
content = entry.text
if entry.entity == "assistant":
# Remove trailing JSON artifacts like {...}
import re
content = re.sub(r"\s*\{.*\}\s*$", "", content).strip()
return prefix + content

49
src/capa_de_integracion/services/dlp_service.py
View File

@@ -1,9 +1,4 @@
"""Copyright 2025 Google. This software is provided as-is, without warranty or
representation for any use or purpose. Your use of it is subject to your
agreement with Google.
Data Loss Prevention service for obfuscating sensitive information.
"""
"""DLP service for detecting and obfuscating sensitive data."""
import logging
import re
@@ -15,6 +10,11 @@ from capa_de_integracion.config import Settings
logger = logging.getLogger(__name__)
# DLP likelihood threshold for filtering findings
LIKELIHOOD_THRESHOLD = 3 # POSSIBLE (values: 0=VERY_UNLIKELY to 5=VERY_LIKELY)
# Minimum length for last 4 characters extraction
MIN_LENGTH_FOR_LAST_FOUR = 4
class DLPService:
"""Service for detecting and obfuscating sensitive data using Google Cloud DLP.
@@ -71,7 +71,10 @@ class DLPService:
)
# Build request
inspect_template_name = f"projects/{self.project_id}/locations/{self.location}/inspectTemplates/{template_id}"
inspect_template_name = (
f"projects/{self.project_id}/locations/{self.location}/"
f"inspectTemplates/{template_id}"
)
parent = f"projects/{self.project_id}/locations/{self.location}"
request = types.InspectContentRequest(
@@ -85,18 +88,18 @@ class DLPService:
response = await self.dlp_client.inspect_content(request=request)
findings_count = len(response.result.findings)
logger.info(f"DLP {template_id} Findings: {findings_count}")
logger.info("DLP %s Findings: %s", template_id, findings_count)
if findings_count > 0:
return self._obfuscate_text(response, text)
return text
obfuscated_text = self._obfuscate_text(response, text)
else:
obfuscated_text = text
except Exception as e:
logger.error(
f"Error during DLP inspection: {e}. Returning original text.",
exc_info=True,
)
except Exception:
logger.exception("Error during DLP inspection. Returning original text.")
return text
else:
return obfuscated_text
def _obfuscate_text(self, response: types.InspectContentResponse, text: str) -> str:
"""Obfuscate sensitive findings in text.
@@ -109,11 +112,11 @@ class DLPService:
Text with sensitive data obfuscated
"""
# Filter findings by likelihood (> POSSIBLE, which is value 3)
# Filter findings by likelihood (> POSSIBLE)
findings = [
finding
for finding in response.result.findings
if finding.likelihood.value > 3
if finding.likelihood.value > LIKELIHOOD_THRESHOLD
]
# Sort by likelihood (descending)
@@ -124,7 +127,9 @@ class DLPService:
info_type = finding.info_type.name
logger.info(
f"InfoType: {info_type} | Likelihood: {finding.likelihood.value}",
"InfoType: %s | Likelihood: %s",
info_type,
finding.likelihood.value,
)
# Obfuscate based on info type
@@ -176,13 +181,15 @@ class DLPService:
def _get_last4(self, quote: str) -> str:
"""Extract last 4 characters from quote (removing spaces)."""
clean_quote = quote.strip().replace(" ", "")
if len(clean_quote) >= 4:
if len(clean_quote) >= MIN_LENGTH_FOR_LAST_FOUR:
return clean_quote[-4:]
return clean_quote
def _clean_direccion(self, text: str) -> str:
"""Clean up consecutive [DIRECCION] tags."""
# Replace multiple [DIRECCION] tags separated by commas or spaces with single tag
"""Clean up consecutive [DIRECCION] tags.
Replace multiple [DIRECCION] tags separated by commas or spaces.
"""
pattern = r"\[DIRECCION\](?:(?:,\s*|\s+)\[DIRECCION\])*"
return re.sub(pattern, "[DIRECCION]", text).strip()

168
src/capa_de_integracion/services/firestore_service.py
View File

@@ -1,5 +1,7 @@
"""Firestore service for conversation and notification persistence."""
import logging
from datetime import datetime
from datetime import UTC, datetime
from google.cloud import firestore
@@ -28,7 +30,8 @@ class FirestoreService:
f"artifacts/{settings.gcp_project_id}/notifications"
)
logger.info(
f"Firestore client initialized for project: {settings.gcp_project_id}",
"Firestore client initialized for project: %s",
settings.gcp_project_id,
)
async def close(self) -> None:
@@ -36,7 +39,7 @@ class FirestoreService:
self.db.close()
logger.info("Firestore client closed")
def _session_ref(self, session_id: str):
def _session_ref(self, session_id: str) -> firestore.DocumentReference:
"""Get Firestore document reference for session."""
return self.db.collection(self.conversations_collection).document(session_id)
@@ -47,19 +50,20 @@ class FirestoreService:
doc = await doc_ref.get()
if not doc.exists:
logger.debug(f"Session not found in Firestore: {session_id}")
logger.debug("Session not found in Firestore: %s", session_id)
return None
data = doc.to_dict()
session = ConversationSession.model_validate(data)
logger.debug(f"Retrieved session from Firestore: {session_id}")
return session
except Exception as e:
logger.debug("Retrieved session from Firestore: %s", session_id)
except Exception:
logger.exception(
f"Error retrieving session {session_id} from Firestore: {e!s}",
"Error retrieving session %s from Firestore:",
session_id,
)
return None
else:
return session
async def get_session_by_phone(self, telefono: str) -> ConversationSession | None:
"""Retrieve most recent conversation session from Firestore by phone number.
@@ -84,33 +88,37 @@ class FirestoreService:
data = doc.to_dict()
session = ConversationSession.model_validate(data)
logger.debug(
f"Retrieved session from Firestore for phone {telefono}: {session.sessionId}",
"Retrieved session from Firestore for phone %s: %s",
telefono,
session.session_id,
)
return session
logger.debug(f"No session found in Firestore for phone: {telefono}")
return None
except Exception as e:
logger.debug("No session found in Firestore for phone: %s", telefono)
except Exception:
logger.exception(
f"Error querying session by phone {telefono} from Firestore: {e!s}",
"Error querying session by phone %s from Firestore:",
telefono,
)
return None
else:
return None
async def save_session(self, session: ConversationSession) -> bool:
"""Save conversation session to Firestore."""
try:
doc_ref = self._session_ref(session.sessionId)
doc_ref = self._session_ref(session.session_id)
data = session.model_dump()
await doc_ref.set(data, merge=True)
logger.debug(f"Saved session to Firestore: {session.sessionId}")
return True
except Exception as e:
logger.debug("Saved session to Firestore: %s", session.session_id)
except Exception:
logger.exception(
f"Error saving session {session.sessionId} to Firestore: {e!s}",
"Error saving session %s to Firestore:",
session.session_id,
)
return False
else:
return True
async def create_session(
self,
@@ -144,11 +152,11 @@ class FirestoreService:
last_message=last_message,
)
doc_ref = self._session_ref(session.sessionId)
doc_ref = self._session_ref(session.session_id)
data = session.model_dump()
await doc_ref.set(data, merge=True)
logger.info(f"Created new session in Firestore: {session_id}")
logger.info("Created new session in Firestore: %s", session_id)
return session
async def save_entry(self, session_id: str, entry: ConversationEntry) -> bool:
@@ -163,14 +171,15 @@ class FirestoreService:
data = entry.model_dump()
await entry_doc.set(data)
logger.debug(f"Saved entry to Firestore for session: {session_id}")
return True
except Exception as e:
logger.debug("Saved entry to Firestore for session: %s", session_id)
except Exception:
logger.exception(
f"Error saving entry for session {session_id} to Firestore: {e!s}",
"Error saving entry for session %s to Firestore:",
session_id,
)
return False
else:
return True
async def get_entries(
self, session_id: str, limit: int = 10,
@@ -195,14 +204,17 @@ class FirestoreService:
# Reverse to get chronological order
entries.reverse()
logger.debug(f"Retrieved {len(entries)} entries for session: {session_id}")
return entries
except Exception as e:
logger.debug(
"Retrieved %s entries for session: %s", len(entries), session_id,
)
except Exception:
logger.exception(
f"Error retrieving entries for session {session_id} from Firestore: {e!s}",
"Error retrieving entries for session %s from Firestore:",
session_id,
)
return []
else:
return entries
async def delete_session(self, session_id: str) -> bool:
"""Delete conversation session and all entries from Firestore."""
@@ -216,14 +228,15 @@ class FirestoreService:
# Delete session document
await doc_ref.delete()
logger.debug(f"Deleted session from Firestore: {session_id}")
return True
except Exception as e:
logger.debug("Deleted session from Firestore: %s", session_id)
except Exception:
logger.exception(
f"Error deleting session {session_id} from Firestore: {e!s}",
"Error deleting session %s from Firestore:",
session_id,
)
return False
else:
return True
async def update_pantalla_contexto(
self, session_id: str, pantalla_contexto: str | None,
@@ -244,31 +257,34 @@ class FirestoreService:
if not doc.exists:
logger.warning(
f"Session {session_id} not found in Firestore. Cannot update pantallaContexto",
"Session %s not found in Firestore. Cannot update pantallaContexto",
session_id,
)
return False
await doc_ref.update(
{
"pantallaContexto": pantalla_contexto,
"lastModified": datetime.now(),
"lastModified": datetime.now(UTC),
},
)
logger.debug(
f"Updated pantallaContexto for session {session_id} in Firestore",
"Updated pantallaContexto for session %s in Firestore",
session_id,
)
return True
except Exception as e:
except Exception:
logger.exception(
f"Error updating pantallaContexto for session {session_id} in Firestore: {e!s}",
"Error updating pantallaContexto for session %s in Firestore:",
session_id,
)
return False
else:
return True
# ====== Notification Methods ======
def _notification_ref(self, notification_id: str):
def _notification_ref(self, notification_id: str) -> firestore.DocumentReference:
"""Get Firestore document reference for notification."""
return self.db.collection(self.notifications_collection).document(
notification_id,
@@ -303,30 +319,33 @@ class FirestoreService:
await doc_ref.update(
{
"notificaciones": firestore.ArrayUnion([entry_dict]),
"ultimaActualizacion": datetime.now(),
"ultima_actualizacion": datetime.now(UTC),
},
)
logger.info(
f"Successfully appended notification entry to session {notification_session_id} in Firestore",
"Successfully appended notification entry "
"to session %s in Firestore",
notification_session_id,
)
else:
# Create new notification session
new_session_data = {
"sessionId": notification_session_id,
"session_id": notification_session_id,
"telefono": phone_number,
"fechaCreacion": datetime.now(),
"ultimaActualizacion": datetime.now(),
"fecha_creacion": datetime.now(UTC),
"ultima_actualizacion": datetime.now(UTC),
"notificaciones": [entry_dict],
}
await doc_ref.set(new_session_data)
logger.info(
f"Successfully created new notification session {notification_session_id} in Firestore",
"Successfully created new notification session %s in Firestore",
notification_session_id,
)
except Exception as e:
logger.error(
f"Error saving notification to Firestore for phone {phone_number}: {e!s}",
exc_info=True,
except Exception:
logger.exception(
"Error saving notification to Firestore for phone %s",
phone_number,
)
raise
@@ -344,7 +363,9 @@ class FirestoreService:
if not doc.exists:
logger.warning(
f"Notification session {session_id} not found in Firestore. Cannot update status",
"Notification session %s not found in Firestore. "
"Cannot update status",
session_id,
)
return
@@ -359,18 +380,21 @@ class FirestoreService:
await doc_ref.update(
{
"notificaciones": updated_notifications,
"ultimaActualizacion": datetime.now(),
"ultima_actualizacion": datetime.now(UTC),
},
)
logger.info(
f"Successfully updated notification status to '{status}' for session {session_id} in Firestore",
"Successfully updated notification status to '%s' "
"for session %s in Firestore",
status,
session_id,
)
except Exception as e:
logger.error(
f"Error updating notification status in Firestore for session {session_id}: {e!s}",
exc_info=True,
except Exception:
logger.exception(
"Error updating notification status in Firestore for session %s",
session_id,
)
raise
@@ -378,18 +402,20 @@ class FirestoreService:
"""Delete notification session from Firestore."""
try:
logger.info(
f"Deleting notification session {notification_id} from Firestore",
"Deleting notification session %s from Firestore",
notification_id,
)
doc_ref = self._notification_ref(notification_id)
await doc_ref.delete()
logger.info(
f"Successfully deleted notification session {notification_id} from Firestore",
"Successfully deleted notification session %s from Firestore",
notification_id,
)
return True
except Exception as e:
logger.error(
f"Error deleting notification session {notification_id} from Firestore: {e!s}",
exc_info=True,
except Exception:
logger.exception(
"Error deleting notification session %s from Firestore",
notification_id,
)
return False
else:
return True

21
src/capa_de_integracion/services/notification_manager.py
View File

@@ -1,3 +1,5 @@
"""Notification manager service for processing push notifications."""
import asyncio
import logging
from uuid import uuid4
@@ -103,7 +105,8 @@ class NotificationManagerService:
# Save notification to Redis (with async Firestore write-back)
await self.redis_service.save_or_append_notification(new_notification_entry)
logger.info(
f"Notification for phone {telefono} cached. Kicking off async Firestore write-back",
"Notification for phone %s cached. Kicking off async Firestore write-back",
telefono,
)
# Fire-and-forget Firestore write (matching Java's .subscribe() behavior)
@@ -113,13 +116,17 @@ class NotificationManagerService:
new_notification_entry,
)
logger.debug(
f"Notification entry persisted to Firestore for phone {telefono}",
"Notification entry persisted to Firestore for phone %s",
telefono,
)
except Exception as e:
logger.error(
f"Background: Error during notification persistence to Firestore for phone {telefono}: {e}",
exc_info=True,
except Exception:
logger.exception(
"Background: Error during notification persistence "
"to Firestore for phone %s",
telefono,
)
# Fire and forget - don't await
asyncio.create_task(save_notification_to_firestore())
_task = asyncio.create_task(save_notification_to_firestore())
# Store reference to prevent premature garbage collection
del _task

48
src/capa_de_integracion/services/quick_reply_content.py
View File

@@ -1,5 +1,8 @@
"""Quick reply content service for loading FAQ screens."""
import json
import logging
from pathlib import Path
from capa_de_integracion.config import Settings
from capa_de_integracion.models.quick_replies import (
@@ -24,9 +27,17 @@ class QuickReplyContentService:
self.quick_replies_path = settings.base_path / "quick_replies"
logger.info(
f"QuickReplyContentService initialized with path: {self.quick_replies_path}",
"QuickReplyContentService initialized with path: %s",
self.quick_replies_path,
)
def _validate_file(self, file_path: Path, screen_id: str) -> None:
"""Validate that the quick reply file exists."""
if not file_path.exists():
logger.warning("Quick reply file not found: %s", file_path)
msg = f"Quick reply file not found for screen_id: {screen_id}"
raise ValueError(msg)
async def get_quick_replies(self, screen_id: str) -> QuickReplyScreen:
"""Load quick reply screen content by ID.
@@ -53,15 +64,11 @@ class QuickReplyContentService:
file_path = self.quick_replies_path / f"{screen_id}.json"
try:
if not file_path.exists():
logger.warning(f"Quick reply file not found: {file_path}")
msg = f"Quick reply file not found for screen_id: {screen_id}"
raise ValueError(
msg,
)
self._validate_file(file_path, screen_id)
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
# Use Path.read_text() for async-friendly file reading
content = file_path.read_text(encoding="utf-8")
data = json.loads(content)
# Parse questions
preguntas_data = data.get("preguntas", [])
@@ -83,22 +90,17 @@ class QuickReplyContentService:
)
logger.info(
f"Successfully loaded {len(preguntas)} quick replies for screen: {screen_id}",
"Successfully loaded %s quick replies for screen: %s",
len(preguntas),
screen_id,
)
return quick_reply
except json.JSONDecodeError as e:
logger.error(f"Error parsing JSON file {file_path}: {e}", exc_info=True)
logger.exception("Error parsing JSON file %s", file_path)
msg = f"Invalid JSON format in quick reply file for screen_id: {screen_id}"
raise ValueError(
msg,
) from e
raise ValueError(msg) from e
except Exception as e:
logger.error(
f"Error loading quick replies for screen {screen_id}: {e}",
exc_info=True,
)
logger.exception("Error loading quick replies for screen %s", screen_id)
msg = f"Error loading quick replies for screen_id: {screen_id}"
raise ValueError(
msg,
) from e
raise ValueError(msg) from e
else:
return quick_reply

42
src/capa_de_integracion/services/rag_service.py
View File

@@ -1,4 +1,8 @@
"""RAG service for calling RAG endpoints with high concurrency."""
import logging
from types import TracebackType
from typing import Self
import httpx
from pydantic import BaseModel, Field
@@ -68,8 +72,11 @@ class RAGService:
)
logger.info(
f"RAGService initialized with endpoint: {self.rag_endpoint_url}, "
f"max_connections: {max_connections}, timeout: {timeout}s",
"RAGService initialized with endpoint: %s, "
"max_connections: %s, timeout: %ss",
self.rag_endpoint_url,
max_connections,
timeout,
)
async def query(self, messages: list[dict[str, str]]) -> str:
@@ -93,7 +100,7 @@ class RAGService:
request = RAGRequest(messages=message_objects)
# Make async HTTP POST request
logger.debug(f"Sending RAG request with {len(messages)} messages")
logger.debug("Sending RAG request with %s messages", len(messages))
response = await self._client.post(
self.rag_endpoint_url,
@@ -108,32 +115,37 @@ class RAGService:
response_data = response.json()
rag_response = RAGResponse(**response_data)
logger.debug(f"RAG response received: {len(rag_response.response)} chars")
return rag_response.response
logger.debug("RAG response received: %s chars", len(rag_response.response))
except httpx.HTTPStatusError as e:
logger.exception(
f"HTTP error calling RAG endpoint: {e.response.status_code} - {e.response.text}",
"HTTP error calling RAG endpoint: %s - %s",
e.response.status_code,
e.response.text,
)
raise
except httpx.RequestError as e:
logger.exception(f"Request error calling RAG endpoint: {e!s}")
except httpx.RequestError:
logger.exception("Request error calling RAG endpoint:")
raise
except Exception as e:
logger.error(
f"Unexpected error calling RAG endpoint: {e!s}", exc_info=True,
)
except Exception:
logger.exception("Unexpected error calling RAG endpoint")
raise
else:
return rag_response.response
async def close(self) -> None:
"""Close the HTTP client and release connections."""
await self._client.aclose()
logger.info("RAGService client closed")
async def __aenter__(self):
async def __aenter__(self) -> Self:
"""Async context manager entry."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
"""Async context manager exit."""
await self.close()

131
src/capa_de_integracion/services/redis_service.py
View File

@@ -1,11 +1,13 @@
"""Redis service for caching conversation sessions and notifications."""
import json
import logging
from datetime import datetime
from datetime import UTC, datetime
from redis.asyncio import Redis
from capa_de_integracion.config import Settings
from capa_de_integracion.models import ConversationSession
from capa_de_integracion.models import ConversationEntry, ConversationSession
from capa_de_integracion.models.notification import Notification, NotificationSession
logger = logging.getLogger(__name__)
@@ -31,7 +33,9 @@ class RedisService:
decode_responses=True,
)
logger.info(
f"Connected to Redis at {self.settings.redis_host}:{self.settings.redis_port}",
"Connected to Redis at %s:%s",
self.settings.redis_host,
self.settings.redis_port,
)
async def close(self) -> None:
@@ -66,29 +70,26 @@ class RedisService:
phone_key = self._phone_to_session_key(session_id_or_phone)
mapped_session_id = await self.redis.get(phone_key)
if mapped_session_id:
# Found phone mapping, get the actual session
session_id = mapped_session_id
else:
# Try as direct session ID
session_id = session_id_or_phone
# Use mapped session ID if found, otherwise use input directly
session_id = mapped_session_id or session_id_or_phone
# Get session by ID
key = self._session_key(session_id)
data = await self.redis.get(key)
if not data:
logger.debug(f"Session not found in Redis: {session_id_or_phone}")
logger.debug("Session not found in Redis: %s", session_id_or_phone)
return None
try:
session_dict = json.loads(data)
session = ConversationSession.model_validate(session_dict)
logger.debug(f"Retrieved session from Redis: {session_id}")
return session
except Exception as e:
logger.exception(f"Error deserializing session {session_id}: {e!s}")
logger.debug("Retrieved session from Redis: %s", session_id)
except Exception:
logger.exception("Error deserializing session %s:", session_id)
return None
else:
return session
async def save_session(self, session: ConversationSession) -> bool:
"""Save conversation session to Redis with TTL.
@@ -99,7 +100,7 @@ class RedisService:
msg = "Redis client not connected"
raise RuntimeError(msg)
key = self._session_key(session.sessionId)
key = self._session_key(session.session_id)
phone_key = self._phone_to_session_key(session.telefono)
try:
@@ -108,15 +109,18 @@ class RedisService:
await self.redis.setex(key, self.session_ttl, data)
# Save phone-to-session mapping
await self.redis.setex(phone_key, self.session_ttl, session.sessionId)
await self.redis.setex(phone_key, self.session_ttl, session.session_id)
logger.debug(
f"Saved session to Redis: {session.sessionId} for phone: {session.telefono}",
"Saved session to Redis: %s for phone: %s",
session.session_id,
session.telefono,
)
return True
except Exception as e:
logger.exception(f"Error saving session {session.sessionId} to Redis: {e!s}")
except Exception:
logger.exception("Error saving session %s to Redis:", session.session_id)
return False
else:
return True
async def delete_session(self, session_id: str) -> bool:
"""Delete conversation session from Redis."""
@@ -128,11 +132,12 @@ class RedisService:
try:
result = await self.redis.delete(key)
logger.debug(f"Deleted session from Redis: {session_id}")
return result > 0
except Exception as e:
logger.exception(f"Error deleting session {session_id} from Redis: {e!s}")
logger.debug("Deleted session from Redis: %s", session_id)
except Exception:
logger.exception("Error deleting session %s from Redis:", session_id)
return False
else:
return result > 0
async def exists(self, session_id: str) -> bool:
"""Check if session exists in Redis."""
@@ -149,7 +154,7 @@ class RedisService:
"""Generate Redis key for conversation messages."""
return f"conversation:messages:{session_id}"
async def save_message(self, session_id: str, message) -> bool:
async def save_message(self, session_id: str, message: ConversationEntry) -> bool:
"""Save a conversation message to Redis sorted set.
Messages are stored in a sorted set with timestamp as score.
@@ -179,13 +184,15 @@ class RedisService:
# Set TTL on the messages key to match session TTL
await self.redis.expire(key, self.session_ttl)
logger.debug(f"Saved message to Redis: {session_id}")
return True
except Exception as e:
logger.debug("Saved message to Redis: %s", session_id)
except Exception:
logger.exception(
f"Error saving message to Redis for session {session_id}: {e!s}",
"Error saving message to Redis for session %s:",
session_id,
)
return False
else:
return True
async def get_messages(self, session_id: str) -> list:
"""Retrieve all conversation messages for a session from Redis.
@@ -210,7 +217,7 @@ class RedisService:
message_strings = await self.redis.zrange(key, 0, -1)
if not message_strings:
logger.debug(f"No messages found in Redis for session: {session_id}")
logger.debug("No messages found in Redis for session: %s", session_id)
return []
# Parse JSON strings to dictionaries
@@ -218,19 +225,23 @@ class RedisService:
for msg_str in message_strings:
try:
messages.append(json.loads(msg_str))
except json.JSONDecodeError as e:
logger.exception(f"Error parsing message JSON: {e!s}")
except json.JSONDecodeError:
logger.exception("Error parsing message JSON:")
continue
logger.debug(
f"Retrieved {len(messages)} messages from Redis for session: {session_id}",
"Retrieved %s messages from Redis for session: %s",
len(messages),
session_id,
)
return messages
except Exception as e:
except Exception:
logger.exception(
f"Error retrieving messages from Redis for session {session_id}: {e!s}",
"Error retrieving messages from Redis for session %s:",
session_id,
)
return []
else:
return messages
# ====== Notification Methods ======
@@ -273,17 +284,17 @@ class RedisService:
updated_session = NotificationSession(
sessionId=notification_session_id,
telefono=phone_number,
fechaCreacion=existing_session.fechaCreacion,
ultimaActualizacion=datetime.now(),
fechaCreacion=existing_session.fecha_creacion,
ultimaActualizacion=datetime.now(UTC),
notificaciones=updated_notifications,
)
else:
# Create new session
updated_session = NotificationSession(
sessionId=notification_session_id,
session_id=notification_session_id,
telefono=phone_number,
fechaCreacion=datetime.now(),
ultimaActualizacion=datetime.now(),
fecha_creacion=datetime.now(UTC),
ultima_actualizacion=datetime.now(UTC),
notificaciones=[new_entry],
)
@@ -296,7 +307,7 @@ class RedisService:
msg = "Redis client not connected"
raise RuntimeError(msg)
key = self._notification_key(session.sessionId)
key = self._notification_key(session.session_id)
phone_key = self._phone_to_notification_key(session.telefono)
try:
@@ -305,15 +316,17 @@ class RedisService:
await self.redis.setex(key, self.notification_ttl, data)
# Save phone-to-session mapping
await self.redis.setex(phone_key, self.notification_ttl, session.sessionId)
await self.redis.setex(phone_key, self.notification_ttl, session.session_id)
logger.debug(f"Cached notification session: {session.sessionId}")
return True
except Exception as e:
logger.debug("Cached notification session: %s", session.session_id)
except Exception:
logger.exception(
f"Error caching notification session {session.sessionId}: {e!s}",
"Error caching notification session %s:",
session.session_id,
)
return False
else:
return True
async def get_notification_session(
self, session_id: str,
@@ -327,19 +340,21 @@ class RedisService:
data = await self.redis.get(key)
if not data:
logger.debug(f"Notification session not found in Redis: {session_id}")
logger.debug("Notification session not found in Redis: %s", session_id)
return None
try:
session_dict = json.loads(data)
session = NotificationSession.model_validate(session_dict)
logger.info(f"Notification session {session_id} retrieved from Redis")
return session
except Exception as e:
logger.info("Notification session %s retrieved from Redis", session_id)
except Exception:
logger.exception(
f"Error deserializing notification session {session_id}: {e!s}",
"Error deserializing notification session %s:",
session_id,
)
return None
else:
return session
async def get_notification_id_for_phone(self, phone: str) -> str | None:
"""Get notification session ID for a phone number."""
@@ -351,7 +366,7 @@ class RedisService:
session_id = await self.redis.get(key)
if session_id:
logger.info(f"Session ID {session_id} found for phone")
logger.info("Session ID %s found for phone", session_id)
else:
logger.debug("Session ID not found for phone")
@@ -367,12 +382,14 @@ class RedisService:
phone_key = self._phone_to_notification_key(phone_number)
try:
logger.info(f"Deleting notification session for phone {phone_number}")
logger.info("Deleting notification session for phone %s", phone_number)
await self.redis.delete(notification_key)
await self.redis.delete(phone_key)
return True
except Exception as e:
except Exception:
logger.exception(
f"Error deleting notification session for phone {phone_number}: {e!s}",
"Error deleting notification session for phone %s:",
phone_number,
)
return False
else:
return True