nlr_common.py · Consciousness Atlas

Source identity

Field	Recorded value
Source type	archive member
Original record	nlr_common.py
Date	Not specified in source
Path or archive member	archives/libsrc-1c3f88063121/NETLIFY_AI_RECOVERY_2026-10-14_21-10-49/scripts/nlr_common.py
Original SHA-256	839570af553b21c544433073470e12eff20766c5c62c0b6257bafc1cbb204ebd
Reading-copy SHA-256	839570af553b21c544433073470e12eff20766c5c62c0b6257bafc1cbb204ebd
Archive source	{"sourceId": "libsrc-1c3f88063121", "member": "NETLIFY_AI_RECOVERY_2026-10-14_21-10-49/scripts/nlr_common.py"}
Recovered source transcript

Download complete source text ↓232 lines · 7.0 KB
"""Shared helpers for the Netlify AI recovery. Read only: only GET requests are ever issued."""
import datetime
import hashlib
import json
import os
import re
import ssl
import threading
import time
import urllib.error
import urllib.parse
import urllib.request

os.umask(0o077)

ROOT = os.environ.get("NLR_ROOT") or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
API = "https://api.netlify.com/api/v1"
UA = "netlify-ai-recovery/1.0 (read only preservation)"
CHECKPOINT = os.path.join(ROOT, "checkpoints", "state.json")


def _load_token():
    t = os.environ.get("NETLIFY_AUTH_TOKEN")
    if t:
        return t
    p = os.path.expanduser("~/Library/Preferences/netlify/config.json")
    d = json.load(open(p))
    return d["users"][d["userId"]]["auth"]["token"]


TOKEN = _load_token()


def now():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


_log_lock = threading.Lock()


def log(msg, name="recovery.log", echo=True):
    line = f"{now()} {msg}"
    with _log_lock:
        with open(os.path.join(ROOT, "logs", name), "a") as f:
            f.write(line + "\n")
    if echo:
        print(line, flush=True)


class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
    """Never forward the bearer token to a different host on redirect."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        new = super().redirect_request(req, fp, code, msg, headers, newurl)
        if new is not None and urllib.parse.urlsplit(newurl).netloc != urllib.parse.urlsplit(req.full_url).netloc:
            for k in list(new.headers):
                if k.lower() == "authorization":
                    del new.headers[k]
            new.unredirected_hdrs.pop("Authorization", None)
        return new


_opener = urllib.request.build_opener(_StripAuthOnRedirect)
_rl_lock = threading.Lock()
_rl = {"remaining": None, "reset": None}


def _update_rl(h):
    if not h:
        return
    try:
        rem = h.get("X-RateLimit-Remaining")
        reset = h.get("X-RateLimit-Reset")
        with _rl_lock:
            if rem is not None:
                _rl["remaining"] = int(rem)
            if reset is not None:
                _rl["reset"] = float(reset)
    except (TypeError, ValueError):
        pass


def request(url, auth=True, headers=None, timeout=180, max_tries=7):
    """GET only. Returns (status or None, headers, body bytes, final_url)."""
    h = {"User-Agent": UA}
    if auth:
        h["Authorization"] = "Bearer " + TOKEN
    if headers:
        h.update(headers)
    delay = 2.0
    last_err = None
    for _ in range(max_tries):
        with _rl_lock:
            rem, reset = _rl["remaining"], _rl["reset"]
        if auth and rem is not None and rem < 10 and reset:
            wait = reset - time.time() + 1
            if 0 < wait < 300:
                log(f"rate limit low ({rem}), waiting {wait:.0f}s", echo=False)
                time.sleep(wait)
        req = urllib.request.Request(url, headers=h, method="GET")
        try:
            with _opener.open(req, timeout=timeout) as r:
                body = r.read()
                if auth:
                    _update_rl(r.headers)
                return r.status, r.headers, body, r.geturl()
        except urllib.error.HTTPError as e:
            body = e.read() if e.fp else b""
            if auth:
                _update_rl(e.headers)
            if e.code == 429 or e.code >= 500:
                ra = e.headers.get("Retry-After") if e.headers else None
                wait = float(ra) if ra and ra.replace(".", "", 1).isdigit() else delay
                last_err = f"HTTP {e.code}"
                time.sleep(min(wait, 120))
                delay = min(delay * 2, 120)
                continue
            return e.code, e.headers, body, url
        except (urllib.error.URLError, TimeoutError, ConnectionError, ssl.SSLError, OSError) as e:
            last_err = type(e).__name__
            time.sleep(delay)
            delay = min(delay * 2, 120)
    log(f"giving up after retries: {redact_url(url)} ({last_err})", echo=False)
    return None, {}, b"", url


def redact_url(u):
    return re.sub(r"(token|signature|sig|X-Amz-[A-Za-z]+|ticket)=[^&]+", r"\1=<redacted>", u)


def api_json(path, params=None):
    url = API + path + ("?" + urllib.parse.urlencode(params) if params else "")
    st, h, body, _ = request(url)
    if st != 200:
        return st, None
    return st, json.loads(body)


def paginate(path, params=None):
    """Follow Link rel=next headers until exhausted. Returns (items, pages, last_status)."""
    params = dict(params or {})
    params.setdefault("per_page", 100)
    url = API + path + "?" + urllib.parse.urlencode(params)
    items, pages = [], 0
    while url:
        st, h, body, _ = request(url)
        if st != 200:
            return items, pages, st
        pages += 1
        data = json.loads(body)
        items.extend(data)
        nxt = None
        link = h.get("Link") if h else None
        if link:
            for part in link.split(","):
                m = re.search(r'<([^>]+)>\s*;\s*rel="next"', part)
                if m:
                    nxt = m.group(1)
        if not nxt and len(data) >= params["per_page"]:
            # Defensive: some endpoints omit Link. Ask for the next page explicitly.
            params["page"] = int(params.get("page", 1)) + 1
            nxt = API + path + "?" + urllib.parse.urlencode(params)
        if nxt and not data:
            nxt = None
        url = nxt
    return items, pages, 200


def write_json(path, obj):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump(obj, f, indent=2, sort_keys=True, ensure_ascii=False)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)


def write_bytes_atomic(path, data):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".part"
    with open(tmp, "wb") as f:
        f.write(data)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)


def sha1(b):
    return hashlib.sha1(b).hexdigest()


def sha256_file(p):
    h = hashlib.sha256()
    with open(p, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


_ck_lock = threading.Lock()


def checkpoint_update(fn):
    """fn(state) mutates state; persisted atomically."""
    with _ck_lock:
        try:
            st = json.load(open(CHECKPOINT))
        except (FileNotFoundError, json.JSONDecodeError):
            st = {"created": now(), "sites": {}, "deploys": {}}
        fn(st)
        st["updated"] = now()
        write_json(CHECKPOINT, st)
        return st


def checkpoint_read():
    try:
        return json.load(open(CHECKPOINT))
    except (FileNotFoundError, json.JSONDecodeError):
        return {"sites": {}, "deploys": {}}


def free_gb(path=ROOT):
    s = os.statvfs(path)
    return s.f_bavail * s.f_frsize / 1e9


def safe_name(s, maxlen=80):
    s = re.sub(r"[^A-Za-z0-9._-]+", "_", s or "").strip("._")
    return (s or "unnamed")[:maxlen]