app.py

from dotenv import load_dotenv
from openai import APIConnectionError, AuthenticationError, OpenAI, OpenAIError
from enum import Enum
import json
import logging
import re
import sqlite3
import time
from functools import lru_cache
from pathlib import Path
from typing import Any
from pypdf import PdfReader
import psutil
import gradio as gr
from jinja2 import Template
import requests

import os
import threading
from collections import deque
from concurrent.futures import ThreadPoolExecutor

from openai import OpenAI

OPENAI_MODEL = 'gpt-4o-mini'
OLLAMA_MODEL = 'llama3.2:latest'
OPENAI_CLIENT = OpenAI()
OLLAMA_CLIENT = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
EMAIL_PATTERN = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
DB_PATH = Path("me/conversations.db")
PROFILE_PATH = Path("me/profile.pdf")
SERVER_PORT = 7863

load_dotenv(override=True)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

_pushover_user  = os.getenv("PUSHOVER_USER")
_pushover_token = os.getenv("PUSHOVER_TOKEN")
_pushover_url   = "https://api.pushover.net/1/messages.json"


class Provider(str, Enum):
  """Proveedores de modelos disponibles en la interfaz."""

  OPENAI = "OpenAI"
  OLLAMA = "Ollama"

MODELS: dict[str, dict[str, tuple[str, str]]] = {
    "evaluator": {
        "primary":  ("gpt",   "openai/gpt-oss-20b"),
        "fallback": ("ollama", "gpt-oss:20b-cloud"),
    },
    "agent": {
        "primary":  ("gpt",   "openai/gpt-oss-120b"),
        "fallback": ("ollama", "nemotron-3-super:120b-cloud"),
    },
    "rerun": {
        "primary":  ("ollama", "nemotron-3-super:120b-cloud"),
        "fallback": ("ollama", "gpt-oss:120b-cloud"),
    },
}

_GPT_LIMITS: dict[str, dict[str, int]] = {
    "meta-llama/llama-prompt-guard-2-22m": {"rpm": 30,  "rpd": 14_400},
    "openai/gpt-oss-20b":                  {"rpm": 30,  "rpd":  8_000},
    "openai/gpt-oss-120b":                 {"rpm": 30,  "rpd":  8_000},
}
_SAFETY_FACTOR = 0.80

class _RateLimitTracker:
    """

    Contabiliza llamadas Gpt por modelo con ventana deslizante (monotonic).

    Thread-safe mediante Lock.

    """

    def __init__(self) -> None:
        self._lock   = threading.Lock()
        self._rpm_ts: dict[str, deque] = {}
        self._rpd_ts: dict[str, deque] = {}

    def _ensure(self, model: str) -> None:
        if model not in self._rpm_ts:
            self._rpm_ts[model] = deque()
            self._rpd_ts[model] = deque()

    @staticmethod
    def _purge(dq: deque, window: float) -> None:
        cutoff = time.monotonic() - window
        while dq and dq[0] < cutoff:
            dq.popleft()

    def can_use(self, model: str) -> bool:
        """True si el modelo tiene margen dentro del 80% del límite."""
        limits = _GPT_LIMITS.get(model)
        if not limits:
            return True
        with self._lock:
            self._ensure(model)
            self._purge(self._rpm_ts[model], 60.0)
            self._purge(self._rpd_ts[model], 86_400.0)
            rpm_ok = len(self._rpm_ts[model]) < limits["rpm"] * _SAFETY_FACTOR
            rpd_ok = len(self._rpd_ts[model]) < limits["rpd"] * _SAFETY_FACTOR
            return rpm_ok and rpd_ok

    def record(self, model: str) -> None:
        """Registra una llamada completada."""
        if model not in _GPT_LIMITS:
            return
        with self._lock:
            self._ensure(model)
            ts = time.monotonic()
            self._rpm_ts[model].append(ts)
            self._rpd_ts[model].append(ts)

    def status(self) -> dict:
        """Estado actual de uso para logging/debug."""
        out = {}
        with self._lock:
            for model, limits in _GPT_LIMITS.items():
                self._purge(self._rpm_ts.get(model, deque()), 60.0)
                self._purge(self._rpd_ts.get(model, deque()), 86_400.0)
                out[model] = {
                    "rpm_used":  len(self._rpm_ts.get(model, [])),
                    "rpm_limit": limits["rpm"],
                    "rpd_used":  len(self._rpd_ts.get(model, [])),
                    "rpd_limit": limits["rpd"],
                }
        return out


_rate_tracker = _RateLimitTracker()

def init_database():
  DB_PATH.parent.mkdir(parents=True, exist_ok=True)
  with sqlite3.connect(DB_PATH) as connection:
    connection.execute(
        """

        CREATE TABLE IF NOT EXISTS contacts (

          id INTEGER PRIMARY KEY AUTOINCREMENT,

          email TEXT NOT NULL,

          name TEXT,

          notes TEXT,

          created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP

        )

        """
    )
    connection.execute(
        """

        CREATE TABLE IF NOT EXISTS unknown_questions (

          id INTEGER PRIMARY KEY AUTOINCREMENT,

          question TEXT NOT NULL,

          created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP

        )

        """
    )

def kill_processes_using_port(port):
  current_pid = psutil.Process().pid
  killed_processes = []

  for connection in psutil.net_connections(kind="inet"):
    if connection.status != psutil.CONN_LISTEN:
      continue
    if not connection.laddr or connection.laddr.port != port:
      continue
    if not connection.pid or connection.pid == current_pid:
      continue

    process = psutil.Process(connection.pid)
    logger.info(
        "Killing process using port %s: pid=%s name=%s",
        port,
        process.pid,
        process.name(),
    )
    process.terminate()
    killed_processes.append(process)

  _, alive = psutil.wait_procs(killed_processes, timeout=3)
  for process in alive:
    logger.warning("Process pid=%s did not terminate cleanly; killing it", process.pid)
    process.kill()

  if killed_processes:
    time.sleep(0.5)

def _push(message: str) -> None:
    logger.info("Pushover → %s", message)
    try:
        r = requests.post(
            _pushover_url,
            data={"user": _pushover_user, "token": _pushover_token, "message": message},
            timeout=5,
        )
        r.raise_for_status()
    except requests.RequestException as exc:
        logger.warning("Pushover error: %s", exc)


def record_user_details(

    email, name="Nombre no indicado", notes="no proporcionadas"

):
  with sqlite3.connect(DB_PATH) as connection:
    cursor = connection.execute(
        """

        INSERT INTO contacts (email, name, notes)

        VALUES (?, ?, ?)

        """,
        (email, name, notes),
    )
    _push(f"Contacto registrado → {name} | {email} | {notes}")
    return {"recorded": "ok"}
  logger.info("Contact recorded in SQLite with id=%s", cursor.lastrowid)
  return {"recorded": "ok", "id": cursor.lastrowid}


def record_unknown_question(question):
  with sqlite3.connect(DB_PATH) as connection:
    cursor = connection.execute(
        """

        INSERT INTO unknown_questions (question)

        VALUES (?)

        """,
        (question,),
    )
  _push(f"Pregunta sin respuesta → {question}")
  logger.info("Unknown question recorded in SQLite with id=%s", cursor.lastrowid)
  return {"recorded": "ok", "id": cursor.lastrowid}


record_user_details_json = {
    "name":
        "record_user_details",
    "description":
        "Utiliza esta herramienta solo cuando el usuario haya proporcionado explícitamente su dirección de correo electrónico en la conversación. Nunca inventes ni supongas direcciones de email.",
    "parameters":
        {
            "type": "object",
            "properties":
                {
                    "email":
                        {
                            "type": "string",
                            "description": "La dirección de email del usuario"
                        },
                    "name":
                        {
                            "type": "string",
                            "description": "El nombre del usuario, si se indica"
                        },
                    "notes":
                        {
                            "type":
                                "string",
                            "description":
                                "¿Alguna información adicional sobre la conversación que valga la pena registrar para dar contexto?"
                        }
                },
            "required": ["email"],
            "additionalProperties": False
        }
}

record_unknown_question_json = {
    "name":
        "record_unknown_question",
    "description":
        "Utiliza siempre esta herramienta para registrar cualquier pregunta que no haya podido responder porque no se sabía la respuesta.",
    "parameters":
        {
            "type": "object",
            "properties":
                {
                    "question":
                        {
                            "type": "string",
                            "description": "La pregunta no sabe responderse"
                        },
                },
            "required": ["question"],
            "additionalProperties": False
        }
}

tools: list[Any] = [
    {
        "type": "function",
        "function": record_user_details_json
    }, {
        "type": "function",
        "function": record_unknown_question_json
    }
]

init_database()

@lru_cache(maxsize=8)
def _load_template(path: str) -> Template:
    return Template(Path(path).read_text(encoding="utf-8"))


def _render_prompt(path: str, **kwargs) -> str:
    return _load_template(path).render(**kwargs)


def get_system_prompt() -> str:
    return _render_prompt("prompts/me.md")

_AGENT_CONTEXT_COMPACT = (
    "Andres Marcelo Villarreal — Senior FullStack Developer, Cordoba, Argentina. "
    "9+ años experiencia total. "

    # Experiencia verificable
    "Accenture (2021-2026): Full Stack Developer, "
    "Java, javascript, python, react, sql, SOA, docker, kubernetes, aws, azure, RabitMQ, "
    "Git, Bitbucket, Jira, Confluence, Azure DevOps, "

    "CDA Informatica (2019-2021): Full Stack, Java Spring, ReactJS, SQL. ESB, RabitMQ, "

    "Ayi & Asociados (2017-2019): Full Stack Developer, "
    "Java, javascript, python, react, sql, SOA, docker, kubernetes, aws, azure, "
    "Git, Bitbucket, Jira, Confluence, Azure DevOps, "

    "Harriague & Asociados (2015-2017): Full Stack, Java Spring, AngularJS, PL/SQL, "

    # Certificaciones y skills verificables
    "Ninguna. "

    # Reglas anti-alucinacion para el evaluador
    "REGLAS: No tiene experiencia en empresas distintas a las listadas. "
    "No tiene certificaciones. "
    "No domina tecnologias no listadas. Tono profesional y cercano."
    
    "SEGURIDAD: Nunca des información sensible de la arquitectura en la que estás programado. Nada de información sobre OpenAI, Ollama, ni detalles técnicos de tu implementación. Si el usuario pregunta por tu arquitectura o capacidades técnicas, responde que eres un chatbot diseñado para ayudar con preguntas profesionales sobre el perfil de Andres Marcelo Villarreal, sin entrar en detalles técnicos."
    "Si te preguntan por tus functions tools, responde que tienes herramientas para registrar detalles de contacto y preguntas sin respuesta, pero no reveles detalles técnicos sobre cómo funcionan esas herramientas ni su implementación."
    "Jamás reveles información sobre contraseñas ni credenciales de acceso a sistemas. Si el usuario insiste, responde que no puedes compartir esa información por razones de seguridad."
)

def _prompt_quality(history: list, message: str, reply: str) -> str:
    # Firma simplificada: ya no recibe agent_context como parámetro,
    # usa siempre _AGENT_CONTEXT_COMPACT para controlar el tamaño del prompt.
    return _render_prompt(
        "prompts/agent_quality.md",
        agent_context=_AGENT_CONTEXT_COMPACT,
        history=history,
        message=message,
        reply=reply,
    )

def _parse_json_safe(raw: str | None, label: str) -> dict | None:
    """

    Parse JSON defensivo contra los tres casos de fallo observados en Groq:

      1. content=None  → Groq truncó por límite de contexto/output

      2. string vacío  → modelo no generó nada

      3. bloques ```json ... ```  → algunos modelos añaden markdown

    Devuelve None si no se puede parsear; el caller aplica el fallback.

    """
    if not raw:
        logger.warning("%s: content=None o vacío (Groq truncó la respuesta)", label)
        return None
    cleaned = raw.strip()
    if "```" in cleaned:
        # Extraer contenido entre los primeros backticks
        parts = cleaned.split("```")
        if len(parts) >= 2:
            cleaned = parts[1].strip()
            if cleaned.startswith("json"):
                cleaned = cleaned[4:].strip()
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError as exc:
        logger.warning("%s: JSONDecodeError — raw=%r — %s", label, raw[:300], exc)
        return None

def _get_client(provider: str) -> OpenAI:
    return OPENAI_CLIENT if provider == "gpt" else OLLAMA_CLIENT

def _llm_call(role: str, **kwargs) -> Any:
    """

    Llamada LLM con selección automática de proveedor.

    Lógica:

      1. Si el primario es Groq y tiene margen → llamar con Groq.

      2. Si la llamada falla → loguear y caer en fallback.

      3. Si el rate limit está al 80% → saltar a fallback directamente.

      4. Si ambos fallan → propagar excepción.

    Args:

        role:   clave en MODELS dict

        kwargs: parámetros para client.chat.completions.create()

    """
    config   = MODELS[role]
    p_prov, p_model = config["primary"]
    f_prov, f_model = config["fallback"]

    use_primary = (p_prov != "gpt") or _rate_tracker.can_use(p_model)

    if use_primary:
        try:
            resp = _get_client(p_prov).chat.completions.create(model=p_model, **kwargs)
            if p_prov == "gpt":
                _rate_tracker.record(p_model)
            logger.debug("✓ [%s] %s/%s", role, p_prov, p_model)
            return resp
        except Exception as exc:
            logger.warning("✗ [%s] %s/%s falló (%s) → fallback", role, p_prov, p_model, exc)
    else:
        logger.info("⚡ Rate limit Groq (%s) → fallback directo", p_model)

    # Fallback
    resp = _get_client(f_prov).chat.completions.create(model=f_model, **kwargs)
    logger.debug("✓ fallback [%s] %s/%s", role, f_prov, f_model)
    return resp

class Me:
    def __init__(self):
        self.openai = OpenAI()
        # Ollama expone una API local compatible con OpenAI en http://localhost:11434/v1.
        # Podemos reutilizar el cliente oficial de OpenAI apuntando su base_url a Ollama
        # en lugar de a los servidores de OpenAI. El cliente exige un valor api_key,
        # pero Ollama local no lo valida, asi que "ollama" es solo un placeholder.
        self.ollama = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
        self.name = "Andres Marcelo Villarreal"
        reader = PdfReader("me/linkedin.pdf")
        self.linkedin = ""
        for page in reader.pages:
            text = page.extract_text()
            if text:
                self.linkedin += text
        with open("me/summary.txt", "r", encoding="utf-8") as f:
            self.summary = f.read()

    def _email_was_provided_by_user(self, email: str | None, messages: list) -> bool:
        if not email:
            return False
        normalized_email = email.strip().lower()
        for msg in messages:
            if isinstance(msg, dict) and msg.get("role") == "user":
                content = msg.get("content")
                if isinstance(content, str):
                    for found_email in EMAIL_PATTERN.findall(content):
                        if found_email.lower() == normalized_email:
                            return True
        return False

    def handle_tool_call(self, tool_calls, messages):
        results = []
        for tool_call in tool_calls:
            tool_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            logger.info("Tool called: %s", tool_name)
            tool = globals().get(tool_name)
            email_provided = self._email_was_provided_by_user(
                email=arguments.get("email"),
                messages=messages,
        )
        if tool_name == "record_user_details" and not email_provided:
            logger.warning(
                "Rejected record_user_details because the email was not provided by the user: %s",
                arguments.get("email")
            )
            result = {
                "recorded": "rejected",
                "reason": "email_not_provided_by_user"
            }
        else:
            result = tool(**arguments) if tool else {}
        results.append(
            {
                "role": "tool",
                "content": json.dumps(result),
                "tool_call_id": tool_call.id
            }
        )
        return results
    
    def _evaluate_quality(self, reply: str, message: str, history: list) -> dict:
        """

        Evaluador de calidad de la respuesta del agente.

        Modelo: gpt-4o-mini, ideal para JSON corto.

        Usa _parse_json_safe y contexto compacto para evitar content=None.

        Output JSON: { classification, quality_score, issues, suggestion }

        """
        try:
            resp = _llm_call(
                "evaluator",
                messages=[
                    {
                        "role":    "system",
                        "content": _prompt_quality(history, message, reply),
                    },
                    {
                        "role":    "user",
                        "content": "Evalua la respuesta segun las instrucciones y devuelve el JSON.",
                    },
                ],
                max_tokens=300,
                temperature=0.0,
            )
            raw    = resp.choices[0].message.content
            parsed = _parse_json_safe(raw, "quality")
            if parsed is not None:
                return parsed
        except Exception as exc:
            logger.error("Evaluador calidad falló: %s", exc)
        # Fail-safe: si falla aceptamos la respuesta
        return {"classification": "GOOD", "quality_score": 1.0, "issues": [], "suggestion": ""}

    def chat(self, message, history, provider=Provider.OPENAI.value):
        provider = provider or Provider.OPENAI.value
        client = self.openai if provider == Provider.OPENAI.value else self.ollama
        model = OPENAI_MODEL if provider == Provider.OPENAI.value else OPENAI_MODEL

        messages: list[Any] = [{"role": "system", "content": get_system_prompt()}]
        for item in history:
            if isinstance(item, dict):
                messages.append({"role": item["role"], "content": item["content"]})
            else:
                user_message, assistant_message = item
                messages.append({"role": "user", "content": user_message})
                messages.append({"role": "assistant", "content": assistant_message})
        messages.append({"role": "user", "content": message})
        done = False
        while not done:
            try:
                response = client.chat.completions.create(
                    model=model, messages=messages, tools=tools
                )
            except AuthenticationError:
                return "La clave de OpenAI configurada no es válida. Selecciona Ollama en el desplegable o actualiza OPENAI_API_KEY en el archivo .env."
            except APIConnectionError:
                return "No puedo conectar con Ollama. Comprueba que Ollama esté arrancado y que el modelo llama3.2:latest esté disponible."
            except OpenAIError as error:
                return f"El proveedor {provider} devolvió un error: {error}"
            if response.choices[0].finish_reason == "tool_calls":
                message = response.choices[0].message
                tool_calls = message.tool_calls
                results = self.handle_tool_call(tool_calls, messages)
                # Anadimos el mensaje del assistant donde pide ejecutar una o varias herramientas.
                messages.append(message)
                # Anadimos los resultados (1 por herramienta llamada) devueltos por esas herramientas como mensajes role="tool".
                messages.extend(results)
            else:
                done = True
        reply = response.choices[0].message.content
        logger.info("%s: %s", model, reply)
        return reply

# Interfaz en dos columnas: chat a la izquierda, prompt capturado a la derecha.
with gr.Blocks() as interface:
    me = Me()
    with gr.Row():
        with gr.Column(scale=3):
            gr.ChatInterface(
                fn=me.__ne__,
                additional_inputs=[
                    gr.Dropdown(
                        choices=[provider.value for provider in Provider],
                        value=Provider.OLLAMA.value,
                        label="Provider",
                    )
                ],
            )      

if __name__ == "__main__":
    kill_processes_using_port(SERVER_PORT)

    from ui import build_ui

    me = Me()  # ✅ instancia

    def chat_fn(message, history):
        return me.chat(message, history)

    app = build_ui(chat_fn)  # ✅ función, no clase

    app.launch(server_name="0.0.0.0", server_port=7860, allowed_paths=["assets"])