Repository Public
1"""Client for the Pangram commercial AI-text detector (https://www.pangram.com), used ONLY to measure detector2behaviour on originals / LLM rewrites / HTR reconstructions (see ``docs/experiments/PANGRAM_EVAL.md``).34A detector score is a measurement of one commercial classifier, never evidence of authorship. Scores must not be5used as a training signal (that would turn the project into a detector-evasion tool).67API contract (verified 2026-11-10, docs.pangram.com/api-reference/{introduction,ai-detection,bulk-api}):8 base https://text.external-api.pangram.com, header ``x-api-key``9 POST /task {text, model} -> {task_id}; GET /task/{id} -> {stage, prediction_short, fraction_ai, ..., windows[]}10 POST /bulk {items:[{id,text}], model} -> {bulk_id, accepted_items, failed_items}; GET /bulk/{id}; GET /bulk/{id}/results11 billing: one unit per started 100-word block per text (Pangram 4.x), >= 1 unit per text; 0.05 USD/unit single,12 20 % discount in bulk (www.pangram.com/pricing, 2026-11-10); bulk launch limit 1 000 units per request.1314Guarantees: the key is read from ``$PANGRAM_API_KEY`` or ``~/Desktop/Cluster/secrets/pangram/.env`` and never logged;15every text is cached on disk by sha256(model + text) so it is never billed twice; every HTTP request that can bill is16appended to a JSONL ledger (timestamp, endpoint, n texts, chars, words, estimated units and USD — never the text);17``max_calls`` caps the number of texts submitted for billing by one client instance.18"""19from __future__ import annotations2021import hashlib22import json23import math24import os25import time26from pathlib import Path27from typing import Any, Callable, Iterable2829import httpx3031ROOT = Path(__file__).resolve().parents[3]32BASE_URL = "https://text.external-api.pangram.com"33DEFAULT_MODEL = "pangram-4.1" # pinned explicitly (the docs advise against relying on "default")34SECRETS_ENV = Path.home() / "Desktop/Cluster/secrets/pangram/.env"35USD_PER_UNIT_SINGLE = 0.0536USD_PER_UNIT_BULK = 0.0437WORDS_PER_UNIT = 10038BULK_MAX_UNITS = 100039RETRY_STATUS = {429, 500, 502, 503, 504}404142class BudgetExceeded(RuntimeError):43 pass444546class PangramError(RuntimeError):47 pass484950def load_api_key(env_file: Path = SECRETS_ENV) -> str:51 key = os.environ.get("PANGRAM_API_KEY", "").strip()52 if key:53 return key54 if env_file.exists():55 for line in env_file.read_text().splitlines():56 if line.strip().startswith("PANGRAM_API_KEY="):57 return line.split("=", 1)[1].strip().strip("\"'")58 raise PangramError(f"PANGRAM_API_KEY not set and not found in {env_file}")596061def billable_units(text: str) -> int:62 """Estimated billable units (started 100-word blocks, minimum 1). Pangram's own word count may differ slightly."""63 return max(1, math.ceil(len(text.split()) / WORDS_PER_UNIT))646566def text_key(text: str, model: str) -> str:67 return hashlib.sha256(f"{model}\x00{text}".encode("utf-8")).hexdigest()686970def normalise(raw: dict[str, Any]) -> dict[str, Any]:71 """Map a Pangram result to project fields.7273 ``ai_likelihood`` = word-weighted mean of the windows' ``ai_assistance_score`` (0 = fully human, 1 = fully74 AI-generated, per the docs); falls back to ``fraction_ai`` when no window is returned. ``label`` is the API's own75 ``prediction_short`` ("AI" / "Human" / "Mixed" for Pangram 4.x)."""76 wins = raw.get("windows") or []77 num = den = 0.078 for w in wins:79 s = w.get("ai_assistance_score")80 if s is None:81 continue82 wt = float(w.get("word_count") or 1)83 num += wt * float(s)84 den += wt85 ai = num / den if den > 0 else float(raw.get("fraction_ai") or 0.0)86 return {87 "ai_likelihood": min(1.0, max(0.0, ai)),88 "label": raw.get("prediction_short") or "",89 "headline": raw.get("headline") or "",90 "fraction_ai": float(raw.get("fraction_ai") or 0.0),91 "fraction_ai_assisted": float(raw.get("fraction_ai_assisted") or 0.0),92 "fraction_human": float(raw.get("fraction_human") or 0.0),93 "version": raw.get("version") or "",94 "stage": raw.get("stage") or "",95 "raw": raw,96 }979899class PangramClient:100 def __init__(self, api_key: str | None = None, cache_dir: Path | str = ROOT / "data/manifests/pangram_cache",101 ledger_path: Path | str = ROOT / "data/manifests/pangram_ledger.jsonl", model: str = DEFAULT_MODEL,102 max_calls: int | None = 1200, min_interval: float = 0.25, max_retries: int = 6,103 poll_interval: float = 1.0, poll_timeout: float = 4 * 3600.0, base_url: str = BASE_URL,104 transport: httpx.BaseTransport | None = None, sleep: Callable[[float], None] = time.sleep,105 run_tag: str = ""):106 self._key = api_key # resolved lazily so cache-only use needs no key107 self.cache_dir = Path(cache_dir)108 self.cache_dir.mkdir(parents=True, exist_ok=True)109 gi = self.cache_dir / ".gitignore" # cached raw responses echo the scored text (licensed passages): never commit110 if not gi.exists():111 gi.write_text("*\n")112 self.ledger_path = Path(ledger_path)113 self.model = model114 self.max_calls = max_calls115 self.min_interval = min_interval116 self.max_retries = max_retries117 self.poll_interval = poll_interval118 self.poll_timeout = poll_timeout119 self.base_url = base_url.rstrip("/")120 self._sleep = sleep121 self._transport = transport122 self._http: httpx.Client | None = None123 self._last = 0.0124 self.run_tag = run_tag125 self.calls = 0 # texts submitted for billing by this instance126 self.cache_hits = 0127 self.est_units = 0128 self.est_usd = 0.0129130 # ---------------------------------------------------------------- plumbing131 def _client(self) -> httpx.Client:132 if self._http is None:133 if self._key is None:134 self._key = load_api_key()135 self._http = httpx.Client(base_url=self.base_url, headers={"x-api-key": self._key}, timeout=60.0,136 transport=self._transport)137 return self._http138139 def close(self) -> None:140 if self._http is not None:141 self._http.close()142 self._http = None143144 def _request(self, method: str, path: str, **kw) -> httpx.Response:145 delay = 2.0146 for attempt in range(self.max_retries + 1):147 wait = self.min_interval - (time.monotonic() - self._last)148 if wait > 0:149 self._sleep(wait)150 self._last = time.monotonic()151 try:152 r = self._client().request(method, path, **kw)153 except httpx.TransportError as e:154 if attempt == self.max_retries:155 raise PangramError(f"{method} {path}: transport error {type(e).__name__}") from e156 self._sleep(delay)157 delay = min(delay * 2, 60.0)158 continue159 if r.status_code in RETRY_STATUS and attempt < self.max_retries:160 ra = r.headers.get("retry-after")161 try:162 pause = float(ra) if ra else delay163 except ValueError:164 pause = delay165 self._sleep(min(pause, 120.0))166 delay = min(delay * 2, 60.0)167 continue168 if r.status_code >= 400:169 raise PangramError(f"{method} {path}: HTTP {r.status_code} {r.text[:200]}")170 return r171 raise PangramError(f"{method} {path}: retries exhausted") # pragma: no cover172173 def _ledger(self, endpoint: str, texts: list[str], usd_per_unit: float, status: str, extra: dict | None = None) -> None:174 units = sum(billable_units(t) for t in texts)175 row = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "endpoint": endpoint, "model": self.model,176 "n_texts": len(texts), "chars": sum(len(t) for t in texts), "words": sum(len(t.split()) for t in texts),177 "est_units": units, "est_usd": round(units * usd_per_unit, 4), "status": status, "run": self.run_tag}178 if extra:179 row.update(extra)180 self.ledger_path.parent.mkdir(parents=True, exist_ok=True)181 with self.ledger_path.open("a") as f:182 f.write(json.dumps(row) + "\n")183184 def _cache_path(self, text: str) -> Path:185 h = text_key(text, self.model)186 return self.cache_dir / h[:2] / f"{h}.json"187188 def cached(self, text: str) -> dict | None:189 p = self._cache_path(text)190 if p.exists():191 return json.loads(p.read_text())192 return None193194 def _store(self, text: str, raw: dict) -> None:195 p = self._cache_path(text)196 p.parent.mkdir(parents=True, exist_ok=True)197 tmp = p.with_suffix(".tmp")198 tmp.write_text(json.dumps({"model": self.model, "sha256": text_key(text, self.model), "raw": raw}))199 tmp.replace(p)200201 def _reserve(self, n: int) -> None:202 if self.max_calls is not None and self.calls + n > self.max_calls:203 raise BudgetExceeded(f"max_calls={self.max_calls} would be exceeded ({self.calls} used, {n} requested)")204 self.calls += n205206 # ---------------------------------------------------------------- public API207 def score(self, text: str) -> dict[str, Any]:208 """Score one text (async task API, polled). Cached results are returned without any HTTP call."""209 if not text or not text.strip():210 raise ValueError("empty text")211 hit = self.cached(text)212 if hit is not None:213 self.cache_hits += 1214 return {**normalise(hit["raw"]), "cached": True}215 self._reserve(1)216 u = billable_units(text)217 self.est_units += u218 self.est_usd += u * USD_PER_UNIT_SINGLE219 try:220 r = self._request("POST", "/task", json={"text": text, "model": self.model})221 except Exception as e:222 self._ledger("POST /task", [text], USD_PER_UNIT_SINGLE, f"error: {str(e)[:120]}")223 raise224 tid = r.json()["task_id"]225 self._ledger("POST /task", [text], USD_PER_UNIT_SINGLE, "submitted", {"task_id": tid})226 raw = self._poll_task(tid)227 if raw.get("stage") != "STAGE_SUCCESS":228 raise PangramError(f"task {tid} failed: {raw.get('headline', '')[:200]}")229 self._store(text, raw)230 return {**normalise(raw), "cached": False}231232 def _poll_task(self, tid: str) -> dict:233 t0 = time.monotonic()234 while True:235 raw = self._request("GET", f"/task/{tid}").json()236 if raw.get("stage") in ("STAGE_SUCCESS", "STAGE_FAILED"):237 return raw238 if time.monotonic() - t0 > self.poll_timeout:239 raise PangramError(f"task {tid}: poll timeout")240 self._sleep(self.poll_interval)241242 def score_many(self, texts: Iterable[str], bulk: bool = True, bulk_max_units: int = 900,243 progress: Callable[[str], None] | None = None) -> dict[str, dict[str, Any]]:244 """Score distinct texts; returns {text: normalised result}. Uncached texts go through the Bulk API (20 %245 cheaper) in chunks of at most ``bulk_max_units`` estimated units (< the 1 000-unit launch limit), or through246 ``score`` one by one when ``bulk=False``. Failed items are reported with ``label='ERROR'`` and not cached."""247 uniq = list(dict.fromkeys(t for t in texts if t and t.strip()))248 out: dict[str, dict[str, Any]] = {}249 todo = []250 for t in uniq:251 hit = self.cached(t)252 if hit is not None:253 self.cache_hits += 1254 out[t] = {**normalise(hit["raw"]), "cached": True}255 else:256 todo.append(t)257 if todo and bulk:258 resumed = self.resume_pending(set(todo), progress)259 out.update({t: r for t, r in resumed.items() if t in set(todo)})260 todo = [t for t in todo if t not in out]261 if not todo:262 return out263 self._reserve(len(todo))264 if not bulk:265 self.calls -= len(todo) # score() reserves per text266 for t in todo:267 out[t] = self.score(t)268 return out269 chunk: list[str] = []270 units = 0271 chunks = []272 for t in todo:273 u = billable_units(t)274 if chunk and units + u > bulk_max_units:275 chunks.append(chunk)276 chunk, units = [], 0277 chunk.append(t)278 units += u279 if chunk:280 chunks.append(chunk)281 for ci, ch in enumerate(chunks):282 if progress:283 progress(f"bulk chunk {ci + 1}/{len(chunks)}: {len(ch)} texts, ~{sum(map(billable_units, ch))} units")284 out.update(self._bulk(ch))285 return out286287 def _pending_dir(self) -> Path:288 return self.cache_dir / "pending_bulk"289290 def _bulk(self, texts: list[str]) -> dict[str, dict[str, Any]]:291 ids = {f"t{i:05d}": t for i, t in enumerate(texts)}292 body = {"items": [{"id": k, "text": v} for k, v in ids.items()], "model": self.model}293 u = sum(billable_units(t) for t in texts)294 try:295 r = self._request("POST", "/bulk", json=body)296 except Exception as e:297 self._ledger("POST /bulk", texts, USD_PER_UNIT_BULK, f"error: {str(e)[:120]}")298 raise299 sub = r.json()300 bid = sub["bulk_id"]301 # persist the id → text map first, so a crash/timeout can be resumed without re-billing302 self._pending_dir().mkdir(parents=True, exist_ok=True)303 (self._pending_dir() / f"{bid}.json").write_text(json.dumps({"bulk_id": bid, "model": self.model, "ids": ids}))304 self._ledger("POST /bulk", texts, USD_PER_UNIT_BULK, "submitted", {"bulk_id": bid, "failed_at_submit": len(sub.get("failed_items") or [])})305 self.est_units += u306 self.est_usd += u * USD_PER_UNIT_BULK307 return self._collect(bid, ids)308309 def resume_pending(self, wanted: set[str] | None = None, progress: Callable[[str], None] | None = None) -> dict[str, dict[str, Any]]:310 """Collect results of bulk jobs submitted earlier but never collected (no new billing)."""311 out: dict[str, dict[str, Any]] = {}312 d = self._pending_dir()313 if not d.exists():314 return out315 for f in sorted(d.glob("*.json")):316 job = json.loads(f.read_text())317 if job.get("model") != self.model:318 continue319 if wanted is not None and not (set(job["ids"].values()) & wanted):320 continue321 if progress:322 progress(f"resuming bulk {job['bulk_id']} ({len(job['ids'])} texts, already billed)")323 out.update(self._collect(job["bulk_id"], job["ids"]))324 return out325326 def _collect(self, bid: str, ids: dict[str, str]) -> dict[str, dict[str, Any]]:327 texts = list(ids.values())328 t0 = time.monotonic()329 while True:330 st = self._request("GET", f"/bulk/{bid}").json()331 if st.get("status") in ("succeeded", "failed", "partial"):332 break333 if time.monotonic() - t0 > self.poll_timeout:334 raise PangramError(f"bulk {bid}: poll timeout (status {st.get('status')}, {st.get('succeeded')}/{st.get('total_items')}); "335 "re-run to resume (pending map kept, no re-billing)")336 self._sleep(max(self.poll_interval, 5.0))337 out: dict[str, dict[str, Any]] = {}338 offset = 0339 while True:340 page = self._request("GET", f"/bulk/{bid}/results", params={"offset": offset, "limit": 1000}).json()341 items = page.get("items") or []342 for it in items:343 t = ids.get(it.get("id"))344 res = it.get("result")345 if t is None:346 continue347 if res and res.get("stage") == "STAGE_SUCCESS":348 self._store(t, res)349 out[t] = {**normalise(res), "cached": False}350 else:351 out[t] = {"ai_likelihood": float("nan"), "label": "ERROR", "error": it.get("error") or (res or {}).get("headline", ""), "raw": res, "cached": False}352 for it in page.get("failed_items") or []:353 t = ids.get(it.get("id"))354 if t is not None and t not in out:355 out[t] = {"ai_likelihood": float("nan"), "label": "ERROR", "error": it.get("error", ""), "raw": None, "cached": False}356 offset += len(items)357 if not items or offset >= int(page.get("total_items") or 0):358 break359 for t in texts:360 out.setdefault(t, {"ai_likelihood": float("nan"), "label": "ERROR", "error": "missing from results", "raw": None, "cached": False})361 n_err = sum(1 for t in texts if out[t]["label"] == "ERROR")362 self._ledger(f"GET /bulk/{bid}/results", [], USD_PER_UNIT_BULK, st.get("status", "?"),363 {"bulk_id": bid, "n_items": len(texts), "n_error": n_err, "items_units": sum(billable_units(t) for t in texts)})364 pf = self._pending_dir() / f"{bid}.json"365 if pf.exists():366 pf.unlink()367 return out368369 def stats(self) -> dict[str, Any]:370 return {"model": self.model, "texts_submitted": self.calls, "cache_hits": self.cache_hits,371 "est_units": self.est_units, "est_usd": round(self.est_usd, 4)}372