Sentinel Gallery

Developers

Call the Pixel Sentinel API from your own server

The same tour in seven languages: get a token, read the account, upload an image, embed it, list, create a download link, set approved sites, read traffic and patrols, and delete. Each client is one file with no dependencies beyond the language's usual HTTP and JSON libraries. This website runs on the Node.js one.

1. Create an API client

In the Pixel Sentinel dashboard, open the API tab, choose Create an API client, and tick the scopes it needs. Copy the client secret straight away: it's shown once. Keep both on your server, never in browser code.

ScopeEndpointsFor
account:readGET /client/v1/accountName, status, credits, views this month, approved sites
images:readGET /client/v1/images, GET /client/v1/images/{id}List (newest first, q filters by name) and read images
images:writePOST /client/v1/images, DELETE /client/v1/images/{id}Upload (multipart, field "file") and delete
links:writePOST /client/v1/images/{id}/download-linkSigned, watermarked full-size download links
domains:writePUT /client/v1/domainsReplace the approved sites list
traffic:readGET /client/v1/trafficThe Traffic report: totals, daily, top sources
patrols:readGET /client/v1/patrols, …/{id}, …/{id}/report.csvPatrols, allowance and reports
patrols:writePOST /client/v1/patrolsSearch the web for copies (sends unprotected copies to Google and TinEye)

2. Get a token

OAuth 2.0 client credentials. The token lasts an hour; the clients below keep it and fetch a new one when it's about to expire, or once when a call answers 401.

curl -u "$PS_CLIENT_ID:$PS_CLIENT_SECRET" -d grant_type=client_credentials \
  https://argus.pixelsentinel.net/oauth/token
# {"access_token":"pst_…","token_type":"Bearer","expires_in":3600,"scope":"account:read images:read …"}

3. Run a sample

Every sample reads PS_CLIENT_ID, PS_CLIENT_SECRET and, optionally, PS_API_BASE. Add --keep to keep the uploaded image, or --patrol to run a patrol on it too.

In samples/python:

pip install -r requirements.txt
python demo.py photo.jpg
pixelsentinel.py
"""A small Pixel Sentinel Client API wrapper (Python 3.9+, `pip install requests`).

Authenticates with OAuth 2.0 client credentials, keeps the access token until shortly before it
expires, and gets a new one once if a call answers 401. Every /client/v1 operation is here.

    from pixelsentinel import PixelSentinel
    ps = PixelSentinel(client_id, client_secret)
    for image in ps.list_images(q="photo--")["images"]:
        print(image["id"], image["name"])
"""
from __future__ import annotations

import mimetypes
import os
import time
from typing import Any, Iterable, Iterator

import requests

DEFAULT_BASE_URL = "https://argus.pixelsentinel.net"


class PixelSentinelError(Exception):
    """An error answer from the API: `status` is the HTTP status, `body` the parsed JSON (if any)."""

    def __init__(self, status: int, message: str, body: Any = None):
        super().__init__(f"{status}: {message}")
        self.status = status
        self.body = body


class PixelSentinel:
    def __init__(self, client_id: str, client_secret: str, base_url: str = DEFAULT_BASE_URL,
                 scope: str | None = None, timeout: float = 60):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip("/")
        self.scope = scope  # ask for fewer scopes than the client has, e.g. "images:read"
        self.timeout = timeout
        self.session = requests.Session()
        self._token: str | None = None
        self._expires = 0.0

    @classmethod
    def from_env(cls) -> "PixelSentinel":
        """PS_CLIENT_ID, PS_CLIENT_SECRET and (optionally) PS_API_BASE."""
        return cls(os.environ["PS_CLIENT_ID"], os.environ["PS_CLIENT_SECRET"],
                   os.environ.get("PS_API_BASE", DEFAULT_BASE_URL))

    # ------------------------------------------------------------------ auth

    def token(self, force: bool = False) -> str:
        """POST /oauth/token with HTTP Basic (client_secret_basic). Tokens last an hour."""
        if not force and self._token and time.time() < self._expires - 60:
            return self._token
        data = {"grant_type": "client_credentials"}
        if self.scope:
            data["scope"] = self.scope
        res = self.session.post(f"{self.base_url}/oauth/token", data=data,
                                auth=(self.client_id, self.client_secret), timeout=self.timeout)
        body = _json(res)
        if not res.ok:
            raise PixelSentinelError(res.status_code, body.get("error_description") or body.get("error") or res.reason, body)
        self._token = body["access_token"]
        self._expires = time.time() + body["expires_in"]
        return self._token

    def _request(self, method: str, path: str, *, raw: bool = False, **kwargs) -> Any:
        for attempt in (1, 2):
            headers = {"Authorization": f"Bearer {self.token(force=attempt == 2)}"}
            res = self.session.request(method, f"{self.base_url}/client/v1{path}", headers=headers,
                                       timeout=self.timeout, **kwargs)
            if res.status_code != 401:
                break
            # The token expired early or the secret was replaced: get a new one, once.
            if "files" in kwargs:
                for f in kwargs["files"].values():
                    f[1].seek(0)
        if raw and res.ok:
            return res
        body = _json(res)
        if not res.ok:
            raise PixelSentinelError(res.status_code, body.get("error") or res.reason, body)
        return body

    # ------------------------------------------------------------------ account

    def get_account(self) -> dict:
        """account:read - name, status, credits, views this month, approved sites."""
        return self._request("GET", "/account")

    def set_domains(self, domains: Iterable[str]) -> list[str]:
        """domains:write - replace the whole approved-sites list ("*.example.com" covers subdomains)."""
        return self._request("PUT", "/domains", json={"domains": list(domains)})["domains"]

    # ------------------------------------------------------------------ images

    def list_images(self, q: str | None = None, limit: int = 25, offset: int = 0) -> dict:
        """images:read - newest first; `q` filters by name. Returns {images, total, offset, limit}."""
        params = {"limit": limit, "offset": offset}
        if q:
            params["q"] = q
        return self._request("GET", "/images", params=params)

    def iter_images(self, q: str | None = None, page_size: int = 100) -> Iterator[dict]:
        """Every image, following the pages."""
        offset = 0
        while True:
            page = self.list_images(q=q, limit=page_size, offset=offset)
            yield from page["images"]
            offset += len(page["images"])
            if not page["images"] or offset >= page["total"]:
                return

    def get_image(self, image_id: str) -> dict:
        """images:read - one image; 404 if it isn't in this account."""
        return self._request("GET", f"/images/{image_id}")["image"]

    def upload_image(self, path: str, name: str | None = None) -> dict:
        """images:write - upload a file. The image's name is the filename sent (`name` overrides it)."""
        name = name or os.path.basename(path)
        mime = mimetypes.guess_type(name)[0] or "application/octet-stream"
        with open(path, "rb") as f:
            return self._request("POST", "/images", files={"file": (name, f, mime)})["image"]

    def delete_image(self, image_id: str) -> None:
        """images:write"""
        self._request("DELETE", f"/images/{image_id}")

    def create_download_link(self, image_id: str, expires_in_hours: float = 168) -> dict:
        """links:write - a signed, watermarked full-size download. Returns {url, expiresAt}."""
        return self._request("POST", f"/images/{image_id}/download-link", json={"expiresInHours": expires_in_hours})

    @staticmethod
    def embed_html(image: dict, alt: str = "") -> str:
        """The script-embed snippet: the <img> loads the scrambled copy, embed.js swaps in the real one."""
        base = image["urls"]["scrambled"].split("/i/")[0]
        return (f'<script src="{base}/embed.js"></script>\n'
                f'<img src="{image["urls"]["scrambled"]}" data-pg="{image["id"]}" '
                f'width="{image["width"]}" height="{image["height"]}" alt="{alt}">')

    # ------------------------------------------------------------------ traffic and patrols

    def traffic(self, days: int = 30, image_id: str | None = None) -> dict:
        """traffic:read - the dashboard's Traffic report."""
        params: dict[str, Any] = {"days": days}
        if image_id:
            params["image"] = image_id
        return self._request("GET", "/traffic", params=params)

    def list_patrols(self) -> dict:
        """patrols:read - recent patrols and this month's allowance."""
        return self._request("GET", "/patrols")

    def start_patrol(self, image_ids: list[str], allow_unprotected: bool = False) -> dict:
        """patrols:write - search the web for copies. Sends unprotected 1000 px copies of these images
        to Google and TinEye, so you must pass allow_unprotected=True deliberately."""
        if allow_unprotected is not True:
            raise ValueError("start_patrol sends unprotected copies to search providers: pass allow_unprotected=True")
        return self._request("POST", "/patrols", json={"imageIds": image_ids, "allowUnprotected": True})

    def get_patrol(self, patrol_id: str) -> dict:
        """patrols:read - the report, with every sighting."""
        return self._request("GET", f"/patrols/{patrol_id}")["patrol"]

    def wait_for_patrol(self, patrol_id: str, every: float = 5, timeout: float = 600) -> dict:
        deadline = time.time() + timeout
        while True:
            patrol = self.get_patrol(patrol_id)
            if patrol["status"] in ("done", "failed") or time.time() > deadline:
                return patrol
            time.sleep(every)

    def patrol_csv(self, patrol_id: str) -> str:
        """patrols:read - the report as CSV."""
        return self._request("GET", f"/patrols/{patrol_id}/report.csv", raw=True).text


def _json(res: requests.Response) -> dict:
    try:
        body = res.json()
        return body if isinstance(body, dict) else {"data": body}
    except ValueError:
        return {}
demo.py
"""A tour of every Client API call. Uploads one image, works with it, and deletes it again.

    export PS_CLIENT_ID=psc_… PS_CLIENT_SECRET=pss_…   # PS_API_BASE defaults to production
    python demo.py path/to/image.jpg [--keep] [--patrol]

--keep leaves the image in the account; --patrol also runs a patrol on it, which uses one patrol
from the plan's allowance and sends an unprotected 1000 px copy to Google and TinEye.
"""
import sys

from pixelsentinel import PixelSentinel, PixelSentinelError


def main(args: list[str]) -> None:
    files = [a for a in args if not a.startswith("--")]
    if not files:
        sys.exit(__doc__)
    ps = PixelSentinel.from_env()

    ps.token()
    print("1. Got an access token")

    account = ps.get_account()
    print(f"2. Account {account['orgName'] or account['id']}: {account['status']}, "
          f"{account['credits']} credits, {account['viewsThisMonth']} views this month")
    print(f"   Approved sites: {', '.join(account['domains']) or '(none yet)'}")

    image = ps.upload_image(files[0])
    print(f"3. Uploaded {image['name']} as {image['id']} ({image['width']}x{image['height']})")

    image = ps.get_image(image["id"])
    print("4. Embed it on an approved site with:\n" + ps.embed_html(image, alt=image["name"]))
    print(f"   Plain image (no script): {image['urls']['plain']}")

    page = ps.list_images(limit=5)
    print(f"5. The account has {page['total']} images; newest: {', '.join(i['name'] for i in page['images'])}")

    link = ps.create_download_link(image["id"], expires_in_hours=24)
    print(f"6. Full-size download link until {link['expiresAt']}:\n   {link['url']}")

    domains = ps.set_domains(account["domains"])
    print(f"7. Saved the approved sites unchanged: {domains}")

    report = ps.traffic(days=7)
    print(f"8. Traffic, last 7 days: {report['totals']}")

    patrols = ps.list_patrols()
    allowance = patrols.get("allowance") or {}
    print(f"9. {len(patrols['patrols'])} patrols so far; {allowance.get('left', 0)} left this month")
    if "--patrol" in args:
        started = ps.start_patrol([image["id"]], allow_unprotected=True)
        patrol = ps.wait_for_patrol(started["patrol"]["id"])
        print(f"   Patrol {patrol['id']}: {patrol['status']}, {patrol['copies']} copies found")
        print(ps.patrol_csv(patrol["id"]))

    if "--keep" in args:
        print(f"10. Kept {image['id']}")
    else:
        ps.delete_image(image["id"])
        print(f"10. Deleted {image['id']}")


if __name__ == "__main__":
    try:
        main(sys.argv[1:])
    except PixelSentinelError as e:
        sys.exit(f"Pixel Sentinel said {e}")
seed.py
"""Uploads a folder of images, skipping names the account already has.

    python seed.py ../../seed-images

The gallery files images by name prefix: photo--…, art--…, graphics--….
"""
import pathlib
import sys

from pixelsentinel import PixelSentinel

EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".avif", ".heic", ".tif", ".tiff"}


def main(folder: str) -> None:
    ps = PixelSentinel.from_env()
    have = {image["name"] for image in ps.iter_images()}
    for path in sorted(pathlib.Path(folder).iterdir()):
        if path.suffix.lower() not in EXTENSIONS:
            continue
        if path.name in have:
            print(f"have     {path.name}")
            continue
        image = ps.upload_image(str(path))
        print(f"uploaded {path.name} -> {image['id']}")


if __name__ == "__main__":
    if len(sys.argv) != 2:
        sys.exit(__doc__)
    main(sys.argv[1])

Things worth knowing

The full reference is the Pixel Sentinel API reference (sign in to your Pixel Sentinel account to read it).