8431784708
- Moved Jinja2 template setup to a shared configuration file (templates_config.py) for consistent usage across routers. - Introduced timezone utilities in a new module (timezone.py) to handle UTC to local time conversions and formatting. - Updated all relevant routers to use the new shared template configuration and timezone filters. - Enhanced templates to utilize local time formatting for various datetime fields, improving user experience with timezone awareness.
327 lines
9.3 KiB
Python
327 lines
9.3 KiB
Python
"""
|
|
Alerts Router
|
|
|
|
API endpoints for managing in-app alerts.
|
|
"""
|
|
|
|
from fastapi import APIRouter, Request, Depends, HTTPException, Query
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional
|
|
from datetime import datetime, timedelta
|
|
|
|
from backend.database import get_db
|
|
from backend.models import Alert, RosterUnit
|
|
from backend.services.alert_service import get_alert_service
|
|
from backend.templates_config import templates
|
|
|
|
router = APIRouter(prefix="/api/alerts", tags=["alerts"])
|
|
|
|
|
|
# ============================================================================
|
|
# Alert List and Count
|
|
# ============================================================================
|
|
|
|
@router.get("/")
|
|
async def list_alerts(
|
|
db: Session = Depends(get_db),
|
|
status: Optional[str] = Query(None, description="Filter by status: active, acknowledged, resolved, dismissed"),
|
|
project_id: Optional[str] = Query(None),
|
|
unit_id: Optional[str] = Query(None),
|
|
alert_type: Optional[str] = Query(None, description="Filter by type: device_offline, device_online, schedule_failed"),
|
|
limit: int = Query(50, le=100),
|
|
offset: int = Query(0, ge=0),
|
|
):
|
|
"""
|
|
List alerts with optional filters.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
|
|
alerts = alert_service.get_all_alerts(
|
|
status=status,
|
|
project_id=project_id,
|
|
unit_id=unit_id,
|
|
alert_type=alert_type,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
|
|
return {
|
|
"alerts": [
|
|
{
|
|
"id": a.id,
|
|
"alert_type": a.alert_type,
|
|
"severity": a.severity,
|
|
"title": a.title,
|
|
"message": a.message,
|
|
"status": a.status,
|
|
"unit_id": a.unit_id,
|
|
"project_id": a.project_id,
|
|
"location_id": a.location_id,
|
|
"created_at": a.created_at.isoformat() if a.created_at else None,
|
|
"acknowledged_at": a.acknowledged_at.isoformat() if a.acknowledged_at else None,
|
|
"resolved_at": a.resolved_at.isoformat() if a.resolved_at else None,
|
|
}
|
|
for a in alerts
|
|
],
|
|
"count": len(alerts),
|
|
"limit": limit,
|
|
"offset": offset,
|
|
}
|
|
|
|
|
|
@router.get("/active")
|
|
async def list_active_alerts(
|
|
db: Session = Depends(get_db),
|
|
project_id: Optional[str] = Query(None),
|
|
unit_id: Optional[str] = Query(None),
|
|
alert_type: Optional[str] = Query(None),
|
|
min_severity: Optional[str] = Query(None, description="Minimum severity: info, warning, critical"),
|
|
limit: int = Query(50, le=100),
|
|
):
|
|
"""
|
|
List only active alerts.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
|
|
alerts = alert_service.get_active_alerts(
|
|
project_id=project_id,
|
|
unit_id=unit_id,
|
|
alert_type=alert_type,
|
|
min_severity=min_severity,
|
|
limit=limit,
|
|
)
|
|
|
|
return {
|
|
"alerts": [
|
|
{
|
|
"id": a.id,
|
|
"alert_type": a.alert_type,
|
|
"severity": a.severity,
|
|
"title": a.title,
|
|
"message": a.message,
|
|
"unit_id": a.unit_id,
|
|
"project_id": a.project_id,
|
|
"created_at": a.created_at.isoformat() if a.created_at else None,
|
|
}
|
|
for a in alerts
|
|
],
|
|
"count": len(alerts),
|
|
}
|
|
|
|
|
|
@router.get("/active/count")
|
|
async def get_active_alert_count(db: Session = Depends(get_db)):
|
|
"""
|
|
Get count of active alerts (for navbar badge).
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
count = alert_service.get_active_alert_count()
|
|
return {"count": count}
|
|
|
|
|
|
# ============================================================================
|
|
# Single Alert Operations
|
|
# ============================================================================
|
|
|
|
@router.get("/{alert_id}")
|
|
async def get_alert(
|
|
alert_id: str,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Get a specific alert.
|
|
"""
|
|
alert = db.query(Alert).filter_by(id=alert_id).first()
|
|
if not alert:
|
|
raise HTTPException(status_code=404, detail="Alert not found")
|
|
|
|
# Get related unit info
|
|
unit = None
|
|
if alert.unit_id:
|
|
unit = db.query(RosterUnit).filter_by(id=alert.unit_id).first()
|
|
|
|
return {
|
|
"id": alert.id,
|
|
"alert_type": alert.alert_type,
|
|
"severity": alert.severity,
|
|
"title": alert.title,
|
|
"message": alert.message,
|
|
"metadata": alert.alert_metadata,
|
|
"status": alert.status,
|
|
"unit_id": alert.unit_id,
|
|
"unit_name": unit.id if unit else None,
|
|
"project_id": alert.project_id,
|
|
"location_id": alert.location_id,
|
|
"schedule_id": alert.schedule_id,
|
|
"created_at": alert.created_at.isoformat() if alert.created_at else None,
|
|
"acknowledged_at": alert.acknowledged_at.isoformat() if alert.acknowledged_at else None,
|
|
"resolved_at": alert.resolved_at.isoformat() if alert.resolved_at else None,
|
|
"expires_at": alert.expires_at.isoformat() if alert.expires_at else None,
|
|
}
|
|
|
|
|
|
@router.post("/{alert_id}/acknowledge")
|
|
async def acknowledge_alert(
|
|
alert_id: str,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Mark alert as acknowledged.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
alert = alert_service.acknowledge_alert(alert_id)
|
|
|
|
if not alert:
|
|
raise HTTPException(status_code=404, detail="Alert not found")
|
|
|
|
return {
|
|
"success": True,
|
|
"alert_id": alert.id,
|
|
"status": alert.status,
|
|
}
|
|
|
|
|
|
@router.post("/{alert_id}/dismiss")
|
|
async def dismiss_alert(
|
|
alert_id: str,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Dismiss alert.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
alert = alert_service.dismiss_alert(alert_id)
|
|
|
|
if not alert:
|
|
raise HTTPException(status_code=404, detail="Alert not found")
|
|
|
|
return {
|
|
"success": True,
|
|
"alert_id": alert.id,
|
|
"status": alert.status,
|
|
}
|
|
|
|
|
|
@router.post("/{alert_id}/resolve")
|
|
async def resolve_alert(
|
|
alert_id: str,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Manually resolve an alert.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
alert = alert_service.resolve_alert(alert_id)
|
|
|
|
if not alert:
|
|
raise HTTPException(status_code=404, detail="Alert not found")
|
|
|
|
return {
|
|
"success": True,
|
|
"alert_id": alert.id,
|
|
"status": alert.status,
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# HTML Partials for HTMX
|
|
# ============================================================================
|
|
|
|
@router.get("/partials/dropdown", response_class=HTMLResponse)
|
|
async def get_alert_dropdown(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Return HTML partial for alert dropdown in navbar.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
alerts = alert_service.get_active_alerts(limit=10)
|
|
|
|
# Calculate relative time for each alert
|
|
now = datetime.utcnow()
|
|
alerts_data = []
|
|
for alert in alerts:
|
|
delta = now - alert.created_at
|
|
if delta.days > 0:
|
|
time_ago = f"{delta.days}d ago"
|
|
elif delta.seconds >= 3600:
|
|
time_ago = f"{delta.seconds // 3600}h ago"
|
|
elif delta.seconds >= 60:
|
|
time_ago = f"{delta.seconds // 60}m ago"
|
|
else:
|
|
time_ago = "just now"
|
|
|
|
alerts_data.append({
|
|
"alert": alert,
|
|
"time_ago": time_ago,
|
|
})
|
|
|
|
return templates.TemplateResponse("partials/alerts/alert_dropdown.html", {
|
|
"request": request,
|
|
"alerts": alerts_data,
|
|
"total_count": alert_service.get_active_alert_count(),
|
|
})
|
|
|
|
|
|
@router.get("/partials/list", response_class=HTMLResponse)
|
|
async def get_alert_list(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
status: Optional[str] = Query(None),
|
|
limit: int = Query(20),
|
|
):
|
|
"""
|
|
Return HTML partial for alert list page.
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
|
|
if status:
|
|
alerts = alert_service.get_all_alerts(status=status, limit=limit)
|
|
else:
|
|
alerts = alert_service.get_all_alerts(limit=limit)
|
|
|
|
# Calculate relative time for each alert
|
|
now = datetime.utcnow()
|
|
alerts_data = []
|
|
for alert in alerts:
|
|
delta = now - alert.created_at
|
|
if delta.days > 0:
|
|
time_ago = f"{delta.days}d ago"
|
|
elif delta.seconds >= 3600:
|
|
time_ago = f"{delta.seconds // 3600}h ago"
|
|
elif delta.seconds >= 60:
|
|
time_ago = f"{delta.seconds // 60}m ago"
|
|
else:
|
|
time_ago = "just now"
|
|
|
|
alerts_data.append({
|
|
"alert": alert,
|
|
"time_ago": time_ago,
|
|
})
|
|
|
|
return templates.TemplateResponse("partials/alerts/alert_list.html", {
|
|
"request": request,
|
|
"alerts": alerts_data,
|
|
"status_filter": status,
|
|
})
|
|
|
|
|
|
# ============================================================================
|
|
# Cleanup
|
|
# ============================================================================
|
|
|
|
@router.post("/cleanup-expired")
|
|
async def cleanup_expired_alerts(db: Session = Depends(get_db)):
|
|
"""
|
|
Cleanup expired alerts (admin/maintenance endpoint).
|
|
"""
|
|
alert_service = get_alert_service(db)
|
|
count = alert_service.cleanup_expired_alerts()
|
|
|
|
return {
|
|
"success": True,
|
|
"cleaned_up": count,
|
|
}
|