import os import datetime from dotenv import load_dotenv from sqlalchemy import create_engine, Column, Integer, String, Numeric, Boolean, DateTime, ForeignKey, CheckConstraint from sqlalchemy.orm import declarative_base, sessionmaker # Load environment variables from .env file load_dotenv() DATABASE_URL = os.getenv("DATABASE_URL") if not DATABASE_URL: DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/postgres" # For Supabase, pool_pre_ping=True is highly recommended to handle stale connections engine = create_engine( DATABASE_URL, pool_pre_ping=True, pool_recycle=3600 ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # ============================================================================= # DIMENSIONES — mapeadas exactamente al DDL de Supabase # ============================================================================= class DimTiempo(Base): __tablename__ = "dim_tiempo" id_tiempo = Column(Integer, primary_key=True) gestion = Column(Integer, nullable=False, default=2026) semestre = Column(Integer, default=1) mes = Column(String(20), default="Mayo") class DimEstudiante(Base): __tablename__ = "dim_estudiante" id_estudiante = Column(Integer, primary_key=True, index=True) nombre_completo = Column(String(200), nullable=False) codigo_estudiante = Column(String(50)) class DimDocente(Base): __tablename__ = "dim_docente" id_docente = Column(Integer, primary_key=True) nombre_completo = Column(String(200), nullable=False, default="Docente Generico") area_especialidad = Column(String(200), default="Generico") class DimModulo(Base): __tablename__ = "dim_modulo" id_modulo = Column(Integer, primary_key=True) nombre_modulo = Column(String(200), nullable=False, default="Modulo Generico") nombre_institucion = Column(String(200), nullable=False, default="GiraGroup") programa = Column(String(200), default="General") class DimOrigenDocumental(Base): """Mapeada a la tabla dim_origen_documental en Supabase.""" __tablename__ = "dim_origen_documental" id_documento = Column(Integer, primary_key=True) tipo_documento = Column(String(10), default="SHEET") nombre_archivo = Column(String(500), default="archivo_generico") fecha_procesamiento = Column(DateTime, default=datetime.datetime.utcnow) class Users(Base): """Tabla de usuarios/autenticación en Supabase.""" __tablename__ = "users" id = Column(Integer, primary_key=True) username = Column(String(100), unique=True, nullable=False, default="sistema") hashed_password = Column(String(255), nullable=False, default="$placeholder$") role = Column(String(20), default="admin") created_at = Column(DateTime, default=datetime.datetime.utcnow) # ============================================================================= # TABLAS DE HECHOS # ============================================================================= class FactRendimientoAcademico(Base): __tablename__ = "fact_rendimiento_academico" id_hecho_aca = Column(Integer, primary_key=True, index=True) id_estudiante = Column(Integer, ForeignKey("dim_estudiante.id_estudiante")) id_docente = Column(Integer, ForeignKey("dim_docente.id_docente")) id_modulo = Column(Integer, ForeignKey("dim_modulo.id_modulo")) id_tiempo = Column(Integer, ForeignKey("dim_tiempo.id_tiempo")) id_documento = Column(Integer, ForeignKey("dim_origen_documental.id_documento")) id_usuario_carga = Column(Integer, ForeignKey("users.id")) nota_final = Column(Numeric(5, 2)) asistencia_pct = Column(Numeric(5, 2)) incumplimiento_actividades_pct = Column(Numeric(5, 2), default=0.00) nivel_confianza_ia = Column(Numeric(5, 4)) requiere_revision = Column(Boolean, default=False) created_at = Column(DateTime, default=datetime.datetime.utcnow) class FactSituacionFinanciera(Base): __tablename__ = "fact_situacion_financiera" id_hecho_fin = Column(Integer, primary_key=True) id_estudiante = Column(Integer, ForeignKey("dim_estudiante.id_estudiante")) id_tiempo = Column(Integer, ForeignKey("dim_tiempo.id_tiempo")) monto_deuda = Column(Numeric(10, 2)) cuotas_impagas = Column(Integer) estado_cartera = Column(String(20)) tipo_alerta = Column(String(20)) fecha_registro = Column(DateTime, default=datetime.datetime.utcnow) def get_db(): """ Dependency injection helper to yield a database session. Guarantees that the session is closed after the request completes. """ db = SessionLocal() try: yield db finally: db.close()