Two Scripts to Simplify ObservePoint Cookie Reports

ObservePoint is a powerhouse for auditing your site, especially for privacy checks. It reveals what cookies load on pages and during user interactions, based on consent choices. But while its interface shines and report-sharing rocks, it doesn’t offer side-by-side views of multiple audits – key for comparing privacy settings. Luckily, you can use ObservePoint’s robust API to pull data and transform it the way you need. Here are two handy scripts to pull cookie inventory data into tidy spreadsheets.

The Context

ObservePoint’s UI is slick, and shareable report links make collaboration a breeze. Yet, when you need to compare audits – like seeing which cookies fire under different consent options – there’s no built-in multi-audit view. Thankfully, the API is robust, letting you extract and shape data however you want.

Script 1: Cookie Inventory from Regular Crawls

Run privacy audits with varying consent settings, then use this script to merge the cookie data into one report.

				
					import requests
import pandas as pd
import re
from datetime import datetime

# API Key (Replace with your actual API key)
API_KEY = '<PASTE YOUR OBSERVEPOINT API KEY HERE>'

# API Headers
HEADERS = {
    'Authorization': f'api_key {API_KEY}',
    'Content-Type': 'application/json'
}

BASE_URL = 'https://app.observepoint.com/api'

# Function to make GET requests
def api_get(endpoint):
    url = f"{BASE_URL}{endpoint}"
    response = requests.get(url, headers=HEADERS)
    if response.status_code != 200:
        print(f"ERROR {response.status_code}: {url}")
    response.raise_for_status()
    return response.json()

# Function to make POST requests
def api_post(endpoint, payload=None):
    url = f"{BASE_URL}{endpoint}"
    response = requests.post(url, headers=HEADERS, json=payload if payload else {})
    if response.status_code != 200:
        print(f"ERROR {response.status_code}: {url}")
    response.raise_for_status()
    return response.json()

# Fetch all audits and filter based on regex
def fetch_audits_by_regex(regex_pattern):
    audits = api_get('/v2/web-audits')
    regex = re.compile(regex_pattern, re.IGNORECASE)
    return [audit for audit in audits if regex.search(audit['name'])]

# Fetch the latest run ID for an audit
def get_latest_run_id(audit_id):
    runs = api_get(f'/v2/web-audits/{audit_id}/runs')
    if runs:
        return runs[0]['id']  # The most recent run ID
    return None

# Fetch audit location based on locationId
def get_audit_location(location_id):
    locations = api_get('/v2/locations')
    for location in locations:
        if location['id'] == location_id:
            return location['label']
    return 'Unknown'

# Fetch all cookies with deduplication
def fetch_all_cookies(audit_id, run_id):
    page = 0
    size = 100
    all_cookies = []
    seen_cookies = set()  # Store unique full-row tuples

    while True:
        endpoint = f'/v3/web-audits/{audit_id}/runs/{run_id}/reports/cookie-inventory/cookies'
        response = api_post(endpoint, {"page": page, "size": size})  # POST request

        cookies = response.get('cookies', [])
        for cookie in cookies:
            # Convert all mutable structures (lists, dicts) into immutable tuples
            def make_hashable(value):
                if isinstance(value, list):
                    return tuple(make_hashable(v) for v in value)
                elif isinstance(value, dict):
                    return tuple(sorted((k, make_hashable(v)) for k, v in value.items()))
                return value

            # Create a hashable tuple for the entire cookie dictionary
            cookie_tuple = tuple(sorted((k, make_hashable(v)) for k, v in cookie.items()))

            if cookie_tuple not in seen_cookies:
                seen_cookies.add(cookie_tuple)
                all_cookies.append(cookie)

        # Check if more pages are available
        pagination = response.get('metadata', {}).get('pagination', {})
        if page >= pagination.get('totalPageCount', 1) - 1:
            break
        page += 1

    return all_cookies



# Fetch tag inventory
def fetch_tags(audit_id, run_id):
    endpoint = f'/v3/web-audits/{audit_id}/runs/{run_id}/reports/tag-inventory/tags'
    return api_post(endpoint).get('tags', [])  # Using POST request for tags too

# Fetch audit run details (to get locationId)
def fetch_audit_run_info(audit_id, run_id):
    endpoint = f'/v3/web-audits/{audit_id}/runs/{run_id}/info'
    return api_get(endpoint)

# Process and compile data
def compile_data(audit_id, run_id, audit_name, audit_date, audit_location):
    cookies = fetch_all_cookies(audit_id, run_id)
    tags = fetch_tags(audit_id, run_id)

    tag_dict = {tag['tagId']: tag for tag in tags}  # Mapping tags by tagId
    data_rows = []

    for cookie in cookies:
        row = {
            "name": cookie.get("name"),
            "domain": cookie.get("domain"),
            "partyType": cookie.get("partyType"),
            "expirationType": cookie.get("expirationType"),
            "sameSite": cookie.get("sameSite"),
            "secure": cookie.get("secure"),
            "httpOnly": cookie.get("httpOnly"),
            "minDurationInMinutes": cookie.get("minDurationInMinutes"),
            "maxDurationInMinutes": cookie.get("maxDurationInMinutes"),
            "filteredPageCount": cookie.get("filteredPageCount"),
            "auditName": audit_name,
            "auditDate": audit_date,
            "auditLocation": audit_location
        }

        # Process initiators
        initiators = cookie.get("initiators", [])
        for i, initiator in enumerate(initiators, start=1):
            row[f"initiatorType{i}"] = initiator.get("initiatorType")
            row[f"initiatorUrlDomain{i}"] = initiator.get("urlDomain")
            row[f"initiatorInstanceCount{i}"] = initiator.get("instanceCount")

            # Match with tag inventory using tagId
            tag_id = initiator.get("tagId")
            if tag_id and tag_id in tag_dict:
                tag = tag_dict[tag_id]
                row[f"initiatorTagName{i}"] = tag.get("tagName")
                row[f"initiatorTagCategory{i}"] = tag.get("tagCategoryName")
                row[f"initiatorTagPresentPageCount{i}"] = tag.get("tagPresentPageCount")

        data_rows.append(row)

    return data_rows

#####################################
######## Main execution #############
#####################################

def main():
    regex_pattern = r'My Journey .*Default Consent State.*'  # Modify Name As Needed
    audits = fetch_audits_by_regex(regex_pattern)

    if not audits:
        print("No audits matched the regex.")
        return

    all_data = []

    for audit in audits:
        audit_id = audit['id']
        run_id = get_latest_run_id(audit_id)

        if not run_id:
            print(f"No run data found for audit: {audit['name']}")
            continue

        audit_info = fetch_audit_run_info(audit_id, run_id)
        audit_location = get_audit_location(audit_info.get("locationId", -1))

        audit_date = audit.get("lastRun")
        audit_data = compile_data(audit_id, run_id, audit['name'], audit_date, audit_location)
        all_data.extend(audit_data)

    # Convert to DataFrame and remove **only exact duplicate rows**
    if all_data:
        df = pd.DataFrame(all_data)
        df.drop_duplicates(keep="first", inplace=True)  # Drop exact row duplicates

        output_filename = fr"C:\....\observepoint_audit_cookie_inventory_{datetime.today().strftime('%Y-%m-%d')}.csv" # Specify the path to the output file
        df.to_csv(output_filename, index=False)
        print(f"Report saved to {output_filename} with duplicates removed.")
    else:
        print("No data to save.")

if __name__ == "__main__":
    main()

Pick audits by name using a regex pattern.
Exports cookie data to a CSV file.
Shows cookies side-by-side across audits.

Here is how the end report can look like after a bit of editing

Script 2: Cookie Inventory from Web Journeys

Web journeys track tags and cookies triggered by specific page interactions. This script pulls data from one or more journeys into a single report.

				
					import requests
import pandas as pd
import re
from datetime import datetime

# API Key (Replace with your actual API key)
API_KEY = '<PASTE YOUR OBSERVEPOINT API KEY HERE>'

# API Headers
HEADERS = {
    'Authorization': f'api_key {API_KEY}',
    'Content-Type': 'application/json'
}

BASE_URL = 'https://app.observepoint.com/api'

# Function to make GET requests
def api_get(endpoint):
    url = f"{BASE_URL}{endpoint}"
    response = requests.get(url, headers=HEADERS)
    if response.status_code != 200:
        print(f"ERROR {response.status_code}: {url}")
    response.raise_for_status()
    return response.json()

# Fetch all web journeys and filter based on regex
def fetch_journeys_by_regex(regex_pattern):
    journeys = api_get('/v3/web-journeys')
    regex = re.compile(regex_pattern, re.IGNORECASE)
    matched_journeys = [journey for journey in journeys if regex.search(journey['name'])]
    print(f"DEBUG: Matched Journeys: {[j['name'] for j in matched_journeys]}")
    return matched_journeys

# Fetch the latest run ID and date for a journey
def get_latest_run_id(journey_id):
    runs = api_get(f'/v2/web-journeys/{journey_id}/runs')
    if runs:
        latest_run = runs[0]
        run_id = latest_run.get('id')
        run_date = latest_run.get('date') or latest_run.get('runDate') or latest_run.get('startedAt', "Unknown")
        return run_id, run_date
    return None, "Unknown"

# Fetch and map journey location
def fetch_journey_location(journey_id, run_id):
    run_info = api_get(f'/v3/web-journeys/{journey_id}/runs/{run_id}/info')
    location_id = run_info.get("locationId", "Unknown")

    if location_id != "Unknown":
        locations = api_get('/v2/locations')
        location_map = {loc["id"]: loc["label"] for loc in locations}
        return location_map.get(location_id, "Unknown")

    return "Unknown"

# Fetch journey actions
def fetch_journey_actions(journey_id):
    return api_get(f'/v3/web-journeys/{journey_id}/actions')

# Fetch cookies for a specific action sequence
def fetch_action_cookies(journey_id, run_id, action_sequence):
    try:
        cookies_response = api_get(f'/v3/web-journeys/{journey_id}/runs/{run_id}/actions/{action_sequence}/cookies')
        all_cookies = []
        for section in ["create", "update", "delete", "unchanged"]:
            if section in cookies_response:
                all_cookies.extend(cookies_response[section])
        return all_cookies
    except Exception as e:
        print(f"Error fetching cookies for Journey {journey_id}, Run {run_id}, Action {action_sequence}: {e}")
        return []

# Fetch tag details and categories
def fetch_tags_and_categories():
    tags = api_get('/v3/tags')
    categories = api_get('/v3/tag-categories')

    tag_map = {tag["id"]: tag["name"] for tag in tags}
    tag_category_map = {tag["id"]: tag["tagCategoryId"] for tag in tags}
    category_map = {cat["id"]: cat["category"] for cat in categories}

    return tag_map, tag_category_map, category_map

# Process and compile data
def compile_data(journey_id, run_id, journey_name, journey_date, journey_location, tag_map, tag_category_map, category_map):
    actions = fetch_journey_actions(journey_id)
    data_rows = []

    for action in actions:
        action_sequence = action.get("sequence")
        action_description = action.get("description")
        action_name = f"{action_sequence} - {action_description}" if action_sequence is not None else action_description

        cookies = fetch_action_cookies(journey_id, run_id, action_sequence)

        if not isinstance(cookies, list):
            print(f"Unexpected cookie format for Journey {journey_id}, Run {run_id}, Action {action_sequence}")
            continue

        for cookie in cookies:
            if not isinstance(cookie, dict):
                print(f"Skipping unexpected cookie format: {cookie}")
                continue

            row = {
                "JourneyID": journey_id,
                "JourneyName": journey_name,
                "RunID": run_id,
                "RunDate": journey_date,
                "RunLocation": journey_location,
                "ActionSequence": action_sequence,
                "ActionDescription": action_description,
                "ActionName": action_name,
                "ActionType": action.get("actionType"),
                "CookieName": cookie.get("name"),
                "CookieValue": cookie.get("value"),
                "CookieDomain": cookie.get("domain"),
                "CookiePath": cookie.get("path"),
                "CookieHttpOnly": cookie.get("httpOnly"),
                "CookieSecure": cookie.get("secure"),
                "CookieSameSite": cookie.get("samesite"),
                "CookieExpires": cookie.get("expires"),
                "CookiePartyType": cookie.get("partyType"),
            }

            # Process initiators (ensuring unique values)
            seen_initiators = set()
            initiator_index = 1

            if "initiators" in cookie and isinstance(cookie["initiators"], list):
                initiators = cookie["initiators"]
                
                # First, process non-tag initiators
                for initiator in initiators:
                    initiator_type = initiator.get("initiatorType")
                    tag_id = initiator.get("tagId")

                    if tag_id is None and initiator_type and initiator_type not in seen_initiators:
                        row[f"InitiatorType"] = initiator_type
                        row[f"InitiatorTagName"] = ""
                        row[f"InitiatorTagCategory"] = ""
                        seen_initiators.add(initiator_type)

                # Next, process tag-based initiators
                for initiator in initiators:
                    tag_id = initiator.get("tagId")
                    if tag_id and tag_id not in seen_initiators:
                        tag_name = tag_map.get(tag_id, "Unknown")
                        tag_category_id = tag_category_map.get(tag_id)
                        tag_category = category_map.get(tag_category_id, "Unknown")

                        row[f"InitiatorType{initiator_index}"] = initiator.get("initiatorType", "Unknown")
                        row[f"InitiatorTagName{initiator_index}"] = tag_name
                        row[f"InitiatorTagCategory{initiator_index}"] = tag_category

                        seen_initiators.add(tag_id)
                        initiator_index += 1

            data_rows.append(row)

    return data_rows



#####################################
######## Main execution #############
#####################################


# Here you specify the regex pattern for your audits
# (or just include your journey full name if you want to pull data from one journey). 
# NOTE: since this script is intended for a privacy / cookie audit workflow,
# your journeys need to be identical as far as their actions are concerned. 
# The only difference in these journeys should be the 
# concent state (opt-in / opt-out OR location from which you are running  
# these journeys - ex GDPR region, California, etc)
  
def main():
    regex_pattern = r'Unity.*' # Add your journey name pattern here
    journeys = fetch_journeys_by_regex(regex_pattern)

    if not journeys:
        print("No journeys matched the regex.")
        return

    tag_map, tag_category_map, category_map = fetch_tags_and_categories()
    all_data = []

    for journey in journeys:
        journey_id = journey['id']
        run_id, run_date = get_latest_run_id(journey_id)

        if not run_id:
            print(f"No run data found for journey: {journey['name']}")
            continue

        journey_location = fetch_journey_location(journey_id, run_id)

        journey_data = compile_data(journey_id, run_id, journey['name'], run_date, journey_location, tag_map, tag_category_map, category_map)
        all_data.extend(journey_data)

    if all_data:
        df = pd.DataFrame(all_data)
        output_filename = fr"C:\....\observepoint_cookie_inventory_journey_report_{datetime.today().strftime('%Y-%m-%d')}.csv" # Specify the path to the output file
        df.to_csv(output_filename, index=False)
        print(f"Report saved to {output_filename}")
    else:
        print("No data to save.")

if __name__ == "__main__":
    main()

Allows to select journeys by name with regex.
Outputs cookie data to a CSV (fields differ slightly from Script 1).
Use pivot table to display cookies side-by-side for comparison.

Here is a sample report showing journeys with cookies side by side

Why It Matters

Combined reports cut through the noise, making analysis, teamwork, and presentations smoother. Hope this boosts your workflow! Drop your thoughts in the comments.

Additional Resources

Here are a few links that will help you get started with ObservePoint API: