Paperless-ngx network scan --> Signal message with direct link

README.md

Receive a Signal notification when your scan is ready

Powered by Paperlessg-ngx

Great for Paperless users that have printer that scan to a network folder. Quickly scanning a document and getting the link to its PDF is now super fast and low effort!

Paperless

Mount the local /scripts directory as a volume in your docker-compose.yml for the webserver container:

services:
(...)
  webserver:
    (...)
    volumes:
    (...)
      - ./scripts:/usr/src/paperless/scripts

Add the following env vars to your .env file (mine is docker-compose.env:

• PAPERLESS_URL – base URL of your Paperless-ngx instance (e.g. https://paperless.example.com)
• PAPERLESS_API_TOKEN – Paperless API token, necessary to generate a share link
• SIGNAL_API_URL – URL of your signal-cli REST API (default: http://signal-api:8080/v2/send)
• SIGNAL_SENDER_NUMBER – Signal phone number used to send messages
• SIGNAL_RECIPIENT_NUMBER – Signal phone number that should receive notifications

Optional env vars:

• SHARE_LINK_EXPIRE_DAYS – days before the share link expires (default: 7)
• SIGNAL_TRIGGER_TAGS – comma-separated list of tags that should trigger a notification (default: notify, I used names that mapped to a phone number)

To support multiple recipients, extend pick_recipients with a tag → phone-number mapping as shown in the inline comment.

Signal

Expects you to run an instance of signal-cli-rest-api. I had to use Podman to run this docker-compose because I run docker rootless on my Proxmox host. I'm using my own Signal number to send messages which was easiest to set up, I might eventually get another phone number for this.

docker-compose.yml

services:
  signal-cli-rest-api:
    image: docker.io/bbernhard/signal-cli-rest-api:latest
    environment:
      - MODE=json-rpc #supported modes: json-rpc, native, normal
      #- AUTO_RECEIVE_SCHEDULE=0 22 * * * #enable this parameter on demand (see description below)
    ports:
      - "9980:8080" #map docker port 8080 to host port 8080.
    volumes:
      - "./signal-cli-config:/home/.local/share/signal-cli" #map "signal-cli-config" folder on host system into docker container. the folder contains the password and cryptographic keys when a new number is registered

post_consume.py

#!/usr/bin/env python3

##
## post_consume script to send message to Signal numbers when a scan is ready. 
## 
## Put this in a ./scripts/ folder where your Paperless-ngx runs. 
##

import os
import sys
import json
import datetime
import urllib.request
import urllib.error


def log(msg: str):
    """Log to Paperless-ngx stdout."""
    sys.stdout.write(f"[post-consume-signal] {msg}\n")
    sys.stdout.flush()


def get_env(name: str, default=None, required: bool = False):
    val = os.environ.get(name, default)
    if required and not val:
        log(f"Missing required env var: {name}, skipping notification.")
        raise RuntimeError(f"Missing env: {name}")
    return val


def http_request(method: str, url: str, headers=None, data=None):
    headers = headers or {}
    if data is not None and not isinstance(data, (bytes, bytearray)):
        data = json.dumps(data).encode("utf-8")

    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            content_type = resp.headers.get("Content-Type", "")
            body = resp.read()
            if "application/json" in content_type:
                return json.loads(body.decode("utf-8"))
            return body
    except urllib.error.HTTPError as e:
        try:
            body = e.read().decode("utf-8", errors="ignore")
        except Exception:
            body = ""
        log(f"HTTP error {e.code} for {url}: {body}")
        raise
    except Exception as e:
        log(f"HTTP exception for {url}: {e}")
        raise


def parse_tags(tag_string: str):
    if not tag_string:
        return []
    return [t.strip() for t in tag_string.split(",") if t.strip()]


def pick_recipients(tags, default_number: str, trigger_tags):
    """
    Map tags to Signal recipient numbers.

    This implementation supports a single recipient number. It sends a
    notification if any of the document's tags match one of the
    configured trigger tags (case-insensitive).

    To extend this for multiple phone numbers, you can replace this
    function with a static mapping, e.g.:

        TAG_RECIPIENTS = {
            "alice": "+11111111111",
            "bob": "+22222222222",
        }
        for t in tags:
            number = TAG_RECIPIENTS.get(t.lower())
            if number:
                recipients.append(number)

    and return the de-duplicated list.
    """
    if not default_number:
        return []

    lower_tags = {t.lower() for t in tags}
    trigger_tags = {t.lower() for t in trigger_tags if t.strip()}

    if lower_tags & trigger_tags:
        return [default_number]

    return []


def ensure_share_link(base_url: str, token: str, document_id: int, expire_days: int):
    """
    Reuse an existing non-expired archive share link if possible,
    otherwise create a new one with the given expiration.
    """
    auth_headers = {
        "Authorization": f"Token {token}",
        "Accept": "application/json",
        "Content-Type": "application/json",
    }

    # 1) Try to reuse existing share link
    url_list = f"{base_url}/api/documents/{document_id}/share_links/"
    try:
        links = http_request("GET", url_list, headers=auth_headers)
    except Exception:
        log("Failed to list existing share links, will try to create a new one.")
        links = []

    now = datetime.datetime.now(datetime.timezone.utc)

    for link in links:
        # Expect: { slug, expiration, document, file_version, ... }
        if link.get("file_version") != "archive":
            continue
        expiration = link.get("expiration")
        if expiration:
            try:
                exp_dt = datetime.datetime.fromisoformat(
                    expiration.replace("Z", "+00:00")
                )
                if exp_dt <= now:
                    continue  # expired
            except ValueError:
                # If parsing fails, be conservative and skip
                continue
        slug = link.get("slug")
        if slug:
            return slug

    # 2) Create new share link
    expiration_dt = now + datetime.timedelta(days=expire_days)
    expiration_iso = expiration_dt.isoformat().replace("+00:00", "Z")

    url_create = f"{base_url}/api/share_links/"
    payload = {
        "document": document_id,
        "file_version": "archive",
        "expiration": expiration_iso,
    }

    link = http_request("POST", url_create, headers=auth_headers, data=payload)
    slug = link.get("slug")
    if not slug:
        raise RuntimeError("Share link creation succeeded but slug is missing.")

    return slug


def get_document_title(base_url: str, token: str, document_id: int, fallback_name: str):
    auth_headers = {
        "Authorization": f"Token {token}",
        "Accept": "application/json",
    }
    url_doc = f"{base_url}/api/documents/{document_id}/"
    try:
        data = http_request("GET", url_doc, headers=auth_headers)
        title = data.get("title") or fallback_name
        return title
    except Exception:
        log("Failed to fetch document title, falling back to filename.")
        return fallback_name


def send_signal(signal_url: str, sender_number: str, recipients, message: str):
    if not recipients:
        log("No recipients provided, nothing to send.")
        return

    body = {
        "message": message,
        "number": sender_number,
        "recipients": recipients,
    }

    headers = {"Content-Type": "application/json"}
    try:
        http_request("POST", signal_url, headers=headers, data=body)
        log(f"Signal message sent to: {', '.join(recipients)}")
    except Exception as e:
        log(f"Failed to send Signal message: {e}")


def main():
    # Paperless passes env vars + positional args. We rely on env vars.
    try:
        document_id_str = get_env("DOCUMENT_ID", required=True)
    except RuntimeError:
        # No doc ID -> nothing we can do; don't break consumption.
        return 0

    document_id = int(document_id_str)
    document_file_name = get_env(
        "DOCUMENT_FILE_NAME", default=f"Document {document_id}"
    )
    document_tags_str = get_env("DOCUMENT_TAGS", default="")

    tags = parse_tags(document_tags_str)

    # Signal configuration (generic, no personal data)
    signal_url = get_env("SIGNAL_API_URL", default="http://signal-api:8080/v2/send")
    signal_sender = get_env("SIGNAL_SENDER_NUMBER", required=True)
    signal_recipient = get_env("SIGNAL_RECIPIENT_NUMBER", required=True)
    trigger_tags_str = get_env("SIGNAL_TRIGGER_TAGS", default="notify")
    trigger_tags = parse_tags(trigger_tags_str)

    recipients = pick_recipients(tags, signal_recipient, trigger_tags)

    if not recipients:
        log(
            f"Document {document_id} has no matching notification tags "
            f"(configured: {trigger_tags}), skipping Signal."
        )
        return 0

    # Paperless API auth
    api_token = get_env("PAPERLESS_API_TOKEN", required=True)

    # Base URL: from PAPERLESS_URL
    paperless_url = get_env("PAPERLESS_URL", required=True)
    base_url = paperless_url.rstrip("/")

    # Share link expiration (days)
    expire_days = int(get_env("SHARE_LINK_EXPIRE_DAYS", default="7"))

    # Fetch title
    title = get_document_title(base_url, api_token, document_id, document_file_name)

    # Ensure we have a share link slug
    slug = ensure_share_link(base_url, api_token, document_id, expire_days)

    # Build final share URL
    share_url = f"{base_url}/share/{slug}"

    # Build message
    message = f"Your document is ready!\n{title}\n{share_url}"

    log(
        f"Prepared notification for document {document_id}: "
        f"title='{title}', recipients={recipients}, share_url={share_url}"
    )

    send_signal(signal_url, signal_sender, recipients, message)

    return 0


if __name__ == "__main__":
    sys.exit(main())