@@ -0,0 +1,45 @@
|
||||
import ipaddress
|
||||
import socket
|
||||
from urllib.parse import urlparse
|
||||
import requests
|
||||
|
||||
|
||||
def validate_notification_url(url: str):
|
||||
parsed = urlparse(url or "")
|
||||
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
|
||||
raise ValueError("Adres ntfy musi być adresem HTTP(S) bez danych logowania")
|
||||
try:
|
||||
addresses = {info[4][0] for info in socket.getaddrinfo(parsed.hostname, parsed.port, type=socket.SOCK_STREAM)}
|
||||
if any(ipaddress.ip_address(address).is_private or ipaddress.ip_address(address).is_loopback or
|
||||
ipaddress.ip_address(address).is_link_local for address in addresses):
|
||||
raise ValueError("Adres ntfy wskazuje na prywatną lub lokalną sieć")
|
||||
except socket.gaierror as exc:
|
||||
raise ValueError("Nie można rozpoznać hosta ntfy") from exc
|
||||
|
||||
def send_notification(url: str, topic: str, title: str, message: str, username: str = None, password: str = None):
|
||||
if not url or not topic:
|
||||
return False
|
||||
validate_notification_url(url)
|
||||
|
||||
full_url = f"{url.rstrip('/')}/{topic}"
|
||||
headers = {
|
||||
"Title": title.encode('utf-8')
|
||||
}
|
||||
|
||||
auth = None
|
||||
if username and password:
|
||||
auth = (username, password)
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
full_url,
|
||||
data=message.encode('utf-8'),
|
||||
headers=headers,
|
||||
auth=auth,
|
||||
timeout=5
|
||||
)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Failed to send ntfy notification: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user