meisnate12
3 years ago
parent
cc8f925afc
commit
9312c5ec35
@ -0,0 +1,83 @@
|
||||
import logging
|
||||
|
||||
from modules.util import Failed
|
||||
|
||||
logger = logging.getLogger("Plex Meta Manager")
|
||||
|
||||
class Webhooks:
|
||||
def __init__(self, config, system_webhooks, library=None, notifiarr=None):
|
||||
self.config = config
|
||||
self.error_webhooks = system_webhooks["error_webhooks"] if "error_webhooks" in system_webhooks else []
|
||||
self.run_start_webhooks = system_webhooks["run_start_webhooks"] if "run_start_webhooks" in system_webhooks else []
|
||||
self.run_end_webhooks = system_webhooks["run_end_webhooks"] if "run_end_webhooks" in system_webhooks else []
|
||||
self.library = library
|
||||
self.notifiarr = notifiarr
|
||||
|
||||
def _request(self, webhooks, json):
|
||||
if self.config.trace_mode:
|
||||
logger.debug("")
|
||||
logger.debug(f"JSON: {json}")
|
||||
for webhook in webhooks:
|
||||
if self.config.trace_mode:
|
||||
logger.debug(f"Webhook: {webhook}")
|
||||
if webhook == "notifiarr":
|
||||
url, params = self.notifiarr.get_url("notification/plex/")
|
||||
response = self.config.get(url, json=json, params=params)
|
||||
else:
|
||||
response = self.config.post(webhook, json=json)
|
||||
response_json = response.json()
|
||||
if self.config.trace_mode:
|
||||
logger.debug(f"Response: {response_json}")
|
||||
if response.status_code >= 400 or ("result" in response_json and response_json["result"] == "error"):
|
||||
raise Failed(f"({response.status_code} [{response.reason}]) {response_json}")
|
||||
|
||||
def start_time_hooks(self, start_time):
|
||||
if self.run_start_webhooks:
|
||||
self._request(self.run_start_webhooks, {"start_time": start_time})
|
||||
|
||||
def end_time_hooks(self, start_time, run_time, stats):
|
||||
if self.run_end_webhooks:
|
||||
self._request(self.run_end_webhooks, {
|
||||
"start_time": start_time,
|
||||
"run_time": run_time,
|
||||
"collections_created": stats["created"],
|
||||
"collections_modified": stats["modified"],
|
||||
"collections_deleted": stats["deleted"],
|
||||
"items_added": stats["added"],
|
||||
"items_removed": stats["removed"],
|
||||
"added_to_radarr": stats["radarr"],
|
||||
"added_to_sonarr": stats["sonarr"],
|
||||
})
|
||||
|
||||
def error_hooks(self, text, library=None, collection=None, critical=True):
|
||||
if self.error_webhooks:
|
||||
json = {"error": str(text), "critical": critical}
|
||||
if library:
|
||||
json["server_name"] = library.PlexServer.friendlyName
|
||||
json["library_name"] = library.name
|
||||
if collection:
|
||||
json["collection"] = str(collection)
|
||||
self._request(self.error_webhooks, json)
|
||||
|
||||
def collection_hooks(self, webhooks, collection, created=False, additions=None, removals=None):
|
||||
if self.library:
|
||||
thumb = None
|
||||
if collection.thumb and next((f for f in collection.fields if f.name == "thumb"), None):
|
||||
thumb = self.config.get_image_encoded(f"{self.library.url}{collection.thumb}?X-Plex-Token={self.library.token}")
|
||||
art = None
|
||||
if collection.art and next((f for f in collection.fields if f.name == "art"), None):
|
||||
art = self.config.get_image_encoded(f"{self.library.url}{collection.art}?X-Plex-Token={self.library.token}")
|
||||
json = {
|
||||
"server_name": self.library.PlexServer.friendlyName,
|
||||
"library_name": self.library.name,
|
||||
"type": "movie" if self.library.is_movie else "show",
|
||||
"collection": collection.title,
|
||||
"created": created,
|
||||
"poster": thumb,
|
||||
"background": art
|
||||
}
|
||||
if additions:
|
||||
json["additions"] = additions
|
||||
if removals:
|
||||
json["removals"] = removals
|
||||
self._request(webhooks, json)
|
||||
Loading…
Reference in new issue