65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
"""signal_alert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import socket
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SignalRPCError(RuntimeError):
|
|
"""signal-cli returned an error or closed without responding."""
|
|
|
|
|
|
def signal_alert(body: str, from_phone: str, to_phone: str, *, timeout: float = 4.0) -> None:
|
|
"""Send a Signal alert through the local signal-cli daemon.
|
|
|
|
Args:
|
|
body: The body of the alert.
|
|
from_phone: The Signal account sending the alert.
|
|
to_phone: The Signal account receiving the alert.
|
|
timeout: Seconds to wait on each socket operation.
|
|
|
|
Raises:
|
|
SignalRPCError: If signal-cli returns an error or closes the
|
|
connection before responding.
|
|
OSError: If the socket is unreachable or an operation times out.
|
|
"""
|
|
signal_rpc_id = "signal-alert"
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"method": "send",
|
|
"params": {
|
|
"account": from_phone,
|
|
"recipient": [to_phone],
|
|
"message": body,
|
|
},
|
|
"id": signal_rpc_id,
|
|
}
|
|
|
|
try:
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
|
|
client.settimeout(timeout)
|
|
client.connect("/run/signal-cli/socket")
|
|
client.sendall(json.dumps(request).encode() + b"\n")
|
|
|
|
buffer = b""
|
|
while chunk := client.recv(65536):
|
|
buffer += chunk
|
|
while b"\n" in buffer:
|
|
line, buffer = buffer.split(b"\n", maxsplit=1)
|
|
response = json.loads(line)
|
|
if response.get("id") != signal_rpc_id:
|
|
continue
|
|
if rpc_error := response.get("error"):
|
|
error = f"Signal RPC returned an error: {rpc_error}"
|
|
raise SignalRPCError(error)
|
|
return
|
|
error = "Signal RPC socket closed before returning a response"
|
|
raise SignalRPCError(error)
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
error = f"Signal RPC failed: {exc!r}"
|
|
raise SignalRPCError(error) from exc
|