File size: 3,030 Bytes
50415aa |
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 |
# database.py
import logging
from pymongo import MongoClient
import certifi
from datetime import datetime
logger = logging.getLogger(__name__)
# Variables globales
mongo_client = None
db = None
analysis_collection = None
def initialize_mongodb_connection():
global mongo_client, db, analysis_collection
try:
cosmos_mongodb_connection_string = os.getenv("MONGODB_CONNECTION_STRING")
if not cosmos_mongodb_connection_string:
logger.error("La variable de entorno MONGODB_CONNECTION_STRING no está configurada")
return False
mongo_client = MongoClient(cosmos_mongodb_connection_string,
tls=True,
tlsCAFile=certifi.where(),
retryWrites=False,
serverSelectionTimeoutMS=5000,
connectTimeoutMS=10000,
socketTimeoutMS=10000)
mongo_client.admin.command('ping')
db = mongo_client['aideatext_db']
analysis_collection = db['text_analysis']
logger.info("Conexión a Cosmos DB MongoDB API exitosa")
return True
except Exception as e:
logger.error(f"Error al conectar con Cosmos DB MongoDB API: {str(e)}", exc_info=True)
return False
def get_student_data(username):
if analysis_collection is None:
logger.error("La conexión a MongoDB no está inicializada")
return None
try:
# Buscar los datos del estudiante
cursor = analysis_collection.find({"username": username}).sort("timestamp", -1)
# Formatear los datos
formatted_data = {
"username": username,
"entries": [],
"entries_count": 0,
"word_count": {},
"arc_diagrams": [],
"network_diagrams": []
}
for entry in cursor:
formatted_data["entries"].append({
"timestamp": entry["timestamp"].isoformat(),
"text": entry["text"]
})
formatted_data["entries_count"] += 1
# Agregar conteo de palabras
for category, count in entry.get("word_count", {}).items():
if category in formatted_data["word_count"]:
formatted_data["word_count"][category] += count
else:
formatted_data["word_count"][category] = count
# Agregar diagramas
formatted_data["arc_diagrams"].extend(entry.get("arc_diagrams", []))
formatted_data["network_diagrams"].append(entry.get("network_diagram", ""))
return formatted_data if formatted_data["entries_count"] > 0 else None
except Exception as e:
logger.error(f"Error al obtener datos del estudiante {username}: {str(e)}")
return None
# Aquí irían las demás funciones relacionadas con la base de datos... |