#!/usr/bin/env python3 """Resumable parallel dumper for ergit.presidencia.gov.co ArcGIS Portal. Pulls every public item from the portal: - items/.json : item metadata (sharing/rest/content/items/) - data/. : item /data payload (PDF/Shapefile/Excel/Image/Code/Web Map JSON/Dashboard JSON/etc.) - resources// : item resources (icons, attached files) - layers//Lk.json : per-layer query dumps for Feature/Map Services - logs/dump.log, status.json Skips files already on disk so it's resumable. """ import json, os, sys, time, urllib.request, urllib.parse, urllib.error, threading from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path ROOT = Path(__file__).parent BASE = "https://ergit.presidencia.gov.co/arcpre/sharing/rest" UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36" WORKERS = 8 TIMEOUT = 60 PAGE_SIZE = 5000 # Feature Service query batch ITEMS_DIR = ROOT / "items" DATA_DIR = ROOT / "data" RES_DIR = ROOT / "resources" LAY_DIR = ROOT / "layers" LOG = ROOT / "logs" / "dump.log" STATUS = ROOT / "logs" / "status.json" INDEX = ROOT / "items_index.json" _lock = threading.Lock() _stats = {"items":0, "data":0, "layers":0, "skip":0, "err":0, "start":time.time()} def log(msg): line = f"[{time.strftime('%H:%M:%S')}] {msg}" with _lock: print(line, flush=True) with open(LOG, "a", encoding="utf-8") as f: f.write(line + "\n") def get(url, binary=False): req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept":"*/*"}) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: return r.read() if binary else r.read().decode("utf-8", errors="replace") def safe_get(url, binary=False, retries=2): for i in range(retries+1): try: return get(url, binary=binary) except urllib.error.HTTPError as e: if e.code in (404, 401, 403): return None if i==retries: raise except Exception: if i==retries: raise time.sleep(1+i) return None EXT = { "PDF":"pdf","Microsoft Excel":"xlsx","Microsoft Word":"docx","Microsoft Powerpoint":"pptx", "CSV":"csv","Image":"img","Shapefile":"zip","File Geodatabase":"zip","GeoJson":"geojson", "Service Definition":"sd","Code Attachment":"zip","KML":"kml","Map Document":"mxd", "Notebook":"ipynb","Form":"json","Web Map":"json","Web Scene":"json","Dashboard":"json", "StoryMap":"json","Web Mapping Application":"json","Web Experience":"json", "Insights Page":"json","Insights Workbook":"json","Insights Model":"json", "Site Application":"json","Site Page":"json","StoryMap Theme":"json", "AppBuilder Extension":"zip","Data Store":"json", } BINARY_TYPES = {"PDF","Microsoft Excel","Microsoft Word","Microsoft Powerpoint","Image", "Shapefile","File Geodatabase","Code Attachment","AppBuilder Extension", "Service Definition","KML"} SERVICE_TYPES = {"Feature Service","Map Service","Image Service","WFS","Geoprocessing Service", "Vector Tile Service","Scene Service","Geometry Service"} def dump_item(it): iid = it["id"] typ = it.get("type","") title = it.get("title","")[:60] # 1. item metadata meta_p = ITEMS_DIR / f"{iid}.json" if not meta_p.exists(): try: meta = safe_get(f"{BASE}/content/items/{iid}?f=json") if meta: meta_p.write_text(meta, encoding="utf-8") with _lock: _stats["items"]+=1 except Exception as e: log(f"ERR meta {iid}: {e}") with _lock: _stats["err"]+=1 # 2. /data payload ext = EXT.get(typ, "bin") is_bin = typ in BINARY_TYPES data_p = DATA_DIR / f"{iid}.{ext}" if not data_p.exists() and typ not in SERVICE_TYPES: try: url = f"{BASE}/content/items/{iid}/data" if is_bin: blob = safe_get(url, binary=True) if blob and len(blob)>0: data_p.write_bytes(blob) with _lock: _stats["data"]+=1 else: txt = safe_get(url) if txt and len(txt)>0: data_p.write_text(txt, encoding="utf-8") with _lock: _stats["data"]+=1 except Exception as e: log(f"ERR data {iid} ({typ}): {e}") with _lock: _stats["err"]+=1 # 3. resources list (cheap) res_idx = RES_DIR / f"{iid}.json" if not res_idx.exists(): try: rj = safe_get(f"{BASE}/content/items/{iid}/resources?f=json&num=100") if rj: res_idx.write_text(rj, encoding="utf-8") # also pull each resource if small list d = json.loads(rj) for r in d.get("resources",[])[:50]: name = r.get("resource","") if not name: continue safe = name.replace("/","__").replace("\\","__")[:100] out = RES_DIR / iid / safe if out.exists(): continue out.parent.mkdir(parents=True, exist_ok=True) try: blob = safe_get(f"{BASE}/content/items/{iid}/resources/{name}", binary=True) if blob: out.write_bytes(blob) except Exception: pass except Exception: pass # 4. Feature/Map Service layer dumps if typ in {"Feature Service","Map Service"}: url = it.get("url","") if url: dump_service(iid, url, typ) def dump_service(iid, url, typ): out_dir = LAY_DIR / iid done_marker = out_dir / "_done.txt" if done_marker.exists(): with _lock: _stats["skip"]+=1 return out_dir.mkdir(parents=True, exist_ok=True) try: meta = safe_get(f"{url}?f=json") if not meta: return (out_dir / "_service.json").write_text(meta, encoding="utf-8") sj = json.loads(meta) layers = sj.get("layers",[]) + sj.get("tables",[]) for L in layers: lid = L.get("id") try: lmeta = safe_get(f"{url}/{lid}?f=json") if lmeta: (out_dir / f"L{lid}_meta.json").write_text(lmeta, encoding="utf-8") # paginate query offset = 0 page = 0 while True: q = (f"{url}/{lid}/query?where=1%3D1&outFields=*&returnGeometry=true" f"&f=geojson&resultOffset={offset}&resultRecordCount={PAGE_SIZE}") qr = safe_get(q) if not qr: break try: qd = json.loads(qr) except: break feats = qd.get("features",[]) if not feats and offset>0: break (out_dir / f"L{lid}_p{page}.geojson").write_text(qr, encoding="utf-8") with _lock: _stats["layers"]+=1 if len(feats) < PAGE_SIZE: break offset += PAGE_SIZE page += 1 if page > 200: break # safety cap except Exception as e: log(f"ERR layer {iid}/{lid}: {e}") done_marker.write_text("ok", encoding="utf-8") except Exception as e: log(f"ERR service {iid}: {e}") with _lock: _stats["err"]+=1 def status_writer(): while True: time.sleep(15) with _lock: s = dict(_stats) s["elapsed"] = int(time.time() - s["start"]) STATUS.write_text(json.dumps(s, indent=2)) def main(): for d in (ITEMS_DIR, DATA_DIR, RES_DIR, LAY_DIR, LOG.parent): d.mkdir(parents=True, exist_ok=True) items = json.load(open(INDEX, encoding="utf-8")) log(f"loaded {len(items)} items from index") threading.Thread(target=status_writer, daemon=True).start() with ThreadPoolExecutor(max_workers=WORKERS) as ex: futs = [ex.submit(dump_item, it) for it in items] done = 0 for f in as_completed(futs): done += 1 if done % 25 == 0: with _lock: s = dict(_stats) log(f"progress {done}/{len(items)} items={s['items']} data={s['data']} layers={s['layers']} err={s['err']}") log(f"DONE final stats: {_stats}") if __name__ == "__main__": main()