Source identity
| Field | Recorded value |
|---|---|
| Source type | archive member |
| Original record | 01_inventory.py |
| Date | Not specified in source |
| Path or archive member | archives/libsrc-1c3f88063121/NETLIFY_AI_RECOVERY_2026-10-14_21-10-49/scripts/01_inventory.py |
| Original SHA-256 | eff76654de0027af54d9759f75435ab27a759da886a9905a14bb96f1f613c67b |
| Reading-copy SHA-256 | eff76654de0027af54d9759f75435ab27a759da886a9905a14bb96f1f613c67b |
| Archive source | {"sourceId": "libsrc-1c3f88063121", "member": "NETLIFY_AI_RECOVERY_2026-10-14_21-10-49/scripts/01_inventory.py"} |
Recovered source transcript
Download complete source text ↓190 lines · 9.0 KB
"""Phase 1: inventory every accessible Netlify project and gather classification evidence.
Read only. Resumable: sites already scanned (per checkpoint) are skipped."""
import concurrent.futures as cf
import html
import json
import os
import re
import sys
import urllib.parse
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from nlr_common import (API, ROOT, api_json, checkpoint_read, checkpoint_update, log, now, paginate,
request, safe_name, write_json)
KNOWN = {"stalwart-stardust-052d42", "megamind-agi-documentation", "feed-the-joe"}
STRONG = ["MEGAMIND", "MEGAMIND-AGI", "Seferim", "SEFERIM_AGI", "G16", "Pantheon", "ThatAIGuy", "That AI Guy",
"DNA consciousness", "consciousness research", "substrate", "federation", "Yggdrasil", "Epona",
"Rhiannon", "Lashon", "MADDIE", "AI_Brain", "FreeAIForEveryone", "NeuralSwarm", "agent farm",
"The Hive", "Skynet", "OmniCron", "Feed The Joe", "feedthejoe", "AGI"]
GENERIC = ["artificial intelligence", "machine learning", "neural network", "neural", "LLM", "large language model",
"GPT", "OpenAI", "Anthropic", "Claude", "chatbot", "AI agent", "agents", "inference", "embedding",
"transformer", "consciousness", "cognitive", "swarm", "Ollama", "llama", "model training",
"knowledge graph", "Yad"]
SITE_FIELDS = ["id", "name", "url", "ssl_url", "admin_url", "custom_domain", "domain_aliases", "created_at",
"updated_at", "account_slug", "account_name", "account_id", "state", "plan", "deploy_url",
"git_provider", "build_image", "functions_region", "managed_dns", "force_ssl", "ssl"]
BUILD_FIELDS = ["provider", "repo_type", "repo_url", "repo_branch", "repo_path", "base", "base_rel_dir", "dir",
"functions_dir", "cmd", "public_repo", "stop_builds", "package_path", "private_logs"]
PUB_FIELDS = ["id", "state", "context", "branch", "commit_ref", "commit_url", "title", "created_at", "published_at",
"updated_at", "deploy_ssl_url", "framework", "manual_deploy"]
SECRETISH = re.compile(r"(?i)(token|secret|password|passwd|api[_-]?key|auth)\s*[=:]\s*\S+")
def safe_site(s):
out = {k: s.get(k) for k in SITE_FIELDS}
bs = s.get("build_settings") or {}
out["build_settings"] = {k: bs.get(k) for k in BUILD_FIELDS if k in bs}
if out["build_settings"].get("cmd"):
out["build_settings"]["cmd"] = SECRETISH.sub(r"\1=<redacted>", out["build_settings"]["cmd"])
pd = s.get("published_deploy") or {}
out["published_deploy"] = {k: pd.get(k) for k in PUB_FIELDS} if pd else None
out["password_protected"] = bool(s.get("password")) or bool(s.get("has_password"))
out["sso_protected"] = bool(s.get("sso_login") or s.get("sso_login_context"))
return out
def site_dir(s):
return os.path.join(ROOT, "sites", f"{safe_name(s.get('account_slug'))}__{safe_name(s['name'])}__{s['id']}")
def raw_file(site_id, path):
url = f"{API}/sites/{site_id}/files/{urllib.parse.quote(path.lstrip('/'), safe='/')}"
return request(url, headers={"Content-Type": "application/vnd.bitballoon.v1.raw"})
def text_of(b):
t = b.decode("utf-8", "replace")
t = re.sub(r"(?is)<(script|style)[^>]*>.*?</\1>", " ", t) if "<html" in t[:2000].lower() else t
return html.unescape(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", t)))
def hits(text, words):
found = {}
for w in words:
pat = r"(?<![A-Za-z0-9])" + re.escape(w) + r"(?![A-Za-z0-9])"
n = len(re.findall(pat, text, flags=0 if w.isupper() and len(w) <= 4 else re.I))
if n:
found[w] = n
return found
def scan_site(s):
meta = safe_site(s)
d = site_dir(s)
ev = {"site_id": s["id"], "scanned_at": now(), "sources_checked": [], "title": None, "description": None,
"strong_hits": {}, "generic_hits": {}, "file_count": None, "total_bytes": None, "errors": []}
corpus = " ".join(filter(None, [s["name"], meta.get("custom_domain"), " ".join(meta.get("domain_aliases") or []),
meta["build_settings"].get("repo_url"),
(meta.get("published_deploy") or {}).get("title")]))
ev["sources_checked"].append("metadata")
files = []
if meta.get("published_deploy"):
files, pages, st = paginate(f"/sites/{s['id']}/files")
if st != 200:
ev["errors"].append(f"listSiteFiles HTTP {st}")
ev["file_count"] = len(files)
ev["total_bytes"] = sum(f.get("size") or 0 for f in files)
corpus += " " + " ".join(f["path"] for f in files)
ev["sources_checked"].append("file paths")
# Pick a small set of text files for content clues.
def prio(f):
p = f["path"].lower()
if p == "/index.html":
return 0
if re.search(r"readme|about|claude\.md|docs?/", p):
return 1
if p.endswith((".md", ".txt")):
return 2
if p.endswith(".html"):
return 3
if p.endswith(("manifest.json", "package.json")):
return 4
if p.endswith(".js"):
return 6
return 9
cands = sorted([f for f in files if (f.get("size") or 0) < 3_000_000 and prio(f) < 9], key=lambda f: (prio(f), f["path"]))
picked, js = [], 0
for f in cands:
if prio(f) == 6:
if js >= 1:
continue
js += 1
picked.append(f)
if len(picked) >= 6:
break
for f in picked:
st, h, body, _ = raw_file(s["id"], f["path"])
if st != 200:
ev["errors"].append(f"raw {f['path']} HTTP {st}")
continue
ev["sources_checked"].append(f["path"])
if f["path"].lower().endswith((".html", ".htm")):
t = body.decode("utf-8", "replace")
m = re.search(r"(?is)<title[^>]*>(.*?)</title>", t)
if m and not ev["title"]:
ev["title"] = html.unescape(m.group(1)).strip()[:200]
m = re.search(r'(?is)<meta[^>]+name=["\']description["\'][^>]+content=["\']([^"\']*)', t)
if m and not ev["description"]:
ev["description"] = html.unescape(m.group(1)).strip()[:300]
corpus += " " + text_of(body)[:400_000]
ev["strong_hits"] = hits(corpus, STRONG)
ev["generic_hits"] = hits(corpus, GENERIC)
write_json(os.path.join(d, "metadata", "site.json"), meta)
write_json(os.path.join(d, "metadata", "classification_evidence.json"), ev)
if files:
write_json(os.path.join(d, "manifests", "current_site_files_listing.json"),
{"retrieved_at": now(), "endpoint": "GET /api/v1/sites/{site_id}/files",
"published_deploy_id": (meta.get("published_deploy") or {}).get("id"), "files": files})
return meta, ev
def main():
log("phase 1 inventory start")
st, accounts = api_json("/accounts")
write_json(os.path.join(ROOT, "metadata_accounts.json"),
[{k: a.get(k) for k in ("id", "name", "slug", "type_name", "created_at", "roles_allowed")} for a in accounts or []])
sites, pages, st = paginate("/sites", {"filter": "all"})
log(f"listSites filter=all: HTTP {st}, {pages} pages, {len(sites)} records")
per_account = {}
for a in accounts or []:
items, p2, st2 = paginate(f"/{a['slug']}/sites")
per_account[a["slug"]] = {"http": st2, "pages": p2, "count": len(items)}
sites.extend(items)
byid = {}
for s in sites:
byid.setdefault(s["id"], s)
log(f"unique sites by id: {len(byid)}; per account listing: {per_account}")
checkpoint_update(lambda c: c.update({"inventory_scope": {"accounts": [a["slug"] for a in accounts or []],
"list_sites_pages": pages, "per_account": per_account,
"unique_sites": len(byid), "at": now()}}))
done = checkpoint_read().get("sites", {})
todo = [s for s in byid.values() if not done.get(s["id"], {}).get("scanned")]
log(f"scanning {len(todo)} sites ({len(byid) - len(todo)} already scanned)")
def work(s):
try:
meta, ev = scan_site(s)
def upd(c):
c["sites"].setdefault(s["id"], {}).update({"name": s["name"], "dir": site_dir(s), "scanned": True,
"strong": ev["strong_hits"], "generic": ev["generic_hits"],
"title": ev["title"]})
checkpoint_update(upd)
return s["name"], None
except Exception as e: # keep going on any single site failure
return s["name"], f"{type(e).__name__}: {e}"
with cf.ThreadPoolExecutor(max_workers=4) as ex:
for i, (name, err) in enumerate(ex.map(work, todo), 1):
if err:
log(f"scan error {name}: {err}")
if i % 20 == 0:
log(f"scanned {i}/{len(todo)}")
log("phase 1 inventory done")
if __name__ == "__main__":
main()