mirror of
https://github.com/marcogll/telegram_new_socias.git
synced 2026-01-13 13:15:16 +00:00
feat: Implement AI-powered permit reason classification and add comprehensive project documentation.
This commit is contained in:
Marco Gallegos
41
modules/ai.py
Normal file
41
modules/ai.py
Normal file
@@ -0,0 +1,41 @@
|
||||
import os
|
||||
import google.generativeai as genai
|
||||
|
||||
def classify_reason(text: str) -> str:
|
||||
"""
|
||||
Clasifica el motivo de un permiso utilizando la API de Gemini.
|
||||
|
||||
Args:
|
||||
text: El motivo del permiso proporcionado por el usuario.
|
||||
|
||||
Returns:
|
||||
La categoría clasificada (EMERGENCIA, MÉDICO, TRÁMITE, PERSONAL) o "OTRO" si no se puede clasificar.
|
||||
"""
|
||||
try:
|
||||
genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
|
||||
|
||||
model = genai.GenerativeModel('gemini-pro')
|
||||
|
||||
prompt = f"""
|
||||
Clasifica el siguiente motivo de solicitud de permiso en una de estas cuatro categorías: EMERGENCIA, MÉDICO, TRÁMITE, PERSONAL.
|
||||
Responde únicamente con la palabra de la categoría en mayúsculas.
|
||||
|
||||
Motivo: "{text}"
|
||||
Categoría:
|
||||
"""
|
||||
|
||||
response = model.generate_content(prompt)
|
||||
|
||||
# Limpiar la respuesta para obtener solo la categoría
|
||||
category = response.text.strip().upper()
|
||||
|
||||
# Validar que la categoría sea una de las esperadas
|
||||
valid_categories = ["EMERGENCIA", "MÉDICO", "TRÁMITE", "PERSONAL"]
|
||||
if category in valid_categories:
|
||||
return category
|
||||
else:
|
||||
return "PERSONAL" # Si la IA devuelve algo inesperado, se asigna a PERSONAL
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error al clasificar con IA: {e}")
|
||||
return "PERSONAL" # En caso de error, se asigna a PERSONAL por defecto
|
||||
@@ -8,12 +8,21 @@ import logging
|
||||
# Configuración de logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
|
||||
# Obtener la URL de la base de datos desde las variables de entorno
|
||||
DATABASE_URL = os.getenv("DATABASE_URL", "mysql+mysqlconnector://user:password@db:3306/vanessa_logs")
|
||||
# Construir la URL de la base de datos desde las variables de entorno individuales
|
||||
try:
|
||||
user = os.getenv("MYSQL_USER")
|
||||
password = os.getenv("MYSQL_PASSWORD")
|
||||
host = "db" # El nombre del servicio de la base de datos en docker-compose
|
||||
database = os.getenv("MYSQL_DATABASE")
|
||||
DATABASE_URL = f"mysql+mysqlconnector://{user}:{password}@{host}:3306/{database}"
|
||||
|
||||
# Crear el motor de la base de datos
|
||||
engine = create_engine(DATABASE_URL)
|
||||
metadata = MetaData()
|
||||
|
||||
# Crear el motor de la base de datos
|
||||
engine = create_engine(DATABASE_URL)
|
||||
metadata = MetaData()
|
||||
except AttributeError:
|
||||
logging.error("Error: Faltan una o más variables de entorno para la base de datos (MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE).")
|
||||
exit(1)
|
||||
|
||||
# Base para los modelos declarativos
|
||||
Base = declarative_base()
|
||||
|
||||
@@ -1,22 +1,69 @@
|
||||
import os
|
||||
import re
|
||||
import requests
|
||||
from telegram import Update, ReplyKeyboardMarkup, ReplyKeyboardRemove
|
||||
import uuid
|
||||
from datetime import datetime, date
|
||||
from telegram import Update
|
||||
from telegram.ext import ContextTypes, ConversationHandler, CommandHandler, MessageHandler, filters
|
||||
from modules.database import log_request
|
||||
from modules.ai import classify_reason
|
||||
|
||||
TIPO_SOLICITITUD, FECHAS, MOTIVO = range(3)
|
||||
|
||||
def _calculate_vacation_metrics(date_string: str) -> dict:
|
||||
"""
|
||||
Calcula métricas de vacaciones a partir de un texto.
|
||||
Asume un formato como "10 al 15 de Octubre".
|
||||
"""
|
||||
today = date.today()
|
||||
current_year = today.year
|
||||
|
||||
# Mapeo de meses en español a número
|
||||
meses = {
|
||||
'enero': 1, 'febrero': 2, 'marzo': 3, 'abril': 4, 'mayo': 5, 'junio': 6,
|
||||
'julio': 7, 'agosto': 8, 'septiembre': 9, 'octubre': 10, 'noviembre': 11, 'diciembre': 12
|
||||
}
|
||||
|
||||
# Regex para "10 al 15 de Octubre"
|
||||
match = re.search(r'(\d{1,2})\s*al\s*(\d{1,2})\s*de\s*(\w+)', date_string, re.IGNORECASE)
|
||||
|
||||
if not match:
|
||||
return {"dias_totales": 0, "dias_anticipacion": 0}
|
||||
|
||||
start_day, end_day, month_str = match.groups()
|
||||
start_day, end_day = int(start_day), int(end_day)
|
||||
month = meses.get(month_str.lower())
|
||||
|
||||
if not month:
|
||||
return {"dias_totales": 0, "dias_anticipacion": 0}
|
||||
|
||||
try:
|
||||
start_date = date(current_year, month, start_day)
|
||||
# Si la fecha ya pasó este año, asumir que es del próximo año
|
||||
if start_date < today:
|
||||
start_date = date(current_year + 1, month, start_day)
|
||||
|
||||
end_date = date(start_date.year, month, end_day)
|
||||
|
||||
dias_totales = (end_date - start_date).days + 1
|
||||
dias_anticipacion = (start_date - today).days
|
||||
|
||||
return {"dias_totales": dias_totales, "dias_anticipacion": dias_anticipacion, "fechas_calculadas": {"inicio": start_date.isoformat(), "fin": end_date.isoformat()}}
|
||||
except ValueError:
|
||||
return {"dias_totales": 0, "dias_anticipacion": 0}
|
||||
|
||||
TIPO_SOLICITUD, FECHAS, MOTIVO = range(3)
|
||||
|
||||
async def start_vacaciones(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
||||
user = update.effective_user
|
||||
log_request(user.id, user.username, "vacaciones", update.message.text)
|
||||
context.user_data['tipo'] = 'Vacaciones'
|
||||
context.user_data['tipo'] = 'VACACIONES'
|
||||
await update.message.reply_text("🌴 **Solicitud de Vacaciones**\n\n¿Para qué fechas las necesitas? (Ej: 10 al 15 de Octubre)")
|
||||
return FECHAS
|
||||
|
||||
async def start_permiso(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
||||
user = update.effective_user
|
||||
log_request(user.id, user.username, "permiso", update.message.text)
|
||||
context.user_data['tipo'] = 'Permiso Especial'
|
||||
context.user_data['tipo'] = 'PERMISO'
|
||||
await update.message.reply_text("⏱️ **Solicitud de Permiso**\n\n¿Para qué día y horario lo necesitas?")
|
||||
return FECHAS
|
||||
|
||||
@@ -29,25 +76,63 @@ async def recibir_motivo_fin(update: Update, context: ContextTypes.DEFAULT_TYPE)
|
||||
motivo = update.message.text
|
||||
datos = context.user_data
|
||||
user = update.effective_user
|
||||
|
||||
# Payload para n8n
|
||||
|
||||
# Generar payload base
|
||||
payload = {
|
||||
"solicitante": user.full_name,
|
||||
"id_telegram": user.id,
|
||||
"record_id": str(uuid.uuid4()),
|
||||
"solicitante": {
|
||||
"id_telegram": user.id,
|
||||
"nombre": user.full_name
|
||||
},
|
||||
"tipo_solicitud": datos['tipo'],
|
||||
"fechas": datos['fechas'],
|
||||
"motivo": motivo
|
||||
"fechas_texto_original": datos['fechas'],
|
||||
"motivo_usuario": motivo,
|
||||
"created_at": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
webhook = os.getenv("WEBHOOK_VACACIONES")
|
||||
if datos['tipo'] == 'PERMISO':
|
||||
webhook = os.getenv("WEBHOOK_PERMISOS")
|
||||
categoria = classify_reason(motivo)
|
||||
payload["categoria_detectada"] = categoria
|
||||
await update.message.reply_text(f"Categoría detectada → **{categoria}** 🚨")
|
||||
|
||||
elif datos['tipo'] == 'VACACIONES':
|
||||
webhook = os.getenv("WEBHOOK_VACACIONES")
|
||||
metrics = _calculate_vacation_metrics(datos['fechas'])
|
||||
|
||||
if metrics["dias_totales"] > 0:
|
||||
payload["metricas"] = metrics
|
||||
|
||||
dias = metrics["dias_totales"]
|
||||
if dias <= 5:
|
||||
status = "RECHAZADO"
|
||||
mensaje = f"🔴 {dias} días es un periodo muy corto. Las vacaciones deben ser de al menos 6 días."
|
||||
elif 6 <= dias <= 11:
|
||||
status = "REVISION_MANUAL"
|
||||
mensaje = f"🟡 Solicitud de {dias} días recibida. Tu manager la revisará pronto."
|
||||
else: # 12+
|
||||
status = "PRE_APROBADO"
|
||||
mensaje = f"🟢 ¡Excelente planeación! Tu solicitud de {dias} días ha sido pre-aprobada."
|
||||
|
||||
payload["status_inicial"] = status
|
||||
await update.message.reply_text(mensaje)
|
||||
else:
|
||||
# Si no se pudieron parsear las fechas
|
||||
payload["status_inicial"] = "ERROR_FECHAS"
|
||||
await update.message.reply_text("🤔 No entendí las fechas. Por favor, usa un formato como '10 al 15 de Octubre'.")
|
||||
|
||||
try:
|
||||
requests.post(webhook, json=payload)
|
||||
await update.message.reply_text(f"✅ Solicitud de *{datos['tipo']}* enviada a tu Manager.")
|
||||
except:
|
||||
if webhook:
|
||||
requests.post(webhook, json=payload)
|
||||
tipo_solicitud_texto = "Permiso" if datos['tipo'] == 'PERMISO' else 'Vacaciones'
|
||||
await update.message.reply_text(f"✅ Solicitud de *{tipo_solicitud_texto}* enviada a tu Manager.")
|
||||
except Exception as e:
|
||||
print(f"Error enviando webhook: {e}")
|
||||
await update.message.reply_text("⚠️ Error enviando la solicitud.")
|
||||
|
||||
return ConversationHandler.END
|
||||
|
||||
|
||||
async def cancelar(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
|
||||
await update.message.reply_text("Solicitud cancelada.")
|
||||
return ConversationHandler.END
|
||||
|
||||
Reference in New Issue
Block a user