feat: Add asynchronous support and improved logging/error handling for calendar functions, introduce a calendar debug script, and refactor role-based menu logic.

This commit is contained in:
Marco Gallegos
2025-12-18 09:24:54 -06:00
parent 54f4e9ee41
commit 9a83fb93bb
6 changed files with 87 additions and 28 deletions

View File

@@ -3,11 +3,14 @@
# Permite buscar espacios libres y crear eventos. # Permite buscar espacios libres y crear eventos.
import datetime import datetime
import logging
from google.oauth2 import service_account from google.oauth2 import service_account
from googleapiclient.discovery import build from googleapiclient.discovery import build
from googleapiclient.errors import HttpError from googleapiclient.errors import HttpError
from config import GOOGLE_SERVICE_ACCOUNT_FILE, CALENDAR_ID from config import GOOGLE_SERVICE_ACCOUNT_FILE, CALENDAR_ID
logger = logging.getLogger(__name__)
# Configuración de los permisos (SCOPES) para acceder al calendario # Configuración de los permisos (SCOPES) para acceder al calendario
SCOPES = ["https://www.googleapis.com/auth/calendar"] SCOPES = ["https://www.googleapis.com/auth/calendar"]
@@ -120,6 +123,7 @@ def get_events(start_time, end_time, calendar_id=CALENDAR_ID):
Obtiene la lista de eventos entre dos momentos. Obtiene la lista de eventos entre dos momentos.
""" """
try: try:
logger.info(f"Llamando a la API de Google Calendar para {calendar_id}")
events_result = ( events_result = (
service.events() service.events()
.list( .list(
@@ -131,7 +135,12 @@ def get_events(start_time, end_time, calendar_id=CALENDAR_ID):
) )
.execute() .execute()
) )
return events_result.get("items", []) events = events_result.get("items", [])
logger.info(f"Se obtuvieron {len(events)} eventos de la API.")
return events
except HttpError as error: except HttpError as error:
print(f"Ocurrió un error al obtener eventos: {error}") logger.error(f"Ocurrió un error al obtener eventos: {error}")
return []
except Exception as e:
logger.error(f"Error inesperado al obtener eventos: {e}")
return [] return []

View File

@@ -2,6 +2,7 @@
# Este es el archivo principal del bot. Aquí se inicia todo y se configuran los comandos. # Este es el archivo principal del bot. Aquí se inicia todo y se configuran los comandos.
import logging import logging
import asyncio
from telegram import Update from telegram import Update
from telegram.ext import ( from telegram.ext import (
Application, Application,
@@ -87,12 +88,25 @@ async def button_dispatcher(update: Update, context: ContextTypes.DEFAULT_TYPE)
} }
if query.data in simple_handlers: if query.data in simple_handlers:
response_text = simple_handlers[query.data]() handler = simple_handlers[query.data]
logger.info(f"Ejecutando simple_handler para: {query.data}")
if asyncio.iscoroutinefunction(handler):
response_text = await handler()
else:
response_text = handler()
logger.info(f"Respuesta obtenida para {query.data}")
await query.edit_message_text(text=response_text, parse_mode='Markdown') await query.edit_message_text(text=response_text, parse_mode='Markdown')
elif query.data in complex_handlers: elif query.data in complex_handlers:
response_text, reply_markup = complex_handlers[query.data]() handler = complex_handlers[query.data]
logger.info(f"Ejecutando complex_handler para: {query.data}")
if asyncio.iscoroutinefunction(handler):
response_text, reply_markup = await handler()
else:
response_text, reply_markup = handler()
logger.info(f"Respuesta obtenida para {query.data}")
await query.edit_message_text(text=response_text, reply_markup=reply_markup, parse_mode='Markdown') await query.edit_message_text(text=response_text, reply_markup=reply_markup, parse_mode='Markdown')
elif query.data.startswith(('approve:', 'reject:')): elif query.data.startswith(('approve:', 'reject:')):
logger.info(f"Ejecutando acción de aprobación: {query.data}")
response_text = handle_approval_action(query.data) response_text = handle_approval_action(query.data)
await query.edit_message_text(text=response_text, parse_mode='Markdown') await query.edit_message_text(text=response_text, parse_mode='Markdown')
elif query.data == 'start_create_tag': elif query.data == 'start_create_tag':

View File

@@ -3,19 +3,26 @@
# Permite obtener y mostrar las actividades programadas para el día. # Permite obtener y mostrar las actividades programadas para el día.
import datetime import datetime
import logging
from google_calendar import get_events from google_calendar import get_events
def get_agenda(): logger = logging.getLogger(__name__)
async def get_agenda():
""" """
Obtiene y muestra la agenda del usuario para el día actual desde Google Calendar. Obtiene y muestra la agenda del usuario para el día actual desde Google Calendar.
""" """
now = datetime.datetime.utcnow() try:
logger.info("Obteniendo agenda...")
now = datetime.datetime.now(datetime.timezone.utc)
start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0) start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0)
end_of_day = start_of_day + datetime.timedelta(days=1) end_of_day = start_of_day + datetime.timedelta(days=1)
logger.info(f"Buscando eventos desde {start_of_day} hasta {end_of_day}")
events = get_events(start_of_day, end_of_day) events = get_events(start_of_day, end_of_day)
if not events: if not events:
logger.info("No se encontraron eventos.")
return "📅 *Agenda para Hoy*\n\nNo tienes eventos programados para hoy." return "📅 *Agenda para Hoy*\n\nNo tienes eventos programados para hoy."
agenda_text = "📅 *Agenda para Hoy*\n\n" agenda_text = "📅 *Agenda para Hoy*\n\n"
@@ -30,4 +37,8 @@ def get_agenda():
summary = event.get("summary", "(Sin título)") summary = event.get("summary", "(Sin título)")
agenda_text += f"• *{time_str}* - {summary}\n" agenda_text += f"• *{time_str}* - {summary}\n"
logger.info("Agenda obtenida con éxito.")
return agenda_text return agenda_text
except Exception as e:
logger.error(f"Error al obtener la agenda: {e}")
return "❌ Error al obtener la agenda. Por favor, intenta de nuevo más tarde."

View File

@@ -55,9 +55,7 @@ def handle_start(user_role):
""" """
welcome_message = "Hola, soy Talía. ¿En qué puedo ayudarte hoy?" welcome_message = "Hola, soy Talía. ¿En qué puedo ayudarte hoy?"
if user_role == "owner": if user_role in ["owner", "admin"]:
menu = get_owner_menu()
elif user_role == "admin":
menu = get_admin_menu() menu = get_admin_menu()
elif user_role == "team": elif user_role == "team":
menu = get_team_menu() menu = get_team_menu()

View File

@@ -13,8 +13,8 @@ from telegram.ext import (
ContextTypes, ContextTypes,
) )
from app.config import VIKUNJA_API_URL, VIKUNJA_API_TOKEN from config import VIKUNJA_API_URL, VIKUNJA_API_TOKEN
from app.permissions import is_admin from permissions import is_admin
# Configuración del logger # Configuración del logger
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)

27
test_calendar_debug.py Normal file
View File

@@ -0,0 +1,27 @@
import datetime
import sys
import os
# Add app directory to path
sys.path.append(os.path.join(os.getcwd(), 'app'))
from google_calendar import get_events
from config import CALENDAR_ID
def test_get_events():
print(f"Testing with CALENDAR_ID: {CALENDAR_ID}")
now = datetime.datetime.now(datetime.timezone.utc)
start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0)
end_of_day = start_of_day + datetime.timedelta(days=1)
print(f"Fetching events from {start_of_day.isoformat()} to {end_of_day.isoformat()}...")
try:
events = get_events(start_of_day, end_of_day)
print(f"Found {len(events)} events.")
for event in events:
print(f"- {event.get('summary')} at {event['start'].get('dateTime', event['start'].get('date'))}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
test_get_events()