notifications.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568
  1. """API routes for notification providers."""
  2. import json
  3. import logging
  4. import time
  5. from collections import defaultdict, deque
  6. from datetime import datetime, timedelta, timezone
  7. from fastapi import APIRouter, Depends, HTTPException, Query
  8. from sqlalchemy import delete, desc, func, select
  9. from sqlalchemy.ext.asyncio import AsyncSession
  10. from backend.app.core.auth import RequirePermissionIfAuthEnabled, ScopedCaller, require_notification_send
  11. from backend.app.core.database import get_db
  12. from backend.app.core.permissions import Permission
  13. from backend.app.models.notification import NotificationLog, NotificationProvider
  14. from backend.app.models.user import User
  15. from backend.app.schemas.notification import (
  16. AppMessage,
  17. AppMessageChannel,
  18. AppMessageResult,
  19. NotificationLogResponse,
  20. NotificationLogStats,
  21. NotificationProviderCreate,
  22. NotificationProviderResponse,
  23. NotificationProviderUpdate,
  24. NotificationTestRequest,
  25. NotificationTestResponse,
  26. )
  27. from backend.app.services.notification_service import notification_service
  28. logger = logging.getLogger(__name__)
  29. router = APIRouter(prefix="/notifications", tags=["notifications"])
  30. def _provider_to_dict(provider: NotificationProvider) -> dict:
  31. """Convert a NotificationProvider model to a response dictionary."""
  32. return {
  33. "id": provider.id,
  34. "name": provider.name,
  35. "provider_type": provider.provider_type,
  36. "enabled": provider.enabled,
  37. "config": json.loads(provider.config) if isinstance(provider.config, str) else provider.config,
  38. # Print lifecycle events
  39. "on_print_start": provider.on_print_start,
  40. "on_print_complete": provider.on_print_complete,
  41. "on_print_failed": provider.on_print_failed,
  42. "on_print_stopped": provider.on_print_stopped,
  43. "on_print_progress": provider.on_print_progress,
  44. "on_print_missing_spool_assignment": provider.on_print_missing_spool_assignment,
  45. "on_billing_charge_failed": provider.on_billing_charge_failed,
  46. # Printer status events
  47. "on_printer_offline": provider.on_printer_offline,
  48. "on_printer_error": provider.on_printer_error,
  49. "on_ai_failure_detection": provider.on_ai_failure_detection,
  50. "on_filament_low": provider.on_filament_low,
  51. "on_maintenance_due": provider.on_maintenance_due,
  52. # Home Assistant sensor alerts (#1148, #2824). Both directions of this
  53. # file are hand-maintained field maps, so a column missing here reads
  54. # back as the schema default (False) no matter what the row holds.
  55. "on_ha_sensor_alert": provider.on_ha_sensor_alert,
  56. "on_location_ha_sensor_alert": provider.on_location_ha_sensor_alert,
  57. # AMS environmental alarms (regular AMS)
  58. "on_ams_humidity_high": provider.on_ams_humidity_high,
  59. "on_ams_temperature_high": provider.on_ams_temperature_high,
  60. "on_ams_drying_suspended": provider.on_ams_drying_suspended,
  61. # AMS-HT environmental alarms
  62. "on_ams_ht_humidity_high": provider.on_ams_ht_humidity_high,
  63. "on_ams_ht_temperature_high": provider.on_ams_ht_temperature_high,
  64. # Build plate detection
  65. "on_plate_not_empty": provider.on_plate_not_empty,
  66. "on_plate_clear_required": provider.on_plate_clear_required,
  67. # Post-print outcome confirmation (#1898)
  68. "on_print_confirm_request": provider.on_print_confirm_request,
  69. # Bed cooled
  70. "on_bed_cooled": provider.on_bed_cooled,
  71. # First layer complete
  72. "on_first_layer_complete": provider.on_first_layer_complete,
  73. # Messages from connected apps
  74. "on_app_message": bool(provider.on_app_message),
  75. # Inventory stock alerts. Absent here, the toggles above always read
  76. # back off no matter what the row holds — the same hand-maintained
  77. # field map the Home Assistant comment warns about.
  78. "on_stock_reorder_alert": provider.on_stock_reorder_alert,
  79. "on_stock_break_alert": provider.on_stock_break_alert,
  80. # Print queue events
  81. "on_queue_job_added": provider.on_queue_job_added,
  82. "on_queue_job_assigned": provider.on_queue_job_assigned,
  83. "on_queue_job_started": provider.on_queue_job_started,
  84. "on_queue_job_waiting": provider.on_queue_job_waiting,
  85. "on_queue_job_skipped": provider.on_queue_job_skipped,
  86. "on_queue_job_failed": provider.on_queue_job_failed,
  87. "on_queue_completed": provider.on_queue_completed,
  88. # Quiet hours
  89. "quiet_hours_enabled": provider.quiet_hours_enabled,
  90. "quiet_hours_start": provider.quiet_hours_start,
  91. "quiet_hours_end": provider.quiet_hours_end,
  92. # Daily digest
  93. "daily_digest_enabled": provider.daily_digest_enabled,
  94. "daily_digest_time": provider.daily_digest_time,
  95. # Printer filter
  96. "printer_id": provider.printer_id,
  97. # Status tracking
  98. "last_success": provider.last_success,
  99. "last_error": provider.last_error,
  100. "last_error_at": provider.last_error_at,
  101. # Timestamps
  102. "created_at": provider.created_at,
  103. "updated_at": provider.updated_at,
  104. }
  105. # ============================================================================
  106. # Provider List/Create Routes (no path parameters)
  107. # ============================================================================
  108. @router.get("/", response_model=list[NotificationProviderResponse])
  109. async def list_notification_providers(
  110. db: AsyncSession = Depends(get_db),
  111. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
  112. ):
  113. """List all notification providers."""
  114. result = await db.execute(select(NotificationProvider).order_by(NotificationProvider.created_at.desc()))
  115. providers = result.scalars().all()
  116. return [_provider_to_dict(provider) for provider in providers]
  117. @router.post("/", response_model=NotificationProviderResponse)
  118. async def create_notification_provider(
  119. provider_data: NotificationProviderCreate,
  120. db: AsyncSession = Depends(get_db),
  121. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_CREATE),
  122. ):
  123. """Create a new notification provider."""
  124. provider = NotificationProvider(
  125. name=provider_data.name,
  126. provider_type=provider_data.provider_type.value,
  127. enabled=provider_data.enabled,
  128. config=json.dumps(provider_data.config),
  129. # Print lifecycle events
  130. on_print_start=provider_data.on_print_start,
  131. on_print_complete=provider_data.on_print_complete,
  132. on_print_failed=provider_data.on_print_failed,
  133. on_print_stopped=provider_data.on_print_stopped,
  134. on_print_progress=provider_data.on_print_progress,
  135. on_print_missing_spool_assignment=provider_data.on_print_missing_spool_assignment,
  136. on_billing_charge_failed=provider_data.on_billing_charge_failed,
  137. # Printer status events
  138. on_printer_offline=provider_data.on_printer_offline,
  139. on_printer_error=provider_data.on_printer_error,
  140. on_ai_failure_detection=provider_data.on_ai_failure_detection,
  141. on_filament_low=provider_data.on_filament_low,
  142. on_maintenance_due=provider_data.on_maintenance_due,
  143. # Home Assistant sensor alerts (#1148, #2824)
  144. on_ha_sensor_alert=provider_data.on_ha_sensor_alert,
  145. on_location_ha_sensor_alert=provider_data.on_location_ha_sensor_alert,
  146. # AMS environmental alarms (regular AMS)
  147. on_ams_humidity_high=provider_data.on_ams_humidity_high,
  148. on_ams_temperature_high=provider_data.on_ams_temperature_high,
  149. on_ams_drying_suspended=provider_data.on_ams_drying_suspended,
  150. # AMS-HT environmental alarms
  151. on_ams_ht_humidity_high=provider_data.on_ams_ht_humidity_high,
  152. on_ams_ht_temperature_high=provider_data.on_ams_ht_temperature_high,
  153. # Build plate detection
  154. on_plate_not_empty=provider_data.on_plate_not_empty,
  155. on_plate_clear_required=provider_data.on_plate_clear_required,
  156. # Post-print outcome confirmation (#1898)
  157. on_print_confirm_request=provider_data.on_print_confirm_request,
  158. # Bed cooled
  159. on_bed_cooled=provider_data.on_bed_cooled,
  160. # First layer complete
  161. on_first_layer_complete=provider_data.on_first_layer_complete,
  162. on_app_message=provider_data.on_app_message,
  163. # Inventory stock alerts
  164. on_stock_reorder_alert=provider_data.on_stock_reorder_alert,
  165. on_stock_break_alert=provider_data.on_stock_break_alert,
  166. # Print queue events
  167. on_queue_job_added=provider_data.on_queue_job_added,
  168. on_queue_job_assigned=provider_data.on_queue_job_assigned,
  169. on_queue_job_started=provider_data.on_queue_job_started,
  170. on_queue_job_waiting=provider_data.on_queue_job_waiting,
  171. on_queue_job_skipped=provider_data.on_queue_job_skipped,
  172. on_queue_job_failed=provider_data.on_queue_job_failed,
  173. on_queue_completed=provider_data.on_queue_completed,
  174. # Quiet hours
  175. quiet_hours_enabled=provider_data.quiet_hours_enabled,
  176. quiet_hours_start=provider_data.quiet_hours_start,
  177. quiet_hours_end=provider_data.quiet_hours_end,
  178. # Daily digest
  179. daily_digest_enabled=provider_data.daily_digest_enabled,
  180. daily_digest_time=provider_data.daily_digest_time,
  181. # Printer filter
  182. printer_id=provider_data.printer_id,
  183. )
  184. db.add(provider)
  185. await db.commit()
  186. await db.refresh(provider)
  187. logger.info("Created notification provider: %s (%s)", provider.name, provider.provider_type)
  188. return _provider_to_dict(provider)
  189. # ============================================================================
  190. # Static Path Routes (must come BEFORE parameterized routes)
  191. # ============================================================================
  192. @router.post("/test-config", response_model=NotificationTestResponse)
  193. async def test_notification_config(
  194. test_request: NotificationTestRequest,
  195. db: AsyncSession = Depends(get_db),
  196. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_CREATE),
  197. ):
  198. """Test notification configuration before saving."""
  199. success, message = await notification_service.send_test_notification(
  200. test_request.provider_type.value, test_request.config, db
  201. )
  202. return NotificationTestResponse(success=success, message=message)
  203. @router.post("/test-all")
  204. async def test_all_notification_providers(
  205. db: AsyncSession = Depends(get_db),
  206. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_UPDATE),
  207. ):
  208. """Send a test notification to all enabled providers."""
  209. result = await db.execute(select(NotificationProvider).where(NotificationProvider.enabled.is_(True)))
  210. providers = result.scalars().all()
  211. if not providers:
  212. return {"tested": 0, "success": 0, "failed": 0, "results": []}
  213. results = []
  214. success_count = 0
  215. failed_count = 0
  216. for provider in providers:
  217. config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
  218. success, message = await notification_service.send_test_notification(provider.provider_type, config, db)
  219. # Update provider status
  220. if success:
  221. provider.last_success = datetime.now(timezone.utc)
  222. success_count += 1
  223. else:
  224. provider.last_error = message
  225. provider.last_error_at = datetime.now(timezone.utc)
  226. failed_count += 1
  227. results.append(
  228. {
  229. "provider_id": provider.id,
  230. "provider_name": provider.name,
  231. "provider_type": provider.provider_type,
  232. "success": success,
  233. "message": message,
  234. }
  235. )
  236. await db.commit()
  237. return {
  238. "tested": len(providers),
  239. "success": success_count,
  240. "failed": failed_count,
  241. "results": results,
  242. }
  243. # ============================================================================
  244. # Notification Log Routes (must come BEFORE /{provider_id} routes)
  245. # ============================================================================
  246. @router.get("/logs", response_model=list[NotificationLogResponse])
  247. async def get_notification_logs(
  248. limit: int = Query(default=100, ge=1, le=500),
  249. offset: int = Query(default=0, ge=0),
  250. provider_id: int | None = Query(default=None),
  251. event_type: str | None = Query(default=None),
  252. success: bool | None = Query(default=None),
  253. days: int | None = Query(default=7, ge=1, le=90, description="Filter logs from the last N days"),
  254. db: AsyncSession = Depends(get_db),
  255. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
  256. ):
  257. """Get notification logs with optional filters."""
  258. query = select(NotificationLog).order_by(desc(NotificationLog.created_at))
  259. # Apply filters
  260. if provider_id is not None:
  261. query = query.where(NotificationLog.provider_id == provider_id)
  262. if event_type is not None:
  263. query = query.where(NotificationLog.event_type == event_type)
  264. if success is not None:
  265. query = query.where(NotificationLog.success == success)
  266. if days is not None:
  267. cutoff = datetime.now(timezone.utc) - timedelta(days=days)
  268. query = query.where(NotificationLog.created_at >= cutoff)
  269. query = query.offset(offset).limit(limit)
  270. result = await db.execute(query)
  271. logs = result.scalars().all()
  272. # Get provider info for each log
  273. response = []
  274. providers_cache: dict[int, NotificationProvider | None] = {}
  275. for log in logs:
  276. if log.provider_id not in providers_cache:
  277. provider_result = await db.execute(
  278. select(NotificationProvider).where(NotificationProvider.id == log.provider_id)
  279. )
  280. providers_cache[log.provider_id] = provider_result.scalar_one_or_none()
  281. provider = providers_cache[log.provider_id]
  282. response.append(
  283. NotificationLogResponse(
  284. id=log.id,
  285. provider_id=log.provider_id,
  286. provider_name=provider.name if provider else None,
  287. provider_type=provider.provider_type if provider else None,
  288. event_type=log.event_type,
  289. title=log.title,
  290. message=log.message,
  291. success=log.success,
  292. error_message=log.error_message,
  293. printer_id=log.printer_id,
  294. printer_name=log.printer_name,
  295. created_at=log.created_at,
  296. )
  297. )
  298. return response
  299. @router.get("/logs/stats", response_model=NotificationLogStats)
  300. async def get_notification_log_stats(
  301. days: int = Query(default=7, ge=1, le=90, description="Statistics for the last N days"),
  302. db: AsyncSession = Depends(get_db),
  303. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
  304. ):
  305. """Get notification log statistics."""
  306. cutoff = datetime.now(timezone.utc) - timedelta(days=days)
  307. # Total counts
  308. total_result = await db.execute(select(func.count(NotificationLog.id)).where(NotificationLog.created_at >= cutoff))
  309. total = total_result.scalar() or 0
  310. success_result = await db.execute(
  311. select(func.count(NotificationLog.id)).where(
  312. NotificationLog.created_at >= cutoff, NotificationLog.success.is_(True)
  313. )
  314. )
  315. success_count = success_result.scalar() or 0
  316. # By event type
  317. event_result = await db.execute(
  318. select(NotificationLog.event_type, func.count(NotificationLog.id))
  319. .where(NotificationLog.created_at >= cutoff)
  320. .group_by(NotificationLog.event_type)
  321. )
  322. by_event_type = {row[0]: row[1] for row in event_result.fetchall()}
  323. # By provider (need to join to get name)
  324. provider_result = await db.execute(
  325. select(NotificationProvider.name, func.count(NotificationLog.id))
  326. .join(NotificationProvider, NotificationLog.provider_id == NotificationProvider.id)
  327. .where(NotificationLog.created_at >= cutoff)
  328. .group_by(NotificationProvider.name)
  329. )
  330. by_provider = {row[0]: row[1] for row in provider_result.fetchall()}
  331. return NotificationLogStats(
  332. total=total,
  333. success_count=success_count,
  334. failure_count=total - success_count,
  335. by_event_type=by_event_type,
  336. by_provider=by_provider,
  337. )
  338. @router.delete("/logs")
  339. async def clear_notification_logs(
  340. older_than_days: int = Query(default=30, ge=1, description="Delete logs older than N days"),
  341. db: AsyncSession = Depends(get_db),
  342. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_DELETE),
  343. ):
  344. """Clear old notification logs."""
  345. cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days)
  346. result = await db.execute(delete(NotificationLog).where(NotificationLog.created_at < cutoff))
  347. await db.commit()
  348. deleted_count = result.rowcount
  349. logger.info("Deleted %s notification logs older than %s days", deleted_count, older_than_days)
  350. return {"deleted": deleted_count, "message": f"Deleted {deleted_count} logs older than {older_than_days} days"}
  351. # ============================================================================
  352. # Provider Instance Routes (parameterized - must come LAST)
  353. # ============================================================================
  354. # Messages from other applications -------------------------------------------
  355. # Per caller, in memory: enough for any real app (Bambuddy Orders sends a few a
  356. # day), and a buggy or hostile one can't flood the channels.
  357. APP_MESSAGE_LIMIT = 20
  358. APP_MESSAGE_WINDOW_SECONDS = 60
  359. _app_message_times: dict[str, deque[float]] = defaultdict(deque)
  360. def _app_sender(caller: ScopedCaller) -> tuple[str, str]:
  361. """(rate-limit key, name shown in the log) for whoever sends the message."""
  362. if caller.api_key is not None:
  363. return f"key:{caller.api_key.id}", caller.api_key.name
  364. if caller.user is not None:
  365. return f"user:{caller.user.id}", caller.user.username
  366. return "anonymous", "app"
  367. def _check_app_message_rate(key: str) -> None:
  368. times = _app_message_times[key]
  369. cutoff = time.monotonic() - APP_MESSAGE_WINDOW_SECONDS
  370. while times and times[0] < cutoff:
  371. times.popleft()
  372. if len(times) >= APP_MESSAGE_LIMIT:
  373. raise HTTPException(status_code=429, detail="Too many messages; try again in a minute")
  374. times.append(time.monotonic())
  375. @router.post("/app-message", response_model=AppMessageResult)
  376. async def send_app_message(
  377. data: AppMessage,
  378. db: AsyncSession = Depends(get_db),
  379. caller: ScopedCaller = Depends(require_notification_send()),
  380. ):
  381. """Send a message through every enabled channel that has "Messages from
  382. connected apps" on. For other applications, e.g. Bambuddy Orders; an API
  383. key needs the "Send notifications" permission."""
  384. key, sender = _app_sender(caller)
  385. _check_app_message_rate(key)
  386. channels = await notification_service.on_app_message(
  387. db, sender=sender, title=data.title, message=data.message, url=data.url
  388. )
  389. return AppMessageResult(channels=channels)
  390. @router.get("/app-message/channels", response_model=list[AppMessageChannel])
  391. async def app_message_channels(
  392. db: AsyncSession = Depends(get_db),
  393. _: ScopedCaller = Depends(require_notification_send()),
  394. ):
  395. """The enabled channels that deliver app messages: names and types only,
  396. so an app can tell its user where its messages will arrive."""
  397. rows = await db.execute(
  398. select(NotificationProvider)
  399. .where(NotificationProvider.enabled.is_(True), NotificationProvider.on_app_message.is_(True))
  400. .order_by(NotificationProvider.name)
  401. )
  402. return [AppMessageChannel(name=p.name, provider_type=p.provider_type) for p in rows.scalars()]
  403. @router.get("/{provider_id}", response_model=NotificationProviderResponse)
  404. async def get_notification_provider(
  405. provider_id: int,
  406. db: AsyncSession = Depends(get_db),
  407. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
  408. ):
  409. """Get a specific notification provider."""
  410. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  411. provider = result.scalar_one_or_none()
  412. if not provider:
  413. raise HTTPException(status_code=404, detail="Notification provider not found")
  414. return _provider_to_dict(provider)
  415. @router.patch("/{provider_id}", response_model=NotificationProviderResponse)
  416. async def update_notification_provider(
  417. provider_id: int,
  418. update_data: NotificationProviderUpdate,
  419. db: AsyncSession = Depends(get_db),
  420. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_UPDATE),
  421. ):
  422. """Update a notification provider."""
  423. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  424. provider = result.scalar_one_or_none()
  425. if not provider:
  426. raise HTTPException(status_code=404, detail="Notification provider not found")
  427. # Update only provided fields
  428. update_dict = update_data.model_dump(exclude_unset=True)
  429. for key, value in update_dict.items():
  430. if key == "config" and value is not None:
  431. setattr(provider, key, json.dumps(value))
  432. elif key == "provider_type" and value is not None:
  433. setattr(provider, key, value.value)
  434. else:
  435. setattr(provider, key, value)
  436. await db.commit()
  437. await db.refresh(provider)
  438. logger.info("Updated notification provider: %s", provider.name)
  439. return _provider_to_dict(provider)
  440. @router.delete("/{provider_id}")
  441. async def delete_notification_provider(
  442. provider_id: int,
  443. db: AsyncSession = Depends(get_db),
  444. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_DELETE),
  445. ):
  446. """Delete a notification provider."""
  447. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  448. provider = result.scalar_one_or_none()
  449. if not provider:
  450. raise HTTPException(status_code=404, detail="Notification provider not found")
  451. name = provider.name
  452. await db.delete(provider)
  453. await db.commit()
  454. logger.info("Deleted notification provider: %s", name)
  455. return {"message": f"Notification provider '{name}' deleted"}
  456. @router.post("/{provider_id}/test", response_model=NotificationTestResponse)
  457. async def test_notification_provider(
  458. provider_id: int,
  459. db: AsyncSession = Depends(get_db),
  460. _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_UPDATE),
  461. ):
  462. """Send a test notification using an existing provider."""
  463. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  464. provider = result.scalar_one_or_none()
  465. if not provider:
  466. raise HTTPException(status_code=404, detail="Notification provider not found")
  467. config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
  468. success, message = await notification_service.send_test_notification(provider.provider_type, config, db)
  469. # Update provider status
  470. if success:
  471. provider.last_success = datetime.now(timezone.utc)
  472. else:
  473. provider.last_error = message
  474. provider.last_error_at = datetime.now(timezone.utc)
  475. await db.commit()
  476. return NotificationTestResponse(success=success, message=message)