code wiki / _hdl_build / nx_cc_ingest.nx
`nx_cc_ingest.nx` source

↩ module page · 231 lines · 14571 B
1// nx_cc_ingest.nx -- THE CC BRIDGE: ingest REAL Common Crawl records into the web scope, INDEX-ONLY, serve-LIVE.
2// Per the operator's July-2026 architecture (store index not pages; link the live url; CC = index-source + rot
3// fallback): CDX-query CC-MAIN-2026-25 for <domain> -> for each record: range-fetch the exact byte-slice ->
4// gz-inflate the member -> pull the HTTP body out of the WARC record -> html->text -> ss_add into dp-web-pub-:
5//   doc:<cid> = extracted TEXT (snippet + BM25; NOT the full page)  ·  url:<cid> = the LIVE captured url
6//   loc:<cid> = "<warcfile>|<offset>|<length>" = the CDX locator for the rot-fallback (dead-live re-fetch)
7// Idempotent by content cid. Composes nx_commoncrawl_query's endpoint + nx_https_fetch_range + nx_gzip_inflate +
8// nx_html_to_text + the seg_store writer. usage: nx_cc_ingest <domain> [max_records]. license_tier: ORIGINAL
9import "nx_corpus_ingest.nx"          // ci_hash / ci_mkurlkey / dss_prefix / dss_mkkey / seg_store / nx_html_to_text
10import "nx_x509_trust_store.nx"
11import "nx_trust_store_load_from_certdata.nx"
12import "nx_https_fetch_follow.nx"     // nx_https_fetch_follow + nx_https_fetch_range
13import "nx_gzip_wrap.nx"              // nx_gzip_inflate -> *NxGzipResult
14import "nx_outlink_harvest.nx"        // olh_scan / olh_outkey -> P1 link-graph edges (feeds nx_pagerank_build)
15
16func gw(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
17func gn(v: i64) -> i64 { if v==0 { sys_write(1,"0" as *u8,1); return 0 } var m: i64=v; if m<0 { sys_write(1,"-" as *u8,1); m=0-m } let t: *u8=sys_mmap(24); var k: i64=0; while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 } let o: *u8=sys_mmap(24); var q: i64=k-1; var x: i64=0; while q>=0 { o[x]=t[q]; x=x+1; q=q-1 } sys_write(1,o,x); return 0 }
18func slen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
19func atoin(s: *u8) -> i64 { var v: i64=0; var i: i64=0; while s[i]!=(0 as u8){ let c: i64=s[i] as i64; if c>=48 { if c<=57 { v=v*10+(c-48) } } i=i+1 } return v }
20// JSON string field within window hay[0..hn): "key": "<value>" (whitespace after colon tolerated). -1 absent.
21func json_str(hay: *u8, hn: i64, key: *u8, out: *u8, cap: i64) -> i64 {
  let kl: i64 = slen(key)
  var i: i64 = 0
  while i + kl + 3 < hn {
      if hay[i] == (34 as u8) {
          var m: i64 = 1; var j: i64 = 0
          while j < kl { if hay[i+1+j] != key[j] { m = 0; j = kl } else { j = j + 1 } }
          if m == 1 { let after: i64 = i + 1 + kl
              if hay[after] == (34 as u8) { if hay[after+1] == (58 as u8) {
                  var p: i64 = after + 2; var stop: i64 = 0
                  while stop == 0 { if p >= hn { stop = 1 } else { if hay[p] == (32 as u8) { p = p + 1 } else { stop = 1 } } }
                  if p < hn { if hay[p] == (34 as u8) {
                      var s: i64 = p + 1; var o: i64 = 0
                      while s < hn { if hay[s] == (34 as u8) { s = hn } else { if o < cap-1 { out[o]=hay[s]; o=o+1 } s=s+1 } }
                      out[o]=0 as u8; return o
                  } }
              } }
          }
      }
      i = i + 1
  }
  return 0 - 1
43}
44// index AFTER the 2nd CRLFCRLF in b[0..n) = start of the HTTP response BODY inside a WARC 'response' record
45// (WARC headers \r\n\r\n HTTP headers \r\n\r\n BODY). -1 if not found.
46func warc_body_start(b: *u8, n: i64) -> i64 {
  var count: i64 = 0; var i: i64 = 0
  while i + 3 < n {
      if b[i]==(13 as u8) { if b[i+1]==(10 as u8) { if b[i+2]==(13 as u8) { if b[i+3]==(10 as u8) {
          count = count + 1
          if count == 2 { return i + 4 }
          i = i + 4
      } else { i = i + 1 } } else { i = i + 1 } } else { i = i + 1 } } else { i = i + 1 }
  }
  return 0 - 1
56}
57// loc:<cid> key builder
58func mk_lockey(cid: i64, out: *u8) -> i64 {
  out[0]=108 as u8; out[1]=111 as u8; out[2]=99 as u8; out[3]=58 as u8   // "loc:"
  var o: i64=4; if cid==0 { out[o]=48 as u8; o=o+1; out[o]=0 as u8; return o }
  let t: *u8=sys_mmap(24); var k: i64=0; var m: i64=cid
  while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
  var j: i64=0; while j<k { out[o]=t[k-1-j]; o=o+1; j=j+1 } out[o]=0 as u8; return o
64}
65
66func main(argc: i64, argv: *i64) -> i64 {
  if argc < 2 { gw("usage: nx_cc_ingest <domain-or-url-prefix> [max_records] [matchtype: domain|prefix]\n" as *u8); return 1 }
  let domain: *u8 = argv[1] as *u8
  var maxrec: i64 = 5
  if argc >= 3 { maxrec = atoin(argv[2] as *u8) }
  if maxrec > 50 { maxrec = 50 }
  // matchType: domain (default, whole-site sample) or prefix (ENTITY-TARGETED: exact URL subtree,
  // e.g. javdatabase.com/idols/julia -- the CC CDX answers path-prefix queries directly)
  var mtype: *u8 = "domain" as *u8
  if argc >= 4 {
      let mt: *u8 = argv[3] as *u8
      if mt[0] == (112 as u8) { mtype = "prefix" as *u8 }
  }
  let r: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, 4194304)
  if r <= 0 { gw("trust store load failed\n" as *u8); return 2 }
  let store: *TrustStore = r as *TrustStore
  gw("=== nx_cc_ingest: real Common Crawl -> dp-web-pub- (INDEX-ONLY, serve-LIVE) domain=" as *u8); gw(domain); gw(" ===\n" as *u8)
83
  // CDX query -> JSON lines
  let cap: i64 = 4194304
  let out: *u8 = sys_mmap(cap)
  let st: *i64 = sys_mmap(8) as *i64
  let cdx: *u8 = sys_mmap(1024); var co: i64 = 0
  let a1: *u8 = "https://index.commoncrawl.org/CC-MAIN-2026-25-index?url=" as *u8
  var pi: i64=0; while a1[pi]!=(0 as u8){ cdx[co]=a1[pi]; co=co+1; pi=pi+1 }
  var di: i64=0; while domain[di]!=(0 as u8){ cdx[co]=domain[di]; co=co+1; di=di+1 }
  let a2: *u8 = "&output=json&matchType=" as *u8
  pi=0; while a2[pi]!=(0 as u8){ cdx[co]=a2[pi]; co=co+1; pi=pi+1 }
  pi=0; while mtype[pi]!=(0 as u8){ cdx[co]=mtype[pi]; co=co+1; pi=pi+1 }
  let a2b: *u8 = "&limit=" as *u8
  pi=0; while a2b[pi]!=(0 as u8){ cdx[co]=a2b[pi]; co=co+1; pi=pi+1 }
  // append maxrec
  let mt: *u8=sys_mmap(24); var mk: i64=0; var mm: i64=maxrec; if mm==0 { mt[0]=48 as u8; mk=1 }
  while mm>0 { mt[mk]=(48+(mm%10)) as u8; mm=mm/10; mk=mk+1 }
  var mj: i64=mk-1; while mj>=0 { cdx[co]=mt[mj]; co=co+1; mj=mj-1 }
  cdx[co]=0 as u8
  let n1: i64 = nx_https_fetch_follow(cdx, store, out, cap, 10, st)
  gw("  cdx status=" as *u8); gn(st[0]); gw(" bytes=" as *u8); gn(n1); gw("\n" as *u8)
  if st[0] != 200 { gw("  (domain absent in this crawl or CDX error)\n" as *u8); return 3 }
105
  // open the web shard + fresh segid
  let prefix: *u8 = sys_mmap(512); dss_prefix("web" as *u8, prefix)
  let segs: *i64 = sys_mmap(4096*8) as *i64
  let nseg: i64 = ss_manifest_cap(prefix, segs, 4096)
  var segid: i64 = 1; var si: i64 = 0
  // ROOT FIX seq1730 (id 1785450987): segs[] holds POINTERS to seg-<id> name strings
  // (nx_seg_store.nx:1234 stores segs[cnt] = name as i64), NOT ids -- so segs[si] + 1 produced
  // MMAP_ADDRESS+1 (~1.4e14), the pointer-shaped poison that PINS a plane forever: every later
  // epoch id sorts BELOW it in supersede order and its rows are silently shadowed while rc=0.
  // Parse the DIGITS, as nx_web_shard_compact.nx:55-58 already does on this SAME array.
  while si < nseg {
      let sg_nm: *u8 = segs[si] as *u8
      var sg_v: i64 = 0
      var sg_ci: i64 = 0
      while sg_nm[sg_ci] != (0 as u8) { let sg_c: i64 = sg_nm[sg_ci] as i64; if sg_c >= 48 { if sg_c <= 57 { sg_v = sg_v * 10 + (sg_c - 48) } } sg_ci = sg_ci + 1 }
      if sg_v >= segid { segid = sg_v + 1 }
      si = si + 1
  }
  let h: *i64 = ss_open(prefix)
125
  let rec: *u8 = sys_mmap(2097152)         // one gzip member (compressed record)
  let rst: *i64 = sys_mmap(8) as *i64
  let text: *u8 = sys_mmap(1048576)
  let key: *u8 = sys_mmap(64); let ukey: *u8 = sys_mmap(64); let lkey: *u8 = sys_mmap(64)
  let pbox: *i64 = sys_mmap(16) as *i64; let lbox: *i64 = sys_mmap(16) as *i64
  let urlv: *u8 = sys_mmap(2048); let fname: *u8 = sys_mmap(1024); let offs: *u8 = sys_mmap(64); let lens: *u8 = sys_mmap(64)
  let durl: *u8 = sys_mmap(2048); let locv: *u8 = sys_mmap(2048)
  let w: *i64 = ss_begin()
  let ceb: *i64 = sys_mmap(OLH_MAXEDGE * 8) as *i64        // reused per page: this page's outlink target cids
  let okey2: *u8 = sys_mmap(64)
  var ingested: i64 = 0; var present: i64 = 0; var fetched: i64 = 0; var edged: i64 = 0
137
  // iterate CDX JSON lines (newline-separated)
  var ls: i64 = 0
  while ls < n1 {
      // find the end of this JSON line (next newline, or buffer end)
      var e2: i64 = ls
      var scan: i64 = 1
      while scan == 1 { if e2 >= n1 { scan = 0 } else { if out[e2] == (10 as u8) { scan = 0 } else { e2 = e2 + 1 } } }
      let linelen: i64 = e2 - ls
      if linelen < 20 { ls = e2 + 1; continue }
      let line: *u8 = (out as i64 + ls) as *u8
      // parse the record's fields
      if json_str(line, linelen, "filename" as *u8, fname, 1024) < 0 { ls = e2 + 1; continue }
      if json_str(line, linelen, "offset" as *u8, offs, 64) < 0 { ls = e2 + 1; continue }
      if json_str(line, linelen, "length" as *u8, lens, 64) < 0 { ls = e2 + 1; continue }
      if json_str(line, linelen, "url" as *u8, urlv, 2048) < 0 { ls = e2 + 1; continue }
      // QUALITY FILTER: only real HTML pages (status 200 + mime text/html); skip robots/sitemaps/feeds.
      let mimev: *u8 = sys_mmap(128); let statv: *u8 = sys_mmap(16)
      if json_str(line, linelen, "status" as *u8, statv, 16) < 0 { ls = e2 + 1; continue }
      if statv[0] != (50 as u8) { ls = e2 + 1; continue }               // status must start '2' (2xx)
      if json_str(line, linelen, "mime" as *u8, mimev, 128) < 0 { ls = e2 + 1; continue }
      var ishtml: i64 = 0
      if mimev[0]==(116 as u8) { if mimev[1]==(101 as u8) { if mimev[2]==(120 as u8) { if mimev[3]==(116 as u8) { if mimev[4]==(47 as u8) { if mimev[5]==(104 as u8) { ishtml = 1 } } } } } }  // "text/h"
      if ishtml == 0 { ls = e2 + 1; continue }
      // skip robots.txt / sitemap / .xml feed urls (not content pages)
      let ul0: i64 = slen(urlv)
      var skipu: i64 = 0
      var ux: i64 = 0
      while ux + 4 <= ul0 {
          if urlv[ux]==(114 as u8) { if urlv[ux+1]==(111 as u8) { if urlv[ux+2]==(98 as u8) { if urlv[ux+3]==(111 as u8) { skipu = 1; ux = ul0 } } } }   // "robo"
          ux = ux + 1
      }
      if ul0 >= 4 { if urlv[ul0-4]==(46 as u8) { if urlv[ul0-3]==(120 as u8) { if urlv[ul0-2]==(109 as u8) { if urlv[ul0-1]==(108 as u8) { skipu = 1 } } } } }  // ".xml"
      if skipu == 1 { ls = e2 + 1; continue }
      let offv: i64 = atoin(offs); let lenv: i64 = atoin(lens)
      if lenv < 10 { ls = e2 + 1; continue }
      if lenv > 2000000 { ls = e2 + 1; continue }
      // data-host url
      var dodo: i64 = 0
      let d1: *u8 = "https://data.commoncrawl.org/" as *u8
      pi=0; while d1[pi]!=(0 as u8){ durl[dodo]=d1[pi]; dodo=dodo+1; pi=pi+1 }
      var fi: i64=0; while fname[fi]!=(0 as u8){ durl[dodo]=fname[fi]; dodo=dodo+1; fi=fi+1 } durl[dodo]=0 as u8
      // range-fetch (minimal hello then Chrome-JA3 fallback)
      var nr: i64 = nx_https_fetch_range(durl, store, rec, 2097152, offv, offv+lenv-1, rst, 0)
      if nr <= 0 { nr = nx_https_fetch_range(durl, store, rec, 2097152, offv, offv+lenv-1, rst, 1) }
      fetched = fetched + 1
      if nr < 18 { ls = e2 + 1; continue }
      // inflate the gzip member
      let gz: *NxGzipResult = nx_gzip_inflate(rec, nr, 1048576)
      if gz.error_code != NX_GZ_OK { ls = e2 + 1; continue }
      let warc: *u8 = gz.output_data
      let warcn: i64 = gz.output_size
      // HTTP body inside the WARC 'response' record
      let bstart: i64 = warc_body_start(warc, warcn)
      if bstart < 0 { ls = e2 + 1; continue }
      let html: *u8 = (warc as i64 + bstart) as *u8
      let htmln: i64 = warcn - bstart
      let tlen: i64 = nx_html_to_text(html, htmln, text, 1048576)
      if tlen < CI_MINDOC { ls = e2 + 1; continue }
      var tn: i64 = tlen; if tn > CI_DOCCAP { tn = CI_DOCCAP }
      let cid: i64 = ci_hash(text, tn)
      dss_mkkey(cid, key)
      var already: i64 = 0
      if (h as i64) != 0 { if ss_hget(h, key, pbox, lbox) == 1 { already = 1 } }
      if already == 1 { present = present + 1 } else {
          if ss_add(w, 1, key, text, tn) < 0 { if ss_commit(prefix, w, segid)==0 {} segid=segid+1; w[1]=0; ss_add(w, 1, key, text, tn) }
          ci_mkurlkey(cid, ukey)
          if ss_add(w, 1, ukey, urlv, slen(urlv)) < 0 { if ss_commit(prefix, w, segid)==0 {} segid=segid+1; w[1]=0; ss_add(w, 1, ukey, urlv, slen(urlv)) }
          // locator row for the rot fallback: "<filename>|<offset>|<length>"
          var lo: i64 = 0
          var lf: i64=0; while fname[lf]!=(0 as u8){ locv[lo]=fname[lf]; lo=lo+1; lf=lf+1 }
          locv[lo]=124 as u8; lo=lo+1                    // '|'
          var lof: i64=0; while offs[lof]!=(0 as u8){ locv[lo]=offs[lof]; lo=lo+1; lof=lof+1 }
          locv[lo]=124 as u8; lo=lo+1
          var lln: i64=0; while lens[lln]!=(0 as u8){ locv[lo]=lens[lln]; lo=lo+1; lln=lln+1 }
          mk_lockey(cid, lkey)
          if ss_add(w, 1, lkey, locv, lo) < 0 { if ss_commit(prefix, w, segid)==0 {} segid=segid+1; w[1]=0; ss_add(w, 1, lkey, locv, lo) }
          // P1 link-graph: harvest this page's outlinks -> out:<ci_hash(url)> = packed target cids. Node
          // identity == the crawler's and == dss_urlcid at serve time, so CC pages now feed PageRank too.
          let cne: i64 = olh_scan(html, htmln, urlv, slen(urlv), ceb, OLH_MAXEDGE)
          if cne > 0 {
              olh_outkey(ci_hash(urlv, slen(urlv)), okey2)
              if ss_add(w, 1, okey2, ceb as *u8, cne * 8) < 0 { if ss_commit(prefix, w, segid)==0 {} segid=segid+1; w[1]=0; ss_add(w, 1, okey2, ceb as *u8, cne * 8) }
              edged = edged + 1
          }
          ingested = ingested + 1
          gw("  + " as *u8); gw(urlv); gw(" (" as *u8); gn(tn); gw(" chars)\n" as *u8)
      }
      ls = e2 + 1
  }
  if w[1] > 0 { if ss_commit(prefix, w, segid)==0 {} }
  gw("CC-INGEST done: fetched=" as *u8); gn(fetched); gw(" ingested=" as *u8); gn(ingested); gw(" present=" as *u8); gn(present); gw(" edged=" as *u8); gn(edged); gw(" -> dp-web-pub-\n" as *u8)
  if ingested + present >= 1 { gw("CC-INGEST GREEN\n" as *u8); return 0 }
  gw("CC-INGEST: nothing ingested (diagnose above)\n" as *u8); return 4
231}
nx_cc_ingest.nx source

`nx_cc_ingest.nx` source
