162 lines
5.0 KiB
Python
162 lines
5.0 KiB
Python
"""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 (
|
|
QuickReplyQuestions,
|
|
QuickReplyScreen,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class QuickReplyContentService:
|
|
"""Service for loading quick reply screen content from JSON files."""
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
"""Initialize quick reply content service.
|
|
|
|
Args:
|
|
settings: Application settings
|
|
|
|
"""
|
|
self.settings = settings
|
|
self.quick_replies_path = settings.base_path / "quick_replies"
|
|
self._cache: dict[str, QuickReplyScreen] = {}
|
|
|
|
logger.info(
|
|
"QuickReplyContentService initialized with path: %s",
|
|
self.quick_replies_path,
|
|
)
|
|
|
|
# Preload all quick reply files into memory
|
|
self._preload_cache()
|
|
|
|
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)
|
|
|
|
def _parse_quick_reply_data(self, data: dict) -> QuickReplyScreen:
|
|
"""Parse JSON data into QuickReplyScreen model.
|
|
|
|
Args:
|
|
data: JSON data dictionary
|
|
|
|
Returns:
|
|
Parsed QuickReplyScreen object
|
|
|
|
"""
|
|
preguntas_data = data.get("preguntas", [])
|
|
preguntas = [
|
|
QuickReplyQuestions(
|
|
titulo=q.get("titulo", ""),
|
|
descripcion=q.get("descripcion"),
|
|
respuesta=q.get("respuesta", ""),
|
|
)
|
|
for q in preguntas_data
|
|
]
|
|
|
|
return QuickReplyScreen(
|
|
header=data.get("header"),
|
|
body=data.get("body"),
|
|
button=data.get("button"),
|
|
header_section=data.get("header_section"),
|
|
preguntas=preguntas,
|
|
)
|
|
|
|
def _preload_cache(self) -> None:
|
|
"""Preload all quick reply files into memory cache at startup.
|
|
|
|
This method runs synchronously at initialization to load all
|
|
quick reply JSON files. Blocking here is acceptable since it
|
|
only happens once at startup.
|
|
|
|
"""
|
|
if not self.quick_replies_path.exists():
|
|
logger.warning(
|
|
"Quick replies directory not found: %s",
|
|
self.quick_replies_path,
|
|
)
|
|
return
|
|
|
|
loaded_count = 0
|
|
failed_count = 0
|
|
|
|
for file_path in self.quick_replies_path.glob("*.json"):
|
|
screen_id = file_path.stem
|
|
try:
|
|
# Blocking I/O is OK at startup
|
|
content = file_path.read_text(encoding="utf-8")
|
|
data = json.loads(content)
|
|
quick_reply = self._parse_quick_reply_data(data)
|
|
|
|
self._cache[screen_id] = quick_reply
|
|
loaded_count += 1
|
|
|
|
logger.debug(
|
|
"Cached %s quick replies for screen: %s",
|
|
len(quick_reply.preguntas),
|
|
screen_id,
|
|
)
|
|
|
|
except json.JSONDecodeError:
|
|
logger.exception("Invalid JSON in file: %s", file_path)
|
|
failed_count += 1
|
|
except Exception:
|
|
logger.exception("Failed to load quick reply file: %s", file_path)
|
|
failed_count += 1
|
|
|
|
logger.info(
|
|
"Quick reply cache initialized: %s screens loaded, %s failed",
|
|
loaded_count,
|
|
failed_count,
|
|
)
|
|
|
|
async def get_quick_replies(self, screen_id: str) -> QuickReplyScreen:
|
|
"""Get quick reply screen content by ID from in-memory cache.
|
|
|
|
This method is non-blocking as it retrieves data from the
|
|
in-memory cache populated at startup.
|
|
|
|
Args:
|
|
screen_id: Screen identifier (e.g., "pagos", "home")
|
|
|
|
Returns:
|
|
Quick reply screen data
|
|
|
|
Raises:
|
|
ValueError: If the quick reply is not found in cache
|
|
|
|
"""
|
|
if not screen_id or not screen_id.strip():
|
|
logger.warning("screen_id is null or empty. Returning empty quick replies")
|
|
return QuickReplyScreen(
|
|
header="empty",
|
|
body=None,
|
|
button=None,
|
|
header_section=None,
|
|
preguntas=[],
|
|
)
|
|
|
|
# Non-blocking: just a dictionary lookup
|
|
quick_reply = self._cache.get(screen_id)
|
|
|
|
if quick_reply is None:
|
|
logger.warning("Quick reply not found in cache for screen: %s", screen_id)
|
|
msg = f"Quick reply not found for screen_id: {screen_id}"
|
|
raise ValueError(msg)
|
|
|
|
logger.info(
|
|
"Retrieved %s quick replies for screen: %s from cache",
|
|
len(quick_reply.preguntas),
|
|
screen_id,
|
|
)
|
|
|
|
return quick_reply
|