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
requestsand setATRIA_API_KEY. - Set
ATRIA_LAUNCHED_AFTERto an inclusive cutoff date inYYYY-MM-DDformat. - Follow the brands you want to monitor. See Competitor Research.
Workflow
| Step | Request | Pagination |
|---|---|---|
| List followed brands | POST /open/v1/inspo-brands/search, scope="followed" | offset and limit |
| Read recent ads | GET /open/v1/ad-library/search, unified_brand_id and launched_after | cursor and page_size |
| Build the digest | Deduplicate results by ad id | Persist 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
quotawhen 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-Afteron HTTP 429. Publish the digest only after the complete run succeeds.
Updated 11 days ago
Did this page help you?
