_galx_prod_ingest.nx source

1// _galx_prod_ingest.nx -- PRODUCTION corpus ingest driver (sovereign, idempotent, additive). 2// Reads knowledge/status/galx_corpus_full.txt (one absolute PNG path per line) and ingests the LINE 3// RANGE [start, start+count) via the proven nx_store_ingest_ingest_cid organ into the stable namespace 4// "knowledge/store/galx-prod-", APPENDING <CID>\t<path> rows to the sidecar. The range arg lets the 5// sovereign Nishi batch-runner (nx_galx_fill) drive the whole corpus in memory-bounded slices: one fresh 6// process per slice, so the kernel reclaims all mmaps on exit (sys_munmap is unavailable, so process- 7// batching is the sovereign memory bound -- NO bash split/awk). Usage: _galx_prod_ingest <start> <count> 8// (no args = whole file). license_tier: ORIGINAL 9import "nx_store_ingest.nx" 10import "nx_seg_store.nx" 11import "nx_syscalls.nx" 12 13func pw(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 14func pn(v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48;k=1}; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1}; var i: i64=0; while i<k{bb[i]=t[k-1-i];i=i+1}; sys_write(1,bb,k); return 0 } 15func gi_atoi(s: *u8) -> i64 { var v: i64=0; var i: i64=0; while s[i]!=(0 as u8){ if s[i]>=(48 as u8){ if s[i]<=(57 as u8){ v=v*10+((s[i]-(48 as u8)) as i64) } } i=i+1 } return v } 16 17func main(argc: i64, argv: *i64) -> i64 { 18 var start: i64 = 0 19 var count: i64 = 1000000000 20 if argc >= 3 { start = gi_atoi(argv[1] as *u8); count = gi_atoi(argv[2] as *u8) } 21 let endl: i64 = start + count 22 let prefix: *u8 = "knowledge/store/galx-prod-" as *u8 23 24 let lszp: *i64 = sys_mmap(16) as *i64 25 let lst: *u8 = sys_read_file("knowledge/status/galx_corpus_full.txt" as *u8, lszp) 26 let lsz: i64 = lszp[0] 27 if (lst as i64) == 0 { pw("PRODINGEST FAIL: cannot read galx_corpus_full.txt\n" as *u8); sys_exit(1); return 1 } 28 29 let sfd: i64 = sys_openat_append("knowledge/status/galx_cid_paths.tsv" as *u8, 0x1a4) 30 if sfd < 0 { pw("PRODINGEST FAIL: sidecar unwritable\n" as *u8); sys_exit(1); return 1 } 31 32 let pathbuf: *u8 = sys_mmap(1024) 33 let cidbuf: *u8 = sys_mmap(96) 34 var ls: i64 = 0 35 var i: i64 = 0 36 var lineno: i64 = 0 37 var newc: i64 = 0 38 var dupc: i64 = 0 39 var skip: i64 = 0 40 while i <= lsz { 41 var atend: i64 = 0 42 if i == lsz { atend = 1 } 43 var isnl: i64 = 0 44 if atend == 0 { if lst[i] == (10 as u8) { isnl = 1 } } 45 if atend == 1 { isnl = 1 } 46 if isnl == 1 { 47 let llen: i64 = i - ls 48 if llen > 0 { 49 if lineno >= start { if lineno < endl { 50 var po: i64 = 0 51 while po < llen { pathbuf[po] = lst[ls + po]; po = po + 1 } 52 pathbuf[llen] = 0 as u8 53 let fszp: *i64 = sys_mmap(16) as *i64 54 let fbuf: *u8 = sys_read_file(pathbuf, fszp) 55 let flen: i64 = fszp[0] 56 if (fbuf as i64) == 0 { skip = skip + 1 } else { 57 if flen < 8 { skip = skip + 1 } else { 58 let r: i64 = nx_store_ingest_ingest_cid(fbuf, flen, prefix, cidbuf) 59 if r < 0 { skip = skip + 1 } else { 60 if r == 1 { newc = newc + 1 } else { dupc = dupc + 1 } 61 sys_write(sfd, cidbuf, 69) 62 sys_write(sfd, "\t" as *u8, 1) 63 sys_write(sfd, pathbuf, llen) 64 sys_write(sfd, "\n" as *u8, 1) 65 } 66 } 67 } 68 }} 69 lineno = lineno + 1 70 } 71 ls = i + 1 72 } 73 i = i + 1 74 } 75 sys_close(sfd) 76 pw("PRODINGEST range [" as *u8); pn(start); pw("," as *u8); pn(endl); pw(") new=" as *u8); pn(newc); pw(" dup=" as *u8); pn(dupc); pw(" skip=" as *u8); pn(skip); pw("\n" as *u8) 77 sys_exit(0) 78 return 0 79}