notifications.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436
  1. """API routes for notification providers."""
  2. import json
  3. import logging
  4. from datetime import datetime, timedelta
  5. from fastapi import APIRouter, Depends, HTTPException, Query
  6. from sqlalchemy import delete, desc, func, select
  7. from sqlalchemy.ext.asyncio import AsyncSession
  8. from backend.app.core.database import get_db
  9. from backend.app.models.notification import NotificationLog, NotificationProvider
  10. from backend.app.schemas.notification import (
  11. NotificationLogResponse,
  12. NotificationLogStats,
  13. NotificationProviderCreate,
  14. NotificationProviderResponse,
  15. NotificationProviderUpdate,
  16. NotificationTestRequest,
  17. NotificationTestResponse,
  18. )
  19. from backend.app.services.notification_service import notification_service
  20. logger = logging.getLogger(__name__)
  21. router = APIRouter(prefix="/notifications", tags=["notifications"])
  22. def _provider_to_dict(provider: NotificationProvider) -> dict:
  23. """Convert a NotificationProvider model to a response dictionary."""
  24. return {
  25. "id": provider.id,
  26. "name": provider.name,
  27. "provider_type": provider.provider_type,
  28. "enabled": provider.enabled,
  29. "config": json.loads(provider.config) if isinstance(provider.config, str) else provider.config,
  30. # Print lifecycle events
  31. "on_print_start": provider.on_print_start,
  32. "on_print_complete": provider.on_print_complete,
  33. "on_print_failed": provider.on_print_failed,
  34. "on_print_stopped": provider.on_print_stopped,
  35. "on_print_progress": provider.on_print_progress,
  36. # Printer status events
  37. "on_printer_offline": provider.on_printer_offline,
  38. "on_printer_error": provider.on_printer_error,
  39. "on_filament_low": provider.on_filament_low,
  40. "on_maintenance_due": provider.on_maintenance_due,
  41. # AMS environmental alarms (regular AMS)
  42. "on_ams_humidity_high": provider.on_ams_humidity_high,
  43. "on_ams_temperature_high": provider.on_ams_temperature_high,
  44. # AMS-HT environmental alarms
  45. "on_ams_ht_humidity_high": provider.on_ams_ht_humidity_high,
  46. "on_ams_ht_temperature_high": provider.on_ams_ht_temperature_high,
  47. # Build plate detection
  48. "on_plate_not_empty": provider.on_plate_not_empty,
  49. # Print queue events
  50. "on_queue_job_added": provider.on_queue_job_added,
  51. "on_queue_job_assigned": provider.on_queue_job_assigned,
  52. "on_queue_job_started": provider.on_queue_job_started,
  53. "on_queue_job_waiting": provider.on_queue_job_waiting,
  54. "on_queue_job_skipped": provider.on_queue_job_skipped,
  55. "on_queue_job_failed": provider.on_queue_job_failed,
  56. "on_queue_completed": provider.on_queue_completed,
  57. # Quiet hours
  58. "quiet_hours_enabled": provider.quiet_hours_enabled,
  59. "quiet_hours_start": provider.quiet_hours_start,
  60. "quiet_hours_end": provider.quiet_hours_end,
  61. # Daily digest
  62. "daily_digest_enabled": provider.daily_digest_enabled,
  63. "daily_digest_time": provider.daily_digest_time,
  64. # Printer filter
  65. "printer_id": provider.printer_id,
  66. # Status tracking
  67. "last_success": provider.last_success,
  68. "last_error": provider.last_error,
  69. "last_error_at": provider.last_error_at,
  70. # Timestamps
  71. "created_at": provider.created_at,
  72. "updated_at": provider.updated_at,
  73. }
  74. # ============================================================================
  75. # Provider List/Create Routes (no path parameters)
  76. # ============================================================================
  77. @router.get("/", response_model=list[NotificationProviderResponse])
  78. async def list_notification_providers(db: AsyncSession = Depends(get_db)):
  79. """List all notification providers."""
  80. result = await db.execute(select(NotificationProvider).order_by(NotificationProvider.created_at.desc()))
  81. providers = result.scalars().all()
  82. return [_provider_to_dict(provider) for provider in providers]
  83. @router.post("/", response_model=NotificationProviderResponse)
  84. async def create_notification_provider(
  85. provider_data: NotificationProviderCreate,
  86. db: AsyncSession = Depends(get_db),
  87. ):
  88. """Create a new notification provider."""
  89. provider = NotificationProvider(
  90. name=provider_data.name,
  91. provider_type=provider_data.provider_type.value,
  92. enabled=provider_data.enabled,
  93. config=json.dumps(provider_data.config),
  94. # Print lifecycle events
  95. on_print_start=provider_data.on_print_start,
  96. on_print_complete=provider_data.on_print_complete,
  97. on_print_failed=provider_data.on_print_failed,
  98. on_print_stopped=provider_data.on_print_stopped,
  99. on_print_progress=provider_data.on_print_progress,
  100. # Printer status events
  101. on_printer_offline=provider_data.on_printer_offline,
  102. on_printer_error=provider_data.on_printer_error,
  103. on_filament_low=provider_data.on_filament_low,
  104. on_maintenance_due=provider_data.on_maintenance_due,
  105. # AMS environmental alarms (regular AMS)
  106. on_ams_humidity_high=provider_data.on_ams_humidity_high,
  107. on_ams_temperature_high=provider_data.on_ams_temperature_high,
  108. # AMS-HT environmental alarms
  109. on_ams_ht_humidity_high=provider_data.on_ams_ht_humidity_high,
  110. on_ams_ht_temperature_high=provider_data.on_ams_ht_temperature_high,
  111. # Build plate detection
  112. on_plate_not_empty=provider_data.on_plate_not_empty,
  113. # Quiet hours
  114. quiet_hours_enabled=provider_data.quiet_hours_enabled,
  115. quiet_hours_start=provider_data.quiet_hours_start,
  116. quiet_hours_end=provider_data.quiet_hours_end,
  117. # Daily digest
  118. daily_digest_enabled=provider_data.daily_digest_enabled,
  119. daily_digest_time=provider_data.daily_digest_time,
  120. # Printer filter
  121. printer_id=provider_data.printer_id,
  122. )
  123. db.add(provider)
  124. await db.commit()
  125. await db.refresh(provider)
  126. logger.info(f"Created notification provider: {provider.name} ({provider.provider_type})")
  127. return _provider_to_dict(provider)
  128. # ============================================================================
  129. # Static Path Routes (must come BEFORE parameterized routes)
  130. # ============================================================================
  131. @router.post("/test-config", response_model=NotificationTestResponse)
  132. async def test_notification_config(
  133. test_request: NotificationTestRequest,
  134. db: AsyncSession = Depends(get_db),
  135. ):
  136. """Test notification configuration before saving."""
  137. success, message = await notification_service.send_test_notification(
  138. test_request.provider_type.value, test_request.config, db
  139. )
  140. return NotificationTestResponse(success=success, message=message)
  141. @router.post("/test-all")
  142. async def test_all_notification_providers(db: AsyncSession = Depends(get_db)):
  143. """Send a test notification to all enabled providers."""
  144. result = await db.execute(select(NotificationProvider).where(NotificationProvider.enabled.is_(True)))
  145. providers = result.scalars().all()
  146. if not providers:
  147. return {"tested": 0, "success": 0, "failed": 0, "results": []}
  148. results = []
  149. success_count = 0
  150. failed_count = 0
  151. for provider in providers:
  152. config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
  153. success, message = await notification_service.send_test_notification(provider.provider_type, config, db)
  154. # Update provider status
  155. if success:
  156. provider.last_success = datetime.utcnow()
  157. success_count += 1
  158. else:
  159. provider.last_error = message
  160. provider.last_error_at = datetime.utcnow()
  161. failed_count += 1
  162. results.append(
  163. {
  164. "provider_id": provider.id,
  165. "provider_name": provider.name,
  166. "provider_type": provider.provider_type,
  167. "success": success,
  168. "message": message,
  169. }
  170. )
  171. await db.commit()
  172. return {
  173. "tested": len(providers),
  174. "success": success_count,
  175. "failed": failed_count,
  176. "results": results,
  177. }
  178. # ============================================================================
  179. # Notification Log Routes (must come BEFORE /{provider_id} routes)
  180. # ============================================================================
  181. @router.get("/logs", response_model=list[NotificationLogResponse])
  182. async def get_notification_logs(
  183. limit: int = Query(default=100, ge=1, le=500),
  184. offset: int = Query(default=0, ge=0),
  185. provider_id: int | None = Query(default=None),
  186. event_type: str | None = Query(default=None),
  187. success: bool | None = Query(default=None),
  188. days: int | None = Query(default=7, ge=1, le=90, description="Filter logs from the last N days"),
  189. db: AsyncSession = Depends(get_db),
  190. ):
  191. """Get notification logs with optional filters."""
  192. query = select(NotificationLog).order_by(desc(NotificationLog.created_at))
  193. # Apply filters
  194. if provider_id is not None:
  195. query = query.where(NotificationLog.provider_id == provider_id)
  196. if event_type is not None:
  197. query = query.where(NotificationLog.event_type == event_type)
  198. if success is not None:
  199. query = query.where(NotificationLog.success == success)
  200. if days is not None:
  201. cutoff = datetime.utcnow() - timedelta(days=days)
  202. query = query.where(NotificationLog.created_at >= cutoff)
  203. query = query.offset(offset).limit(limit)
  204. result = await db.execute(query)
  205. logs = result.scalars().all()
  206. # Get provider info for each log
  207. response = []
  208. providers_cache: dict[int, NotificationProvider | None] = {}
  209. for log in logs:
  210. if log.provider_id not in providers_cache:
  211. provider_result = await db.execute(
  212. select(NotificationProvider).where(NotificationProvider.id == log.provider_id)
  213. )
  214. providers_cache[log.provider_id] = provider_result.scalar_one_or_none()
  215. provider = providers_cache[log.provider_id]
  216. response.append(
  217. NotificationLogResponse(
  218. id=log.id,
  219. provider_id=log.provider_id,
  220. provider_name=provider.name if provider else None,
  221. provider_type=provider.provider_type if provider else None,
  222. event_type=log.event_type,
  223. title=log.title,
  224. message=log.message,
  225. success=log.success,
  226. error_message=log.error_message,
  227. printer_id=log.printer_id,
  228. printer_name=log.printer_name,
  229. created_at=log.created_at,
  230. )
  231. )
  232. return response
  233. @router.get("/logs/stats", response_model=NotificationLogStats)
  234. async def get_notification_log_stats(
  235. days: int = Query(default=7, ge=1, le=90, description="Statistics for the last N days"),
  236. db: AsyncSession = Depends(get_db),
  237. ):
  238. """Get notification log statistics."""
  239. cutoff = datetime.utcnow() - timedelta(days=days)
  240. # Total counts
  241. total_result = await db.execute(select(func.count(NotificationLog.id)).where(NotificationLog.created_at >= cutoff))
  242. total = total_result.scalar() or 0
  243. success_result = await db.execute(
  244. select(func.count(NotificationLog.id)).where(
  245. NotificationLog.created_at >= cutoff, NotificationLog.success.is_(True)
  246. )
  247. )
  248. success_count = success_result.scalar() or 0
  249. # By event type
  250. event_result = await db.execute(
  251. select(NotificationLog.event_type, func.count(NotificationLog.id))
  252. .where(NotificationLog.created_at >= cutoff)
  253. .group_by(NotificationLog.event_type)
  254. )
  255. by_event_type = {row[0]: row[1] for row in event_result.fetchall()}
  256. # By provider (need to join to get name)
  257. provider_result = await db.execute(
  258. select(NotificationProvider.name, func.count(NotificationLog.id))
  259. .join(NotificationProvider, NotificationLog.provider_id == NotificationProvider.id)
  260. .where(NotificationLog.created_at >= cutoff)
  261. .group_by(NotificationProvider.name)
  262. )
  263. by_provider = {row[0]: row[1] for row in provider_result.fetchall()}
  264. return NotificationLogStats(
  265. total=total,
  266. success_count=success_count,
  267. failure_count=total - success_count,
  268. by_event_type=by_event_type,
  269. by_provider=by_provider,
  270. )
  271. @router.delete("/logs")
  272. async def clear_notification_logs(
  273. older_than_days: int = Query(default=30, ge=1, description="Delete logs older than N days"),
  274. db: AsyncSession = Depends(get_db),
  275. ):
  276. """Clear old notification logs."""
  277. cutoff = datetime.utcnow() - timedelta(days=older_than_days)
  278. result = await db.execute(delete(NotificationLog).where(NotificationLog.created_at < cutoff))
  279. await db.commit()
  280. deleted_count = result.rowcount
  281. logger.info(f"Deleted {deleted_count} notification logs older than {older_than_days} days")
  282. return {"deleted": deleted_count, "message": f"Deleted {deleted_count} logs older than {older_than_days} days"}
  283. # ============================================================================
  284. # Provider Instance Routes (parameterized - must come LAST)
  285. # ============================================================================
  286. @router.get("/{provider_id}", response_model=NotificationProviderResponse)
  287. async def get_notification_provider(
  288. provider_id: int,
  289. db: AsyncSession = Depends(get_db),
  290. ):
  291. """Get a specific notification provider."""
  292. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  293. provider = result.scalar_one_or_none()
  294. if not provider:
  295. raise HTTPException(status_code=404, detail="Notification provider not found")
  296. return _provider_to_dict(provider)
  297. @router.patch("/{provider_id}", response_model=NotificationProviderResponse)
  298. async def update_notification_provider(
  299. provider_id: int,
  300. update_data: NotificationProviderUpdate,
  301. db: AsyncSession = Depends(get_db),
  302. ):
  303. """Update a notification provider."""
  304. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  305. provider = result.scalar_one_or_none()
  306. if not provider:
  307. raise HTTPException(status_code=404, detail="Notification provider not found")
  308. # Update only provided fields
  309. update_dict = update_data.model_dump(exclude_unset=True)
  310. for key, value in update_dict.items():
  311. if key == "config" and value is not None:
  312. setattr(provider, key, json.dumps(value))
  313. elif key == "provider_type" and value is not None:
  314. setattr(provider, key, value.value)
  315. else:
  316. setattr(provider, key, value)
  317. await db.commit()
  318. await db.refresh(provider)
  319. logger.info(f"Updated notification provider: {provider.name}")
  320. return _provider_to_dict(provider)
  321. @router.delete("/{provider_id}")
  322. async def delete_notification_provider(
  323. provider_id: int,
  324. db: AsyncSession = Depends(get_db),
  325. ):
  326. """Delete a notification provider."""
  327. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  328. provider = result.scalar_one_or_none()
  329. if not provider:
  330. raise HTTPException(status_code=404, detail="Notification provider not found")
  331. name = provider.name
  332. await db.delete(provider)
  333. await db.commit()
  334. logger.info(f"Deleted notification provider: {name}")
  335. return {"message": f"Notification provider '{name}' deleted"}
  336. @router.post("/{provider_id}/test", response_model=NotificationTestResponse)
  337. async def test_notification_provider(
  338. provider_id: int,
  339. db: AsyncSession = Depends(get_db),
  340. ):
  341. """Send a test notification using an existing provider."""
  342. result = await db.execute(select(NotificationProvider).where(NotificationProvider.id == provider_id))
  343. provider = result.scalar_one_or_none()
  344. if not provider:
  345. raise HTTPException(status_code=404, detail="Notification provider not found")
  346. config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
  347. success, message = await notification_service.send_test_notification(provider.provider_type, config, db)
  348. # Update provider status
  349. if success:
  350. provider.last_success = datetime.utcnow()
  351. else:
  352. provider.last_error = message
  353. provider.last_error_at = datetime.utcnow()
  354. await db.commit()
  355. return NotificationTestResponse(success=success, message=message)