TypeScript 55.4%
Python 43.2%
SQL 1.2%
1#!/usr/bin/env python32"""Quick pre-validation probe for candidate sensor URLs (registry authoring helper).34Usage: python3 scripts/gen-probe-urls.py <urls.txt> [--out report.tsv] [--parallel 12]5One URL per line (blank lines and # comments ignored). For each URL a curl GET is issued (20 s timeout,6browser-like User-Agent, redirects followed) and the body is classified:7 FEED n=<items> RSS 2.0 / Atom / RDF with <item>/<entry> count8 JSON <shape> JSON document — reports top-level keys / array length / guessed items path9 HTML <bytes> server-rendered HTML (size of visible text)10 ERR <code|reason> HTTP error, timeout, TLS failure, WAF challenge…11The result is a TSV: url, class, detail, http code, final url, content-type. It is only a triage aid — the real12gate is apps/engine/src/validate.ts.13"""14import concurrent.futures15import json16import re17import subprocess18import sys1920UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15"212223def guess_items(obj, path=""):24 """Find the first array of dicts inside a JSON object (depth ≤ 3)."""25 if isinstance(obj, list):26 if obj and isinstance(obj[0], dict):27 return path or "", len(obj), sorted(obj[0].keys())[:12]28 return None29 if isinstance(obj, dict):30 for k, v in obj.items():31 r = guess_items(v, f"{path}.{k}" if path else k)32 if r:33 return r34 return None353637def probe(url: str):38 try:39 p = subprocess.run(40 ["curl", "-sSL", "--max-time", "25", "--connect-timeout", "10", "-A", UA,41 "-H", "Accept: application/rss+xml, application/atom+xml, application/json, text/html;q=0.9, */*;q=0.8",42 "-w", "\n@@@%{http_code}@@@%{url_effective}@@@%{content_type}", url],43 capture_output=True, timeout=40)44 out = p.stdout.decode("utf-8", "replace")45 if "@@@" not in out:46 err = p.stderr.decode("utf-8", "replace").strip().splitlines()47 return url, "ERR", (err[-1] if err else "no response")[:80], "", "", ""48 body, meta = out.rsplit("\n@@@", 1)49 code, final, ctype = (meta.split("@@@") + ["", ""])[:3]50 except subprocess.TimeoutExpired:51 return url, "ERR", "timeout", "", "", ""52 if code and code != "200":53 return url, "ERR", f"http {code}", code, final, ctype54 b = body.lstrip(" \r\n\t")55 if b.startswith("{") or b.startswith("["):56 try:57 obj = json.loads(b)58 except Exception as e: # noqa: BLE00159 return url, "ERR", f"bad json {str(e)[:40]}", code, final, ctype60 g = guess_items(obj)61 if g:62 return url, "JSON", f"items@{g[0] or '(root)'} n={g[1]} keys={','.join(g[2])}", code, final, ctype63 keys = list(obj.keys())[:10] if isinstance(obj, dict) else f"array[{len(obj)}]"64 return url, "JSON", f"no list; keys={keys}", code, final, ctype65 if re.search(r"<(rss|feed|rdf:RDF)[\s>]", b[:4000]):66 n = len(re.findall(r"<(item|entry)[\s>]", b))67 return url, "FEED", f"n={n}", code, final, ctype68 if re.search(r"<(urlset|sitemapindex)[\s>]", b[:2000]):69 n = len(re.findall(r"<loc>", b))70 return url, "SITEMAP", f"n={n}", code, final, ctype71 low = b[:6000].lower()72 if "cf-chl" in low or "just a moment" in low or "incapsula" in low or "_incapsula_resource" in low or "captcha" in low or "access denied" in low:73 return url, "ERR", "waf challenge", code, final, ctype74 text = re.sub(r"<script.*?</script>|<style.*?</style>", " ", b, flags=re.S | re.I)75 text = re.sub(r"<[^>]+>", " ", text)76 text = re.sub(r"\s+", " ", text).strip()77 title = re.search(r"<title[^>]*>(.*?)</title>", b, flags=re.S | re.I)78 return url, "HTML", f"text={len(text)} title={(title.group(1).strip()[:60] if title else '')!r}", code, final, ctype798081def main() -> None:82 src = sys.argv[1]83 out = sys.argv[sys.argv.index("--out") + 1] if "--out" in sys.argv else None84 par = int(sys.argv[sys.argv.index("--parallel") + 1]) if "--parallel" in sys.argv else 1285 urls = [l.strip() for l in open(src) if l.strip() and not l.startswith("#")]86 rows = []87 with concurrent.futures.ThreadPoolExecutor(max_workers=par) as ex:88 for r in ex.map(probe, urls):89 rows.append(r)90 print("\t".join(r), flush=True)91 if out:92 with open(out, "w") as f:93 for r in rows:94 f.write("\t".join(r) + "\n")959697if __name__ == "__main__":98 main()99