Upload 9 files

app_whatsapp.py

@@ -0,0 +1,33 @@
+# app_whatsapp.py
+from fastapi import FastAPI, Form
+from fastapi.responses import PlainTextResponse
+from core_pipeline import procesar_mensaje
+from config import logger
+
+app = FastAPI(title="Asistente Financiero WhatsApp")
+
+# Twilio manda POST x-www-form-urlencoded a este endpoint
+# Configura tu webhook en Twilio:  https://TU-SERVIDOR/ngrok/etc/whatsapp
+
+
+@app.post("/whatsapp", response_class=PlainTextResponse)
+async def whatsapp_webhook(
+    Body: str = Form(...),
+    From: str = Form(None),
+    WaId: str = Form(None),
+):
+    """
+    Webhook de Twilio WhatsApp.
+    Body = mensaje de texto
+    From = número del usuario (whatsapp:+51...)
+    WaId = ID de WhatsApp del usuario
+    """
+    logger.info("===== WhatsApp WEBHOOK =====")
+    logger.info("From: %s | WaId: %s | Body: %s", From, WaId, Body)
+
+    resultado = procesar_mensaje(Body)
+    respuesta_texto = resultado["respuesta"]
+
+    # Respondemos en texto plano (Twilio lo acepta),
+    # si quieres TwiML puedes devolver XML.
+    return respuesta_texto

config.py

@@ -0,0 +1,31 @@
+# config.py
+import os
+from dotenv import load_dotenv
+import logging
+
+# Carga variables de entorno desde .env
+load_dotenv()
+
+# ========== LOGGING ==========
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
+logging.basicConfig(
+    level=LOG_LEVEL,
+    format="%(asctime)s | %(name)s | %(levelname)s | %(message)s"
+)
+
+logger = logging.getLogger("finanzas_app")
+
+# ========== TWILIO ==========
+TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")  # para validar firma si quieres
+TWILIO_WHATSAPP_NUMBER = os.getenv("TWILIO_WHATSAPP_NUMBER", "whatsapp:+14155238886")
+
+# ========== SUPABASE ==========
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY")
+SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", SUPABASE_ANON_KEY)
+
+# Usuario por defecto (tu UUID de la tabla usuarios)
+DEFAULT_USER_ID = os.getenv(
+    "DEFAULT_USER_ID",
+    "c6f4a4b6-1234-45ab-b0a2-88ac4ed4d111"
+)

core_pipeline.py

@@ -0,0 +1,89 @@
+# core_pipeline.py
+from typing import Dict, Any
+from config import logger
+from nlp_intent import predecir_intencion
+from nlp_ner import extraer_entidades
+from nlp_category import predecir_categoria
+from db_supabase import insertar_gasto, insertar_ingreso
+import re
+
+
+def _parse_monto_str(monto_str: str) -> float:
+    # Extrae número de algo como "50", "50.00", "S/ 50.90"
+    if not monto_str:
+        return 0.0
+    numeros = re.findall(r"\d+[.,]?\d*", monto_str)
+    if not numeros:
+        return 0.0
+    valor = numeros[0].replace(",", ".")
+    try:
+        return float(valor)
+    except ValueError:
+        return 0.0
+
+
+def procesar_mensaje(texto: str) -> Dict[str, Any]:
+    """
+    Pipeline completo:
+      1. Predice intención
+      2. Extrae entidades (monto, fecha)
+      3. Predice categoría usando SetFit con el TEXTO COMPLETO
+      4. Aplica acción en Supabase según intención
+      5. Devuelve dict con info + mensaje para usuario
+    """
+    logger.info("==== Procesando mensaje ====")
+    logger.info("Texto: %s", texto)
+
+    # 1. INTENCIÓN
+    intencion = predecir_intencion(texto)
+
+    # 2. ENTIDADES
+    ents = extraer_entidades(texto)
+    monto = _parse_monto_str(ents.get("monto"))
+    fecha = ents.get("fecha")
+
+    # 3. CATEGORÍA (texto completo, no la categoria_texto de NER)
+    categoria_final = predecir_categoria(texto)
+
+    # 4. LÓGICA DE NEGOCIO
+    respuesta = ""
+
+    if intencion == "agregar_gasto":
+        insertar_gasto(
+            monto=monto,
+            categoria_str=categoria_final,
+            fecha_str=fecha,
+            descripcion=texto
+        )
+        respuesta = f"Anoté un gasto de S/ {monto:.2f} en la categoría '{categoria_final}'."
+
+    elif intencion == "agregar_ingreso":
+        insertar_ingreso(
+            monto=monto,
+            categoria_ingreso=categoria_final,
+            fecha_str=fecha,
+            descripcion=texto
+        )
+        respuesta = f"Registré un ingreso de S/ {monto:.2f} como '{categoria_final}'."
+
+    elif intencion == "agregar_aporte":
+        # TODO: integrar con metas_ahorro
+        respuesta = "Detecté que quieres registrar un aporte a una meta de ahorro. Aún no he sido conectado a metas_ahorro 😅."
+
+    elif intencion == "puedo_gastar":
+        # TODO: leer presupuestos_mensuales y responder según límites
+        respuesta = "Según tu presupuesto, todavía no tengo conectada la lógica para validar si puedes gastar eso 😅, pero la intención está detectada."
+
+    else:
+        respuesta = f"Detecté intención '{intencion}' con categoría '{categoria_final}', pero aún no tengo lógica asociada."
+
+    logger.info("[PIPELINE] Resultado: intencion=%s, monto=%s, categoria=%s, fecha=%s",
+                intencion, monto, categoria_final, fecha)
+
+    return {
+        "intencion": intencion,
+        "monto": monto,
+        "categoria_final": categoria_final,
+        "fecha": fecha,
+        "respuesta": respuesta,
+    }

db_supabase.py

@@ -0,0 +1,95 @@
+# db_supabase.py
+from supabase import create_client, Client
+from datetime import date
+from typing import Optional
+from config import SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, DEFAULT_USER_ID, logger
+
+supabase: Client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
+
+
+def get_or_create_categoria(nombre: str, tipo: str = "gasto") -> Optional[int]:
+    """
+    Busca una categoría por nombre (case insensitive).
+    Si no existe, la crea.
+    Devuelve id_categoria o None si algo falla.
+    """
+    logger.info("[DB] get_or_create_categoria: %s (%s)", nombre, tipo)
+
+    try:
+        res = (
+            supabase.table("categorias")
+            .select("id_categoria")
+            .eq("id_usuario", DEFAULT_USER_ID)
+            .ilike("nombre_categoria", nombre)
+            .execute()
+        )
+        data = res.data or []
+        if data:
+            return data[0]["id_categoria"]
+
+        # Crear nueva categoría
+        insert_res = (
+            supabase.table("categorias")
+            .insert(
+                {
+                    "id_usuario": DEFAULT_USER_ID,
+                    "nombre_categoria": nombre,
+                    "tipo_categoria": "ingreso" if tipo.startswith("ingreso") else "gasto",
+                    "descripcion": f"Creada automáticamente para {nombre}",
+                }
+            )
+            .execute()
+        )
+        return insert_res.data[0]["id_categoria"]
+    except Exception as e:
+        logger.error("[DB] Error get_or_create_categoria: %s", e, exc_info=True)
+        return None
+
+
+def insertar_gasto(monto: float, categoria_str: str, fecha_str: Optional[str], descripcion: str):
+    logger.info("[DB] Insertar gasto: monto=%s, cat=%s, fecha=%s", monto, categoria_str, fecha_str)
+
+    id_categoria = get_or_create_categoria(categoria_str, tipo="gasto")
+
+    payload = {
+        "id_usuario": DEFAULT_USER_ID,
+        "id_categoria": id_categoria,
+        "id_tipo": 2,  # por ahora, por defecto "Variable"
+        "monto": monto,
+        "descripcion": descripcion,
+    }
+
+    if fecha_str:
+        payload["fecha"] = fecha_str
+    else:
+        payload["fecha"] = str(date.today())
+
+    try:
+        supabase.table("gastos").insert(payload).execute()
+        logger.info("[DB] Gasto insertado correctamente.")
+    except Exception as e:
+        logger.error("[DB] Error al insertar gasto: %s", e, exc_info=True)
+
+
+def insertar_ingreso(monto: float, categoria_ingreso: str, fecha_str: Optional[str], descripcion: str):
+    logger.info("[DB] Insertar ingreso: monto=%s, cat=%s, fecha=%s", monto, categoria_ingreso, fecha_str)
+
+    id_categoria = get_or_create_categoria(categoria_ingreso, tipo="ingreso")
+
+    payload = {
+        "id_usuario": DEFAULT_USER_ID,
+        "id_categoria": id_categoria,
+        "monto": monto,
+        "descripcion": descripcion,
+    }
+
+    if fecha_str:
+        payload["fecha"] = fecha_str
+    else:
+        payload["fecha"] = str(date.today())
+
+    try:
+        supabase.table("ingresos").insert(payload).execute()
+        logger.info("[DB] Ingreso insertado correctamente.")
+    except Exception as e:
+        logger.error("[DB] Error al insertar ingreso: %s", e, exc_info=True)

dockerfile

@@ -0,0 +1,41 @@
+# ---------------------------------------------------
+# Base image: Python optimizado para ML + CPU
+# ---------------------------------------------------
+FROM python:3.10-slim
+
+# ---------------------------------------------------
+# Instalar dependencias del sistema
+# ---------------------------------------------------
+RUN apt-get update && apt-get install -y \
+    git \
+    libsndfile1 \
+    && rm -rf /var/lib/apt/lists/*
+
+# ---------------------------------------------------
+# Crear directorio de la app
+# ---------------------------------------------------
+WORKDIR /app
+
+# ---------------------------------------------------
+# Copiar requirements.txt e instalar dependencias
+# ---------------------------------------------------
+COPY requirements.txt /app/requirements.txt
+
+RUN pip install --upgrade pip
+RUN pip install -r /app/requirements.txt
+
+# ---------------------------------------------------
+# Copiar TODO el proyecto dentro del contenedor
+# ---------------------------------------------------
+COPY . /app
+
+# ---------------------------------------------------
+# Puerto para HuggingFace Spaces
+# ---------------------------------------------------
+EXPOSE 7860
+
+# ---------------------------------------------------
+# Command para iniciar FastAPI en HF Spaces
+# HuggingFace espera que corras en el puerto 7860
+# ---------------------------------------------------
+CMD ["uvicorn", "app_whatsapp:app", "--host", "0.0.0.0", "--port", "7860"]

nlp_category.py

@@ -0,0 +1,71 @@
+# nlp_category.py
+import os
+import logging
+import unicodedata
+from setfit import SetFitModel
+
+logger = logging.getLogger(__name__)
+
+# ======================
+# 1. Cargar modelo SetFit
+# ======================
+
+SETFIT_MODEL_PATH = "Alecit1234/modelo_finanzas_peru_v1"
+
+logger.info("Cargando modelo SetFit desde HuggingFace...")
+
+model_setfit = SetFitModel.from_pretrained(
+    SETFIT_MODEL_PATH,
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+logger.info("Modelo SetFit cargado correctamente.")
+
+
+# ===========================
+# 2. Categorías (Label Map)
+# ===========================
+
+CATEGORY_LABEL_MAP = {
+    0:"comida", 1:"supermercado", 2:"transporte", 3:"taxi", 4:"entretenimiento",
+    5:"educacion", 6:"tecnologia", 7:"servicios", 8:"fitness", 9:"imprevistos",
+    10:"delivery", 11:"mascotas", 12:"familia", 13:"salud",
+    14:"ingreso_beca", 15:"ingreso_trabajo", 16:"ingreso_familia",
+    17:"ingreso_venta", 18:"ingreso_freelance", 19:"ingreso_extra"
+}
+
+
+# ======================
+# 3. Normalizar texto
+# ======================
+
+def _normalizar(texto: str) -> str:
+    texto = texto.lower()
+    texto = unicodedata.normalize("NFD", texto)
+    return "".join(c for c in texto if unicodedata.category(c) != "Mn")
+
+
+# ======================
+# 4. Predicción categoría
+# ======================
+
+def predecir_categoria(texto: str) -> str:
+    """
+    Predice categoría usando SOLO el texto completo.
+    NER aporta monto y fecha, pero NO categoría.
+    """
+
+    if not texto or texto.strip() == "":
+        logger.warning("Texto vacío recibido en predecir_categoria()")
+        return "otros"
+
+    texto_norm = _normalizar(texto)
+
+    logger.debug(f"[SETFIT] Texto normalizado: {texto_norm}")
+
+    pred_id = model_setfit.predict([texto_norm])[0].item()
+    categoria = CATEGORY_LABEL_MAP[pred_id]
+
+    logger.info(f"[SETFIT] Categoría predicha: {categoria} (id={pred_id})")
+
+    return categoria

nlp_intent.py

@@ -0,0 +1,62 @@
+# nlp_intent.py
+import os
+import json
+import torch
+import logging
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+from huggingface_hub import hf_hub_download
+
+logger = logging.getLogger(__name__)
+
+# ========== RUTA DEL MODELO ==========
+INTENT_MODEL_PATH = "Alecit1234/modelo_intenciones"
+
+logger.info("Cargando modelo de intenciones desde HuggingFace Hub: %s", INTENT_MODEL_PATH)
+
+# ========== CARGA DEL MODELO ==========
+tokenizer_int = AutoTokenizer.from_pretrained(
+    INTENT_MODEL_PATH,
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+model_int = AutoModelForSequenceClassification.from_pretrained(
+    INTENT_MODEL_PATH,
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+# ========== LABEL MAP ==========
+label_map_path = hf_hub_download(
+    repo_id=INTENT_MODEL_PATH,
+    filename="label_map.json",
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+with open(label_map_path, "r", encoding="utf-8") as f:
+    INTENT_LABEL_MAP = json.load(f)
+
+
+# ========== FUNCIÓN PRINCIPAL ==========
+def predecir_intencion(texto: str) -> str:
+    """Predice intención de un texto usando modelo de clasificación."""
+    if not texto or texto.strip() == "":
+        logger.warning("[INTENT] Texto vacío recibido. Asignando intención 'otros'.")
+        return "otros"
+
+    logger.debug("[INTENT] Texto de entrada: %s", texto)
+
+    inputs = tokenizer_int(
+        texto,
+        return_tensors="pt",
+        truncation=True,
+        max_length=64,
+        padding="max_length"
+    )
+
+    with torch.no_grad():
+        logits = model_int(**inputs).logits
+        pred_id = torch.argmax(logits, dim=1).item()
+
+    intent = INTENT_LABEL_MAP[str(pred_id)]
+
+    logger.info("[INTENT] Predicción: %s (id=%s)", intent, pred_id)
+    return intent

nlp_ner.py

@@ -0,0 +1,100 @@
+# nlp_ner.py
+import os
+import json
+import logging
+import torch
+from transformers import AutoTokenizer, AutoModelForTokenClassification
+from huggingface_hub import hf_hub_download
+
+logger = logging.getLogger(__name__)
+
+# =====================================
+# 1. NOMBRE DEL MODELO EN HUGGINGFACE
+# =====================================
+NER_MODEL_PATH = "Alecit1234/modelo_ner"
+
+logger.info("Cargando modelo NER desde HuggingFace Hub: %s", NER_MODEL_PATH)
+
+# =====================================
+# 2. CARGAR TOKENIZER & MODEL
+# =====================================
+tokenizer_ner = AutoTokenizer.from_pretrained(
+    NER_MODEL_PATH,
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+model_ner = AutoModelForTokenClassification.from_pretrained(
+    NER_MODEL_PATH,
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+# =====================================
+# 3. CARGAR LABEL_MAP DESDE HF
+# =====================================
+label_map_path = hf_hub_download(
+    repo_id=NER_MODEL_PATH,
+    filename="label_map.json",
+    use_auth_token=os.getenv("HUGGINGFACE_AUTH_TOKEN")
+)
+
+with open(label_map_path, "r", encoding="utf-8") as f:
+    NER_LABELS = json.load(f)
+
+logger.info("Etiquetas NER cargadas: %s", NER_LABELS)
+
+
+# =====================================
+# 4. FUNCIÓN: EXTRAER ENTIDADES
+# =====================================
+def extraer_entidades(texto: str) -> dict:
+    """
+    Extrae entidades: monto, fecha, categoria_texto (solo referencia)
+    """
+
+    logger.debug("[NER] Procesando texto: %s", texto)
+
+    inputs = tokenizer_ner(
+        texto,
+        return_tensors="pt",
+        truncation=True,
+        max_length=64
+    )
+
+    with torch.no_grad():
+        outputs = model_ner(**inputs)
+
+    preds = outputs.logits.argmax(dim=-1)[0].tolist()
+    tokens = tokenizer_ner.convert_ids_to_tokens(inputs["input_ids"][0])
+
+    entidades = {"monto": None, "categoria_texto": None, "fecha": None}
+
+    palabra = ""
+    tipo_actual = None
+
+    for tok, pred_id in zip(tokens, preds):
+        label = NER_LABELS[str(pred_id)]
+
+        # Cuando cambia la etiqueta
+        if label == "O":
+            if tipo_actual and palabra:
+                entidades[tipo_actual] = palabra
+            palabra = ""
+            tipo_actual = None
+            continue
+
+        # Asignación del tipo
+        if label == "MONEY":
+            tipo_actual = "monto"
+        elif label == "CATEGORY":
+            tipo_actual = "categoria_texto"
+        elif label == "DATE":
+            tipo_actual = "fecha"
+
+        palabra += tok.replace("▁", "")
+
+    # Último token acumulado
+    if tipo_actual and palabra:
+        entidades[tipo_actual] = palabra
+
+    logger.info("[NER] Resultado entidades: %s", entidades)
+    return entidades

requirements.txt

@@ -0,0 +1,17 @@
+fastapi==0.110.0
+uvicorn[standard]==0.29.0
+python-dotenv==1.0.1
+
+supabase==2.4.3
+
+transformers==4.38.2
+huggingface-hub==0.20.3
+sentence-transformers==2.2.2
+setfit==1.0.3
+
+torch==2.0.1
+numpy
+pydantic>=1.10
+requests
+python-multipart
+accelerate