File size: 4,267 Bytes
9ec37fb |
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 141 |
# modules/database/writing_progress_mongo_db.py
from .mongo_db import get_collection, insert_document
from datetime import datetime, timezone
import logging
logger = logging.getLogger(__name__)
COLLECTION_NAME = 'writing_progress'
def store_writing_baseline(username, metrics, text):
"""
Guarda la línea base de escritura de un usuario.
Args:
username: ID del usuario
metrics: Diccionario con métricas iniciales
text: Texto analizado
"""
try:
document = {
'username': username,
'type': 'baseline',
'metrics': metrics,
'text': text,
'timestamp': datetime.now(timezone.utc).isoformat(),
'iteration': 0 # Línea base siempre es iteración 0
}
# Verificar si ya existe una línea base
collection = get_collection(COLLECTION_NAME)
existing = collection.find_one({
'username': username,
'type': 'baseline'
})
if existing:
# Actualizar línea base existente
result = collection.update_one(
{'_id': existing['_id']},
{'$set': document}
)
success = result.modified_count > 0
else:
# Insertar nueva línea base
result = collection.insert_one(document)
success = result.inserted_id is not None
logger.info(f"Línea base {'actualizada' if existing else 'creada'} para usuario: {username}")
return success
except Exception as e:
logger.error(f"Error al guardar línea base: {str(e)}")
return False
def store_writing_progress(username, metrics, text):
"""
Guarda una nueva iteración de progreso.
"""
try:
# Obtener último número de iteración
collection = get_collection(COLLECTION_NAME)
last_progress = collection.find_one(
{'username': username},
sort=[('iteration', -1)]
)
next_iteration = (last_progress['iteration'] + 1) if last_progress else 1
document = {
'username': username,
'type': 'progress',
'metrics': metrics,
'text': text,
'timestamp': datetime.now(timezone.utc).isoformat(),
'iteration': next_iteration
}
result = collection.insert_one(document)
success = result.inserted_id is not None
if success:
logger.info(f"Progreso guardado para {username}, iteración {next_iteration}")
return success
except Exception as e:
logger.error(f"Error al guardar progreso: {str(e)}")
return False
def get_writing_baseline(username):
"""
Obtiene la línea base de un usuario.
"""
try:
collection = get_collection(COLLECTION_NAME)
return collection.find_one({
'username': username,
'type': 'baseline'
})
except Exception as e:
logger.error(f"Error al obtener línea base: {str(e)}")
return None
def get_writing_progress(username, limit=None):
"""
Obtiene el historial de progreso de un usuario.
Args:
username: ID del usuario
limit: Número máximo de registros a retornar
"""
try:
collection = get_collection(COLLECTION_NAME)
cursor = collection.find(
{
'username': username,
'type': 'progress'
},
sort=[('iteration', -1)]
)
if limit:
cursor = cursor.limit(limit)
return list(cursor)
except Exception as e:
logger.error(f"Error al obtener progreso: {str(e)}")
return []
def get_latest_writing_metrics(username):
"""
Obtiene las métricas más recientes (línea base o progreso).
"""
try:
collection = get_collection(COLLECTION_NAME)
return collection.find_one(
{'username': username},
sort=[('timestamp', -1)]
)
except Exception as e:
logger.error(f"Error al obtener métricas recientes: {str(e)}")
return None |