https://dev.kronoz.win/r/marzneshin.git
1import logging
2from datetime import datetime as dt
3from datetime import timedelta as td
4from typing import Any, Dict, List
5
6import aiohttp
7from fastapi.encoders import jsonable_encoder
8from requests import Session
9
10import config
11from app.db.models import NotificationReminder
12from app.utils.notification import queue
13from app.db import GetDB
14
15logger = logging.getLogger(__name__)
16
17session = Session()
18
19headers = {"x-webhook-secret": config.WEBHOOK_SECRET} if config.WEBHOOK_SECRET else None
20
21
22async def send(data: List[Dict[Any, Any]]) -> bool:
23 """Send the notification to the webhook address provided by config.WEBHOOK_ADDRESS
24
25 Args:
26 data (List[Dict[Any, Any]]): list of json encoded notifications
27
28 Returns:
29 bool: returns True if an ok response received
30 """
31 try:
32 logger.debug(f"Sending {len(data)} webhook updates to {config.WEBHOOK_ADDRESS}")
33
34 r = await aiohttp.request(
35 "post", config.WEBHOOK_ADDRESS, json=data, headers=headers
36 )
37 if r.ok:
38 return True
39 logger.error(r)
40 except Exception as err:
41 logger.error(err)
42 return False
43
44
45async def send_notifications():
46 if not queue:
47 return
48
49 notifications_to_send = list()
50 try:
51 while notification := queue.popleft():
52 if notification.tries > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
53 continue
54 if notification.send_at > dt.utcnow().timestamp():
55 queue.append(
56 notification
57 ) # add it to the queue again for the next check
58 continue
59 notifications_to_send.append(notification)
60 except IndexError: # if the queue is empty
61 pass
62
63 if not notifications_to_send:
64 return
65 if not await send([jsonable_encoder(notif) for notif in notifications_to_send]):
66 for notification in notifications_to_send:
67 if (notification.tries + 1) > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
68 continue
69 notification.tries += 1
70 notification.send_at = ( # schedule notification for n seconds later
71 dt.utcnow() + td(seconds=config.RECURRENT_NOTIFICATIONS_TIMEOUT)
72 ).timestamp()
73 queue.append(notification)
74
75
76def delete_expired_reminders() -> None:
77 with GetDB() as db:
78 db.query(NotificationReminder).filter(
79 NotificationReminder.expires_at < dt.utcnow()
80 ).delete()
81 db.commit()