code wiki / _hdl_build / nx_cc_warc_fetch.nx
nx_cc_warc_fetch.nx source
↩ module page · 146 lines · 7651 B
1// nx_cc_warc_fetch.nx -- sovereign Common Crawl bulk downloader (feeds nx_cc_warc_ingest).
2// Two verbs:
3// nx_cc_warc_fetch paths <crawl-id> [n] -> fetch crawl-data/<id>/warc.paths.gz, print first n paths
4// nx_cc_warc_fetch get <cc-warc-path> <outfile> -> chunked RESUMABLE download of one ~1GB .warc.gz
5// get: resumes from the existing outfile size (append-only, additive rule 13); sequential single-stream
6// range fetches with backoff on non-206 (polite to data.commoncrawl.org, which throttles bursts with 503);
7// EOF = a short chunk. Own TLS-1.3 + Mozilla trust store; zero curl. license_tier: ORIGINAL
8import "nx_x509_trust_store.nx"
9import "nx_trust_store_load_from_certdata.nx"
10import "nx_https_fetch_follow.nx" // nx_https_fetch_follow + nx_https_fetch_range
11import "nx_gzip_wrap.nx" // inflate warc.paths.gz
12const CWF_MAGIC_4194304: i64 = 4194304
13const CWF_MAGIC_1024: i64 = 1024
14const CWF_MAGIC_2048: i64 = 2048
15const CWF_MAGIC_131072: i64 = 131072
16
17const CWF_CHUNK: i64 = 4194304 // 4MiB range per request: big enough to amortize TLS, small enough to retry cheaply
18const CWF_MAXRETRY: i64 = 6 // attempts per chunk before giving up
19const CWF_BACKOFF_MS: i64 = 3000 // linear backoff step between retries (3s, 6s, 9s...) rides the 503 throttle
20const CWF_PATHCAP: i64 = 4194304 // compressed warc.paths.gz cap
21const CWF_PATHINFL: i64 = 33554432 // inflated warc.paths cap (~80k lines x ~100B)
22const CWF_PROGRESS: i64 = 16 // print progress every N chunks (= 64MiB)
23
24func cf_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
25func cf_num(v: i64) -> i64 {
26 let bb: *u8 = sys_mmap(28); var m: i64 = v
27 if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m }
28 let t: *u8 = sys_mmap(28); var k: i64 = 0
29 if m == 0 { t[0] = 48 as u8; k = 1 }
30 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
31 var i: i64 = 0; while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }
32 sys_write(1, bb, k); return 0
33}
34func cf_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
35func cf_cat(d: *u8, o: i64, s: *u8) -> i64 { var i: i64 = 0; while s[i] != (0 as u8) { d[o+i] = s[i]; i = i + 1 } return o + i }
36func cf_atoi(s: *u8) -> i64 {
37 var v: i64 = 0; var i: i64 = 0
38 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 }
39 return v
40}
41func cf_eq(a: *u8, b: *u8) -> i64 {
42 var i: i64 = 0
43 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
44 if b[i] == (0 as u8) { return 1 }
45 return 0
46}
47
48func main(argc: i64, argv: *i64) -> i64 {
49 if argc < 3 { cf_puts("usage: nx_cc_warc_fetch paths <crawl-id> [n] | get <cc-warc-path> <outfile>\n" as *u8); return 1 }
50 let verb: *u8 = argv[1] as *u8
51 let r: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, CWF_MAGIC_4194304)
52 if r <= 0 { cf_puts("trust store load failed\nverdict=FAIL\n" as *u8); return 2 }
53 let store: *TrustStore = r as *TrustStore
54 let st: *i64 = sys_mmap(8) as *i64
55
56 if cf_eq(verb, "paths" as *u8) == 1 {
57 let crawlid: *u8 = argv[2] as *u8
58 var nshow: i64 = 10
59 if argc >= 4 { nshow = cf_atoi(argv[3] as *u8) }
60 if nshow < 1 { nshow = 1 }
61 let url: *u8 = sys_mmap(CWF_MAGIC_1024)
62 var o: i64 = 0
63 o = cf_cat(url, o, "https://data.commoncrawl.org/crawl-data/" as *u8)
64 o = cf_cat(url, o, crawlid)
65 o = cf_cat(url, o, "/warc.paths.gz" as *u8)
66 url[o] = 0 as u8
67 let buf: *u8 = sys_mmap(CWF_PATHCAP)
68 let n1: i64 = nx_https_fetch_follow(url, store, buf, CWF_PATHCAP, 5, st)
69 cf_puts("paths fetch status=" as *u8); cf_num(st[0]); cf_puts(" bytes=" as *u8); cf_num(n1); cf_puts("\n" as *u8)
70 if st[0] != 200 { cf_puts("verdict=FAIL\n" as *u8); return 3 }
71 let gz: *NxGzipResult = nx_gzip_inflate(buf, n1, CWF_PATHINFL)
72 if gz.error_code != NX_GZ_OK { cf_puts("paths gunzip failed\nverdict=FAIL\n" as *u8); return 3 }
73 let p: *u8 = gz.output_data
74 let pn: i64 = gz.output_size
75 var line: i64 = 0; var i: i64 = 0; var lstart: i64 = 0
76 while i < pn { if p[i] == (10 as u8) {
77 if line < nshow { sys_write(1, (p as i64 + lstart) as *u8, i - lstart); sys_write(1, "\n" as *u8, 1) }
78 line = line + 1; lstart = i + 1
79 }
80 i = i + 1
81 }
82 cf_puts("total-paths=" as *u8); cf_num(line); cf_puts("\nverdict=PASS\n" as *u8)
83 return 0
84 }
85
86 if cf_eq(verb, "get" as *u8) == 1 {
87 if argc < 4 { cf_puts("usage: nx_cc_warc_fetch get <cc-warc-path> <outfile>\n" as *u8); return 1 }
88 let ccpath: *u8 = argv[2] as *u8
89 let outfile: *u8 = argv[3] as *u8
90 let url: *u8 = sys_mmap(CWF_MAGIC_2048)
91 var o: i64 = 0
92 o = cf_cat(url, o, "https://data.commoncrawl.org/" as *u8)
93 o = cf_cat(url, o, ccpath)
94 url[o] = 0 as u8
95 // resume point = existing file size (0 if absent)
96 var from: i64 = 0
97 let stb: *u8 = sys_mmap(160)
98 if sys_fstatat(outfile, stb) == 0 {
99 let szp: *i64 = ((stb as i64) + 48) as *i64
100 from = szp[0]
101 }
102 cf_puts("=== nx_cc_warc_fetch get " as *u8); cf_puts(ccpath); cf_puts(" resume-from=" as *u8); cf_num(from); cf_puts(" ===\n" as *u8)
103 let fd: i64 = sys_openat_append(outfile, 420) // 0644 append-only
104 if fd < 0 { cf_puts("cannot open outfile\nverdict=FAIL\n" as *u8); return 2 }
105 let buf: *u8 = sys_mmap(CWF_CHUNK + CWF_MAGIC_131072)
106 var chunks: i64 = 0
107 var go: i64 = 1
108 var failed: i64 = 0
109 while go == 1 {
110 var nr: i64 = 0
111 var attempt: i64 = 0
112 var okchunk: i64 = 0
113 while attempt < CWF_MAXRETRY {
114 nr = nx_https_fetch_range(url, store, buf, CWF_CHUNK + CWF_MAGIC_131072, from, from + CWF_CHUNK - 1, st, 0)
115 if nr <= 0 { nr = nx_https_fetch_range(url, store, buf, CWF_CHUNK + CWF_MAGIC_131072, from, from + CWF_CHUNK - 1, st, 1) }
116 if st[0] == 206 { if nr > 0 { okchunk = 1; attempt = CWF_MAXRETRY } }
117 if okchunk == 0 {
118 // 416 = requested range past EOF -> the file is complete (resume landed exactly at end)
119 if st[0] == 416 { go = 0; attempt = CWF_MAXRETRY } else {
120 attempt = attempt + 1
121 cf_puts(" retry " as *u8); cf_num(attempt); cf_puts(" status=" as *u8); cf_num(st[0]); cf_puts(" at=" as *u8); cf_num(from); cf_puts("\n" as *u8)
122 sys_sleep_ms(CWF_BACKOFF_MS * attempt)
123 }
124 }
125 }
126 if go == 1 {
127 if okchunk == 0 { failed = 1; go = 0 } else {
128 sys_write(fd, buf, nr)
129 from = from + nr
130 chunks = chunks + 1
131 if (chunks % CWF_PROGRESS) == 0 { cf_puts(" ... " as *u8); cf_num(from); cf_puts(" bytes\n" as *u8) }
132 if nr < CWF_CHUNK { go = 0 } // short chunk = EOF
133 }
134 }
135 }
136 sys_close(fd)
137 if failed == 1 {
138 cf_puts("FETCH FAILED at=" as *u8); cf_num(from); cf_puts(" (resume by re-running)\nverdict=FAIL\n" as *u8)
139 return 4
140 }
141 cf_puts("FETCH DONE bytes=" as *u8); cf_num(from); cf_puts("\nverdict=PASS\n" as *u8)
142 return 0
143 }
144 cf_puts("unknown verb\nverdict=FAIL\n" as *u8)
145 return 1
146}