"""IMAP Proxy dashboard router - status and metrics API.""" from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse, JSONResponse from datetime import datetime, timezone # Public router - no auth required for status/metrics public_router = APIRouter(prefix="/api/public/imap", tags=["imap-proxy"]) # Private router - requires auth for dashboard HTML (if needed in future) router = APIRouter(prefix="/imap", tags=["imap-proxy"]) # Will be set by main app at startup _proxy_instance = None def set_proxy(proxy): """Register the active IMAP proxy instance for dashboard queries.""" global _proxy_instance _proxy_instance = proxy @public_router.get("/status") async def imap_status(): """IMAP proxy status as JSON (public endpoint).""" if not _proxy_instance: return { "status": "not_configured", "message": "No IMAP proxy configured. Set IMAP_USER and IMAP_PASSWORD env vars.", } return _proxy_instance.get_status() @public_router.get("/metrics") async def imap_metrics(): """Recent email processing metrics (public endpoint).""" from .proxy import get_metrics return get_metrics() # Keep old routes for backward compatibility during transition @router.get("/status") async def imap_status_legacy(): """DEPRECATED: Use /api/public/imap/status instead.""" return await imap_status() @router.get("/metrics") async def imap_metrics_legacy(): """DEPRECATED: Use /api/public/imap/metrics instead.""" return await imap_metrics() @router.get("/dashboard", response_class=HTMLResponse) async def imap_dashboard(): """IMAP proxy dashboard as embeddable HTML fragment.""" status = {} metrics = {} if _proxy_instance: status = _proxy_instance.get_status() else: status = {"status": "not_configured"} from .proxy import get_metrics metrics = get_metrics() # Build status pill status_val = status.get("status", "unknown") pill_map = { "connected": ("status-healthy", "● Connected"), "disconnected": ("status-degraded", "○ Disconnected"), "auth_error": ("status-critical", "✕ Auth Failed"), "connection_error": ("status-degraded", "○ Conn Error"), "stopped": ("status-degraded", "■ Stopped"), "not_configured": ("status-critical", "○ Not Configured"), } pill_class, pill_label = pill_map.get(status_val, ("status-degraded", status_val)) provider_label = status.get("provider_label", status.get("provider", "—")) uptime = status.get("uptime_seconds") uptime_str = _format_uptime(uptime) if uptime else "—" last_fetch = status.get("last_fetch_at") last_fetch_str = _format_time_ago(last_fetch) if last_fetch else "Never" emails_count = status.get("emails_fetched", 0) last_error = status.get("last_error", "") recent = metrics.get("recent_emails", []) recent_html = "" if recent: for e in reversed(recent[-10:]): subj = _esc(e.get("subject", "(no subject)")) sender = _esc(e.get("from", "unknown")) ft = e.get("fetched_at", "") ago = _format_time_ago(ft) if ft else "" # Truncate sender to just the name part if it's an email sender_clean = sender.split('<')[0].strip() if '<' in sender else sender if len(sender_clean) > 25: sender_clean = sender_clean[:22] + "..." recent_html += f'''
{subj}
''' else: recent_html = '''
📭

No emails processed yet

''' error_html = "" if last_error: error_html = f'''
{_esc(last_error)}
''' # Provider icon mapping provider_icon = { "gmail": "📧 Gmail", "icloud": "☁️ iCloud", "outlook": "📧 Outlook", "yahoo": "📧 Yahoo" }.get(status.get("provider", "").lower(), f"📧 {provider_label}") return HTMLResponse(content=f"""

📧 Email

{pill_label}
{pill_label}
{uptime_str}
Provider {provider_icon}
Account {_esc(status.get('username', '—'))}
Emails Processed {emails_count}
Last Fetch {last_fetch_str}
{error_html}

Recent Emails

""") def _format_uptime(seconds): if seconds < 60: return f"{int(seconds)}s" if seconds < 3600: return f"{int(seconds // 60)}m" if seconds < 86400: return f"{int(seconds // 3600)}h {int((seconds % 3600) // 60)}m" return f"{int(seconds // 86400)}d" @public_router.get("/dashboard", response_class=HTMLResponse) async def imap_dashboard_public(): """Public IMAP proxy dashboard fragment (no auth required).""" return await imap_dashboard() def _format_time_ago(iso_str): try: then = datetime.fromisoformat(iso_str.replace("Z", "+00:00")) now = datetime.now(timezone.utc) diff = (now - then).total_seconds() if diff < 60: return "just now" if diff < 3600: return f"{int(diff // 60)}m ago" if diff < 86400: return f"{int(diff // 3600)}h ago" return f"{int(diff // 86400)}d ago" except Exception: return iso_str or "—" def _esc(text): if not text: return "" return str(text).replace("&", "&").replace("<", "<").replace(">", ">")