feat: implement financial unpivot ingestion endpoint and add new dimension and fact tables to schema

app.py

@@ -4,7 +4,10 @@ import logging
 import re
 from fastapi import FastAPI, Depends, HTTPException, status, Query
 from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, ConfigDict
+import rapidfuzz
+import pandas as pd
+from typing import List, Optional, Dict, Any
 from sqlalchemy.orm import Session
 from database import (
     get_db,
@@ -584,6 +587,83 @@ def procesar_registro_financiero(payload: FinancePayload, db: Session = Depends(
         db.rollback()
         raise HTTPException(status_code=500, detail=str(e))
 
+class UnpivotFinancePayload(BaseModel):
+    id_estudiante: int
+    raw_data: Dict[str, Any]
+
+@app.post("/api/v1/ingest/finance_unpivot", status_code=status.HTTP_201_CREATED)
+def procesar_lote_financiero_unpivot(payloads: List[UnpivotFinancePayload], db: Session = Depends(get_db)):
+    """
+    Recibe un lote de datos financieros con columnas de meses (ej. 'MONTO ENERO 2024')
+    y utiliza pandas para despivotarlos antes de insertarlos en FactCobranzasProyectadas.
+    """
+    try:
+        from database import FactCobranzasProyectadas, DimTiempo
+        
+        # 1. Convertir payloads a DataFrame
+        df_list = []
+        for p in payloads:
+            row = p.raw_data.copy()
+            row['id_estudiante'] = p.id_estudiante
+            df_list.append(row)
+            
+        if not df_list:
+            return {"status": "success", "inserted": 0}
+            
+        df = pd.DataFrame(df_list)
+        
+        # 2. Identificar columnas de meses (Empiezan con 'MONTO ')
+        monto_cols = [c for c in df.columns if c.startswith('MONTO ')]
+        id_cols = [c for c in df.columns if c not in monto_cols]
+        
+        # 3. Despivotar (Melt)
+        df_melted = df.melt(id_vars=id_cols, value_vars=monto_cols, var_name='mes_anio', value_name='monto_esperado')
+        
+        # Filtrar nulos o ceros si no son necesarios
+        df_melted['monto_esperado'] = pd.to_numeric(df_melted['monto_esperado'], errors='coerce')
+        df_melted = df_melted.dropna(subset=['monto_esperado'])
+        
+        # 4. Insertar en base de datos
+        inserted_count = 0
+        for index, row in df_melted.iterrows():
+            mes_raw = str(row['mes_anio']).replace('MONTO ', '').strip() # Ej 'ENERO 2024'
+            parts = mes_raw.split()
+            gestion = int(parts[1]) if len(parts) > 1 else 2024
+            mes_str = parts[0] if len(parts) > 0 else 'Enero'
+            
+            # Buscar o crear tiempo
+            tiempo = db.query(DimTiempo).filter(DimTiempo.gestion == gestion, DimTiempo.mes == mes_str).first()
+            if not tiempo:
+                tiempo = DimTiempo(gestion=gestion, mes=mes_str)
+                db.add(tiempo)
+                db.commit()
+                db.refresh(tiempo)
+                
+            nuevo_cobro = FactCobranzasProyectadas(
+                id_estudiante=row['id_estudiante'],
+                id_tiempo=tiempo.id_tiempo,
+                monto_esperado=row['monto_esperado'],
+                estado_pago='PROYECTADO'
+            )
+            db.add(nuevo_cobro)
+            inserted_count += 1
+            
+        db.commit()
+        return {"status": "success", "inserted": inserted_count}
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/api/v1/ingest/surveys", status_code=status.HTTP_201_CREATED)
+def procesar_lote_encuestas(payloads: List[Dict[str, Any]], db: Session = Depends(get_db)):
+    """Ruta para encuestas (Placeholder para lógica de NLP sobre comentarios)"""
+    return {"status": "success", "message": "Ruta de encuestas lista para implementación"}
+
+@app.post("/api/v1/ingest/marketing", status_code=status.HTTP_201_CREATED)
+def procesar_lote_marketing(payloads: List[Dict[str, Any]], db: Session = Depends(get_db)):
+    """Ruta para OKRs de Marketing (Placeholder)"""
+    return {"status": "success", "message": "Ruta de marketing lista para implementación"}
+
 @app.post("/api/v1/ingesta/financiera/bulk", status_code=status.HTTP_201_CREATED)
 def procesar_lote_financiero(payloads: List[FinancePayload], db: Session = Depends(get_db), current_user: Users = Depends(get_current_user)):
     if current_user.role not in ["analista_datos_marketing", "admin"]:

database.py

@@ -55,6 +55,13 @@ class DimEstudiante(Base):
     id_estudiante     = Column(Integer, primary_key=True, index=True)
     nombre_completo   = Column(String(200), nullable=False)
     codigo_estudiante = Column(String(50))
+    genero            = Column(String(20))
+    fecha_nacimiento  = Column(DateTime)
+    pais              = Column(String(100))
+    ciudad            = Column(String(100))
+    nivel_academico   = Column(String(100))
+    carrera           = Column(String(200))
+    institucion_origen= Column(String(200))
 
 
 class DimDocente(Base):
@@ -101,6 +108,13 @@ class LogAuditoriaNlp(Base):
     usuario_auditor = Column(Integer, ForeignKey("users.id"))
     created_at = Column(DateTime, default=datetime.datetime.utcnow)
 
+class DimCategoriaFinanciera(Base):
+    """Nueva dimension para Egresos, Ingresos y OKRs Financieros"""
+    __tablename__ = "dim_categoria_financiera"
+    id_categoria = Column(Integer, primary_key=True)
+    nombre_categoria = Column(String(200), nullable=False)
+    tipo = Column(String(50)) # 'INGRESO', 'EGRESO', 'RENTABILIDAD', 'EBITDA'
+
 
 # =============================================================================
 # TABLAS DE HECHOS
@@ -134,6 +148,49 @@ class FactSituacionFinanciera(Base):
     tipo_alerta    = Column(String(20))
     fecha_registro = Column(DateTime, default=datetime.datetime.utcnow)
 
+class FactCobranzasProyectadas(Base):
+    """Almacena montos despivotados de ingresos mensuales proyectados."""
+    __tablename__ = "fact_cobranzas_proyectadas"
+    id_hecho_cobro = Column(Integer, primary_key=True, index=True)
+    id_estudiante  = Column(Integer, ForeignKey("dim_estudiante.id_estudiante"))
+    id_tiempo      = Column(Integer, ForeignKey("dim_tiempo.id_tiempo")) # Referencia al mes proyectado
+    monto_esperado = Column(Numeric(10, 2))
+    estado_pago    = Column(String(50)) # Ej. PAGADO, PENDIENTE
+    created_at     = Column(DateTime, default=datetime.datetime.utcnow)
+
+class FactEvaluacionDocente(Base):
+    """Respuestas a encuestas de satisfacción docente y NPS."""
+    __tablename__ = "fact_evaluacion_docente"
+    id_hecho_eval = Column(Integer, primary_key=True)
+    id_docente    = Column(Integer, ForeignKey("dim_docente.id_docente"))
+    id_modulo     = Column(Integer, ForeignKey("dim_modulo.id_modulo"))
+    id_estudiante = Column(Integer, ForeignKey("dim_estudiante.id_estudiante"), nullable=True) # A veces anonimo
+    id_tiempo     = Column(Integer, ForeignKey("dim_tiempo.id_tiempo"))
+    pregunta_bloque = Column(String(500))
+    puntuacion    = Column(Numeric(4, 2)) # Ej. 1 al 5
+    comentario    = Column(String(1000))
+
+class FactMarketingInscripciones(Base):
+    """Métricas de OKRs de Marketing y ventas."""
+    __tablename__ = "fact_marketing_inscripciones"
+    id_hecho_mkt  = Column(Integer, primary_key=True)
+    id_modulo     = Column(Integer, ForeignKey("dim_modulo.id_modulo"))
+    id_tiempo     = Column(Integer, ForeignKey("dim_tiempo.id_tiempo"))
+    leads         = Column(Integer, default=0)
+    reservas      = Column(Integer, default=0)
+    inscritos     = Column(Integer, default=0)
+    costo_programa= Column(Numeric(12, 2))
+
+class FactRentabilidadPresupuesto(Base):
+    """Indicadores de Rentabilidad, Egresos y EBITDA (Ejecutado vs Meta)."""
+    __tablename__ = "fact_rentabilidad_presupuesto"
+    id_hecho_rent = Column(Integer, primary_key=True)
+    id_modulo     = Column(Integer, ForeignKey("dim_modulo.id_modulo"))
+    id_tiempo     = Column(Integer, ForeignKey("dim_tiempo.id_tiempo"))
+    id_categoria  = Column(Integer, ForeignKey("dim_categoria_financiera.id_categoria"))
+    monto_ejecutado = Column(Numeric(14, 2))
+    monto_meta      = Column(Numeric(14, 2))
+
 
 def get_db():
     """