46 lines
1.2 KiB
Python
46 lines
1.2 KiB
Python
"""Push opcional via ntfy.sh (sem instalar servidor na VM122)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
NTFY_BASE_URL = os.getenv("DESK_NTFY_BASE_URL", "https://ntfy.sh").rstrip("/")
|
|
|
|
|
|
def _ascii_header(value: str) -> str:
|
|
"""HTTP headers exigem latin-1; remove acentos e tracos especiais."""
|
|
return (
|
|
(value or "")
|
|
.replace("\u2014", "-")
|
|
.replace("\u2013", "-")
|
|
.encode("ascii", "ignore")
|
|
.decode("ascii")
|
|
)
|
|
|
|
|
|
def push(topic: str, title: str, message: str, priority: str = "default") -> bool:
|
|
topic = (topic or "").strip()
|
|
if not topic:
|
|
return False
|
|
url = f"{NTFY_BASE_URL}/{topic}"
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=message.encode("utf-8"),
|
|
method="POST",
|
|
headers={
|
|
"Title": _ascii_header(title),
|
|
"Priority": priority,
|
|
"Tags": "key",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=8) as resp:
|
|
return 200 <= resp.status < 300
|
|
except (urllib.error.URLError, TimeoutError, OSError):
|
|
return False
|
|
|
|
|
|
def subscribe_url(topic: str) -> str:
|
|
return f"{NTFY_BASE_URL}/{topic}"
|