signal-webhook-handler/signal_webhook/signal.py

import logging
from . import configuration
import socket
import random
import json

logger = logging.getLogger(__name__)


def signal_send(recipients: list[str], message: str) -> None:
    payload = {
        "jsonrpc": "2.0",
        "method": "send",
        "params": {
            "message": message,
            "recipient": recipients,
        },
        "id": random.randint(0, (1 << 24)),
    }
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as signal_json:
        signal_json.connect(configuration.SIGNAL_SOCKET)
        signal_json.send(json.dumps(payload).encode("utf-8"))