1"""Google Ads Transparency Center client — the `anji` RPC surface, plain HTTP.
2
3Verified against the live endpoints on 2026-08-06 from a datacenter IP, no browser and no proxy.
4
5What this surface is, and what it is not:
6
7* It is **not** WIZ ``batchexecute``. There is no rpcid, no ``f.sid`` and no ``bl``. The methods are
8 addressed by name: ``POST /anji/_/rpc/SearchService/<Method>``.
9* Only **form-encoded ``f.req=<json>``** is accepted. A ``application/json+protobuf`` body is
10 refused with a converter error that names the request class
11 (``…reporting.SearchCreativesRequest``) — which is, in fact, the only informative error this
12 surface produces.
13* **A wrong field shape returns HTTP 200 and an empty ``{}``.** No error, no hint. That is why the
14 field numbering below is stated as measured fact rather than derived at runtime: there is no
15 signal to derive it from, and a caller must never be told "this advertiser has no ads" when the
16 truth is "we sent the wrong request".
17
18The page itself is useless for discovery — 2.5 MB of generic framework with no app strings, and a
19full browser TLS fingerprint returns the same shell byte for byte.
20"""
21
22from __future__ import annotations
23
24import gzip
25import json
26import logging
27import random
28import time
29import urllib.error
30import urllib.parse
31import urllib.request
32import zlib
33
34BASE = "https://adstransparency.google.com/anji/_/rpc/%s"
35UA = (
36 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
37 "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
38)
39
40MIN_GAP = 1.1
41BACKOFF = (
42 4.0,
43)
44
45log = logging.getLogger("ads-transparency")
46
47
48class AdsError(RuntimeError):
49 """The request failed in a way retrying will not fix."""
50
51
52class AdsBlocked(AdsError):
53 """Rate-limited — rotate the egress IP rather than waiting it out."""
54
55
56def _decode(blob: bytes) -> str:
57 """Error bodies come back gzipped even when the success path does not, and reading them raw
58 turns a perfectly clear message into mojibake — which is exactly how these errors got missed
59 the first time round."""
60 for attempt in (gzip.decompress, lambda b: zlib.decompress(b, -15), lambda b: b):
61 try:
62 return attempt(blob).decode("utf-8", "replace")
63 except Exception:
64 continue
65 return blob.decode("utf-8", "replace")
66
67
68def advertiser_id(text: str) -> str | None:
69 """Pull an `AR…` id out of an id, a Transparency Center URL, or return None."""
70 t = (text or "").strip()
71 if not t:
72 return None
73 if t.startswith("AR") and t[2:].isdigit():
74 return t
75 for part in t.replace("?", "/").replace("&", "/").split("/"):
76 if part.startswith("AR") and part[2:].isdigit():
77 return part
78 return None
79
80
81class AdsClient:
82 """One session, bound to one egress IP for its life. To rotate, build a new client."""
83
84 def __init__(self, min_gap: float = MIN_GAP, proxy_url: str | None = None):
85 handlers = []
86 if proxy_url:
87 handlers.append(
88 urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url})
89 )
90 self._op = urllib.request.build_opener(*handlers)
91 self.proxy_url = proxy_url
92 self.min_gap = float(min_gap)
93 self._last = 0.0
94 self.requests_made = 0
95
96 def _pace(self) -> None:
97 wait = self.min_gap - (time.monotonic() - self._last)
98 if wait > 0:
99 time.sleep(wait + random.uniform(0, 0.3))
100 self._last = time.monotonic()
101
102 def _rpc(self, method: str, payload: dict) -> dict:
103 body = urllib.parse.urlencode(
104 {"f.req": json.dumps(payload, separators=(",", ":"))}
105 ).encode()
106 req = urllib.request.Request(
107 BASE % method,
108 data=body,
109 headers={
110 "User-Agent": UA,
111 "Content-Type": "application/x-www-form-urlencoded",
112 "Accept-Encoding": "gzip",
113 "Referer": "https://adstransparency.google.com/",
114 "X-Same-Domain": "1",
115 },
116 )
117 for pause in (None,) + BACKOFF:
118 if pause:
119 log.warning(
120 "rate-limited by Google — waiting %.0fs then retrying once", pause
121 )
122 time.sleep(pause)
123 self._pace()
124 self.requests_made += 1
125 try:
126 raw = _decode(self._op.open(req, timeout=45).read())
127 except urllib.error.HTTPError as exc:
128 text = _decode(exc.read())
129 if exc.code == 429:
130 continue
131
132
133 raise AdsError(
134 "HTTP %s from %s: %s" % (exc.code, method, text[:300])
135 ) from None
136 except urllib.error.URLError as exc:
137 raise AdsError("%s unreachable: %s" % (method, exc)) from None
138 if not raw.strip():
139 return {}
140 try:
141 return json.loads(raw)
142 except ValueError:
143 raise AdsError(
144 "%s returned non-JSON: %s" % (method, raw[:200])
145 ) from None
146 raise AdsBlocked("rate-limited by Google on %s" % method)
147
148
149 def search_advertisers(self, keyword: str, limit: int = 10) -> list[dict]:
150 """Advertisers whose name matches `keyword`, with their id, country and ad count."""
151 data = self._rpc(
152 "SearchService/SearchSuggestions", {"1": keyword, "2": limit, "3": limit}
153 )
154 rows = []
155 for raw in data.get("1") or []:
156
157 r = raw.get("1", raw) if isinstance(raw, dict) else {}
158 count = ((r.get("4") or {}).get("2") or {}).get("1")
159 rows.append(
160 {
161 "advertiserName": r.get("1"),
162 "advertiserId": r.get("2"),
163 "advertiserCountry": r.get("3"),
164
165
166
167 "declaredAdCount": int(count)
168 if str(count or "").isdigit()
169 else None,
170 "advertiserUrl": (
171 "https://adstransparency.google.com/advertiser/%s" % r.get("2")
172 if r.get("2")
173 else None
174 ),
175 }
176 )
177 return [r for r in rows if r["advertiserId"]]
178
179
180 def creatives_page(
181 self,
182 advertiser: str,
183 count: int = 40,
184 token: str | None = None,
185 region: int | None = None,
186 domain: str = "",
187 ) -> tuple[list[dict], str | None]:
188 """One page of ads. `region` is Google's own enum and is left out by default — a wrong
189 value returns zero rows with no error, which is indistinguishable from an advertiser that
190 simply has no ads."""
191 payload: dict = {
192 "2": max(1, min(int(count), 100)),
193 "3": {"12": {"1": domain, "2": True}, "13": {"1": [advertiser]}},
194 "7": {"1": 1},
195 }
196 if region is not None:
197 payload["3"]["8"] = [int(region)]
198 if token:
199 payload["4"] = token
200 data = self._rpc("SearchService/SearchCreatives", payload)
201 rows = [self._creative(c) for c in (data.get("1") or [])]
202 return [r for r in rows if r.get("creativeId")], data.get("2")
203
204 @staticmethod
205 def _ts(node) -> str | None:
206 """{1: unixSeconds, 2: nanos} -> ISO 8601. The seconds arrive as a STRING."""
207 try:
208 return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(node["1"])))
209 except Exception:
210 return None
211
212 @staticmethod
213 def _creative(c: dict) -> dict:
214 """One creative record, as measured on live data.
215
216 `4` is a format code and `3` is the content, which comes in two observed flavours: `3.1.4`
217 is a rendered-preview URL, `3.3.2` is an HTML `<img>` snippet. The format is reported from
218 the content actually present rather than from the code, because the code's full enum is not
219 known and inventing labels for unseen values would be a guess dressed up as data. The raw
220 code is passed through as `formatCode` so a caller can group by it regardless.
221 """
222 adv, cid = c.get("1"), c.get("2")
223 content = c.get("3") or {}
224 preview = image_html = fmt = None
225 if isinstance(content.get("1"), dict):
226 fmt, preview = "rendered_preview", content["1"].get("4")
227 elif isinstance(content.get("3"), dict):
228 fmt = "image"
229 image_html = content["3"].get("2")
230 if image_html:
231 marker = 'src="'
232 k = image_html.find(marker)
233 if k >= 0:
234 preview = image_html[k + len(marker):].split('"', 1)[0]
235 return {
236 "advertiserId": adv,
237 "advertiserName": c.get("12"),
238 "creativeId": cid,
239 "format": fmt,
240 "formatCode": c.get("4"),
241 "previewUrl": preview,
242 "imageHtml": image_html,
243 "creativeUrl": (
244 "https://adstransparency.google.com/advertiser/%s/creative/%s" % (adv, cid)
245 if adv and cid
246 else None
247 ),
248 "firstShownAt": AdsClient._ts(c.get("6")),
249 "lastShownAt": AdsClient._ts(c.get("7")),
250 }
