130 lines
4.4 KiB
Python
Executable File
130 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# ollie:prompt
|
|
# ## gui_notify
|
|
#
|
|
# Show desktop notifications.
|
|
#
|
|
# **Args**: `[summary]` or `[summary, body]` or `[summary, body, "--timeout=ms"]`
|
|
#
|
|
# ```
|
|
# call_tool: calls=[{tool: "gui_notify", args: ["Build complete", "All tests passed"]}]
|
|
# ```
|
|
# ollie:end
|
|
"""Send interactive notifications via org.freedesktop.Notifications D-Bus interface.
|
|
|
|
If --actions are specified, waits for user response and prints chosen action label.
|
|
Timeout defaults to 10000ms (10s) for action notifications, 5000ms otherwise.
|
|
"""
|
|
import sys
|
|
import argparse
|
|
import subprocess
|
|
import json
|
|
|
|
def parse_args():
|
|
p = argparse.ArgumentParser(description="Desktop notification with optional actions")
|
|
p.add_argument("summary", help="Notification title")
|
|
p.add_argument("body", nargs="?", default="", help="Notification body text")
|
|
p.add_argument("--actions", default="", help="Comma-separated action labels")
|
|
p.add_argument("--timeout", type=int, default=0, help="Timeout in ms (0=default)")
|
|
p.add_argument("--icon", default="ollie", help="Icon name (default: ollie)")
|
|
p.add_argument("--urgency", choices=["low", "normal", "critical"], default="normal")
|
|
return p.parse_args()
|
|
|
|
def main():
|
|
args = parse_args()
|
|
|
|
# Build actions array: ["id1", "label1", "id2", "label2", ...]
|
|
actions_list = []
|
|
if args.actions:
|
|
labels = [a.strip() for a in args.actions.split(",") if a.strip()]
|
|
for i, label in enumerate(labels):
|
|
actions_list.append(str(i))
|
|
actions_list.append(label)
|
|
|
|
timeout = args.timeout
|
|
if timeout == 0:
|
|
timeout = 10000 if actions_list else 5000
|
|
|
|
urgency_map = {"low": 0, "normal": 1, "critical": 2}
|
|
|
|
# Build gdbus call
|
|
# Notify(app_name, replaces_id, icon, summary, body, actions, hints, timeout)
|
|
actions_str = "[" + ", ".join(f"'{a}'" for a in actions_list) + "]"
|
|
hints_str = "{'urgency': <byte %d>}" % urgency_map[args.urgency]
|
|
|
|
cmd = [
|
|
"gdbus", "call", "--session",
|
|
"--dest", "org.freedesktop.Notifications",
|
|
"--object-path", "/org/freedesktop/Notifications",
|
|
"--method", "org.freedesktop.Notifications.Notify",
|
|
"ollie", # app_name
|
|
"0", # replaces_id (uint32)
|
|
args.icon, # icon
|
|
args.summary, # summary
|
|
args.body, # body
|
|
actions_str, # actions
|
|
hints_str, # hints
|
|
str(timeout), # 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)
|
|
|
|
# Parse notification ID from "(uint32 N,)"
|
|
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) # extra 2s grace
|
|
|
|
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:
|
|
# Parse: /org/freedesktop/Notifications: org.freedesktop.Notifications.ActionInvoked (uint32 N, 'action_id')
|
|
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()
|