DEV Community

Atlas Data
Atlas Data

Posted on

Track a competitor's ads on Google, LinkedIn and TikTok with Python

#python #marketing #webscraping #tutorial

Google, LinkedIn and TikTok all publish public ad libraries:

All three are useful for competitive research: what a brand is promoting, which formats it uses, which landing pages it sends people to, and how long an ad stays live. None of them has a convenient export, and the three sites look nothing alike.

In this tutorial we'll write one Python script (~120 lines) that:

  1. fetches a brand's ads from all three libraries in parallel,
  2. maps them to one common table (platform, ad_id, advertiser, format, text, first_shown, last_shown, landing_url, ad_url),
  3. keeps only ads from the brand itself (keyword searches return other advertisers too),
  4. remembers which ads it has already seen, so each run tells you what is new.

The example brand is Canva, because it runs ads on all three platforms. Swap in any company you like.

Disclosure: the three scrapers used here are Apify Actors published by Atlas Data, the author of this post. They are pay-per-result, and every call below sets a hard cost cap. You could also write the scraping yourself; the parts on normalizing and diffing apply either way.

Setup 

pip install apify-client   # tested with apify-client 3.4
export APIFY_TOKEN=...      # https://console.apify.com/settings/integrations
Enter fullscreen mode Exit fullscreen mode

The three Actors:

Platform Apify Store Docs & examples
Google Ads Transparency atlasdata/google-ads-transparency-scraper-reliable GitHub
LinkedIn Ad Library atlasdata/linkedin-ad-library-scraper-reliable GitHub
TikTok Ad Library atlasdata/tiktok-ad-library-scraper-reliable GitHub

1. One input per platform

Each library is searched a little differently, and that affects what you get back:

  • Google works best with the advertiser's domain (canva.com). You can also pass an advertiser name, an advertiser ID (AR...) or a Transparency Center URL. region is an ISO country code.
  • LinkedIn has an advertisers field that matches the company page name. dateRange takes values such as last-30-days, current-month or custom.
  • TikTok searchTerms match both the advertiser name and the ad text. Wrapping the term in double quotes makes it an exact phrase. countries must be EU/EEA, CH, GB or TR, because those are the only markets TikTok publishes. 
BRAND_NAME = "Canva"         # used for TikTok + LinkedIn searches
BRAND_DOMAIN = "canva.com"    # used for Google Ads Transparency
MAX_ADS = 15                  # per platform, keep it small while testing

JOBS = {
    "google": (
        "atlasdata/google-ads-transparency-scraper-reliable",
        {"searchTerms": [BRAND_DOMAIN], "maxAdsPerSearch": MAX_ADS, "region": "DE"},
    ),
    "linkedin": (
        "atlasdata/linkedin-ad-library-scraper-reliable",
        {"advertisers": [BRAND_NAME], "maxAdsPerSearch": MAX_ADS, "dateRange": "last-30-days"},
    ),
    "tiktok": (
        "atlasdata/tiktok-ad-library-scraper-reliable",
        {"searchTerms": [f'"{BRAND_NAME}"'], "countries": ["DE", "FR"], "maxAdsPerSearch": MAX_ADS,
         "dateRange": "last-30-days", "includeDetails": True},
    ),
}
Enter fullscreen mode Exit fullscreen mode

On TikTok, includeDetails: True opens each ad's detail page. That adds the landing page, CTA, objective, targeting and impressions per country, at the cost of a slower run.

2. Run the three scrapers in parallel

ActorClient.call() starts a run and waits for it to finish. The calls mostly spend their time waiting, so a thread pool runs all three at once.

def run_actor(platform: str) -> tuple[str, list[dict]]:
    actor_id, run_input = JOBS[platform]
    run = client.actor(actor_id).call(
        run_input=run_input,
        max_total_charge_usd=Decimal("0.50"),  # hard cost cap per run
        logger=None,                           # don't stream the run log
    )
    if run is None or run.status != "SUCCEEDED":
        # One blocked platform shouldn't sink the whole report: warn and move on.
        print(f"{platform}: run {run and run.status} - {run and run.status_message}")
        return platform, []
    # Each Actor also writes a short summary (per-search status, errors) to OUTPUT.
    summary = client.key_value_store(run.default_key_value_store_id).get_record("OUTPUT")
    if summary and summary["value"].get("status") != "SUCCEEDED":
        print(f"{platform}: {summary['value'].get('status')} - check the run's OUTPUT record")
    items = list(client.dataset(run.default_dataset_id).iterate_items())
    return platform, items
Enter fullscreen mode Exit fullscreen mode

Two notes:

  • In apify-client v3, call() returns a Run model, so you read run.default_dataset_id. The old run["defaultDatasetId"] from v1/v2 code no longer works.
  • Public ad libraries do block automated traffic from time to time, LinkedIn's Cloudflare in particular. A run can end PARTIAL, for example when a few detail pages were refused, or FAILED when every attempt was blocked. The OUTPUT record says which. For a daily report, it is better to log the problem and keep the other platforms' data than to crash.

3. Normalize to one schema

Every platform has its own field names. This function maps them onto one row:

def normalize(platform: str, ad: dict) -> dict:
    """Map each platform's fields onto one common row."""
    if platform == "google":
        return {
            "platform": "google",
            "ad_id": ad["creativeId"],
            "advertiser": ad["advertiserName"],
            "format": ad["format"],
            "text": " | ".join(filter(None, [ad.get("headline"), ad.get("description")])),
            "first_shown": (ad.get("firstShown") or "")[:10],
            "last_shown": (ad.get("lastShown") or "")[:10],
            "landing_url": ad.get("targetDomain"),
            "ad_url": ad["adUrl"],
        }
    if platform == "linkedin":
        return {
            "platform": "linkedin",
            "ad_id": ad["adId"],
            "advertiser": ad["advertiserName"],
            "format": ad.get("format"),
            "text": (ad.get("adText") or "")[:300],
            "first_shown": ad.get("firstShown") or "",  # LinkedIn often hides dates
            "last_shown": ad.get("lastShown") or "",
            "landing_url": ad.get("destinationUrl"),
            "ad_url": ad["adUrl"],
        }
    return {  # tiktok
        "platform": "tiktok",
        "ad_id": ad["adId"],
        "advertiser": ad["advertiserName"],
        "format": ad.get("mediaType"),
        "text": (ad.get("adText") or "")[:300],
        "first_shown": ad.get("firstShown"),
        "last_shown": ad.get("lastShown"),
        "landing_url": ad.get("landingPageUrl"),
        "ad_url": ad["adUrl"],
    }
Enter fullscreen mode Exit fullscreen mode

Things you'll notice in real data:

  • Google text ads often come back with an empty headline/description. The Transparency Center stores many text ads as rendered images, and imageUrl points to that image. Dates (firstShown, lastShown) are reliable, and regions[] shows where the ad ran.
  • LinkedIn shows a first/last-shown date and impression ranges only for some ads (mostly those served in the EU), so expect blanks. adText, headline, ctaText and format are almost always present.
  • TikTok gives the most detail: estimatedAudience, impressionsByCountry, targetAges, targetGenders and more, if you want to add columns.

4. Keep the brand's own ads and diff against the last run

A keyword search for "Canva" on TikTok also returns template creators and agencies whose ads mention Canva. A domain search on Google can return resellers that link to canva.com. If you only want the brand's own ads, filter on the advertiser name:

def main() -> None:
    with ThreadPoolExecutor(max_workers=3) as pool:
        results = list(pool.map(run_actor, JOBS))

    rows = [normalize(platform, ad) for platform, items in results for ad in items]
    print("fetched:", {platform: len(items) for platform, items in results})

    # Keyword and domain searches can also return *other* advertisers
    # (resellers, partners, anyone mentioning the brand). Keep the brand's own ads.
    rows = [r for r in rows if BRAND_NAME.lower() in (r["advertiser"] or "").lower()]
    print("from", BRAND_NAME, "itself:", len(rows))

    # Compare with the previous snapshot to find ads we have not seen before.
    seen_file = Path("seen_ads.json")
    seen = set(json.loads(seen_file.read_text())) if seen_file.exists() else set()
    new_rows = [r for r in rows if f'{r["platform"]}:{r["ad_id"]}' not in seen]
    seen_file.write_text(json.dumps(sorted(seen | {f'{r["platform"]}:{r["ad_id"]}' for r in rows})))

    with open("competitor_ads.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else ["platform"])
        writer.writeheader()
        writer.writerows(rows)

    print(f"{len(rows)} ads saved, {len(new_rows)} new since last run")
    for r in new_rows[:10]:
        print(f'[{r["platform"]}] {r["first_shown"]} {r["format"]}: {r["text"][:80]!r} -> {r["landing_url"]}')
Enter fullscreen mode Exit fullscreen mode

Leave out the filter if you do want to see who else is bidding on a competitor's name. That is often the more interesting list.

Running it 

$ python competitor_ads.py
fetched: {'google': 15, 'linkedin': 15, 'tiktok': 15}
from Canva itself: 39
39 ads saved, 39 new since last run
[google] 2024-01-14 TEXT: '' -> canva.com
[google] 2023-10-20 TEXT: '' -> canva.com
...
Enter fullscreen mode Exit fullscreen mode

Of the 45 ads fetched, 39 came from Canva's own accounts: 15 on Google, 15 on LinkedIn and 9 on TikTok. The other 6 TikTok ads were from third parties that mention Canva. The CSV has one row per ad across all three platforms, ready for a spreadsheet or pandas.

Run it again tomorrow, for example from cron or a GitHub Actions schedule, and only the creatives that appeared since then are reported as new. With 15 ads per platform, a run takes about two minutes, and its cost is capped by max_total_charge_usd.

Full script 

"""Collect a competitor's ads from TikTok, LinkedIn and Google into one table."""
import csv
import json
import os
from concurrent.futures import ThreadPoolExecutor
from decimal import Decimal
from pathlib import Path

from apify_client import ApifyClient

client = ApifyClient(os.environ["APIFY_TOKEN"])

BRAND_NAME = "Canva"         # used for TikTok + LinkedIn searches
BRAND_DOMAIN = "canva.com"    # used for Google Ads Transparency
MAX_ADS = 15                  # per platform, keep it small while testing

JOBS = {
    "google": (
        "atlasdata/google-ads-transparency-scraper-reliable",
        {"searchTerms": [BRAND_DOMAIN], "maxAdsPerSearch": MAX_ADS, "region": "DE"},
    ),
    "linkedin": (
        "atlasdata/linkedin-ad-library-scraper-reliable",
        {"advertisers": [BRAND_NAME], "maxAdsPerSearch": MAX_ADS, "dateRange": "last-30-days"},
    ),
    "tiktok": (
        "atlasdata/tiktok-ad-library-scraper-reliable",
        {"searchTerms": [f'"{BRAND_NAME}"'], "countries": ["DE", "FR"], "maxAdsPerSearch": MAX_ADS,
         "dateRange": "last-30-days", "includeDetails": True},
    ),
}


def run_actor(platform: str) -> tuple[str, list[dict]]:
    actor_id, run_input = JOBS[platform]
    run = client.actor(actor_id).call(
        run_input=run_input,
        max_total_charge_usd=Decimal("0.50"),  # hard cost cap per run
        logger=None,                           # don't stream the run log
    )
    if run is None or run.status != "SUCCEEDED":
        # One blocked platform shouldn't sink the whole report: warn and move on.
        print(f"{platform}: run {run and run.status} - {run and run.status_message}")
        return platform, []
    # Each Actor also writes a short summary (per-search status, errors) to OUTPUT.
    summary = client.key_value_store(run.default_key_value_store_id).get_record("OUTPUT")
    if summary and summary["value"].get("status") != "SUCCEEDED":
        print(f"{platform}: {summary['value'].get('status')} - check the run's OUTPUT record")
    items = list(client.dataset(run.default_dataset_id).iterate_items())
    return platform, items


def normalize(platform: str, ad: dict) -> dict:
    """Map each platform's fields onto one common row."""
    if platform == "google":
        return {
            "platform": "google",
            "ad_id": ad["creativeId"],
            "advertiser": ad["advertiserName"],
            "format": ad["format"],
            "text": " | ".join(filter(None, [ad.get("headline"), ad.get("description")])),
            "first_shown": (ad.get("firstShown") or "")[:10],
            "last_shown": (ad.get("lastShown") or "")[:10],
            "landing_url": ad.get("targetDomain"),
            "ad_url": ad["adUrl"],
        }
    if platform == "linkedin":
        return {
            "platform": "linkedin",
            "ad_id": ad["adId"],
            "advertiser": ad["advertiserName"],
            "format": ad.get("format"),
            "text": (ad.get("adText") or "")[:300],
            "first_shown": ad.get("firstShown") or "",  # LinkedIn often hides dates
            "last_shown": ad.get("lastShown") or "",
            "landing_url": ad.get("destinationUrl"),
            "ad_url": ad["adUrl"],
        }
    return {  # tiktok
        "platform": "tiktok",
        "ad_id": ad["adId"],
        "advertiser": ad["advertiserName"],
        "format": ad.get("mediaType"),
        "text": (ad.get("adText") or "")[:300],
        "first_shown": ad.get("firstShown"),
        "last_shown": ad.get("lastShown"),
        "landing_url": ad.get("landingPageUrl"),
        "ad_url": ad["adUrl"],
    }


def main() -> None:
    with ThreadPoolExecutor(max_workers=3) as pool:
        results = list(pool.map(run_actor, JOBS))

    rows = [normalize(platform, ad) for platform, items in results for ad in items]
    print("fetched:", {platform: len(items) for platform, items in results})

    # Keyword and domain searches can also return *other* advertisers
    # (resellers, partners, anyone mentioning the brand). Keep the brand's own ads.
    rows = [r for r in rows if BRAND_NAME.lower() in (r["advertiser"] or "").lower()]
    print("from", BRAND_NAME, "itself:", len(rows))

    # Compare with the previous snapshot to find ads we have not seen before.
    seen_file = Path("seen_ads.json")
    seen = set(json.loads(seen_file.read_text())) if seen_file.exists() else set()
    new_rows = [r for r in rows if f'{r["platform"]}:{r["ad_id"]}' not in seen]
    seen_file.write_text(json.dumps(sorted(seen | {f'{r["platform"]}:{r["ad_id"]}' for r in rows})))

    with open("competitor_ads.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else ["platform"])
        writer.writeheader()
        writer.writerows(rows)

    print(f"{len(rows)} ads saved, {len(new_rows)} new since last run")
    for r in new_rows[:10]:
        print(f'[{r["platform"]}] {r["first_shown"]} {r["format"]}: {r["text"][:80]!r} -> {r["landing_url"]}')


if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

Where to go next

  • Add impressionsByCountry (LinkedIn/TikTok) or regions (Google) as columns to see where spend is concentrated.
  • Download videoUrl and imageUrls for a swipe file. Some media URLs are signed and expire, so download them soon after the run.
  • Each repo has Node.js and curl versions of the same calls under examples/: Google, LinkedIn, TikTok.
  • If you'd rather ask an AI assistant ("what has Canva been advertising on LinkedIn this month?"), the same tools can be used over MCP. The setup for Claude Desktop and Cursor is in atlas-data-mcp.

Top comments (0)