46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
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
|