112 lines
3.8 KiB
Python
Executable File
112 lines
3.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# args_json: {"type":"object","required":["summary"],"properties":{"summary":{"type":"string","description":"Notification title"},"body":{"type":"string","description":"Notification body"},"actions":{"type":"string","description":"Comma-separated action labels"},"timeout":{"type":"integer","description":"Timeout in ms (default: 5000, 10000 with actions)"},"icon":{"type":"string","description":"Icon name (default: ollie)"},"urgency":{"type":"string","description":"low|normal|critical (default: normal)"}}}
|
|
# ollie:prompt
|
|
# ## gui_notify
|
|
#
|
|
# Show desktop notifications.
|
|
#
|
|
# ```
|
|
# gui_notify(summary="Build complete", body="All tests passed")
|
|
# ```
|
|
# ollie:end
|
|
"""Send interactive notifications via org.freedesktop.Notifications D-Bus interface."""
|
|
import sys
|
|
import os
|
|
import subprocess
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from _lib.args import parse_args
|
|
|
|
def main():
|
|
args = parse_args()
|
|
summary = args.require("summary")
|
|
body = args.get("body", "")
|
|
actions_str = args.get("actions", "")
|
|
timeout = args.get_int("timeout", 0)
|
|
icon = args.get("icon", "ollie")
|
|
urgency = args.get("urgency", "normal")
|
|
|
|
# Build actions array
|
|
actions_list = []
|
|
if actions_str:
|
|
labels = [a.strip() for a in actions_str.split(",") if a.strip()]
|
|
for i, label in enumerate(labels):
|
|
actions_list.append(str(i))
|
|
actions_list.append(label)
|
|
|
|
if timeout == 0:
|
|
timeout = 10000 if actions_list else 5000
|
|
|
|
urgency_map = {"low": 0, "normal": 1, "critical": 2}
|
|
|
|
actions_gdbus = "[" + ", ".join(f"'{a}'" for a in actions_list) + "]"
|
|
hints_gdbus = "{'urgency': <byte %d>}" % urgency_map.get(urgency, 1)
|
|
|
|
cmd = [
|
|
"gdbus", "call", "--session",
|
|
"--dest", "org.freedesktop.Notifications",
|
|
"--object-path", "/org/freedesktop/Notifications",
|
|
"--method", "org.freedesktop.Notifications.Notify",
|
|
"ollie", "0", icon, summary, body,
|
|
actions_gdbus, hints_gdbus, str(timeout),
|
|
]
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
print(f"ERROR: {result.stderr.strip()}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
nid = result.stdout.strip().strip("()").split(",")[0].replace("uint32 ", "")
|
|
|
|
if not actions_list:
|
|
print(f"notification:{nid}")
|
|
return
|
|
|
|
# Wait for ActionInvoked or NotificationClosed signal
|
|
labels = [actions_list[i+1] for i in range(0, len(actions_list), 2)]
|
|
|
|
monitor_cmd = [
|
|
"gdbus", "monitor", "--session",
|
|
"--dest", "org.freedesktop.Notifications",
|
|
"--object-path", "/org/freedesktop/Notifications",
|
|
]
|
|
|
|
import signal
|
|
|
|
class TimeoutError(Exception):
|
|
pass
|
|
|
|
def alarm_handler(signum, frame):
|
|
raise TimeoutError()
|
|
|
|
signal.signal(signal.SIGALRM, alarm_handler)
|
|
signal.alarm((timeout // 1000) + 2)
|
|
|
|
try:
|
|
proc = subprocess.Popen(monitor_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
for line in proc.stdout:
|
|
if "ActionInvoked" in line and nid in line:
|
|
parts = line.split("'")
|
|
if len(parts) >= 2:
|
|
action_id = parts[1]
|
|
try:
|
|
idx = int(action_id)
|
|
print(labels[idx])
|
|
except (ValueError, IndexError):
|
|
print(action_id)
|
|
proc.terminate()
|
|
return
|
|
if "NotificationClosed" in line and nid in line:
|
|
print("dismissed")
|
|
proc.terminate()
|
|
return
|
|
except TimeoutError:
|
|
proc.terminate()
|
|
print("timeout")
|
|
except Exception as e:
|
|
print(f"ERROR: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|