121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
import requests
|
|
import json
|
|
import os
|
|
from PIL import Image
|
|
from PIL.PngImagePlugin import PngInfo
|
|
import numpy as np
|
|
|
|
from datetime import datetime
|
|
import folder_paths
|
|
|
|
class WebhookImage:
|
|
def __init__(self):
|
|
self.output_dir = folder_paths.get_output_directory()
|
|
self.type = 'output'
|
|
self.prefix_append = ''
|
|
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
"""
|
|
Defines the input types for the node.
|
|
|
|
Returns:
|
|
dict: A dictionary specifying required and hidden inputs.
|
|
"""
|
|
return {
|
|
"required": {
|
|
'images': ('IMAGE', ),
|
|
"webhook_url": ("STRING", {'default': '<yourwebhook>'}),
|
|
"verify_ssl": ("BOOLEAN", {"default": True}),
|
|
"safe_prompt": (["enable", "disable"], {"default": "disable"}),
|
|
"notification_text": ('STRING', {'default': 'Your image is ready.'}),
|
|
"json_format": ('STRING', {'default': '{"text": "<notification_text>"}'}),
|
|
"send_notification": (["enable", "disable"], {"default": "disable"}),
|
|
"timeout": ('FLOAT', {'default': 3, 'min': 0, 'max': 60}),
|
|
'image_preview': (['disabled', 'enabled'], {'default': 'enabled'}),
|
|
},
|
|
"optional": {
|
|
"positive_text_opt": ("STRING", {"forceInput": True}),
|
|
"negative_text_opt": ("STRING", {"forceInput": True}),
|
|
},
|
|
"hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"},
|
|
}
|
|
|
|
FUNCTION = 'hookImage'
|
|
OUTPUT_NODE = True
|
|
RETURN_TYPES = tuple()
|
|
CATEGORY = "notifications"
|
|
|
|
@staticmethod
|
|
def sendTxtMessage( webhook_url, notification_text, json_format, timeout, verify_ssl ):
|
|
"""
|
|
handles sending text message to the webhook
|
|
|
|
Args:
|
|
notification_text (str):
|
|
"""
|
|
payload = json_format.replace("<notification_text>", notification_text)
|
|
payload = json.loads(payload)
|
|
res = requests.post( webhook_url, json=payload, timeout=timeout, verify=verify_ssl)
|
|
res.raise_for_status()
|
|
|
|
def hookImage(self, images, webhook_url, notification_text, json_format, timeout, verify_ssl, negative_text_opt=None, positive_text_opt=None, extra_pnginfo=None, prompt=None, safe_prompt="disable", image_preview="disable", send_notification="disable" ):
|
|
counter = 1
|
|
|
|
|
|
results = list()
|
|
|
|
for image in images:
|
|
i = 255. * image.cpu().numpy()
|
|
img = Image.fromarray(np.clip(i, 0, 255).astype(np.uint8))
|
|
|
|
metadata = PngInfo()
|
|
metadata.add_text("webhook_url", webhook_url )
|
|
metadata.add_text("notification_text", notification_text)
|
|
|
|
if prompt is not None:
|
|
if safe_prompt == 'enable':
|
|
metadata.add_text("prompt", json.dumps(prompt))
|
|
|
|
if positive_text_opt is not None:
|
|
metadata.add_text("positive_text_opt", json.dumps(positive_text_opt))
|
|
if negative_text_opt is not None:
|
|
metadata.add_text("negative_text_opt", json.dumps(negative_text_opt))
|
|
if extra_pnginfo is not None:
|
|
for x in extra_pnginfo:
|
|
metadata.add_text(x, json.dumps(extra_pnginfo[x]))
|
|
|
|
timestamp = datetime.now().strftime('%c')
|
|
file = f"telegram_notification_{counter:05}_{timestamp}.png"
|
|
|
|
image_path = os.path.join(self.output_dir, file)
|
|
img.save(image_path, format="PNG", pnginfo=metadata)
|
|
|
|
results.append({ 'filename': file, 'subfolder': '', 'type': self.type})
|
|
counter += 1
|
|
|
|
try:
|
|
with open(image_path, 'rb') as image_file:
|
|
files = {
|
|
'file': ('photo', image_file, 'image/png')
|
|
}
|
|
|
|
res = requests.post( webhook_url, files=files, timeout=timeout, verify=verify_ssl)
|
|
res.raise_for_status()
|
|
|
|
if res.status_code == 200:
|
|
print("Image posted successfully!")
|
|
else:
|
|
print(f"Failed to post image. Status code: {res.status_code}")
|
|
|
|
except OSError as e:
|
|
print(f'An error occurred while sending: {e}')
|
|
else:
|
|
if send_notification == 'enable':
|
|
WebhookImage.sendTxtMessage( webhook_url, notification_text, json_format, timeout, verify_ssl)
|
|
|
|
if image_preview == 'disabled':
|
|
results = list()
|
|
return { 'ui': { 'images': results } }
|