Mercurial > hg > cc > cirrus_work
annotate bin/warc.py @ 46:44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
author | Henry S. Thompson <ht@inf.ed.ac.uk> |
---|---|
date | Wed, 05 Jul 2023 19:32:36 +0100 |
parents | 69be1131bcc5 |
children | d0d2fd9830d6 |
rev | line source |
---|---|
39 | 1 #!/usr/bin/env python3 |
46
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
2 '''Stream a warc format file, unzipping if necessary, invoking a |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
3 callback on each record. Callback can be limited by WARC-Type, record |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
4 part''' |
39 | 5 |
42
689a0e311cd2
make warc.py a library, separate out testing
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
39
diff
changeset
|
6 import sys,io |
689a0e311cd2
make warc.py a library, separate out testing
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
39
diff
changeset
|
7 from isal import igzip |
39 | 8 |
42
689a0e311cd2
make warc.py a library, separate out testing
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
39
diff
changeset
|
9 def warc(filename,callback,types=['response'],whole=False,parts=7,debug=False): |
39 | 10 types=[(t if isinstance(t,bytes) else bytes(t,'utf8')) for t in types] |
11 nb=0 | |
46
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
12 if filename.endswith(".gz"): |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
13 stream=igzip.IGzipFile(filename=filename) |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
14 else: |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
15 stream=open(filename,'rb',0) |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
16 bufsize=2*1024*1024 |
44d3a4f4ea51
support on-board unzipping, reduce buffer size to 2MB
Henry S. Thompson <ht@inf.ed.ac.uk>
parents:
43
diff
changeset
|
17 buf=bytearray(bufsize) |
39 | 18 l=b'\r\n' |
19 while not stream.closed: | |
20 bp=0 | |
21 while l==b'\r\n': | |
22 l=stream.readline() | |
23 nb+=(ln:=len(l)) | |
43 | 24 if ln==0: |
25 break | |
39 | 26 if l!=b'WARC/1.0\r\n': |
27 raise ValueError("Not a WARC file? At %s: %s[%s]"%(nb-len(l), | |
28 l.decode('latin-1'),len(l))) | |
29 wtype=None | |
30 length=None | |
31 state=1 | |
32 tr=None # Was this record truncated? | |
33 while l!=b'\r\n': | |
34 # WARC header | |
35 if parts & 1: | |
36 buf[bp:(bp:=bp+ln)]=l | |
37 l=stream.readline() | |
38 nb+=(ln:=len(l)) | |
39 if l.startswith(b"Content-Length: "): | |
40 length=wl=int(l[16:].rstrip()) | |
41 elif l.startswith(b"WARC-Truncated: "): | |
42 tr=l[16:].rstrip() | |
43 tr="EMPTY" if tr=="" else tr | |
44 elif l.startswith(b'WARC-Type: '): | |
45 wtype = l[11:-2] | |
46 start_2=bp | |
47 if (wtype in types): | |
48 if whole: | |
49 buf[bp:(bp:=bp+ln)]=l | |
50 elif (parts & 1): | |
51 callback(wtype,buf[:start_2],1) | |
52 if parts==1: | |
53 start_2=0 | |
54 else: | |
55 start_2=bp | |
56 else: | |
57 start_2=0 | |
58 bv=memoryview(buf)[start_2:start_2+length] | |
59 ii=0 | |
60 while True and not stream.closed: | |
61 if (i:=stream.readinto(bv))==0: | |
62 break | |
63 ii+=i | |
64 if ii>=length: | |
65 break | |
66 bv=memoryview(buf)[start_2+ii:start_2+length] | |
67 if ii!=length: | |
68 raise ValueError("Chunk read losing, from %s got %s expected %s"%(nb,ii,length)) | |
69 nb+=length | |
70 if wtype in types: | |
71 if whole: | |
72 callback(wtype,buf[0:start_2+length],7) | |
73 continue | |
74 # Only output parts (1 = WARC header, 2 = HTTP header, 4 = body) that are wanted | |
75 bl=None # for HTTP Content-Length for the length of the body? | |
76 L_start=start_2 | |
77 state=2 | |
78 bv=memoryview(buf)[start_2:start_2+length] | |
79 with io.BytesIO(bv) as rec_text: | |
80 for L in rec_text: | |
81 if state==2: | |
82 # HTTP header | |
83 wl -= len(L) | |
84 if not (L==b"" or L.startswith(b"\r")): | |
85 # Non-empty, it's (a continuation of) a header | |
86 if bl is None and L.startswith(b"Content-Length: "): | |
87 bl=int(L[16:].rstrip()) | |
88 else: | |
89 # Blank line, HTTP header is finished | |
90 if parts & 2: | |
91 callback(wtype,buf[start_2:start_2+L_start],2) | |
92 state=4 | |
93 # The above is just for sanity, because we do _not_ | |
94 # continue with the outer loop, | |
95 # since we can now block-output the entire rest of the | |
96 # input buffer. | |
97 if bl is not None: | |
98 if bl!=wl: | |
99 print("length mismatch: %s %s %s here: %s given: %s trunc: %s"%\ | |
100 (length,offset,filename,wl,bl,tr),file=sys.stderr) | |
101 # HTTP body | |
102 balance=start_2+rec_text.tell() | |
103 #print(balance,bl,wl,ll,ll-balance,file=sys.stderr) | |
104 # Output whatever is left | |
105 if parts & 4: | |
106 callback(wtype,buf[balance:balance+wl],4) | |
107 state=1 | |
108 | |
109 L_start=rec_text.tell() |