File size: 3,160 Bytes
3657e14 |
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 |
# modules/database/semantic_mongo_live_db.py
import logging
from datetime import datetime, timezone
import base64
# Importaciones locales
from .mongo_db import get_collection, insert_document, find_documents
logger = logging.getLogger(__name__)
COLLECTION_NAME = 'student_semantic_live_analysis'
def store_student_semantic_live_result(username, text, analysis_result, lang_code='en'):
"""
Guarda el resultado del análisis semántico en vivo en MongoDB.
"""
try:
if not username or not text or not analysis_result:
logger.error("Datos insuficientes para guardar el análisis")
return False
# Preparar el gráfico conceptual
concept_graph_data = None
if 'concept_graph' in analysis_result and analysis_result['concept_graph'] is not None:
try:
if isinstance(analysis_result['concept_graph'], bytes):
concept_graph_data = base64.b64encode(analysis_result['concept_graph']).decode('utf-8')
else:
logger.warning("El gráfico conceptual no está en formato bytes")
except Exception as e:
logger.error(f"Error al codificar gráfico conceptual: {str(e)}")
# Crear documento para MongoDB
analysis_document = {
'username': username,
'timestamp': datetime.now(timezone.utc),
'text': text,
'analysis_type': 'semantic_live',
'key_concepts': analysis_result.get('key_concepts', []),
'concept_centrality': analysis_result.get('concept_centrality', {}),
'concept_graph': concept_graph_data,
'language': lang_code
}
# Insertar en MongoDB
result = insert_document(COLLECTION_NAME, analysis_document)
if result:
logger.info(f"Análisis semántico en vivo guardado para {username}")
return True
logger.error("No se pudo insertar el documento en MongoDB")
return False
except Exception as e:
logger.error(f"Error al guardar el análisis semántico en vivo: {str(e)}")
return False
def get_student_semantic_live_analysis(username, limit=10):
"""
Recupera los análisis semánticos en vivo de un estudiante.
"""
try:
query = {
"username": username,
"analysis_type": "semantic_live"
}
projection = {
"timestamp": 1,
"text": 1,
"key_concepts": 1,
"concept_graph": 1,
"_id": 1
}
results = find_documents(
COLLECTION_NAME,
query,
projection=projection,
sort=[("timestamp", -1)],
limit=limit
)
logger.info(f"Recuperados {len(results)} análisis semánticos en vivo para {username}")
return results
except Exception as e:
logger.error(f"Error recuperando análisis semántico en vivo: {str(e)}")
return []
__all__ = [
'store_student_semantic_live_result',
'get_student_semantic_live_analysis'
] |