File size: 4,901 Bytes
85ccd2a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 |
# /modules/database/chat_mongo_db.py
from .mongo_db import insert_document, find_documents, get_collection
from datetime import datetime, timezone
import logging
import re
logger = logging.getLogger(__name__)
COLLECTION_NAME = 'chat_history-v3'
def clean_text_content(text: str) -> str:
"""
Limpia y normaliza el texto para almacenamiento seguro en UTF-8.
Args:
text: Texto a limpiar
Returns:
str: Texto limpio y normalizado
"""
if not text:
return text
# Caracteres especiales a eliminar (incluyendo los bloques y otros caracteres problemáticos)
special_chars = ["▌", "\u2588", "\u2580", "\u2584", "\u258C", "\u2590"]
# Eliminar caracteres especiales
for char in special_chars:
text = text.replace(char, "")
# Normalizar espacios y saltos de línea
text = re.sub(r'\s+', ' ', text).strip()
# Asegurar codificación UTF-8
try:
text = text.encode('utf-8', errors='strict').decode('utf-8')
except UnicodeError:
text = text.encode('utf-8', errors='ignore').decode('utf-8')
logger.warning("Se encontraron caracteres no UTF-8 en el texto")
return text
def get_chat_history(username: str, analysis_type: str = 'sidebar', limit: int = None) -> list:
"""
Recupera el historial del chat con codificación UTF-8 segura.
"""
try:
query = {
"username": username,
"analysis_type": analysis_type
}
collection = get_collection(COLLECTION_NAME)
if collection is None:
logger.error("No se pudo obtener la colección de chat")
return []
cursor = collection.find(query).sort("timestamp", -1)
if limit:
cursor = cursor.limit(limit)
conversations = []
for chat in cursor:
try:
# Limpiar y asegurar UTF-8 en cada mensaje
cleaned_messages = []
for msg in chat.get('messages', []):
try:
cleaned_messages.append({
'role': msg.get('role', 'unknown'),
'content': clean_text_content(msg.get('content', ''))
})
except Exception as msg_error:
logger.error(f"Error procesando mensaje: {str(msg_error)}")
continue
conversations.append({
'timestamp': chat['timestamp'],
'messages': cleaned_messages
})
except Exception as e:
logger.error(f"Error formateando chat: {str(e)}")
continue
return conversations
except Exception as e:
logger.error(f"Error al recuperar historial de chat: {str(e)}")
return []
def store_chat_history(username: str, messages: list, analysis_type: str = 'sidebar', metadata: dict = None) -> bool:
"""
Guarda el historial del chat con codificación UTF-8 segura.
"""
try:
collection = get_collection(COLLECTION_NAME)
if collection is None:
logger.error("No se pudo obtener la colección de chat")
return False
# Limpiar y formatear cada mensaje
formatted_messages = []
for msg in messages:
try:
formatted_messages.append({
'role': msg.get('role', 'unknown'),
'content': clean_text_content(msg.get('content', '')),
'timestamp': datetime.now(timezone.utc).isoformat()
})
except Exception as msg_error:
logger.error(f"Error procesando mensaje para almacenar: {str(msg_error)}")
continue
chat_document = {
'username': username,
'timestamp': datetime.now(timezone.utc).isoformat(),
'messages': formatted_messages,
'analysis_type': analysis_type,
'metadata': metadata or {}
}
# Verificación adicional de UTF-8 antes de insertar
try:
import json
json.dumps(chat_document, ensure_ascii=False)
except UnicodeEncodeError as e:
logger.error(f"Error de codificación en documento: {str(e)}")
return False
result = collection.insert_one(chat_document)
if result.inserted_id:
logger.info(f"Chat guardado para {username} con ID: {result.inserted_id}")
return True
logger.error("No se pudo insertar el documento")
return False
except Exception as e:
logger.error(f"Error al guardar historial: {str(e)}")
return False |