A script for automatically sending photos from a camera via FTP, created for cameras that send photos to an FTP server.

camera_ntfy.py

#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["httpx", "pyftpdlib"]
# ///

import os
import mimetypes
import tempfile
import logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

import httpx
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer

topic = os.getenv("FTP_PASSWORD")
ftp_port = int(os.getenv("FTP_PORT", 2121))
passive_ports = os.getenv("PASSIVE_PORTS")
masquerade_address = os.getenv("MASQUERADE_ADDRESS")
ntfy_url = os.getenv("NTFY_URL", "https://ntfy.sh")


class ImageForwardHandler(FTPHandler):
    def on_file_received(self, file_path):
        ext = os.path.splitext(file_path)[1].lower()
        if ext not in {".png", ".jpg", ".jpeg"}:
            logging.info("Ignoring non‑image file: %s", file_path)
            os.remove(file_path)
            return

        content_type = (
            mimetypes.guess_type(file_path)[0] or "application/octet-stream"
        )
        filename = os.path.basename(file_path)
        url = f"{ntfy_url.rstrip('/')}/{topic}"

        try:
            with open(file_path, "rb") as fh:
                payload = fh.read()

            headers = {
                "Title": "The camera detect movement!",
                "Filename": filename,
                "Content-Type": content_type,
                "Priority": "max"
            }

            logging.info("Forwarding %s to %s", filename, url)
            response = httpx.put(
                url,
                content=payload,
                headers=headers,
                timeout=10,
            )
            response.raise_for_status()
            logging.info("ntfy response: %s", response.status_code)
        except Exception as exc:
            logging.exception("Failed to forward %s: %s", filename, exc)
        finally:
            os.remove(file_path)


def main():
    if not topic:
        raise RuntimeError("Set FTP_PASSWORD environment variable")

    homedir = tempfile.gettempdir()
    authorizer = DummyAuthorizer()
    authorizer.add_user(
        "user",
        topic,
        homedir=homedir,
        perm="elradfmw",
    )

    handler = ImageForwardHandler
    handler.authorizer = authorizer
    handler.banner = "FTP to ntfy bridge ready."

    if passive_ports:
        try:
            start, end = map(int, passive_ports.split("-"))
            handler.passive_ports = range(start, end + 1)
            logging.info("Passive port range: %d-%d", start, end)
        except ValueError:
            raise RuntimeError("Invalid PASSIVE_PORTS format, expected START-END")

    if masquerade_address:
        handler.masquerade_address = masquerade_address
        logging.info("Masquerade address: %s", masquerade_address)

    address = ("0.0.0.0", ftp_port)
    server = FTPServer(address, handler)

    logging.info("Starting FTP server on %s:%s", *address)
    logging.info("ntfy topic URL: %s/%s", ntfy_url.rstrip("/"), topic)
    server.serve_forever()


if __name__ == "__main__":
    main()