!pip install bigquery_magics==0.12.1 -q

try:
    from google.colab import auth
    auth.authenticate_user()
except ImportError:
    pass  # Not running in Google Colab

import bigquery_magics
%load_ext bigquery_magics

bigquery_magics.context.project = 'httparchive'

%%bigquery
-- ============================================================
-- Step 1: Base tables
-- Source: httparchive.crawl.pages  |  date 2026-02-01  |  all clients  |  root pages
--
-- crawl.pages is scanned ONCE and deduplicated by domain. Both mobile and desktop
-- clients crawl the same pages; ads.txt / sellers.json are server-side so their
-- metrics are identical across clients. ANY_VALUE picks one client's JSON blob.
--
-- The //[ads] custom metric is stored in custom_metrics.other as JSON.
-- Path conventions inside that JSON object:
--   ads.txt data  →  $.ads.ads.*
--   sellers.json  →  $.ads.sellers.*
--   app-ads.txt   →  $.ads.app_ads.*  (not included in this analysis)
--
-- Crawl limitations relevant to this data:
--   • Google's sellers.json (~120 MB) is explicitly excluded: the JS metric
--     has a 5-second HTTP timeout + file-size limit. google.com therefore
--     appears in millions of ads.txt DIRECT entries but has no
--     RepresentsPublisher edges in this graph.
--   • Confidential sellers (is_confidential=1) are counted but their domains
--     are not stored — those publisher domains are invisible in this dataset.
--   • Domain deduplication only (JS Set): no seller_id is retained, so
--     cross-checks between ads.txt and sellers.json are domain-level only.
-- ============================================================

-- Materialise the crawl scan once so both base tables share a single read.
-- GROUP BY NET.REG_DOMAIN(page) deduplicates mobile / desktop rows for the same domain.
CREATE TEMP TABLE pages AS
SELECT
  NET.REG_DOMAIN(page) AS page,
  ANY_VALUE(custom_metrics.other.ads) AS metrics
FROM `httparchive.crawl.pages`
WHERE date = '2026-02-01'
  AND is_root_page = TRUE
GROUP BY NET.REG_DOMAIN(page);

-- Persist crawled domains so later cells can derive is_crawled without re-scanning.
-- A domain is "crawled" when the HTTP Archive fetched its root page; only crawled
-- domains have ads.txt / sellers.json data in this pipeline.
CREATE OR REPLACE TABLE scratchspace.crawled_pages AS
SELECT page AS domain FROM pages;

-- 1a. ads.txt relationships
CREATE OR REPLACE TABLE scratchspace.ads_txt_relationships AS
SELECT
  NET.REG_DOMAIN(REGEXP_EXTRACT(NORMALIZE_AND_CASEFOLD(domain),
    r'\b[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b')) AS platform_domain,
  page AS publisher_domain,
  'direct' AS relationship
FROM pages,
  UNNEST(JSON_VALUE_ARRAY(metrics.ads.account_types.direct.domains)) AS domain
WHERE SAFE.INT64(metrics.ads.account_count) > 0

UNION ALL

SELECT
  NET.REG_DOMAIN(REGEXP_EXTRACT(NORMALIZE_AND_CASEFOLD(domain),
    r'\b[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b')) AS platform_domain,
  page AS publisher_domain,
  'reseller' AS relationship
FROM pages,
  UNNEST(JSON_VALUE_ARRAY(metrics.ads.account_types.reseller.domains)) AS domain
WHERE SAFE.INT64(metrics.ads.account_count) > 0;

-- 1b. sellers.json relationships
-- $.both seller type means the partner acts as both publisher AND intermediary;
-- it is inserted into both groups so both RepresentsPublisher and SellsThrough
-- edges are populated.
CREATE OR REPLACE TABLE scratchspace.sellers_json_relationships AS
-- $.publisher → represents_publisher
SELECT page AS ssp_domain,
  NET.REG_DOMAIN(REGEXP_EXTRACT(NORMALIZE_AND_CASEFOLD(domain),
    r'\b[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b')) AS target_domain,
  'represents_publisher' AS seller_type
FROM pages,
  UNNEST(JSON_VALUE_ARRAY(metrics.ads.sellers.seller_types.publisher.domains)) AS domain
WHERE SAFE.INT64(metrics.ads.sellers.seller_count) > 0

UNION ALL

-- $.both → represents_publisher direction
SELECT page AS ssp_domain,
  NET.REG_DOMAIN(REGEXP_EXTRACT(NORMALIZE_AND_CASEFOLD(domain),
    r'\b[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b')) AS target_domain,
  'represents_publisher' AS seller_type
FROM pages,
  UNNEST(JSON_VALUE_ARRAY(metrics.ads.sellers.seller_types.both.domains)) AS domain
WHERE SAFE.INT64(metrics.ads.sellers.seller_count) > 0

UNION ALL

-- $.intermediary → intermediary
SELECT page AS ssp_domain,
  NET.REG_DOMAIN(REGEXP_EXTRACT(NORMALIZE_AND_CASEFOLD(domain),
    r'\b[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b')) AS target_domain,
  'intermediary' AS seller_type
FROM pages,
  UNNEST(JSON_VALUE_ARRAY(metrics.ads.sellers.seller_types.intermediary.domains)) AS domain
WHERE SAFE.INT64(metrics.ads.sellers.seller_count) > 0

UNION ALL

-- $.both → intermediary direction
SELECT page AS ssp_domain,
  NET.REG_DOMAIN(REGEXP_EXTRACT(NORMALIZE_AND_CASEFOLD(domain),
    r'\b[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b')) AS target_domain,
  'intermediary' AS seller_type
FROM pages,
  UNNEST(JSON_VALUE_ARRAY(metrics.ads.sellers.seller_types.both.domains)) AS domain
WHERE SAFE.INT64(metrics.ads.sellers.seller_count) > 0;

import time
import requests
from google.cloud import bigquery

PROJECT = "httparchive"
STAGING = f"{PROJECT}.scratchspace.sellers_json_staging"
TARGET = f"{PROJECT}.scratchspace.sellers_json_relationships"

HEADERS = {"User-Agent": "sellers-json-fetcher/1.0 (research; +https://httparchive.org)"}

# Overrides for SSPs that serve sellers.json from a different domain than
# the one used in publishers' ads.txt files.
URL_OVERRIDES = {
    "google.com":    "https://storage.googleapis.com/adx-rtb-dictionaries/sellers.json",
    "appnexus.com":  "https://xandr.com/sellers.json",
    "rhythmone.com": "https://nexxen.com/sellers.json",
}

client = bigquery.Client(project=PROJECT)

# Build target list from actual data: top 50 platforms by ads.txt mention count.
rows = client.query("""
    SELECT platform_domain AS domain
    FROM `httparchive.scratchspace.ads_txt_relationships`
    GROUP BY platform_domain
    ORDER BY COUNT(*) DESC
    LIMIT 50
""").result()

SUPPLEMENTAL_SELLERS = {
    row.domain: URL_OVERRIDES.get(row.domain, f"https://{row.domain}/sellers.json")
    for row in rows
    if row.domain
}

print(f"Targeting {len(SUPPLEMENTAL_SELLERS)} platforms:")
for domain, url in SUPPLEMENTAL_SELLERS.items():
    print(f"  {domain} → {url}")

total_inserted = 0

for ssp_domain, url in SUPPLEMENTAL_SELLERS.items():
    print(f"\nFetching {ssp_domain} …")
    try:
        resp = requests.get(url, headers=HEADERS, timeout=120)
        resp.raise_for_status()
        sellers = resp.json().get("sellers", [])
    except Exception as e:
        print(f"  ✗ {e}")
        time.sleep(0.5)
        continue

    raw_rows = []
    for s in sellers:
        if s.get("is_confidential", 0) == 1:
            continue
        domain = (s.get("domain") or "").strip().lower()  # guard against explicit null
        if not domain:
            continue
        st = s.get("seller_type", "").upper()
        if st in ("PUBLISHER", "BOTH"):
            raw_rows.append({"ssp_domain": ssp_domain, "raw_domain": domain, "seller_type": "represents_publisher"})
        if st in ("INTERMEDIARY", "BOTH"):
            raw_rows.append({"ssp_domain": ssp_domain, "raw_domain": domain, "seller_type": "intermediary"})

    if not raw_rows:
        print(f"  ✗ No valid sellers parsed")
        continue

    print(f"  {len(raw_rows):,} entries — staging and normalizing …")
    staging_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        schema=[
            bigquery.SchemaField("ssp_domain", "STRING"),
            bigquery.SchemaField("raw_domain", "STRING"),
            bigquery.SchemaField("seller_type", "STRING"),
        ],
    )
    client.load_table_from_json(raw_rows, STAGING, job_config=staging_config).result()

    sql = (
        f"INSERT INTO `{TARGET}` (ssp_domain, target_domain, seller_type) "
        f"SELECT ssp_domain, NET.REG_DOMAIN(raw_domain), seller_type "
        f"FROM `{STAGING}` WHERE NET.REG_DOMAIN(raw_domain) IS NOT NULL"
    )
    job = client.query(sql)
    job.result()
    inserted = job.num_dml_affected_rows or 0
    total_inserted += inserted
    print(f"  ✓ {inserted:,} rows appended")

client.delete_table(STAGING, not_found_ok=True)
print(f"\nDone — {total_inserted:,} total rows appended to {TARGET}")

%%bigquery
-- ============================================================
-- Step 2: Node tables
-- is_crawled = TRUE when the HTTP Archive fetched that domain as a root page
-- and we therefore have its ads.txt / sellers.json data in this pipeline.
-- ============================================================

-- Publisher: web page domains that declare ads.txt or appear as publisher
--   entries in sellers.json. These are the inventory sources in the ecosystem.
CREATE OR REPLACE TABLE scratchspace.Publisher AS
SELECT
  d.domain,
  c.domain IS NOT NULL AS is_crawled
FROM (
  SELECT publisher_domain AS domain FROM scratchspace.ads_txt_relationships
  UNION DISTINCT
  -- SSPs declare these as publishers they represent; include as Publisher nodes.
  SELECT target_domain AS domain FROM scratchspace.sellers_json_relationships
  WHERE seller_type = 'represents_publisher'
) d
LEFT JOIN scratchspace.crawled_pages c USING (domain)
WHERE d.domain IS NOT NULL;

ALTER TABLE scratchspace.Publisher ADD PRIMARY KEY (domain) NOT ENFORCED;

-- AdTechPlatform: DSPs, SSPs, and intermediary platforms.
--   A domain can appear in both Publisher and AdTechPlatform since many SSPs
--   also operate publisher properties alongside their platform business.
CREATE OR REPLACE TABLE scratchspace.AdTechPlatform AS
SELECT
  d.domain,
  c.domain IS NOT NULL AS is_crawled
FROM (
  SELECT platform_domain AS domain FROM scratchspace.ads_txt_relationships
  UNION DISTINCT
  SELECT ssp_domain AS domain FROM scratchspace.sellers_json_relationships
  UNION DISTINCT
  SELECT target_domain AS domain FROM scratchspace.sellers_json_relationships
  WHERE seller_type = 'intermediary'
) d
LEFT JOIN scratchspace.crawled_pages c USING (domain)
WHERE d.domain IS NOT NULL;

ALTER TABLE scratchspace.AdTechPlatform ADD PRIMARY KEY (domain) NOT ENFORCED;

%%bigquery
-- ============================================================
-- Step 3: Edge tables
-- Each edge source is wrapped in SELECT DISTINCT to collapse any residual
-- duplicates (e.g. same domain appearing twice in a JSON array).
-- GENERATE_UUID() then runs once per unique key pair.
-- ============================================================

-- DirectBid: AdTechPlatform → Publisher
--   Source: ads.txt DIRECT. Publisher explicitly authorises the platform to sell
--   their inventory. Highest trust — typically commands higher CPMs.
CREATE OR REPLACE TABLE scratchspace.DirectBid AS
SELECT GENERATE_UUID() AS edge_id, platform_domain, publisher_domain
FROM (
  SELECT DISTINCT platform_domain, publisher_domain
  FROM scratchspace.ads_txt_relationships
  WHERE relationship = 'direct'
    AND platform_domain IS NOT NULL
    AND publisher_domain IS NOT NULL
);

ALTER TABLE scratchspace.DirectBid ADD PRIMARY KEY (edge_id) NOT ENFORCED;

-- ResellerBid: AdTechPlatform → Publisher
--   Source: ads.txt RESELLER. Publisher allows indirect reselling.
--   Reseller entries without a sellers.json counterpart are a compliance red flag.
CREATE OR REPLACE TABLE scratchspace.ResellerBid AS
SELECT GENERATE_UUID() AS edge_id, platform_domain, publisher_domain
FROM (
  SELECT DISTINCT platform_domain, publisher_domain
  FROM scratchspace.ads_txt_relationships
  WHERE relationship = 'reseller'
    AND platform_domain IS NOT NULL
    AND publisher_domain IS NOT NULL
);

ALTER TABLE scratchspace.ResellerBid ADD PRIMARY KEY (edge_id) NOT ENFORCED;

-- RepresentsPublisher: AdTechPlatform → Publisher
--   Source: sellers.json $.publisher and $.both. SSP claims to represent this publisher.
CREATE OR REPLACE TABLE scratchspace.RepresentsPublisher AS
SELECT GENERATE_UUID() AS edge_id, ssp_domain, publisher_domain
FROM (
  SELECT DISTINCT ssp_domain, target_domain AS publisher_domain
  FROM scratchspace.sellers_json_relationships
  WHERE seller_type = 'represents_publisher'
    AND ssp_domain IS NOT NULL
    AND target_domain IS NOT NULL
);

ALTER TABLE scratchspace.RepresentsPublisher ADD PRIMARY KEY (edge_id) NOT ENFORCED;

-- SellsThrough: AdTechPlatform → AdTechPlatform
--   Source: sellers.json $.intermediary and $.both. Supply chain hop between SSPs.
--   These are the edges the original WITH RECURSIVE CTE was tracing depth-first.
CREATE OR REPLACE TABLE scratchspace.SellsThrough AS
SELECT GENERATE_UUID() AS edge_id, source_platform, dest_platform
FROM (
  SELECT DISTINCT ssp_domain AS source_platform, target_domain AS dest_platform
  FROM scratchspace.sellers_json_relationships
  WHERE seller_type = 'intermediary'
    AND ssp_domain IS NOT NULL
    AND target_domain IS NOT NULL
);

ALTER TABLE scratchspace.SellsThrough ADD PRIMARY KEY (edge_id) NOT ENFORCED;

%%bigquery
-- ============================================================
-- Step 4: Property Graph
-- ============================================================
CREATE OR REPLACE PROPERTY GRAPH scratchspace.AdsGraph
  NODE TABLES (
    scratchspace.Publisher
      KEY (domain)
      LABEL Publisher,
    scratchspace.AdTechPlatform
      KEY (domain)
      LABEL AdTechPlatform
  )
  EDGE TABLES (
    scratchspace.DirectBid
      KEY (edge_id)
      SOURCE KEY (platform_domain) REFERENCES AdTechPlatform (domain)
      DESTINATION KEY (publisher_domain) REFERENCES Publisher (domain)
      LABEL DirectBid,
    scratchspace.ResellerBid
      KEY (edge_id)
      SOURCE KEY (platform_domain) REFERENCES AdTechPlatform (domain)
      DESTINATION KEY (publisher_domain) REFERENCES Publisher (domain)
      LABEL ResellerBid,
    scratchspace.RepresentsPublisher
      KEY (edge_id)
      SOURCE KEY (ssp_domain) REFERENCES AdTechPlatform (domain)
      DESTINATION KEY (publisher_domain) REFERENCES Publisher (domain)
      LABEL RepresentsPublisher,
    scratchspace.SellsThrough
      KEY (edge_id)
      SOURCE KEY (source_platform) REFERENCES AdTechPlatform (domain)
      DESTINATION KEY (dest_platform) REFERENCES AdTechPlatform (domain)
      LABEL SellsThrough
  );

%%bigquery
GRAPH scratchspace.AdsGraph
MATCH (platform:AdTechPlatform)-[:DirectBid]->(publisher:Publisher)
RETURN
  platform.domain AS platform,
  COUNT(DISTINCT publisher.domain) AS direct_reach
ORDER BY direct_reach DESC
LIMIT 50;

%%bigquery
GRAPH scratchspace.AdsGraph
MATCH (platform:AdTechPlatform)-[:ResellerBid]->(publisher:Publisher)
RETURN
  platform.domain AS platform,
  COUNT(DISTINCT publisher.domain) AS reseller_reach
ORDER BY reseller_reach DESC
LIMIT 50;

%%bigquery --graph
GRAPH scratchspace.AdsGraph
MATCH p = (publisher:Publisher)<-[:ResellerBid]-(entry:AdTechPlatform)
            -[:SellsThrough]->(mid:AdTechPlatform)
            -[:SellsThrough]->(downstream:AdTechPlatform)
RETURN TO_JSON(p) AS path
LIMIT 100;

%%bigquery
GRAPH scratchspace.AdsGraph
MATCH (platform:AdTechPlatform)-[:ResellerBid]->(publisher:Publisher)
OPTIONAL MATCH (platform)-[confirms:RepresentsPublisher]->(publisher)
RETURN
  platform.domain AS platform,
  platform.is_crawled AS platform_crawled,
  COUNT(DISTINCT publisher.domain) AS reseller_reach,
  COUNTIF(confirms IS NOT NULL) AS confirmed_in_sellers_json
ORDER BY reseller_reach DESC
LIMIT 50;