25 lines
1.6 KiB
Python
25 lines
1.6 KiB
Python
import json, os, ssl, sys
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError, URLError
|
|
PROTO='jarvis.skill.invoke.v1'
|
|
def truthy(v): return str(v or '').strip().lower() in ('1','true','yes','on')
|
|
def fail(c,m): return {'protocol':PROTO,'success':False,'error':{'code':c,'message':m}}
|
|
def ok(d,m='OK',mut=False): return {'protocol':PROTO,'success':True,'data':d,'message':m,'mutated':mut}
|
|
def main(inv):
|
|
x=inv.get('input') or {}; base=os.getenv('NTFY_BASE_URL','https://ntfy.sh').strip().rstrip('/'); topic=(x.get('topic') or os.getenv('NTFY_DEFAULT_TOPIC','')).strip()
|
|
if not topic: return fail('CONFIG_MISSING','NTFY_DEFAULT_TOPIC oder input.topic fehlt')
|
|
h={'Content-Type':'text/plain; charset=utf-8'}; token=os.getenv('NTFY_TOKEN','').strip()
|
|
if token: h['Authorization']='Bearer '+token
|
|
if x.get('title'): h['Title']=x['title']
|
|
if x.get('priority'): h['Priority']=x['priority']
|
|
if x.get('tags'): h['Tags']=x['tags']
|
|
ctx=ssl._create_unverified_context() if base.startswith('https://') and not truthy(os.getenv('NTFY_VERIFY_TLS','true')) else None
|
|
try:
|
|
with urlopen(Request(base+'/'+topic,data=x['message'].encode(),headers=h,method='POST'),timeout=6,context=ctx) as r: r.read()
|
|
except HTTPError as e: raise RuntimeError(f'ntfy HTTP {e.code}: {e.read().decode(errors="replace")[:500]}')
|
|
except URLError as e: raise RuntimeError(f'ntfy nicht erreichbar: {e.reason}')
|
|
return ok({'sent':True,'topic':topic},'Benachrichtigung gesendet.',True)
|
|
try: res=main(json.load(sys.stdin))
|
|
except Exception as e: res=fail('NTFY_ERROR',str(e))
|
|
json.dump(res,sys.stdout,ensure_ascii=False)
|