rex/integrations.py
Rama master · réplica de lectura del repositorio original.
"""
Integration module for Rex AI Assistant.
Sets up default scheduler jobs and event bus subscriptions for email,
calendar, and other automated tasks.
"""
import logging
from rex.calendar_service import get_calendar_service
from rex.email_service import get_email_service
from rex.openclaw.event_bus import Event, get_event_bus
from rex.scheduler import ScheduledJob, get_scheduler
logger = logging.getLogger(__name__)
def setup_email_job() -> ScheduledJob:
"""
Set up the default email checking job.
Returns:
Created ScheduledJob
"""
scheduler = get_scheduler()
event_bus = get_event_bus()
# Register email check callback
def check_email(job: ScheduledJob) -> None:
"""Check unread email per configured owner and publish scoped events.
Each owner is processed in an isolated context; private email fields
are published only on the owner-scoped topic (see
``rex.integrations._setup.setup_email_job``, the canonical copy).
"""
logger.info("Running scheduled email check")
try:
from rex.email_accounts import EmailAccountResolver
owners = EmailAccountResolver.load().configured_user_ids()
except Exception as e:
logger.error(f"Error resolving email owners: {e}", exc_info=True)
return
if not owners:
logger.debug("No email account owners configured; skipping scheduled email check")
return
try:
email_service = get_email_service()
except Exception as e:
logger.debug(f"Email service unavailable for scheduled check: {e}")
return
for owner in owners:
try:
unread_emails = email_service.fetch_unread(limit=10, user_id=owner)
if unread_emails:
for email in unread_emails:
email.category = email_service.categorize(email)
event_bus.publish(
Event(
event_type=f"email.unread.user.{owner}",
payload={
"count": len(unread_emails),
"user_id": owner,
"emails": [email.model_dump() for email in unread_emails],
},
)
)
event_bus.publish(
Event(
event_type="email.unread",
payload={"count": len(unread_emails), "user_id": owner},
)
)
logger.info(
f"Published email.unread event with {len(unread_emails)} emails "
f"for user {owner}"
)
else:
logger.debug(f"No unread emails found for user {owner}")
except Exception as e:
logger.error(f"Error checking emails for user {owner}: {e}", exc_info=True)
scheduler.register_callback("check_email", check_email)
# Add job (runs every 10 minutes = 600 seconds)
job = scheduler.add_job(
job_id="email_check",
name="Check Email",
schedule="interval:600",
callback_name="check_email",
enabled=True,
)
logger.info("Email check job registered")
return job
def setup_calendar_job() -> ScheduledJob:
"""
Set up the default calendar sync job.
Returns:
Created ScheduledJob
"""
scheduler = get_scheduler()
event_bus = get_event_bus()
# Register calendar sync callback
def sync_calendar(job: ScheduledJob) -> None:
"""Sync calendar and publish event."""
logger.info("Running scheduled calendar sync")
try:
calendar_service = get_calendar_service()
if not calendar_service.connected:
calendar_service.connect()
# Get upcoming events (next 7 days)
events = calendar_service.get_upcoming_events(days=7)
# Publish event
event = Event(
event_type="calendar.update",
payload={"count": len(events), "events": [e.model_dump() for e in events]}, # type: ignore[attr-defined]
)
event_bus.publish(event)
logger.info(f"Published calendar.update event with {len(events)} events")
except Exception as e:
logger.error(f"Error syncing calendar: {e}", exc_info=True)
scheduler.register_callback("sync_calendar", sync_calendar)
# Add job (runs every hour = 3600 seconds)
job = scheduler.add_job(
job_id="calendar_sync",
name="Sync Calendar",
schedule="interval:3600",
callback_name="sync_calendar",
enabled=True,
)
logger.info("Calendar sync job registered")
return job
def setup_default_event_handlers() -> None:
"""Set up default event bus handlers for logging."""
from rex.openclaw.event_bridge import EventBridge
event_bus = EventBridge()
def log_email_event(event: Event) -> None:
"""Log email events."""
count = event.payload.get("count", 0)
logger.info(f"Email event received: {count} unread email(s)")
def log_calendar_event(event: Event) -> None:
"""Log calendar events."""
count = event.payload.get("count", 0)
logger.info(f"Calendar event received: {count} upcoming event(s)")
# Subscribe handlers
event_bus.subscribe("email.unread", log_email_event)
event_bus.subscribe("calendar.update", log_calendar_event)
logger.info("Default event handlers registered")
def initialize_scheduler_system(start_scheduler: bool = False) -> None:
"""
Initialize the scheduler system with default jobs and event handlers.
Args:
start_scheduler: If True, start the scheduler background thread
"""
logger.info("Initializing scheduler system")
# Set up default jobs — failures are non-fatal to allow degraded operation
try:
setup_email_job()
except Exception as exc:
logger.debug("Email job setup skipped: %s", exc)
try:
setup_calendar_job()
except Exception as exc:
logger.debug("Calendar job setup skipped: %s", exc)
_try_register_retention_jobs()
# Set up event handlers
setup_default_event_handlers()
# Start scheduler if requested
if start_scheduler:
scheduler = get_scheduler()
scheduler.start()
logger.info("Scheduler started")
logger.info("Scheduler system initialized")
def _try_register_retention_jobs() -> None:
"""Register retention cleanup jobs from config when available.
Failures are logged at DEBUG and do not prevent scheduler initialization.
"""
try:
from rex.config_manager import load_config
from rex.retention import setup_retention_jobs
raw_config = load_config()
results = setup_retention_jobs(raw_config)
if any(results.values()):
logger.info("Retention cleanup jobs registered: %s", results)
else:
logger.debug("No retention cleanup jobs registered (check config)")
except Exception as exc:
logger.debug("Retention cleanup jobs setup skipped: %s", exc)
def shutdown_scheduler_system() -> None:
"""Shutdown the scheduler system."""
logger.info("Shutting down scheduler system")
scheduler = get_scheduler()
scheduler.stop()
logger.info("Scheduler stopped")