Track visits from AI assistants to your website

import base64
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from clickruntime import on_trigger, http, credentials, ai

def api(name, version):
    return build(name, version, credentials=Credentials(credentials.connection("google", "work")), cache_discovery=False)

def send_gmail(to, subject, text):
    raw = base64.urlsafe_b64encode(f"To: {to}\r\nSubject: {subject}\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n{text}".encode()).decode()
    api("gmail", "v1").users().messages().send(userId="me", body={"raw": raw}).execute()

PROPERTY = "123456789"

def ga(body):
    url = f"https://analyticsdata.googleapis.com/v1beta/properties/{PROPERTY}:runReport"
    return http.post(url, headers={"Authorization": f"Bearer {credentials.connection('google', 'work')}"}, json=body).json().get("rows", [])

WEEKS = [{"startDate": "7daysAgo", "endDate": "yesterday"}, {"startDate": "14daysAgo", "endDate": "8daysAgo"}]

AI_SOURCES = "chatgpt|openai|claude|anthropic|perplexity|gemini|copilot|you\\.com"
FILTER = {"filter": {"fieldName": "sessionSource", "stringFilter": {"matchType": "FULL_REGEXP", "value": f".*({AI_SOURCES}).*"}}}

@on_trigger
def run(event):
    by_source = ga({"dateRanges": WEEKS, "dimensions": [{"name": "sessionSource"}], "metrics": [{"name": "sessions"}], "dimensionFilter": FILTER})
    pages = ga({"dateRanges": WEEKS[:1], "dimensions": [{"name": "landingPage"}], "metrics": [{"name": "sessions"}], "dimensionFilter": FILTER,
                "orderBys": [{"metric": {"metricName": "sessions"}, "desc": True}], "limit": 10})
    lines = [f"{r['dimensionValues'][0]['value']} ({'last week' if r['dimensionValues'][1]['value'] == 'date_range_0' else 'week before'}): "
             f"{r['metricValues'][0]['value']} sessions" for r in by_source]
    report = "\n".join(lines + ["", "Top landing pages:"] + [f"{p['dimensionValues'][0]['value']}: {p['metricValues'][0]['value']}" for p in pages])
    try:
        report = ai.generate(report, system="Summarise the trend in AI assistant referrals in 2 sentences.", max_output_tokens=120).text + "\n\n" + report
    except ai.AIError:
        pass
    send_gmail("you@example.com", "Visits from AI assistants last week", report)
    return {"sources": len(by_source)}