Daily New-Ad Digest

Build a cron job that surfaces yesterday's new ads from your followed brands.

Collect recent launches from followed brands and deduplicate them into one digest.

Before you start

  • Install requests and set ATRIA_API_KEY.
  • Set ATRIA_LAUNCHED_AFTER to an inclusive cutoff date in YYYY-MM-DD format.
  • Follow the brands you want to monitor. See Competitor Research.

Workflow

StepRequestPagination
List followed brandsPOST /open/v1/inspo-brands/search, scope="followed"offset and limit
Read recent adsGET /open/v1/ad-library/search, unified_brand_id and launched_aftercursor and page_size
Build the digestDeduplicate results by ad idPersist emitted IDs between runs.

Python example

The script prints recent ads once each, even if an advertiser appears under multiple brands. It also reports locked brands.

import os
from uuid import UUID

import requests

BASE = "https://api.tryatria.com/open/v1"
HEADERS = {"X-API-Key": os.environ["ATRIA_API_KEY"]}


def call(method, path, **kwargs):
    response = requests.request(
        method,
        BASE + path,
        headers=HEADERS,
        timeout=30,
        **kwargs,
    )
    response.raise_for_status()
    payload = response.json()
    if payload.get("code") != 0:
        raise RuntimeError(
            f"API error {payload.get('code')}: {payload.get('message')}"
        )
    return payload["data"]


def followed_brands():
    offset, limit = 0, 100
    while offset + limit <= 1000:
        page = call(
            "POST",
            "/inspo-brands/search",
            json={
                "scope": "followed",
                "followed": {"offset": offset, "limit": limit},
            },
        )
        yield from page["items"]

        if not page["has_more"]:
            if page["total"] > offset + len(page["items"]):
                raise RuntimeError("Result window exceeded; narrow the search")
            return
        offset += limit
    raise RuntimeError("Result window exceeded; narrow the search")


def recent_ads(brand_id):
    params = {
        "unified_brand_id": str(UUID(brand_id)),
        "launched_after": os.environ["ATRIA_LAUNCHED_AFTER"],
        "order": "newest",
        "page_size": 50,
    }
    while True:
        page = call("GET", "/ad-library/search", params=params)
        yield from page["items"]

        cursor = page.get("cursor")
        if not cursor:
            return
        params["cursor"] = cursor


seen = set()
for brand in followed_brands():
    if brand["locked"]:
        print("Locked brand:", brand["id"], brand.get("name") or "")

    for ad in recent_ads(brand["id"]):
        if ad["id"] in seen:
            continue
        seen.add(ad["id"])
        print(
            brand.get("name") or "",
            ad["id"],
            ad.get("title") or "",
            ad.get("start_date") or "",
        )

Notes

  • Followed search includes locked rows. Check the response's quota when presenting allowance limits.
  • Brand search serves the first 1000 matches. The script stops with an error if the followed list exceeds that window.
  • Brand-scoped ads use tracked coverage; an empty result does not prove the competitor launched nothing.
  • Launch dates are not discovery dates. Use overlapping windows and persist emitted ad IDs to catch late-indexed ads without repeating notifications.
  • Honor Retry-After on HTTP 429. Publish the digest only after the complete run succeeds.

Did this page help you?