57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""signal_alert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import socket
|
|
from os import getenv
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def signal_alert(body: str) -> None:
|
|
"""Send a signal alert.
|
|
|
|
Args:
|
|
body (str): The body of the alert.
|
|
"""
|
|
from_phone = getenv("SIGNAL_ALERT_FROM_PHONE")
|
|
to_phone = getenv("SIGNAL_ALERT_TO_PHONE")
|
|
if not from_phone or not to_phone:
|
|
logger.info("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set")
|
|
return
|
|
|
|
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(4.0)
|
|
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:
|
|
if error := response.get("error"):
|
|
logger.error("Signal RPC returned an error: %s", error)
|
|
return
|
|
|
|
logger.error("Signal RPC socket closed before returning a response")
|
|
except OSError, json.JSONDecodeError:
|
|
logger.exception("Unable to communicate with signal-cli")
|