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:
    def to_dest(recipient: str) -> dict[str, str]:
        if recipient.startswith("G:"):
            return {"groupId": recipient[2:]}
        return {"recipient": [recipient]}

    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as signal_json:
        signal_json.connect(configuration.SIGNAL_SOCKET)
        for recipient in recipients:
            payload = {
                "jsonrpc": "2.0",
                "method": "send",
                "params": {"message": message, **to_dest(recipient)},
                "id": random.randint(0, (1 << 24)),
            }
            str_payload = json.dumps(payload).encode("utf-8")
            logger.debug("Sending payload: %s", str_payload)
            signal_json.send(str_payload)
        logger.info("Sent messages to %s", ", ".join(recipients))