310 lines
10 KiB
Python
310 lines
10 KiB
Python
import asyncio
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import List, TypedDict
|
|
import uuid
|
|
|
|
import httpx
|
|
|
|
from .config import Config
|
|
from .ItemDTO import ItemDTO
|
|
from .Singleton import Singleton
|
|
from .Utils import ag, validateUUID
|
|
from .version import APP_VERSION
|
|
from .encoder import Encoder
|
|
|
|
LOG = logging.getLogger("notifications")
|
|
|
|
|
|
class targetRequestHeader(TypedDict):
|
|
"""Request header details for a notification target."""
|
|
|
|
key: str
|
|
value: str
|
|
|
|
|
|
class targetRequest(TypedDict):
|
|
"""Request details for a notification target."""
|
|
|
|
type: str
|
|
method: str
|
|
url: str
|
|
headers: list[targetRequestHeader] = []
|
|
|
|
|
|
class Target(TypedDict):
|
|
"""Notification target details."""
|
|
|
|
id: str
|
|
name: str
|
|
on: List[str]
|
|
request: targetRequest
|
|
|
|
|
|
class Notification(metaclass=Singleton):
|
|
targets: list[Target] = []
|
|
"""Notification targets to send events to."""
|
|
|
|
allowed_events: tuple = (
|
|
"added",
|
|
"completed",
|
|
"error",
|
|
"not_live",
|
|
"canceled",
|
|
"cleared",
|
|
"log_info",
|
|
"log_success",
|
|
)
|
|
"""Events that can be sent to the targets."""
|
|
|
|
_instance = None
|
|
|
|
def __init__(self):
|
|
Notification._instance = self
|
|
|
|
config = Config.get_instance()
|
|
|
|
self._debug = config.debug
|
|
self.file: str = os.path.join(config.config_path, "notifications.json")
|
|
self._client: httpx.AsyncClient = httpx.AsyncClient()
|
|
self._encoder: Encoder = Encoder()
|
|
|
|
if os.path.exists(self.file) and os.path.getsize(self.file) > 10:
|
|
self.load()
|
|
|
|
@staticmethod
|
|
def get_instance() -> "Notification":
|
|
if Notification._instance is None:
|
|
Notification._instance = Notification()
|
|
|
|
return Notification._instance
|
|
|
|
def getTargets(self) -> list[Target]:
|
|
"""Get the list of notification targets."""
|
|
return self.targets
|
|
|
|
def clearTargets(self) -> "Notification":
|
|
"""Clear the list of notification targets."""
|
|
self.targets.clear()
|
|
return self
|
|
|
|
def save(self, targets: list[Target] | None = None) -> "Notification":
|
|
"""
|
|
Save notification targets to the file.
|
|
|
|
Args:
|
|
targets (list[Target]|None): The list of targets to save.
|
|
|
|
Returns:
|
|
Notification: The Notification instance.
|
|
"""
|
|
LOG.info(f"Saving notification targets to '{self.file}'.")
|
|
try:
|
|
with open(self.file, "w") as f:
|
|
f.write(json.dumps(targets or self.targets, indent=4))
|
|
except Exception as e:
|
|
LOG.error(f"Error saving notification targets to '{self.file}'. '{e}'")
|
|
pass
|
|
|
|
return self
|
|
|
|
def load(self):
|
|
"""Load or reload notification targets from the file."""
|
|
if len(self.targets) > 0:
|
|
LOG.info("Clearing existing notification targets.")
|
|
self.targets.clear()
|
|
|
|
modified = False
|
|
|
|
targets = []
|
|
|
|
if not os.path.exists(self.file) or os.path.getsize(self.file) < 10:
|
|
return
|
|
|
|
LOG.info(f"Loading notification targets from '{self.file}'.")
|
|
|
|
try:
|
|
with open(self.file, "r") as f:
|
|
targets = json.load(f)
|
|
except Exception as e:
|
|
LOG.error(f"Error loading notification targets from '{self.file}'. '{e}'")
|
|
pass
|
|
|
|
for target in targets:
|
|
try:
|
|
try:
|
|
Notification.validate(target)
|
|
except ValueError as e:
|
|
LOG.error(f"Invalid notification target '{target}'. '{e}'")
|
|
continue
|
|
|
|
if "on" not in target:
|
|
target["on"] = []
|
|
modified = True
|
|
|
|
if "type" not in target["request"]:
|
|
target["request"]["type"] = "json"
|
|
modified = True
|
|
|
|
if "method" not in target["request"]:
|
|
target["request"]["method"] = "POST"
|
|
modified = True
|
|
|
|
if "id" not in target or validateUUID(target["id"], version=4) is False:
|
|
target["id"] = str(uuid.uuid4())
|
|
modified = True
|
|
|
|
target = Target(
|
|
id=target.get("id"),
|
|
name=target.get("name"),
|
|
on=target.get("on", []),
|
|
request=targetRequest(
|
|
type=target.get("request", {}).get("type", "json"),
|
|
method=target.get("request", {}).get("method", "POST"),
|
|
url=target.get("request", {}).get("url"),
|
|
headers=[
|
|
targetRequestHeader(key=h.get("key"), value=h.get("value"))
|
|
for h in target.get("request", {}).get("headers", [])
|
|
],
|
|
),
|
|
)
|
|
|
|
self.targets.append(target)
|
|
|
|
LOG.info(
|
|
f"Will send '{target['on'] if len(target['on']) > 0 else 'all'}' as {target['request']['type']} notification events to '{target['name']}'."
|
|
)
|
|
except Exception as e:
|
|
LOG.error(f"Error loading notification target '{target}'. '{e}'")
|
|
pass
|
|
|
|
if modified:
|
|
self.save()
|
|
|
|
@staticmethod
|
|
def validate(target: Target | dict) -> bool:
|
|
"""
|
|
Validate a notification target.
|
|
|
|
Args:
|
|
target (Target|dict): The target to validate.
|
|
|
|
Returns:
|
|
bool: True if the target is valid, False otherwise.
|
|
"""
|
|
|
|
if "id" not in target or validateUUID(target["id"], version=4) is False:
|
|
raise ValueError("Invalid notification target. No ID found.")
|
|
|
|
if "name" not in target:
|
|
raise ValueError("Invalid notification target. No name found.")
|
|
|
|
if "request" not in target:
|
|
raise ValueError("Invalid notification target. No request details found.")
|
|
|
|
if "url" not in target["request"]:
|
|
raise ValueError("Invalid notification target. No URL found.")
|
|
|
|
if "method" in target["request"] and target["request"]["method"].upper() not in ["POST", "PUT"]:
|
|
raise ValueError("Invalid notification target. Invalid method found.")
|
|
|
|
if "type" in target["request"] and target["request"]["type"].lower() not in ["json", "form"]:
|
|
raise ValueError("Invalid notification target. Invalid type found.")
|
|
|
|
if "on" in target:
|
|
if not isinstance(target["on"], list):
|
|
raise ValueError("Invalid notification target. Invalid 'on' event list found.")
|
|
|
|
for e in target["on"]:
|
|
if e not in Notification.allowed_events:
|
|
raise ValueError(f"Invalid notification target. Invalid event '{e}' found.")
|
|
|
|
if "headers" in target["request"]:
|
|
if not isinstance(target["request"]["headers"], list):
|
|
raise ValueError("Invalid notification target. Invalid headers list found.")
|
|
|
|
for h in target["request"]["headers"]:
|
|
if "key" not in h:
|
|
raise ValueError("Invalid notification target. No header key found.")
|
|
if "value" not in h:
|
|
raise ValueError("Invalid notification target. No header value found.")
|
|
|
|
return True
|
|
|
|
async def send(self, event: str, item: ItemDTO | dict) -> list[dict]:
|
|
if len(self.targets) < 1:
|
|
return []
|
|
|
|
if not isinstance(item, ItemDTO) and not isinstance(item, dict):
|
|
LOG.debug(f"Received invalid item type '{type(item)}' with event '{event}'.")
|
|
return []
|
|
|
|
tasks = []
|
|
|
|
for target in self.targets:
|
|
if len(target["on"]) > 0 and event not in target["on"] and "test" != event:
|
|
continue
|
|
|
|
tasks.append(self._send(event, target, item))
|
|
|
|
return await asyncio.gather(*tasks)
|
|
|
|
async def _send(self, event: str, target: Target, item: ItemDTO | dict) -> dict:
|
|
req: targetRequest = target["request"]
|
|
|
|
try:
|
|
itemId = ag(item, ["id", "_id"], "??")
|
|
except Exception:
|
|
itemId = "??"
|
|
|
|
try:
|
|
LOG.info(f"Sending Notification event '{event}' id '{itemId}' to '{target.get('name')}'.")
|
|
reqBody = {
|
|
"method": req["method"].upper(),
|
|
"url": req["url"],
|
|
"headers": {
|
|
"User-Agent": f"YTPTube/{APP_VERSION}",
|
|
"Content-Type": "application/json"
|
|
if "json" == req["type"].lower()
|
|
else "application/x-www-form-urlencoded",
|
|
},
|
|
}
|
|
|
|
if len(req["headers"]) > 0:
|
|
for h in req["headers"]:
|
|
reqBody["headers"][h["key"]] = h["value"]
|
|
|
|
reqBody["json" if "json" == req["type"].lower() else "data"] = {
|
|
"event": event,
|
|
"created_at": datetime.now(tz=timezone.utc).isoformat(),
|
|
"payload": item.__dict__ if isinstance(item, ItemDTO) else item,
|
|
}
|
|
|
|
if "form" == req["type"].lower():
|
|
reqBody["data"]["payload"] = self._encoder.encode(reqBody["data"]["payload"])
|
|
|
|
response = await self._client.request(**reqBody)
|
|
|
|
respData = {"url": req["url"], "status": response.status_code, "text": response.text}
|
|
|
|
msg = f"Notification target '{target['name']}' Responded to event '{event}' id '{itemId}' with status '{response.status_code}'."
|
|
if self._debug and respData.get("text"):
|
|
msg += f" body '{respData.get('text','??')}'."
|
|
|
|
LOG.info(msg)
|
|
|
|
return respData
|
|
except Exception as e:
|
|
LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target['name']}'. '{e}'.")
|
|
return {"url": req["url"], "status": 500, "text": str(e)}
|
|
|
|
def emit(self, event, data, **kwargs):
|
|
if len(self.targets) < 1:
|
|
return False
|
|
|
|
if event not in self.allowed_events and "test" != event:
|
|
return False
|
|
|
|
return self.send(event, data)
|