From 896cde206338f15dcdbd6364d71b46029495dbd9 Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 12 Oct 2026 16:07:52 +0100 Subject: [PATCH 1/9] fix: replace --fetch with --select and expect a warc in input Signed-off-by: Luca Foppiano --- README.md | 22 +- pyproject.toml | 1 - tests/conftest.py | 2 +- tests/test_fetch.py | 581 ---------------------------------------- tests/test_ia.py | 24 +- tests/test_select.py | 259 ++++++++++++++++++ uv.lock | 24 +- warc2zip.py | 622 +++++++++---------------------------------- 8 files changed, 405 insertions(+), 1130 deletions(-) delete mode 100644 tests/test_fetch.py create mode 100644 tests/test_select.py diff --git a/README.md b/README.md index 1ba00a9..69d6034 100644 --- a/README.md +++ b/README.md @@ -56,14 +56,12 @@ of those it tried. | Flag | Description | Default | |---------------------------|----------------------------------------------------------------------------------------|-----------------------------------------| | `input_file` | Path or URI to a `.warc.gz` file: local, `s3://`, `http(s)://` or `ia:///` (positional, required) | | -| `--output` | Path to the output zip file (the output `.warc.gz` with `--fetch`) | `{basename}_{hex}.zip` in the current directory, same hex and `_partial` rule as the root directory inside (`{basename}_{hex}.warc.gz` with `--fetch`) | +| `--output` | Path to the output zip file | `{basename}_{hex}.zip` in the current directory, same hex and `_partial` rule as the root directory inside | | `--dry-run` | Print summary without creating output. The scan always stops after at most 10 capture records, so it never streams the whole file; a lower `--limit` is respected | | | `--limit ` | Limit to N capture records, with their full set of associated request/metadata records | No limit, all records are processed | | `--format {flat,sidecar}` | Output format (see [Output Formats](#output-formats) below) | `flat` | | `--metadata-only` | Write every CSV, manifest and sidecar but no payload files | Off, payloads are written | -| `--fetch` | Treat `input_file` as a `manifest.csv` and download every row's byte range into one `.warc.gz` (see [Building and downloading a subset](#building-and-downloading-a-subset)). Not combinable with `--limit` or `--format` | Off | -| `--rate ` | `--fetch` only: requests per second per host, `0` for unlimited | [cdx_toolkit](https://github.com/commoncrawl/cdx_toolkit)'s per-host pacing for http(s), `2` for s3 and ia | -| `--retries ` | `--fetch` only: connection failures tolerated per request (http(s), throttling is retried without limit) or retries per request (s3, ia) | `100` / `8` | +| `--select ` | Convert only the captures listed in a `manifest.csv` from an earlier run on the same WARC (see [Building and downloading a subset](#building-and-downloading-a-subset)) | Off, every capture is converted | ### Small Examples @@ -390,9 +388,9 @@ Both `source_uri` and the `warc_record_*` fields are pseudo-headers — computed ## Building and downloading a subset -The point of `warc_filename` + `warc_record_offset` + `warc_record_length` is that `manifest.csv` alone is enough to fetch any capture again. Each record is its own gzip member, so those bytes are a standalone WARC — one HTTP range request per record, no reprocessing of the source file. +`warc_filename` + `warc_record_offset` + `warc_record_length` make every `manifest.csv` row enough to fetch that capture again on its own. Each record is its own gzip member, so those bytes are a standalone WARC — one HTTP range request per record, no reprocessing of the source file. -The intended workflow is: convert once with `--metadata-only` (a few tens of KB instead of gigabytes), filter the CSV however you like, then pull only the captures you kept. +The intended workflow is: convert once with `--metadata-only` (a few tens of KB instead of gigabytes), filter the CSV however you like, then convert the same WARC again with `--select` to get a zip with only the captures you kept. ### Get metadata only — no payloads ```bash @@ -419,18 +417,16 @@ or with [miller](https://miller.readthedocs.io/en/latest/): mlr --csv filter '$http_status_code == 200' manifest.csv > subset.csv ``` -### Fetch what survived +### Convert what you kept ```bash -warc2zip subset.csv --fetch --output subset.warc.gz +warc2zip 'https://data.commoncrawl.org/crawl-data/.../CC-MAIN-....warc.gz' --select subset.csv --output subset.zip ``` -Every row is fetched by byte range from its own `source_uri` (https, s3, ia or a local file). Nearby rows share one request, http(s) requests go through [cdx_toolkit](https://github.com/commoncrawl/cdx_toolkit)'s Common Crawl-aware retry and pacing (`--rate`, `--retries`), and each source's own `warcinfo` record leads the output. Each record is stamped with `WARC-Source-URI` and `WARC-Source-Range`, cdx_toolkit's convention, so a re-converted subset says on every row where the record sat in the original. `subset.warc.gz` goes straight back into `warc2zip` — or into any other WARC tool. +The zip has the usual layout with only the listed captures, each one complete: payload, response, request and metadata records, plus the WARC's `warcinfo`. The WARC is read in full again, as for `--metadata-only`. Rows are matched by `warc_record_offset` (checked against `warc_record_id` when the row has one), so a `manifest.csv` from a selected zip works with `--select` too. -Three caveats: +Select from the same WARC the manifest came from. It does not have to be at the same location: a local copy of an `s3://` file works, because rows are matched by file name. Rows that name another WARC are reported and skipped; run once per WARC for a subset that spans several. -- Fetch with `source_uri`, not `warc_filename`. For Common Crawl the latter is a bare basename like `CC-MAIN-20260618163205-20260618193205-00999.warc.gz`; the full path is `crawl-data/{crawl}/segments/{segment}/warc/{basename}`, and **the segment is not recorded anywhere in the WARC** — `warcinfo.csv` gives you the crawl (`_body.ispartof`) but you would need the crawl's `warc.paths.gz` to resolve the rest. -- Offsets address the file named by `source_uri`, nothing else. A fetched subset keeps the original's `warcinfo` record, so its `warc_filename` names the original while its offsets index the subset. -- Offsets stay valid under `--limit`: limiting only stops the read early, it never rewrites them. +Offsets stay valid under `--limit`: limiting only stops the read early, it never rewrites them. ## WARC examples for testing diff --git a/pyproject.toml b/pyproject.toml index fb87524..6a3dd42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,6 @@ version = "0.1.0" description = "Convert gzipped WARC files into zip-of-zips archives" requires-python = ">=3.10" dependencies = [ - "cdx_toolkit>=0.9.39", "fsspec[http]", "tqdm>=4.67.3", "warcio[s3]==1.8.1", diff --git a/tests/conftest.py b/tests/conftest.py index 94f5ff6..f49df3b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,4 @@ -"""Shared fixtures: the synthetic three-capture WARC used by the end-to-end and --fetch tests.""" +"""Shared fixtures: the synthetic three-capture WARC used by the end-to-end, --select and ia:// tests.""" import io from pathlib import Path diff --git a/tests/test_fetch.py b/tests/test_fetch.py deleted file mode 100644 index daf42ed..0000000 --- a/tests/test_fetch.py +++ /dev/null @@ -1,581 +0,0 @@ -"""--fetch: rebuilding a subset .warc.gz from manifest rows. - -Everything runs offline: the sources are local paths, which fsspec serves through the same -cat_file(start, end) range read that https:// and s3:// use. -""" - -import csv -import io -import shutil -import sys -import zipfile -from pathlib import Path - -import pytest -from conftest import CAPTURES -from fsspec.implementations.local import LocalFileSystem -from warcio.archiveiterator import ArchiveIterator -from warcio.statusandheaders import StatusAndHeaders -from warcio.warcwriter import WARCWriter - -import warc2zip -from warc2zip import ( - FETCH_MAX_BACKOFF, - FetchLengthMismatch, - FetchRow, - RateLimiter, - annotate_record_slice, - cli, - coalesce_ranges, - default_fetch_output_path, - fetch_main, - fetch_with_retry, - http_range, - leading_warcinfo, - main, - set_host_interval, - validate_record_slice, -) - - -def zip_csv(zip_path, suffix): - with zipfile.ZipFile(zip_path) as zf: - name = next(n for n in zf.namelist() if n.endswith(suffix)) - return list(csv.reader(io.StringIO(zf.read(name).decode("utf-8")))) - - -def manifest_rows(zip_path): - header, *rows = zip_csv(zip_path, "manifest.csv") - return [dict(zip(header, row)) for row in rows] - - -def warcinfo_range(zip_path): - values = {name: value for key, name, value in zip_csv(zip_path, "warcinfo.csv")[1:] if key == "warcinfo"} - return int(values["warc_record_offset"]), int(values["warc_record_length"]) - - -def request_range(zip_path, filename): - """(offset, length) of the request record joined to a payload file.""" - values = {name: value for key, name, value in zip_csv(zip_path, "request_warc_headers.csv")[1:] if key == filename} - return int(values["warc_record_offset"]), int(values["warc_record_length"]) - - -def slice_of(raw, row): - offset, length = int(row["warc_record_offset"]), int(row["warc_record_length"]) - return raw[offset : offset + length] - - -def fetch_row(row, **overrides): - values = { - "source_uri": row["source_uri"], - "offset": int(row["warc_record_offset"]), - "length": int(row["warc_record_length"]), - "record_id": row["warc_record_id"], - "target_uri": row["warc_target_uri"], - "line": 2, - } - values.update(overrides) - return FetchRow(**values) - - -def write_subset(path, rows, fieldnames=None): - with open(path, "w", newline="", encoding="utf-8") as fh: - writer = csv.DictWriter(fh, fieldnames=fieldnames or list(rows[0].keys()), quoting=csv.QUOTE_ALL) - writer.writeheader() - writer.writerows(rows) - return path - - -def record_types_and_ids(path): - with open(path, "rb") as fh: - return [(r.rec_type, r.rec_headers.get_header("WARC-Record-ID")) for r in ArchiveIterator(fh)] - - -def parsed_records(data): - """(rec_type, WARC headers, payload) per record; the payload is read before advancing.""" - return [ - (r.rec_type, dict(r.rec_headers.headers), r.content_stream().read()) - for r in ArchiveIterator(io.BytesIO(data)) - ] - - -def without_source_headers(headers): - return {k: v for k, v in headers.items() if not k.startswith("WARC-Source-")} - - -def raw_blocks(data): - """Each record's content block as written, HTTP headers and body, without warcio's HTTP parse.""" - return [r.content_stream().read() for r in ArchiveIterator(io.BytesIO(data), no_record_parse=True)] - - -def assert_is_stamped_copy(fetched, original, source_uri, row): - """A fetched record is the original plus the two provenance headers, same payload.""" - rec_type, headers, payload = fetched - o, n = int(row["warc_record_offset"]), int(row["warc_record_length"]) - assert rec_type == "response" - assert without_source_headers(headers) == original[1] - assert headers["WARC-Source-URI"] == source_uri - assert headers["WARC-Source-Range"] == f"bytes={o}-{o + n - 1}" - assert payload == original[2] - - -def assert_blocks_untouched(fetched_member, original_member): - """The stamped record's content block is the wire bytes: obs-folds, NULs and all.""" - assert raw_blocks(fetched_member) == raw_blocks(original_member) - - -def write_warc_without_warcinfo(path): - with open(path, "wb") as fh: - writer = WARCWriter(fh, gzip=True) - payload = b"bare" - writer.write_record( - writer.create_warc_record( - "https://bare.example.com/", - "response", - payload=io.BytesIO(payload), - length=len(payload), - http_headers=StatusAndHeaders("200 OK", [("Content-Type", "text/html")], protocol="HTTP/1.1"), - ) - ) - return path - - -@pytest.fixture -def metadata_zip(warc_path, tmp_path): - out = tmp_path / "meta.zip" - assert main(str(warc_path), str(out), metadata_only=True) == 0 - return out - - -# --- end to end --------------------------------------------------------------------------- - - -def test_fetch_rebuilds_the_subset(warc_path, metadata_zip, tmp_path): - """The source's warcinfo member verbatim, then each kept row's record stamped with its origin.""" - rows = manifest_rows(metadata_zip) - kept = [rows[0], rows[2]] - subset = write_subset(tmp_path / "subset.csv", kept) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out)) == 0 - - raw = warc_path.read_bytes() - offset, length = warcinfo_range(metadata_zip) - fetched = out.read_bytes() - assert fetched.startswith(raw[offset : offset + length]) - records = parsed_records(fetched) - assert [t for t, _, _ in records] == ["warcinfo", "response", "response"] - for record, row in zip(records[1:], kept): - assert_is_stamped_copy(record, parsed_records(slice_of(raw, row))[0], str(warc_path), row) - - -def test_fetched_subset_reconverts_and_still_names_the_original_warc(warc_path, metadata_zip, tmp_path): - """The README caveat: a derived WARC keeps the original's warcinfo, while its offsets index itself.""" - rows = manifest_rows(metadata_zip) - subset = write_subset(tmp_path / "subset.csv", [rows[1]]) - out = tmp_path / "subset.warc.gz" - assert fetch_main(str(subset), str(out)) == 0 - - check = tmp_path / "check.zip" - assert main(str(out), str(check)) == 0 - check_rows = manifest_rows(check) - - assert [row["warc_record_id"] for row in check_rows] == [rows[1]["warc_record_id"]] - assert check_rows[0]["warc_filename"] == "test.warc.gz" - assert check_rows[0]["source_uri"] == str(out) - record = next(iter(ArchiveIterator(io.BytesIO(slice_of(out.read_bytes(), check_rows[0]))))) - assert record.rec_headers.get_header("WARC-Record-ID") == rows[1]["warc_record_id"] - # ...and the CSV now says where the record sat in the original. - o, n = int(rows[1]["warc_record_offset"]), int(rows[1]["warc_record_length"]) - provenance = {name: value for _, name, value in zip_csv(check, "response_warc_headers.csv")[1:]} - assert provenance["warc_source_uri"] == str(warc_path) - assert provenance["warc_source_range"] == f"bytes={o}-{o + n - 1}" - - -def test_rows_are_grouped_by_source_and_ordered_by_offset(warc_path, metadata_zip, tmp_path): - """Two sources: each gets its own warcinfo, in first-appearance order, rows sorted within.""" - other = tmp_path / "other.warc.gz" - shutil.copy(warc_path, other) - rows = manifest_rows(metadata_zip) - from_other = [dict(row, source_uri=str(other)) for row in rows] - # CSV order deliberately scrambled: other first, then this file's rows descending. - subset = write_subset(tmp_path / "subset.csv", [from_other[1], rows[2], rows[0], from_other[0]]) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out)) == 0 - - warcinfo_id = record_types_and_ids(warc_path)[0][1] - assert record_types_and_ids(out) == [ - ("warcinfo", warcinfo_id), - ("response", rows[0]["warc_record_id"]), - ("response", rows[1]["warc_record_id"]), - ("warcinfo", warcinfo_id), - ("response", rows[0]["warc_record_id"]), - ("response", rows[2]["warc_record_id"]), - ] - - -def test_source_without_warcinfo_yields_responses_only(tmp_path, capsys): - bare = write_warc_without_warcinfo(tmp_path / "bare.warc.gz") - meta = tmp_path / "meta.zip" - assert main(str(bare), str(meta), metadata_only=True) == 0 - subset = write_subset(tmp_path / "subset.csv", manifest_rows(meta)) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out)) == 0 - - assert [t for t, _ in record_types_and_ids(out)] == ["response"] - assert "no leading warcinfo record" in capsys.readouterr().err - - -def test_blank_offset_row_is_skipped_with_a_warning(metadata_zip, tmp_path, capsys): - rows = manifest_rows(metadata_zip) - rows[1]["warc_record_offset"] = "" - subset = write_subset(tmp_path / "subset.csv", rows) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out)) == 1 - - err = capsys.readouterr().err - assert "line 3" in err and "1 row(s) could not be fetched" in err - assert [t for t, _ in record_types_and_ids(out)] == ["warcinfo", "response", "response"] - - -def test_duplicate_rows_are_fetched_once(metadata_zip, tmp_path, capsys): - rows = manifest_rows(metadata_zip) - subset = write_subset(tmp_path / "subset.csv", [rows[0], rows[0]]) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out)) == 0 - - assert [t for t, _ in record_types_and_ids(out)] == ["warcinfo", "response"] - assert "1 duplicate row(s) dropped" in capsys.readouterr().err - - -def test_a_server_that_ignores_range_is_caught_not_retried(metadata_zip, tmp_path, monkeypatch, capsys): - """fsspec never checks for a 206, so the length check is the only guard against a whole-file answer.""" - monkeypatch.setattr(LocalFileSystem, "cat_file", lambda self, path, start=None, end=None, **kw: Path(path).read_bytes()) - subset = write_subset(tmp_path / "subset.csv", manifest_rows(metadata_zip)[:1]) - out = tmp_path / "subset.warc.gz" - sleeps = [] - - assert fetch_main(str(subset), str(out), sleep=sleeps.append) == 1 - - assert sleeps == [] - assert "got" in capsys.readouterr().err - assert [t for t, _ in record_types_and_ids(out)] == ["warcinfo"] - - -def test_csv_without_manifest_columns_is_a_usage_error(tmp_path, capsys): - subset = write_subset(tmp_path / "subset.csv", [{"url": "x", "warc_record_offset": "0"}]) - with pytest.raises(SystemExit) as exc: - fetch_main(str(subset), str(tmp_path / "out.warc.gz")) - assert exc.value.code == 2 - assert "source_uri" in capsys.readouterr().err - - -# --- pure pieces -------------------------------------------------------------------------- - - -def row_at(offset, length): - return FetchRow("src", offset, length, "", "", 0) - - -def test_coalesce_ranges_merges_near_rows_and_splits_on_gap_or_span(): - a, b, c = row_at(0, 100), row_at(150, 100), row_at(10_000, 50) - assert coalesce_ranges([a, b, c], max_gap=100, max_span=10**6) == [(0, 250, [a, b]), (10_000, 10_050, [c])] - assert coalesce_ranges([a, b], max_gap=10, max_span=10**6) == [(0, 100, [a]), (150, 250, [b])] - assert coalesce_ranges([a, b], max_gap=100, max_span=200) == [(0, 100, [a]), (150, 250, [b])] - assert coalesce_ranges([a]) == [(0, 100, [a])] - assert coalesce_ranges([]) == [] - - -class FakeHTTPError(Exception): - """Shaped like aiohttp.ClientResponseError: a .status and .headers.""" - - def __init__(self, status, headers=None): - super().__init__(f"HTTP {status}") - self.status = status - self.headers = headers or {} - - -def failing(exc): - def fetch(): - raise exc - - return fetch - - -def test_fetch_with_retry_honours_retry_after_then_succeeds(capsys): - outcomes = [FakeHTTPError(503, {"Retry-After": "3"}), FakeHTTPError(429, {"Retry-After": "3"}), b"ok"] - - def fetch(): - outcome = outcomes.pop(0) - if isinstance(outcome, Exception): - raise outcome - return outcome - - sleeps = [] - assert fetch_with_retry(fetch, retries=3, label="x", sleep=sleeps.append) == b"ok" - assert sleeps == [3.0, 3.0] - assert capsys.readouterr().err.count("retrying in 3.0 s") == 2 - - -def test_fetch_with_retry_backs_off_exponentially_then_gives_up(): - sleeps = [] - with pytest.raises(FakeHTTPError): - fetch_with_retry(failing(FakeHTTPError(503)), retries=3, sleep=sleeps.append, rng=lambda: 0.5) - assert sleeps == [2.0, 4.0, 8.0] # 2**attempt * (0.5 + rng) - - -def test_fetch_with_retry_caps_the_backoff(): - sleeps = [] - with pytest.raises(FakeHTTPError): - fetch_with_retry(failing(FakeHTTPError(503)), retries=8, sleep=sleeps.append, rng=lambda: 0.5) - assert max(sleeps) == FETCH_MAX_BACKOFF - - -@pytest.mark.parametrize( - "exc", - [FileNotFoundError("404"), PermissionError("403"), FakeHTTPError(416), FakeHTTPError(400), FetchLengthMismatch("short")], -) -def test_deterministic_errors_are_not_retried(exc): - calls = [] - - def fetch(): - calls.append(1) - raise exc - - with pytest.raises(type(exc)): - fetch_with_retry(fetch, retries=5, sleep=lambda s: pytest.fail("slept")) - assert len(calls) == 1 - - -@pytest.mark.parametrize("exc", [ConnectionResetError("reset"), TimeoutError("timeout"), OSError("eio")]) -def test_connection_errors_are_retried(exc): - sleeps = [] - with pytest.raises(type(exc)): - fetch_with_retry(failing(exc), retries=1, sleep=sleeps.append) - assert len(sleeps) == 1 - - -def test_rate_limiter_spaces_requests_per_key(): - clock = [100.0] - sleeps = [] - - def sleep(seconds): - sleeps.append(seconds) - clock[0] += seconds - - limiter = RateLimiter(2.0, clock=lambda: clock[0], sleep=sleep) - limiter.wait("a") - limiter.wait("b") - assert sleeps == [] - limiter.wait("a") - assert sleeps == [0.5] - clock[0] += 10 - limiter.wait("a") - assert sleeps == [0.5] - - -def test_rate_limiter_zero_disables(): - limiter = RateLimiter(0, clock=lambda: 0.0, sleep=lambda s: pytest.fail("slept")) - limiter.wait("a") - limiter.wait("a") - - -def test_leading_warcinfo_cuts_the_first_member_only_when_it_is_complete(warc_path, metadata_zip, tmp_path): - raw = warc_path.read_bytes() - offset, length = warcinfo_range(metadata_zip) - assert offset == 0 - assert leading_warcinfo(raw[: 64 * 1024]) == raw[:length] - assert leading_warcinfo(raw[: length + 10]) == raw[:length] # probe ends inside the second member - assert leading_warcinfo(raw[: length - 5]) is None # probe ends inside the warcinfo itself - assert leading_warcinfo(b"not a warc") is None - assert leading_warcinfo(b"") is None - bare = write_warc_without_warcinfo(tmp_path / "bare.warc.gz") - assert leading_warcinfo(bare.read_bytes()) is None - - -def test_validate_record_slice_rejects_impostors(warc_path, metadata_zip): - rows = manifest_rows(metadata_zip) - raw = warc_path.read_bytes() - row = fetch_row(rows[0]) - good = slice_of(raw, rows[0]) - - assert validate_record_slice(good, row) is None - assert validate_record_slice(b"not a warc at all", row) # warcio sniffs this as an ARC response - assert "WARC-Record-ID" in validate_record_slice(slice_of(raw, rows[1]), row) - assert validate_record_slice(good + b"trailing garbage bytes", row) - assert validate_record_slice(good[:-1], row) - offset, length = request_range(metadata_zip, rows[0]["filename"]) - assert "request" in validate_record_slice(raw[offset : offset + length], row) - - # ARC rows carry no record id, so the target URI is what identifies the record. - assert validate_record_slice(good, fetch_row(rows[0], record_id="")) is None - assert "WARC-Target-URI" in validate_record_slice(good, fetch_row(rows[0], record_id="", target_uri="https://x/")) - - -def test_default_fetch_output_path(): - assert default_fetch_output_path("subset.csv", run_id="abcd") == Path("subset_abcd.warc.gz") - assert default_fetch_output_path("/tmp/dir/Manifest.CSV", run_id="abcd") == Path("Manifest_abcd.warc.gz") - assert default_fetch_output_path("-", run_id="abcd") == Path("stdin_abcd.warc.gz") - assert default_fetch_output_path("subset.csv").parent == Path(".") - - -# --- cli ---------------------------------------------------------------------------------- - - -@pytest.mark.parametrize( - "argv", - [ - ["subset.csv", "--fetch", "--limit", "1"], - ["subset.csv", "--fetch", "--format", "sidecar"], - ["x.warc.gz", "--rate", "1"], - ["x.warc.gz", "--retries", "2"], - ["subset.csv", "--fetch", "--dry-run"], - ], -) -def test_cli_refuses_flags_that_would_otherwise_be_ignored(argv, monkeypatch): - monkeypatch.setattr(sys, "argv", ["warc2zip", *argv]) - with pytest.raises(SystemExit) as exc: - cli() - assert exc.value.code == 2 - - -def test_cli_fetch_and_default_format_still_flat(warc_path, metadata_zip, tmp_path, monkeypatch): - subset = write_subset(tmp_path / "subset.csv", manifest_rows(metadata_zip)[:1]) - out = tmp_path / "subset.warc.gz" - monkeypatch.setattr(sys, "argv", ["warc2zip", str(subset), "--fetch", "--output", str(out), "--rate", "0"]) - assert cli() == 0 - assert [t for t, _ in record_types_and_ids(out)] == ["warcinfo", "response"] - - check = tmp_path / "check.zip" - monkeypatch.setattr(sys, "argv", ["warc2zip", str(out), "--output", str(check)]) - assert cli() == 0 - with zipfile.ZipFile(check) as zf: - names = {n.rsplit("/", 1)[-1] for n in zf.namelist()} - assert "manifest.csv" in names and not any(".request." in n for n in names) # flat, not sidecar - assert len(CAPTURES) == 3 # the fixture shape the row indexes above rely on - - -# --- http(s) transport through cdx_toolkit ------------------------------------------------ - - -class FakeResponse: - def __init__(self, content=b"", status_code=200, headers=None): - self.content = content - self.status_code = status_code - self.headers = headers or {} - - -def fake_myrequests_get(warc_bytes, log, redirect_from=None, ignore_range=False): - """A `myrequests_get` stand-in serving Range requests for `warc_bytes`, optionally after a 302.""" - - def get(url, params=None, headers=None, **kwargs): - log.append((url, headers["Range"], kwargs)) - if url == redirect_from: - return FakeResponse(b"moved", 302, {"Location": "/cdn/x.warc.gz"}) - if ignore_range: - return FakeResponse(warc_bytes, 200) - start, end = (int(v) for v in headers["Range"][len("bytes=") :].split("-")) - return FakeResponse(warc_bytes[start : end + 1], 206) - - return get - - -def https_rows(rows, uri): - return [dict(row, source_uri=uri) for row in rows] - - -def test_https_sources_go_through_cdx_toolkit(warc_path, metadata_zip, tmp_path, monkeypatch): - log = [] - monkeypatch.setattr(warc2zip, "myrequests_get", fake_myrequests_get(warc_path.read_bytes(), log)) - uri = "https://data.example.org/crawl-data/x.warc.gz" - rows = manifest_rows(metadata_zip) - subset = write_subset(tmp_path / "subset.csv", https_rows([rows[0], rows[2]], uri)) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out), retries=3) == 0 - - raw = warc_path.read_bytes() - offset, length = warcinfo_range(metadata_zip) - fetched = out.read_bytes() - assert fetched.startswith(raw[offset : offset + length]) - records = parsed_records(fetched) - assert [t for t, _, _ in records] == ["warcinfo", "response", "response"] - for record, row in zip(records[1:], [rows[0], rows[2]]): - assert_is_stamped_copy(record, parsed_records(slice_of(raw, row))[0], uri, row) - # One probe plus one coalesced span, both Range requests to cdx_toolkit's transport. - assert [(u, r) for u, r, _ in log] == [ - (uri, "bytes=0-65535"), - (uri, f"bytes={rows[0]['warc_record_offset']}-{int(rows[2]['warc_record_offset']) + int(rows[2]['warc_record_length']) - 1}"), - ] - assert all(kw == {"raise_error_after_n_errors": 3} for _, _, kw in log) - - -def test_http_range_follows_redirects_by_hand(warc_path): - """myrequests_get passes allow_redirects=False, so a 302 would otherwise come back as the slice.""" - log = [] - raw = warc_path.read_bytes() - get = fake_myrequests_get(raw, log, redirect_from="https://hub.example/resolve/x.warc.gz?download=true") - - data = http_range("https://hub.example/resolve/x.warc.gz?download=true", 10, 20, get=get) - - assert data == raw[10:20] - assert [u for u, _, _ in log] == ["https://hub.example/resolve/x.warc.gz?download=true", "https://hub.example/cdn/x.warc.gz"] - assert {r for _, r, _ in log} == {"bytes=10-19"} - - -def test_http_range_gives_up_on_a_redirect_loop(warc_path): - log = [] - get = fake_myrequests_get(warc_path.read_bytes(), log, redirect_from="https://hub.example/cdn/x.warc.gz") - with pytest.raises(RuntimeError, match="redirects"): - http_range("https://hub.example/cdn/x.warc.gz", 0, 10, redirects=2, get=get) - assert len(log) == 3 - - -def test_https_server_ignoring_range_is_caught(warc_path, metadata_zip, tmp_path, monkeypatch, capsys): - log = [] - monkeypatch.setattr(warc2zip, "myrequests_get", fake_myrequests_get(warc_path.read_bytes(), log, ignore_range=True)) - rows = https_rows(manifest_rows(metadata_zip)[:1], "https://data.example.org/x.warc.gz") - subset = write_subset(tmp_path / "subset.csv", rows) - out = tmp_path / "subset.warc.gz" - - assert fetch_main(str(subset), str(out)) == 1 - - assert "got" in capsys.readouterr().err - assert [t for t, _ in record_types_and_ids(out)] == ["warcinfo"] # the probe tolerates a long answer - - -def test_set_host_interval_maps_rate_onto_cdx_toolkit(monkeypatch): - from cdx_toolkit.myrequests import retry_info - - host = "data.example.org" - monkeypatch.delitem(retry_info, host, raising=False) - set_host_interval(host, None) - assert host not in retry_info # None leaves cdx_toolkit's own defaults alone - set_host_interval(host, 4) - assert retry_info[host]["minimum_interval"] == 0.25 - set_host_interval(host, 0) - assert retry_info[host]["minimum_interval"] == 0.0 - assert retry_info["data.commoncrawl.org"]["minimum_interval"] == 0.55 - - -def test_stamping_keeps_every_original_header_and_the_payload(warc_path, metadata_zip): - rows = manifest_rows(metadata_zip) - raw = warc_path.read_bytes() - row = rows[1] - original = parsed_records(slice_of(raw, row))[0] - - stamped = annotate_record_slice(slice_of(raw, row), str(warc_path), int(row["warc_record_offset"]), int(row["warc_record_length"])) - - records = parsed_records(stamped) - assert len(records) == 1 - assert_is_stamped_copy(records[0], original, str(warc_path), row) - assert_blocks_untouched(stamped, slice_of(raw, row)) - # The fixture's obs-folded header is the case a parsed re-serialisation would unfold. - folded = annotate_record_slice(slice_of(raw, rows[2]), str(warc_path), 0, 1) - assert b"X-Fold: a\r\n b" in raw_blocks(folded)[0] - assert records[0][1]["WARC-Target-URI"] == CAPTURES[1][0] - assert records[0][2] == CAPTURES[1][2] diff --git a/tests/test_ia.py b/tests/test_ia.py index f9cbc22..4c941eb 100644 --- a/tests/test_ia.py +++ b/tests/test_ia.py @@ -1,5 +1,5 @@ """ia://item/file inputs: the Internet Archive filesystem, its credentials, and the two paths -through it (a conversion with main(), a re-fetch with fetch_main()). +through it (a full conversion, and a --select conversion of the same URI). Everything runs offline against the `ia_server` fixture, which reproduces archive.org's shape: a /download/ URL that 302s to a data node on another origin, serving byte ranges, optionally only @@ -8,14 +8,13 @@ import pytest from conftest import CAPTURES -from test_fetch import assert_is_stamped_copy, manifest_rows, parsed_records, slice_of, warcinfo_range, write_subset +from test_select import manifest_rows, without_filename, write_subset import warc2zip from warc2zip import ( InternetArchiveFileSystem, cli, default_output_path, - fetch_main, format_ia_permission_error, ia_config_path, in_domain, @@ -170,9 +169,9 @@ def test_authorization_is_scoped_to_archive_org(): # --- end to end against the stand-in server -------------------------------------------------- -def test_ia_uri_converts_and_refetches(ia_server, warc_path, tmp_path): +def test_ia_uri_converts_and_selects(ia_server, warc_path, tmp_path): """main() streams the file by ranges through the redirect; the manifest carries the ia:// - URI verbatim, so --fetch can go back to it.""" + URI verbatim, and --select against the same URI keeps just the listed captures.""" uri = ia_server.add("testitem", warc_path) out = tmp_path / "out.zip" assert main(uri, str(out)) == 0 @@ -184,17 +183,10 @@ def test_ia_uri_converts_and_refetches(ia_server, warc_path, tmp_path): assert item_reads, "data-node reads should be range requests" assert all(hit.path.startswith(("/download/", "/items/")) for hit in ia_server.hits) - subset = write_subset(tmp_path / "subset.csv", [rows[0], rows[2]]) - fetched_path = tmp_path / "subset.warc.gz" - assert fetch_main(str(subset), str(fetched_path)) == 0 - raw = warc_path.read_bytes() - offset, length = warcinfo_range(out) - fetched = fetched_path.read_bytes() - assert fetched.startswith(raw[offset : offset + length]) - records = parsed_records(fetched) - assert [t for t, _, _ in records] == ["warcinfo", "response", "response"] - for record, row in zip(records[1:], [rows[0], rows[2]]): - assert_is_stamped_copy(record, parsed_records(slice_of(raw, row))[0], uri, row) + kept = [rows[0], rows[2]] + selected = tmp_path / "selected.zip" + assert main(uri, str(selected), select=str(write_subset(tmp_path / "subset.csv", kept))) == 0 + assert [without_filename(r) for r in manifest_rows(selected)] == [without_filename(r) for r in kept] def test_authorization_survives_the_cross_origin_redirect(ia_server, warc_path, tmp_path, monkeypatch): diff --git a/tests/test_select.py b/tests/test_select.py new file mode 100644 index 0000000..a7363d5 --- /dev/null +++ b/tests/test_select.py @@ -0,0 +1,259 @@ +"""--select: a zip with only the captures a (filtered) manifest.csv names, each one complete.""" + +import csv +import io +import shutil +import sys +import zipfile + +import pytest +from conftest import CAPTURES +from test_arc import arc_path # noqa: F401 - fixture + +from warc2zip import SelectRow, cli, main, row_names_input + + +def zip_csv(zip_path, suffix): + with zipfile.ZipFile(zip_path) as zf: + name = next(n for n in zf.namelist() if n.endswith("/" + suffix)) + return list(csv.reader(io.StringIO(zf.read(name).decode("utf-8")))) + + +def manifest_rows(zip_path): + header, *rows = zip_csv(zip_path, "manifest.csv") + return [dict(zip(header, row)) for row in rows] + + +def without_filename(row): + return {k: v for k, v in row.items() if k != "filename"} + + +def keyed_by_record_id(zip_path, name): + """A header CSV's (header_name, header_value) rows per capture, keyed by the capture's + WARC-Record-ID rather than its filename, which a selected zip numbers afresh.""" + ids = {r["filename"]: r["warc_record_id"] for r in manifest_rows(zip_path)} + rows = {} + for filename, header_name, header_value in zip_csv(zip_path, name)[1:]: + rows.setdefault(ids[filename], []).append((header_name, header_value)) + return rows + + +def write_subset(path, rows, fieldnames=None): + with open(path, "w", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=fieldnames or list(rows[0].keys()), quoting=csv.QUOTE_ALL) + writer.writeheader() + writer.writerows(rows) + return path + + +def select(warc, rows, tmp_path, name="selected.zip", **kwargs): + out = tmp_path / name + status = main(str(warc), str(out), select=str(write_subset(tmp_path / f"{name}.csv", rows)), **kwargs) + return status, out + + +@pytest.fixture +def full_zip(warc_path, tmp_path): + out = tmp_path / "full.zip" + assert main(str(warc_path), str(out)) == 0 + return out + + +# --- end to end --------------------------------------------------------------------------- + + +def test_only_the_selected_captures_each_one_complete(warc_path, full_zip, tmp_path, capsys): + """The workflow: --metadata-only, edit the CSV, select from the same WARC. Every CSV holds + exactly the selected captures' rows from a full run, request and metadata included.""" + rows = manifest_rows(full_zip) + kept = [rows[0], rows[2]] + + status, out = select(warc_path, kept, tmp_path) + + assert status == 0 + assert [without_filename(r) for r in manifest_rows(out)] == [without_filename(r) for r in kept] + for name in ("response_warc_headers.csv", "response_http_headers.csv", "request_warc_headers.csv", + "request_http_headers.csv", "metadata.csv"): + full = keyed_by_record_id(full_zip, name) + assert keyed_by_record_id(out, name) == {r["warc_record_id"]: full[r["warc_record_id"]] for r in kept}, name + assert zip_csv(out, "warcinfo.csv") == zip_csv(full_zip, "warcinfo.csv") + err = capsys.readouterr().err + assert "no response record" not in err # the dropped captures' records are not orphans + + +def test_payloads_are_the_selected_ones(warc_path, full_zip, tmp_path): + rows = manifest_rows(full_zip) + _, out = select(warc_path, [rows[1]], tmp_path) + with zipfile.ZipFile(full_zip) as full, zipfile.ZipFile(out) as selected: + full_payload = full.read(next(n for n in full.namelist() if n.endswith("/" + rows[1]["filename"]))) + payloads = [n for n in selected.namelist() if n.rsplit("/", 1)[-1][:1].isdigit()] + assert [selected.read(n) for n in payloads] == [full_payload] + + +def test_the_selected_manifest_selects_again(warc_path, full_zip, tmp_path): + _, first = select(warc_path, [manifest_rows(full_zip)[1]], tmp_path, "first.zip") + status, again = select(warc_path, manifest_rows(first), tmp_path, "again.zip") + assert status == 0 + assert manifest_rows(again) == manifest_rows(first) + + +def test_select_with_sidecar_format_and_metadata_only(warc_path, full_zip, tmp_path): + status, out = select(warc_path, manifest_rows(full_zip)[:1], tmp_path, output_format="sidecar", metadata_only=True) + assert status == 0 + with zipfile.ZipFile(out) as zf: + basenames = [n.rsplit("/", 1)[-1] for n in zf.namelist()] + suffixes = {b.split(".", 1)[-1] for b in basenames if b[:1].isdigit()} + assert {"html.request.warc", "html.metadata.warc-fields", "html.response.http"} <= suffixes + assert "html" not in suffixes # metadata-only: no payload + + +def test_limit_counts_selected_captures(warc_path, full_zip, tmp_path): + rows = manifest_rows(full_zip) + status, out = select(warc_path, [rows[1], rows[2]], tmp_path, limit=1) + assert status == 0 + assert [r["warc_record_id"] for r in manifest_rows(out)] == [rows[1]["warc_record_id"]] + + +def test_arc_rows_are_selected_by_offset(arc_path, tmp_path): # noqa: F811 + """ARC rows carry no record id (warcio mints a new one every run), so the offset is the key.""" + full = tmp_path / "full.zip" + assert main(str(arc_path), str(full)) == 0 + rows = manifest_rows(full) + assert rows[1]["warc_record_id"] == "" + + status, out = select(arc_path, [rows[1]], tmp_path) + + assert status == 0 + assert [without_filename(r) for r in manifest_rows(out)] == [without_filename(rows[1])] + + +def test_same_warc_under_another_path_is_selected(warc_path, full_zip, tmp_path): + """Case 2: the manifest came from one spelling of the file (s3://...), the input is another (a local copy).""" + moved = tmp_path / "elsewhere" / warc_path.name + moved.parent.mkdir() + shutil.copy(warc_path, moved) + rows = [dict(r, source_uri=f"s3://bucket/crawl/{warc_path.name}") for r in manifest_rows(full_zip)] + + status, out = select(moved, rows, tmp_path) + + assert status == 0 + assert len(manifest_rows(out)) == len(rows) + + +def test_rows_of_another_warc_are_noted_not_selected(warc_path, full_zip, tmp_path, capsys): + """Case 3: a subset mixing WARCs is a mistake to report; its foreign offsets are never looked up.""" + rows = manifest_rows(full_zip) + foreign = [dict(r, source_uri="s3://bucket/other.warc.gz", warc_filename="other.warc.gz") for r in rows[1:]] + + status, out = select(warc_path, [rows[0], *foreign], tmp_path) + + assert status == 0 + assert [r["warc_record_id"] for r in manifest_rows(out)] == [rows[0]["warc_record_id"]] + assert "2 row(s) name another WARC and were not selected (other.warc.gz: 2)" in capsys.readouterr().err + + +def test_a_subset_of_only_another_warc_selects_nothing_and_fails(warc_path, full_zip, tmp_path, capsys): + rows = [dict(r, source_uri="other.warc.gz", warc_filename="other.warc.gz") for r in manifest_rows(full_zip)] + status, _ = select(warc_path, rows, tmp_path) + assert status == 1 + assert "selected nothing" in capsys.readouterr().err + + +def test_an_offset_that_is_no_capture_is_reported(warc_path, full_zip, tmp_path, capsys): + rows = manifest_rows(full_zip) + status, out = select(warc_path, [rows[0], dict(rows[1], warc_record_offset="7")], tmp_path) + assert status == 1 + assert len(manifest_rows(out)) == 1 + assert "1 row(s) of" in capsys.readouterr().err + + +def test_a_record_id_that_disagrees_with_the_offset_is_not_selected(warc_path, full_zip, tmp_path, capsys): + """Same name, different file: the offset lands on a capture, but not the one the row describes.""" + rows = manifest_rows(full_zip) + status, out = select(warc_path, [dict(rows[0], warc_record_id="")], tmp_path) + assert status == 1 + assert manifest_rows(out) == [] + assert "not " in capsys.readouterr().err + + +def test_blank_offset_row_is_skipped_with_a_warning(warc_path, full_zip, tmp_path, capsys): + rows = manifest_rows(full_zip) + rows[1]["warc_record_offset"] = "" + status, out = select(warc_path, rows, tmp_path) + assert status == 1 + err = capsys.readouterr().err + assert "line 3" in err and "1 row(s) could not be selected" in err + assert len(manifest_rows(out)) == 2 + + +def test_duplicate_rows_select_once(warc_path, full_zip, tmp_path): + rows = manifest_rows(full_zip) + status, out = select(warc_path, [rows[0], rows[0]], tmp_path) + assert status == 0 + assert len(manifest_rows(out)) == 1 + + +def test_a_csv_cut_down_to_the_offset_column_still_selects(warc_path, full_zip, tmp_path): + rows = manifest_rows(full_zip) + status, out = select(warc_path, [{"warc_record_offset": rows[2]["warc_record_offset"]}], tmp_path) + assert status == 0 + assert [r["warc_record_id"] for r in manifest_rows(out)] == [rows[2]["warc_record_id"]] + + +def test_csv_without_offset_column_is_a_usage_error(warc_path, tmp_path, capsys): + subset = write_subset(tmp_path / "s.csv", [{"url": "x"}]) + with pytest.raises(SystemExit) as exc: + main(str(warc_path), str(tmp_path / "out.zip"), select=str(subset)) + assert exc.value.code == 2 + assert "warc_record_offset" in capsys.readouterr().err + + +# --- row_names_input ---------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "source_uri, warc_filename, input_file, input_warc_filename, expected", + [ + # Case 1: the manifest came from this very file. + ("/data/test.warc.gz", "test.warc.gz", "/data/test.warc.gz", "test.warc.gz", True), + # Case 2: same file, another spelling. + ("s3://commoncrawl/crawl-data/x/test.warc.gz", "test.warc.gz", "./test.warc.gz", "test.warc.gz", True), + ("https://data.commoncrawl.org/x/test.warc.gz", "test.warc.gz", "/tmp/test.warc.gz", "", True), + # A renamed copy: only its warcinfo still says what it is. + ("s3://bucket/test.warc.gz", "test.warc.gz", "/tmp/renamed.warc.gz", "test.warc.gz", True), + # Case 3: another WARC. + ("s3://bucket/other.warc.gz", "other.warc.gz", "/data/test.warc.gz", "test.warc.gz", False), + # A CSV cut down to the offset column names no file: nothing to compare. + ("", "", "/data/test.warc.gz", "test.warc.gz", True), + ], +) +def test_row_names_input(source_uri, warc_filename, input_file, input_warc_filename, expected): + row = SelectRow(0, "", source_uri, warc_filename, 2) + assert row_names_input(row, input_file, input_warc_filename) is expected + + +# --- cli ---------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "argv", + [ + ["x.warc.gz", "--select", "s.csv", "--dry-run"], + ["-", "--select", "-"], + ["s.csv", "--fetch"], + ["x.warc.gz", "--rate", "1"], + ], +) +def test_cli_refuses_flags_that_would_otherwise_be_ignored(argv, monkeypatch): + monkeypatch.setattr(sys, "argv", ["warc2zip", *argv]) + with pytest.raises(SystemExit) as exc: + cli() + assert exc.value.code == 2 + + +def test_cli_select(warc_path, full_zip, tmp_path, monkeypatch): + subset = write_subset(tmp_path / "s.csv", manifest_rows(full_zip)[:1]) + out = tmp_path / "selected.zip" + monkeypatch.setattr(sys, "argv", ["warc2zip", str(warc_path), "--select", str(subset), "--output", str(out)]) + assert cli() == 0 + assert [r["warc_target_uri"] for r in manifest_rows(out)] == [CAPTURES[0][0]] diff --git a/uv.lock b/uv.lock index f5cdd63..b7d5027 100644 --- a/uv.lock +++ b/uv.lock @@ -208,19 +208,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fb/51/08f32aea872253173f513ba68122f4300966290677c8e59887b4ffd5d957/botocore-1.42.70-py3-none-any.whl", hash = "sha256:54ed9d25f05f810efd22b0dfda0bb9178df3ad8952b2e4359e05156c9321bd3c", size = 14671393, upload-time = "2026-03-17T19:43:06.777Z" }, ] -[[package]] -name = "cdx-toolkit" -version = "0.9.39" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "requests" }, - { name = "warcio" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/c2/92/ab18e28629867f3af01f64974ef75075dd5775ad9cbd1dca42faba1504af/cdx_toolkit-0.9.39.tar.gz", hash = "sha256:7fd85cd5b00ec3ac57bfa21e7f70ba633e1df59760eb54dae3a85e63596af4bc", size = 92815, upload-time = "2026-06-01T11:49:02.752Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/93/22/5cda1ba55a88188b5b19ae60961244759b8b0cc5fe97fb29516afef32adf/cdx_toolkit-0.9.39-py3-none-any.whl", hash = "sha256:577ae9a9578f33ce4bae6c887074f0f70331bf4d6a66a26d1e20836b3004e431", size = 32685, upload-time = "2026-06-01T11:49:01.507Z" }, -] - [[package]] name = "certifi" version = "2026.2.25" @@ -486,6 +473,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d5/1f/5f4a3cd9e4440e9d9bc78ad0a91a1c8d46b4d429d5239ebe6793c9fe5c41/fsspec-2026.3.0-py3-none-any.whl", hash = "sha256:d2ceafaad1b3457968ed14efa28798162f1638dbb5d2a6868a2db002a5ee39a4", size = 202595, upload-time = "2026-03-27T19:11:13.595Z" }, ] +[package.optional-dependencies] +http = [ + { name = "aiohttp" }, +] + [[package]] name = "idna" version = "3.11" @@ -974,8 +966,7 @@ name = "warc2zip" version = "0.1.0" source = { editable = "." } dependencies = [ - { name = "cdx-toolkit" }, - { name = "fsspec" }, + { name = "fsspec", extra = ["http"] }, { name = "tqdm" }, { name = "warcio", extra = ["s3"] }, ] @@ -988,8 +979,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "cdx-toolkit", specifier = ">=0.9.39" }, - { name = "fsspec" }, + { name = "fsspec", extras = ["http"] }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" }, { name = "ruff", marker = "extra == 'dev'" }, { name = "tqdm", specifier = ">=4.67.3" }, diff --git a/warc2zip.py b/warc2zip.py index bcbecc1..0b452e0 100644 --- a/warc2zip.py +++ b/warc2zip.py @@ -2,7 +2,6 @@ import asyncio import configparser import csv -import functools import io import json import mimetypes @@ -14,17 +13,14 @@ import sys import time import zipfile -import zlib from collections import Counter from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from urllib.parse import urljoin, urlparse +from urllib.parse import urlparse import aiohttp from aiohttp import hdrs -from cdx_toolkit.myrequests import get_retries, myrequests_get, retry_info -from fsspec.core import url_to_fs from fsspec.implementations.http import HTTPFileSystem from fsspec.registry import register_implementation from fsspec.utils import stringify_path @@ -32,13 +28,12 @@ from warcio.archiveiterator import ArchiveIterator from warcio.recordloader import ARC2WARCHeadersParser from warcio.utils import fsspec_open -from warcio.warcwriter import WARCWriter # --- Internet Archive input: ia:/// ------------------------------------------ # # Read over https://archive.org/download/, not over IA's S3-like API (s3.us.archive.org): that # endpoint ignores Range headers and answers GET with a 307 to a plain-http data node, so neither -# fsspec's block reads nor --fetch could use it (a 100-byte read pulled the whole 1.6 GB file). The +# fsspec's block reads nor --select could use it (a 100-byte read pulled the whole 1.6 GB file). The # download URL 302s to a data node that honours Range; it is also what the `internetarchive` # package itself downloads from. So ia:// is HTTPFileSystem plus a path mapping and credentials. # @@ -304,8 +299,7 @@ def format_ia_permission_error(input_file, credentials=None): EXTRACTED_RECORD_TYPES = ("warcinfo", "response", "revisit", "request", "metadata") -# Transient-failure policy for reading a remote input. Names match the --fetch helpers so the -# two definitions merge into one. +# Transient-failure policy for reading a remote input (see CountingStream). FETCH_DEFAULT_RETRIES = 8 # retries after the first attempt FETCH_MAX_BACKOFF = 60.0 # seconds; the cap on one exponential-backoff wait FETCH_RETRYABLE_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504}) @@ -1077,7 +1071,8 @@ def write_sidecar_files(zip_file, root_dir, group): zip_file.writestr(f"{base}.metadata.warc-fields", "\n\n".join(body_parts)) -def main(input_file, output_path=None, dry_run=False, limit=None, output_format="flat", metadata_only=False): +def main(input_file, output_path=None, dry_run=False, limit=None, output_format="flat", metadata_only=False, + select=None): file_size = get_file_size(input_file) partial = limit is not None @@ -1150,6 +1145,14 @@ def main(input_file, output_path=None, dry_run=False, limit=None, output_format= if output_path is None: output_path = str(default_output_path(input_file, partial, run_id)) + # --select (`select` is a manifest.csv path): the WARC is still read end to end, and a capture + # whose response/revisit offset is not in the CSV is dropped whole — its request and metadata + # records too, wherever they sit relative to it. `wanted` is resolved at the first capture, + # once the warcinfo record has said what the input calls itself. + select_rows, select_skipped = read_select_rows(select) if select else ([], 0) + wanted = None # {offset: SelectRow} still to be seen + dropped_ids = set() # record ids of dropped captures, and of the requests they name + # response -> Record id <-> metadata - with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as outer_zip: # Pass 1: Read WARC, write payloads immediately, buffer only headers @@ -1194,7 +1197,28 @@ def main(input_file, output_path=None, dry_run=False, limit=None, output_format= root_dir = build_root_dir_name(fallback_crawl_name, partial, run_id) if rec_type in ("response", "revisit"): + payload = record.content_stream().read() record_id = record.rec_headers.get_header("WARC-Record-ID") + if select: + if wanted is None: + wanted = selected_offsets(select_rows, input_file, warc_filename) + row = wanted.pop(record_iter.get_record_offset(), None) + if row and row.record_id and row.record_id != record_id: + print(f"warning: {select} line {row.line}: the record at offset {row.offset} is " + f"{record_id}, not {row.record_id}; skipped", file=sys.stderr) + select_skipped += 1 + row = None + if row is None: + record_types[rec_type] -= 1 + dropped_ids.add(record_id) + for request_id in (record.rec_headers.get_header("WARC-Concurrent-To"), + record.rec_headers.get_header("WARC-Refers-To")): + if request_id: + dropped_ids.add(request_id) + record_types["request"] -= len(pending_requests.pop(request_id, [])) + # Metadata that arrived before its response. + record_types["metadata"] -= len(groups.pop(record_id, RecordGroup()).metadata_entries) + continue target_uri = record.rec_headers.get_header("WARC-Target-URI") # ARC has no record IDs; warcio mints one per record so the grouping dict # below still gets a unique key. Flagged so it never reaches a CSV. @@ -1211,7 +1235,6 @@ def main(input_file, output_path=None, dry_run=False, limit=None, output_format= group.refers_to_record_id = record.rec_headers.get_header("WARC-Refers-To") or "" group.refers_to_target_uri = record.rec_headers.get_header("WARC-Refers-To-Target-URI") or "" group.refers_to_date = record.rec_headers.get_header("WARC-Refers-To-Date") or "" - payload = record.content_stream().read() group.response_mime_type = detect_mime_type(record) if rec_type == "revisit": # Synthetic key for the CSVs; no such file is ever written to the zip, @@ -1255,6 +1278,9 @@ def main(input_file, output_path=None, dry_run=False, limit=None, output_format= elif rec_type == "request": body = record.content_stream().read() request_record_id = record.rec_headers.get_header("WARC-Record-ID") + if request_record_id in dropped_ids: # its capture was not selected + record_types[rec_type] -= 1 + continue req_entry = { "warc_headers": archive_header_pairs(record), "http_headers": list(record.http_headers.headers) if record.http_headers else [], @@ -1272,6 +1298,9 @@ def main(input_file, output_path=None, dry_run=False, limit=None, output_format= elif rec_type == "metadata": body = record.content_stream().read() concurrent_to = record.rec_headers.get_header("WARC-Concurrent-To") + if concurrent_to in dropped_ids: # its capture was not selected + record_types[rec_type] -= 1 + continue group = groups.setdefault(concurrent_to, RecordGroup()) body_text = body.decode("utf-8", errors="replace") @@ -1428,521 +1457,126 @@ def main(input_file, output_path=None, dry_run=False, limit=None, output_format= if skipped: print(f"warning: {skipped} CSV row(s) could not be written (see warnings above)", file=sys.stderr) - return skipped + int(download_size_mismatch) + select_failures = 0 + if select: + captures = record_types["response"] + record_types["revisit"] + print(f"Selected {captures} of {len(select_rows) + select_skipped} row(s) from {select}") + unmatched = len(select_rows) if wanted is None else len(wanted) + if unmatched and not limit_reached: + print(f"warning: {unmatched} row(s) of {select} match no capture in {input_file}", file=sys.stderr) + select_skipped += unmatched + if select_skipped: + print(f"warning: {select_skipped} row(s) could not be selected (see warnings above)", file=sys.stderr) + elif not captures: + print(f"warning: {select} selected nothing from {input_file}", file=sys.stderr) + select_failures = select_skipped or int(not captures) + + return skipped + int(download_size_mismatch) + select_failures # --------------------------------------------------------------------------- -# --fetch: re-download the rows of a (filtered) manifest.csv as one .warc.gz +# --select: convert only the captures a (filtered) manifest.csv names # --------------------------------------------------------------------------- -FETCH_COLUMNS = ("source_uri", "warc_record_offset", "warc_record_length") -FETCH_DEFAULT_RATE = 2.0 # requests per second, per host -FETCH_DEFAULT_RETRIES = 8 # retries after the first attempt -FETCH_MAX_BACKOFF = 60.0 # seconds; the cap on one exponential-backoff wait -FETCH_RETRYABLE_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504}) -FETCH_MAX_GAP = 64 * 1024 # bytes between two rows that still get fetched in one request -FETCH_MAX_SPAN = 64 * 1024 * 1024 # bytes held in memory for one request -FETCH_HEAD_BYTES = 64 * 1024 # probe size for the source's leading warcinfo record -FETCH_MAX_REDIRECTS = 5 # cdx_toolkit's myrequests_get does not follow redirects; http_range does -FETCH_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) - @dataclass -class FetchRow: - source_uri: str - offset: int - length: int +class SelectRow: + offset: int # the response/revisit record's warc_record_offset: the key a capture is selected by record_id: str # blank for ARC rows (the id was minted, so the manifest leaves it out) - target_uri: str + source_uri: str + warc_filename: str line: int # CSV line number, for warnings -class FetchLengthMismatch(Exception): - """The server returned a different number of bytes than the range asked for. +def read_select_rows(csv_path): + """Parse a manifest.csv (filtered or not) into SelectRows. - fsspec sends the Range header but never checks for a 206, so a server that ignores Range - hands back the whole file as the "slice". Deterministic, so never retried. - """ - - -def read_fetch_rows(csv_path): - """Parse a manifest.csv (filtered or not) into FetchRows. - - Missing a required column is a usage error (exit 2). A row with a blank or non-numeric - offset/length (main() writes "" when the offset was unknown), or with no fetchable - `source_uri` (stdin input), is warned about and counted, never fatal. - Returns (rows, skipped). + A CSV without a warc_record_offset column is a usage error (exit 2). A row with a blank or + non-numeric offset (main() writes "" when the offset was unknown) is warned about and + counted, never fatal. Returns (rows, skipped). """ if csv_path == "-": - return _read_fetch_rows(io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8", newline=""), "stdin") + return _read_select_rows(io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8", newline=""), "stdin") with open(csv_path, newline="", encoding="utf-8") as fh: - return _read_fetch_rows(fh, csv_path) + return _read_select_rows(fh, csv_path) -def _read_fetch_rows(fh, label): +def _read_select_rows(fh, label): reader = csv.DictReader(fh) - missing = [c for c in FETCH_COLUMNS if c not in (reader.fieldnames or [])] - if missing: - print(f"error: {label}: not a manifest.csv, missing column(s): {', '.join(missing)}", file=sys.stderr) + if "warc_record_offset" not in (reader.fieldnames or []): + print(f"error: {label}: not a manifest.csv, missing column: warc_record_offset", file=sys.stderr) raise SystemExit(2) rows, skipped = [], 0 for row in reader: - source_uri = (row.get("source_uri") or "").strip() try: offset = int(row.get("warc_record_offset") or "") - length = int(row.get("warc_record_length") or "") except ValueError: - offset, length = -1, 0 - - if not source_uri or source_uri == "-": - problem = "no fetchable source_uri" - elif offset < 0 or length <= 0: - problem = "no usable warc_record_offset/warc_record_length" - else: - problem = None - if problem: - print(f"warning: {label} line {reader.line_num}: {problem}, skipped", file=sys.stderr) + offset = -1 + if offset < 0: + print(f"warning: {label} line {reader.line_num}: no usable warc_record_offset, skipped", file=sys.stderr) skipped += 1 continue - rows.append( - FetchRow( - source_uri=source_uri, + SelectRow( offset=offset, - length=length, record_id=(row.get("warc_record_id") or "").strip(), - target_uri=(row.get("warc_target_uri") or "").strip(), + source_uri=(row.get("source_uri") or "").strip(), + warc_filename=(row.get("warc_filename") or "").strip(), line=reader.line_num, ) ) return rows, skipped -def group_fetch_rows(rows): - """Rows keyed by source_uri (first-appearance order), each list sorted by offset. - - Exact (offset, length) duplicates are dropped with one warning per source: fetching the - same member twice would put the same record in the output twice. Output order is - therefore source-then-offset, not CSV order. - """ - groups = {} - for row in rows: - groups.setdefault(row.source_uri, []).append(row) - - for source_uri, source_rows in groups.items(): - source_rows.sort(key=lambda r: (r.offset, r.length)) - deduped, seen = [], set() - for row in source_rows: - key = (row.offset, row.length) - if key not in seen: - seen.add(key) - deduped.append(row) - if len(deduped) != len(source_rows): - print(f"warning: {source_uri}: {len(source_rows) - len(deduped)} duplicate row(s) dropped", file=sys.stderr) - groups[source_uri] = deduped - return groups - - -def coalesce_ranges(rows, max_gap=FETCH_MAX_GAP, max_span=FETCH_MAX_SPAN): - """Merge offset-sorted rows into (start, end, rows) spans fetched with one request each. - - A response record is never adjacent to the next one in a Common Crawl WARC (its request - and metadata records sit in between), so merging only touching ranges would merge nothing. - Instead rows closer than `max_gap` share a request and each row is sliced back out of the - span locally: the output stays response-only and byte-exact, and a status-filtered subset - collapses into a fraction of the requests, which is what data.commoncrawl.org throttles on. - `max_gap` bounds the bytes wasted per merge, `max_span` the bytes held in memory. - """ - spans = [] - for row in rows: - end = row.offset + row.length - if spans: - start, span_end, span_rows = spans[-1] - if row.offset - span_end <= max_gap and end - start <= max_span: - spans[-1] = (start, max(span_end, end), span_rows + [row]) - continue - spans.append((row.offset, end, [row])) - return spans - - -class RateLimiter: - """Minimum interval between requests, tracked per key (a host). rate <= 0 disables it.""" - - def __init__(self, rate, clock=time.monotonic, sleep=time.sleep): - self.interval = 1.0 / rate if rate and rate > 0 else 0.0 - self.clock = clock - self.sleep = sleep - self.next_allowed = {} - - def wait(self, key): - if not self.interval: - return - now = self.clock() - delay = self.next_allowed.get(key, now) - now - if delay > 0: - self.sleep(delay) - now += delay - self.next_allowed[key] = now + self.interval - - -def error_status(exc): - """HTTP status carried by an exception (aiohttp's ClientResponseError), else None.""" - status = getattr(exc, "status", None) - return status if isinstance(status, int) else None +def row_names_input(row, input_file, input_warc_filename): + """True when `row` describes a record of `input_file`, so its offset can be looked up in it. + The workflow is one WARC end to end: convert it with --metadata-only, edit manifest.csv, + select from the same WARC. The same file may be spelled differently the second time (the + manifest came from s3://commoncrawl/..., the file is now local), so full URIs cannot be + compared. `input_warc_filename` is the WARC-Filename of the input's own warcinfo record, or + "" when it has none. A row this rejects is reported once, by name. -def is_retryable(exc): - """Transient failures only: throttling, server errors, timeouts, dropped connections. - - A 404/403 (fsspec raises FileNotFoundError / PermissionError), any other 4xx (416 means - the offsets do not fit this file) and a length mismatch are deterministic and re-raised - at once. aiohttp's non-OSError exceptions (payload/disconnect) are recognised by module, - so aiohttp is not imported here. + A row matches when either name it carries (its source_uri's basename, its warc_filename) + is either name the input goes by (its own basename, its warcinfo's WARC-Filename). Both + sides are needed: a derived WARC keeps the original's warcinfo, and a renamed copy keeps + only the warcinfo. A row naming no file at all (a CSV cut down to the offset column) has + nothing to compare and is accepted; the offset lookup and the record id check still guard it. """ - if isinstance(exc, FetchLengthMismatch): - return False - status = error_status(exc) - if status is not None: - return status in FETCH_RETRYABLE_STATUSES - if isinstance(exc, (FileNotFoundError, PermissionError)): - return False - if isinstance(exc, (OSError, TimeoutError, asyncio.TimeoutError)): + claimed = {input_basename(row.source_uri), row.warc_filename} - {""} + if not claimed: return True - return type(exc).__module__.split(".")[0] == "aiohttp" - - -def retry_after_seconds(exc): - """The Retry-After delay the server asked for, in seconds; only the delta-seconds form.""" - headers = getattr(exc, "headers", None) - if not headers: - return None - try: - return max(0.0, float(headers.get("Retry-After"))) - except (TypeError, ValueError): - return None + return bool(claimed & ({input_basename(input_file), input_warc_filename} - {""})) -def fetch_with_retry(fetch, *, retries=FETCH_DEFAULT_RETRIES, label="", sleep=time.sleep, rng=random.random): - """Call the zero-argument `fetch` until it returns, retrying transient failures. +def selected_offsets(rows, input_file, input_warc_filename): + """{offset: row} for the rows that name this input; one note for the rows that do not. - Waits Retry-After when the server sends one, otherwise an exponential backoff (2^attempt - seconds, capped at FETCH_MAX_BACKOFF) with jitter in [0.5, 1.5). One stderr line per retry. - Re-raises on a non-retryable error or once `retries` retries are used up. + Offsets are the key because they are the one identifier every manifest row has — ARC rows + leave warc_record_id blank, since warcio mints a different one on every run. """ - attempts = max(1, retries + 1) - for attempt in range(1, attempts + 1): - try: - return fetch() - except Exception as exc: - if attempt == attempts or not is_retryable(exc): - raise - delay = retry_after_seconds(exc) - if delay is None: - delay = min(FETCH_MAX_BACKOFF, 2.0**attempt) * (0.5 + rng()) - print( - f"warning: {label}: attempt {attempt}/{attempts} failed ({exc}), retrying in {delay:.1f} s", - file=sys.stderr, - ) - sleep(delay) - - -def leading_warcinfo(head): - """The first gzip member of `head`, verbatim, if it is a warcinfo record; else None. - - `head` is the first FETCH_HEAD_BYTES of the source. Going through open_archive_iterator() - means an ARC's filedesc:// record qualifies too. A warcinfo record that does not fit in - the probe, or a source whose first record is something else, yields None: nothing is - ever synthesised in its place. - """ - try: - iterator = open_archive_iterator(io.BytesIO(head)) - record = next(iter(iterator), None) - if record is None: - return None - rec_type = record.rec_type - record.content_stream().read() - length = iterator.get_record_length() - except Exception: # noqa: BLE001 - any parse failure means "no usable warcinfo" - return None - if rec_type != "warcinfo" or not length or length > len(head): - return None - member = head[:length] - # warcio reads a truncated gzip member without complaint; only a member whose trailer is - # inside the probe is a complete record. - if member.startswith(b"\x1f\x8b") and not _complete_gzip_member(member): - return None - return member - - -def _complete_gzip_member(data): - decompressor = zlib.decompressobj(16 + zlib.MAX_WBITS) - try: - decompressor.decompress(data) - except zlib.error: - return False - return decompressor.eof - - -def validate_record_slice(data, row): - """Check that `data` is exactly the response record the manifest row describes. - - Returns None when it is, else a short reason. "Parses as a response" is not enough: - warcio's sniffer reads arbitrary bytes as an ARC record of type response, so the - WARC-Record-ID (or, for ARC rows whose id is blank by design, the WARC-Target-URI) is - compared against the row, the parsed record must span the whole slice, and the gzip - member must be complete (warcio reads a truncated one without complaint). - """ - try: - iterator = open_archive_iterator(io.BytesIO(data)) - records = [] - for record in iterator: - record.content_stream().read() - records.append( - ( - record.rec_type, - record.rec_headers.get_header("WARC-Record-ID") or "", - record.rec_headers.get_header("WARC-Target-URI") or "", - iterator.get_record_length(), - ) - ) - except Exception as exc: # noqa: BLE001 - the reason is reported, not swallowed - return f"not a WARC record: {exc}" - - if len(records) != 1: - return f"expected one record, found {len(records)}" - rec_type, record_id, target_uri, length = records[0] - if rec_type != "response": - return f"expected a response record, found {rec_type}" - if length != len(data): - return f"record spans {length} of {len(data)} bytes" - if data.startswith(b"\x1f\x8b") and not _complete_gzip_member(data): - return "truncated gzip member" - if row.record_id: - if record_id != row.record_id: - return f"WARC-Record-ID {record_id} is not {row.record_id}" - elif target_uri != row.target_uri: - return f"WARC-Target-URI {target_uri} is not {row.target_uri}" - return None - - -def default_fetch_output_path(csv_path, run_id=None): - """Default .warc.gz path for --fetch: {csv basename minus .csv}_{hex}.warc.gz, in cwd. - - Same hex as default_output_path() and for the same reason: filtering one manifest twice - into `subset.csv` must not overwrite the first subset. `-` is labelled stdin. - """ - basename = input_basename(csv_path) - label = basename[:-4] if basename.lower().endswith(".csv") else basename - return Path(f"{label or 'unknown'}_{run_id or new_run_id()}.warc.gz") - - -def http_range(url, start, end, retries=None, redirects=FETCH_MAX_REDIRECTS, get=None): - """Bytes [start, end) over http(s), through cdx_toolkit's `myrequests_get`. - - That function carries Common Crawl's own retry policy (429/5xx backed off without limit, - connection failures up to `retries`) and per-host pacing, so neither is reimplemented here. - It does not follow redirects (`allow_redirects=False`), so 3xx answers are followed by hand, - re-entering `myrequests_get` each time so the new host is paced too. A 200 with the whole - file, from a server that ignores Range, is left to the caller's length check. - """ - get = get or myrequests_get # resolved at call time so tests can substitute the transport - headers = {"Range": f"bytes={start}-{end - 1}"} - for _ in range(redirects + 1): - resp = get(url, headers=headers, raise_error_after_n_errors=retries) - location = resp.headers.get("Location") - if resp.status_code in FETCH_REDIRECT_STATUSES and location: - url = urljoin(url, location) - continue - return resp.content - raise RuntimeError(f"more than {redirects} redirects") - - -def set_host_interval(host, rate): - """Map --rate onto cdx_toolkit's per-host schedule; rate=None keeps cdx_toolkit's own defaults. - - cdx_toolkit paces requests from a module-level table keyed by hostname (0.55 s between - requests to data.commoncrawl.org, 3 s to a host it does not know). `get_retries()` creates - the host's entry from the default one, and the interval is then overwritten in place — the - table is the only place the pacing can be set. - """ - if rate is None: - return - get_retries(host) - retry_info[host]["minimum_interval"] = 1.0 / rate if rate > 0 else 0.0 - - -def check_length(data, start, end): - if len(data) != end - start: - raise FetchLengthMismatch(f"asked for {end - start} bytes at offset {start}, got {len(data)}") - return data - - -def fetch_range(fs, path, start, end, exact=True): - """Bytes [start, end) of a file through fsspec (s3 and local paths). - - With `exact`, a short or long answer raises FetchLengthMismatch (see there). The warcinfo - probe passes exact=False because it deliberately asks past the end of small files. - """ - data = fs.cat_file(path, start=start, end=end) - return check_length(data, start, end) if exact else data - - -def annotate_record_slice(data, source_uri, offset, length): - """Re-serialise one record with WARC-Source-URI / WARC-Source-Range added. - - The names and values are cdx_toolkit's convention for extracts, so a subset made this way - matches one from `cdxt warc`, and a re-converted subset carries every record's original - coordinates in response_warc_headers.csv. The WARC header block is rewritten by warcio and - the member recompressed, so it is no longer the source's bytes, but the content block (HTTP - headers and body) is copied through untouched (a verbatim switch was tried and dropped: nothing - in the workflow needs the wire bytes, and the curl loop in the README gives them anyway). - Cost measured on CC records: ~1.4 ms CPU and ~43 bytes of output per record. - """ - # no_record_parse=True leaves the HTTP layer unparsed, so warcio writes the content block - # through byte for byte (a parsed one is re-serialised: obs-folds unfolded, Content-Length - # recomputed, WARC-Block-Digest silently wrong). Only the WARC header block is rewritten. - record = next(iter(ArchiveIterator(io.BytesIO(data), no_record_parse=True, arc2warc=True))) - record.rec_headers.replace_header("WARC-Source-URI", source_uri) - record.rec_headers.replace_header("WARC-Source-Range", f"bytes={offset}-{offset + length - 1}") - buffer = io.BytesIO() - WARCWriter(buffer, gzip=True).write_record(record) - return buffer.getvalue() - - -def _is_local_filesystem(fs): - protocol = fs.protocol if isinstance(fs.protocol, (tuple, list)) else (fs.protocol,) - return "file" in protocol - - -class _FetchSource: - """One source WARC: its transport, retry policy, pacing key and transfer counters. - - http(s) goes through cdx_toolkit (`http_range`), which retries and paces on its own; s3 and - local paths go through fsspec with `fetch_with_retry` and `RateLimiter` around them. - """ - - def __init__(self, source_uri, limiter, rate=None, retries=None): - self.source_uri = source_uri - parsed = urlparse(source_uri) - self.via_http = parsed.scheme.lower() in ("http", "https") - self.retries = retries - self.requests = 0 - self.transferred = 0 - if self.via_http: - self.fs = self.path = self.limiter = None - set_host_interval(parsed.hostname, rate) - self.host = parsed.hostname - else: - self.fs, self.path = url_to_fs(source_uri) - # Local files bypass the limiter; for s3 the key is the bucket. - self.host = None if _is_local_filesystem(self.fs) else parsed.netloc - self.limiter = limiter - - def get(self, start, end, exact=True): - self.requests += 1 - if self.via_http: - data = http_range(self.source_uri, start, end, retries=self.retries) - if exact: - check_length(data, start, end) + wanted, foreign = {}, Counter() + for row in rows: + if row_names_input(row, input_file, input_warc_filename): + wanted.setdefault(row.offset, row) else: - if self.host: - self.limiter.wait(self.host) - data = fetch_range(self.fs, self.path, start, end, exact) - self.transferred += len(data) - return data - - def fetch(self, start, end, exact=True, label="", sleep=time.sleep): - if self.via_http: - return self.get(start, end, exact) - retries = FETCH_DEFAULT_RETRIES if self.retries is None else self.retries - return fetch_with_retry( - functools.partial(self.get, start, end, exact), retries=retries, label=label, sleep=sleep - ) - - -def fetch_main(csv_path, output_path=None, rate=None, retries=None, sleep=time.sleep): - """--fetch: download every manifest row's byte range and concatenate the raw members. - - Each source's own warcinfo record is copied first (see leading_warcinfo), then its rows - in offset order. A span that fails after retries skips all its rows with a warning and - the run goes on, like the CSV writers: the file is a valid .warc.gz at every member - boundary, and the return value (skipped rows) makes cli() exit 1. `rate` and `retries` - are None by default so each transport keeps its own defaults (see _FetchSource). Every - record is stamped with WARC-Source-URI / WARC-Source-Range (see annotate_record_slice); the - warcinfo record is copied as-is. - """ - rows, skipped = read_fetch_rows(csv_path) - groups = group_fetch_rows(rows) - if output_path is None: - output_path = str(default_fetch_output_path(csv_path, new_run_id())) - - limiter = RateLimiter(FETCH_DEFAULT_RATE if rate is None else rate, sleep=sleep) - total_rows = sum(len(source_rows) for source_rows in groups.values()) - written = warcinfo_count = 0 - sources = [] - - with open(output_path, "wb") as out, tqdm(total=total_rows, unit="rec", desc="Fetching") as pbar: - for source_uri, source_rows in groups.items(): - source = _FetchSource(source_uri, limiter, rate=rate, retries=retries) - sources.append(source) - - try: - head = source.fetch(0, FETCH_HEAD_BYTES, exact=False, label=f"{source_uri} (warcinfo probe)", sleep=sleep) - except Exception as exc: # noqa: BLE001 - reported below; the rows are still attempted - print(f"warning: {source_uri}: could not read the leading warcinfo record ({exc})", file=sys.stderr) - head = b"" - warcinfo = leading_warcinfo(head) - if warcinfo: - out.write(warcinfo) - warcinfo_count += 1 - else: - print(f"note: {source_uri}: no leading warcinfo record, subset starts at its first response", - file=sys.stderr) - - for start, end, span_rows in coalesce_ranges(source_rows): - label = f"{source_uri} bytes {start}-{end - 1}" - try: - data = source.fetch(start, end, label=label, sleep=sleep) - except Exception as exc: # noqa: BLE001 - skip-and-warn, the run continues - print(f"warning: {label}: could not be fetched ({exc}), {len(span_rows)} row(s) skipped", - file=sys.stderr) - skipped += len(span_rows) - pbar.update(len(span_rows)) - continue - for row in span_rows: - chunk = data[row.offset - start : row.offset - start + row.length] - problem = validate_record_slice(chunk, row) - if problem: - print(f"warning: {csv_path} line {row.line}: fetched bytes are not the expected record " - f"({problem}), skipped", file=sys.stderr) - skipped += 1 - else: - out.write(annotate_record_slice(chunk, source_uri, row.offset, row.length)) - written += 1 - pbar.update(1) - - request_count = sum(s.requests for s in sources) - transferred = sum(s.transferred for s in sources) - print( - f"Created {output_path}: {written} records from {len(groups)} source(s), " - f"{request_count} requests, {tqdm.format_sizeof(transferred, suffix='B')} transferred " - f"({warcinfo_count} warcinfo record{'' if warcinfo_count == 1 else 's'})" - ) - if skipped: - print(f"warning: {skipped} row(s) could not be fetched (see warnings above)", file=sys.stderr) - return skipped + foreign[row.warc_filename or input_basename(row.source_uri) or "?"] += 1 + if foreign: + names = ", ".join(f"{name}: {n}" for name, n in foreign.most_common()) + print(f"note: {sum(foreign.values())} row(s) name another WARC and were not selected ({names})", + file=sys.stderr) + return wanted def cli(): parser = argparse.ArgumentParser(description="Convert a gzipped WARC file into a zip-of-zips archive.") - parser.add_argument("input_file", help="Path to a .warc.gz file, or '-' for stdin (a manifest.csv with --fetch)") + parser.add_argument("input_file", help="Path or URI of a .warc.gz file, or '-' for stdin") parser.add_argument( "--output", default=None, help="Output zip path (default: {basename}_{hex}.zip in the current directory, with the same hex as the " - "root directory inside and _partial appended when --limit is set). With --fetch: the output .warc.gz " - "(default: {basename}_{hex}.warc.gz)", + "root directory inside and _partial appended when --limit is set)", ) mode = parser.add_mutually_exclusive_group() @@ -1954,11 +1588,13 @@ def cli(): help="Write every CSV, manifest and sidecar but no payload files. The WARC is still " "streamed in full, so this saves output size, not transfer (use --limit for that).", ) - mode.add_argument( - "--fetch", - action="store_true", - help="Treat input_file as a manifest.csv (filtered or not) and download every row's byte range from its " - "source_uri into one .warc.gz, with retries and per-host rate limiting.", + parser.add_argument( + "--select", + metavar="CSV", + default=None, + help="Convert only the captures listed in this manifest.csv (from an earlier run on the same WARC, " + "filtered however you like; '-' for stdin), each with its request and metadata records. The WARC " + "is still read in full.", ) parser.add_argument( "--limit", @@ -1973,31 +1609,14 @@ def cli(): help="Output format: 'flat' (counter-named files + global CSVs) or 'sidecar' (domain dirs + per-file " "metadata). Default: flat", ) - parser.add_argument( - "--rate", - type=float, - default=None, - help="--fetch only: requests per second per host, 0 for unlimited. Default: cdx_toolkit's per-host " - f"pacing for http(s) (about 1.8/s for data.commoncrawl.org, 0.33/s elsewhere), {FETCH_DEFAULT_RATE:g}/s for s3", - ) - parser.add_argument( - "--retries", - type=int, - default=None, - help="--fetch only: for http(s), connection failures tolerated per request (default 100; throttling " - f"and server errors are retried without limit, as cdx_toolkit does); for s3, retries per request " - f"(default {FETCH_DEFAULT_RETRIES})", - ) args = parser.parse_args() - # Flags that would be silently ignored are refused instead: --format defaults to None so an - # explicit value is detectable here, and becomes "flat" only when it reaches main(). - if args.fetch: - if args.limit is not None or args.format is not None: - parser.error("--limit and --format do not apply to --fetch") - return 1 if fetch_main(args.input_file, args.output, rate=args.rate, retries=args.retries) else 0 - if args.rate is not None or args.retries is not None: - parser.error("--rate and --retries only apply to --fetch") + # Flags that would be silently ignored are refused instead. + if args.select: + if args.dry_run: + parser.error("--dry-run does not apply to --select") + if args.select == "-" and args.input_file == "-": + parser.error("the WARC and the --select CSV cannot both be stdin") # Default output naming lives in main() (see default_output_path). try: @@ -2008,6 +1627,7 @@ def cli(): limit=args.limit, output_format=args.format or "flat", metadata_only=args.metadata_only, + select=args.select, ) except PermissionError as exc: # archive.org's 401/403 (see InternetArchiveFileSystem); other inputs keep the traceback. From b41690097da6306cb66f50431b74d449e667f71b Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 12 Oct 2026 16:28:53 +0100 Subject: [PATCH 2/9] docs: add note to stick out --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 69d6034..ac9338f 100644 --- a/README.md +++ b/README.md @@ -424,7 +424,8 @@ warc2zip 'https://data.commoncrawl.org/crawl-data/.../CC-MAIN-....warc.gz' --sel The zip has the usual layout with only the listed captures, each one complete: payload, response, request and metadata records, plus the WARC's `warcinfo`. The WARC is read in full again, as for `--metadata-only`. Rows are matched by `warc_record_offset` (checked against `warc_record_id` when the row has one), so a `manifest.csv` from a selected zip works with `--select` too. -Select from the same WARC the manifest came from. It does not have to be at the same location: a local copy of an `s3://` file works, because rows are matched by file name. Rows that name another WARC are reported and skipped; run once per WARC for a subset that spans several. +> [!TIP] +> Select from the same WARC the manifest came from. It does not have to be at the same location: a local copy of an `s3://` file works, because rows are matched by file name. Rows that name another WARC are reported and skipped; run once per WARC for a subset that spans several. Offsets stay valid under `--limit`: limiting only stops the read early, it never rewrites them. From a359e36d3fe463d27f95142b1a36696bfd2e198c Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 12 Oct 2026 16:36:08 +0100 Subject: [PATCH 3/9] docs: add better examples --- README.md | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index ac9338f..b0e4350 100644 --- a/README.md +++ b/README.md @@ -405,16 +405,32 @@ List the column names: head -1 manifest.csv | tr ',' '\n' | tr -d '"\r' | nl ``` -Filter by column name and value, e.g. get all the `200` responses from the column `http_status_code`: +Filter by column name and value. Here some examples with `awk` and [miller](https://miller.readthedocs.io/en/latest/): + +#### get all the records of a specific mime-type, e.g. pdf documents + ```bash -awk -F'","' -v col="http_status_code" -v val="200" ' +awk -F'","' -v col="detected_mime_type" -v val="application/pdf" ' NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } { v=$c; gsub(/["\r]/,"",v) } v==val ' manifest.csv > subset.csv ```` or with [miller](https://miller.readthedocs.io/en/latest/): ``` -mlr --csv filter '$http_status_code == 200' manifest.csv > subset.csv +mlr --csv filter '$detected_mime_type == "application/pdf"' manifest.csv > subset.csv +``` + +#### get all records with http response code `200` from the column `http_status_code`: + +```bash +awk -F'","' -v col="http_status_code" -v val="200" ' +NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } +{ v=$c; gsub(/["\r]/,"",v) } v==val +' manifest.csv > subset.csv +```` +or with [miller](https://miller.readthedocs.io/en/latest/): +``` +mlr --csv filter '$http_status_code == "200"' manifest.csv > subset.csv ``` ### Convert what you kept From 5e63931a4d364e082fbe9e53564f706f59516fde Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 12 Oct 2026 16:37:11 +0100 Subject: [PATCH 4/9] docs: cosmetics --- README.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b0e4350..f2eeb85 100644 --- a/README.md +++ b/README.md @@ -409,12 +409,13 @@ Filter by column name and value. Here some examples with `awk` and [miller](http #### get all the records of a specific mime-type, e.g. pdf documents +with `awk`: ```bash awk -F'","' -v col="detected_mime_type" -v val="application/pdf" ' NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } { v=$c; gsub(/["\r]/,"",v) } v==val ' manifest.csv > subset.csv -```` +``` or with [miller](https://miller.readthedocs.io/en/latest/): ``` mlr --csv filter '$detected_mime_type == "application/pdf"' manifest.csv > subset.csv @@ -422,12 +423,13 @@ mlr --csv filter '$detected_mime_type == "application/pdf"' manifest.csv > subse #### get all records with http response code `200` from the column `http_status_code`: +with `awk`: ```bash awk -F'","' -v col="http_status_code" -v val="200" ' NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } { v=$c; gsub(/["\r]/,"",v) } v==val ' manifest.csv > subset.csv -```` +``` or with [miller](https://miller.readthedocs.io/en/latest/): ``` mlr --csv filter '$http_status_code == "200"' manifest.csv > subset.csv From c162b6dcd2b9919831c27694021816b5c4c71aa5 Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 12 Oct 2026 16:49:36 +0100 Subject: [PATCH 5/9] chore: ruffing --- pyproject.toml | 2 +- uv.lock | 41 +++++++++++++++++++++-------------------- 2 files changed, 22 insertions(+), 21 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 6a3dd42..b6e27a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ dependencies = [ ] [project.optional-dependencies] -dev = ["ruff", "pytest>=8.0"] +dev = ["ruff==0.14.3", "pytest>=8.0"] [project.scripts] warc2zip = "warc2zip:cli" diff --git a/uv.lock b/uv.lock index b7d5027..b98fa47 100644 --- a/uv.lock +++ b/uv.lock @@ -831,27 +831,28 @@ wheels = [ [[package]] name = "ruff" -version = "0.15.6" +version = "0.14.3" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/51/df/f8629c19c5318601d3121e230f74cbee7a3732339c52b21daa2b82ef9c7d/ruff-0.15.6.tar.gz", hash = "sha256:8394c7bb153a4e3811a4ecdacd4a8e6a4fa8097028119160dffecdcdf9b56ae4", size = 4597916, upload-time = "2026-03-12T23:05:47.51Z" } +sdist = { url = "https://files.pythonhosted.org/packages/75/62/50b7727004dfe361104dfbf898c45a9a2fdfad8c72c04ae62900224d6ecf/ruff-0.14.3.tar.gz", hash = "sha256:4ff876d2ab2b161b6de0aa1f5bd714e8e9b4033dc122ee006925fbacc4f62153", size = 5558687, upload-time = "2025-10-31T00:26:26.878Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/9e/2f/4e03a7e5ce99b517e98d3b4951f411de2b0fa8348d39cf446671adcce9a2/ruff-0.15.6-py3-none-linux_armv6l.whl", hash = "sha256:7c98c3b16407b2cf3d0f2b80c80187384bc92c6774d85fefa913ecd941256fff", size = 10508953, upload-time = "2026-03-12T23:05:17.246Z" }, - { url = "https://files.pythonhosted.org/packages/70/60/55bcdc3e9f80bcf39edf0cd272da6fa511a3d94d5a0dd9e0adf76ceebdb4/ruff-0.15.6-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:ee7dcfaad8b282a284df4aa6ddc2741b3f4a18b0555d626805555a820ea181c3", size = 10942257, upload-time = "2026-03-12T23:05:23.076Z" }, - { url = "https://files.pythonhosted.org/packages/e7/f9/005c29bd1726c0f492bfa215e95154cf480574140cb5f867c797c18c790b/ruff-0.15.6-py3-none-macosx_11_0_arm64.whl", hash = "sha256:3bd9967851a25f038fc8b9ae88a7fbd1b609f30349231dffaa37b6804923c4bb", size = 10322683, upload-time = "2026-03-12T23:05:33.738Z" }, - { url = "https://files.pythonhosted.org/packages/5f/74/2f861f5fd7cbb2146bddb5501450300ce41562da36d21868c69b7a828169/ruff-0.15.6-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:13f4594b04e42cd24a41da653886b04d2ff87adbf57497ed4f728b0e8a4866f8", size = 10660986, upload-time = "2026-03-12T23:05:53.245Z" }, - { url = "https://files.pythonhosted.org/packages/c1/a1/309f2364a424eccb763cdafc49df843c282609f47fe53aa83f38272389e0/ruff-0.15.6-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:e2ed8aea2f3fe57886d3f00ea5b8aae5bf68d5e195f487f037a955ff9fbaac9e", size = 10332177, upload-time = "2026-03-12T23:05:56.145Z" }, - { url = "https://files.pythonhosted.org/packages/30/41/7ebf1d32658b4bab20f8ac80972fb19cd4e2c6b78552be263a680edc55ac/ruff-0.15.6-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:70789d3e7830b848b548aae96766431c0dc01a6c78c13381f423bf7076c66d15", size = 11170783, upload-time = "2026-03-12T23:06:01.742Z" }, - { url = "https://files.pythonhosted.org/packages/76/be/6d488f6adca047df82cd62c304638bcb00821c36bd4881cfca221561fdfc/ruff-0.15.6-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:542aaf1de3154cea088ced5a819ce872611256ffe2498e750bbae5247a8114e9", size = 12044201, upload-time = "2026-03-12T23:05:28.697Z" }, - { url = "https://files.pythonhosted.org/packages/71/68/e6f125df4af7e6d0b498f8d373274794bc5156b324e8ab4bf5c1b4fc0ec7/ruff-0.15.6-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:1c22e6f02c16cfac3888aa636e9eba857254d15bbacc9906c9689fdecb1953ab", size = 11421561, upload-time = "2026-03-12T23:05:31.236Z" }, - { url = "https://files.pythonhosted.org/packages/f1/9f/f85ef5fd01a52e0b472b26dc1b4bd228b8f6f0435975442ffa4741278703/ruff-0.15.6-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:98893c4c0aadc8e448cfa315bd0cc343a5323d740fe5f28ef8a3f9e21b381f7e", size = 11310928, upload-time = "2026-03-12T23:05:45.288Z" }, - { url = "https://files.pythonhosted.org/packages/8c/26/b75f8c421f5654304b89471ed384ae8c7f42b4dff58fa6ce1626d7f2b59a/ruff-0.15.6-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:70d263770d234912374493e8cc1e7385c5d49376e41dfa51c5c3453169dc581c", size = 11235186, upload-time = "2026-03-12T23:05:50.677Z" }, - { url = "https://files.pythonhosted.org/packages/fc/d4/d5a6d065962ff7a68a86c9b4f5500f7d101a0792078de636526c0edd40da/ruff-0.15.6-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:55a1ad63c5a6e54b1f21b7514dfadc0c7fb40093fa22e95143cf3f64ebdcd512", size = 10635231, upload-time = "2026-03-12T23:05:37.044Z" }, - { url = "https://files.pythonhosted.org/packages/d6/56/7c3acf3d50910375349016cf33de24be021532042afbed87942858992491/ruff-0.15.6-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:8dc473ba093c5ec238bb1e7429ee676dca24643c471e11fbaa8a857925b061c0", size = 10340357, upload-time = "2026-03-12T23:06:04.748Z" }, - { url = "https://files.pythonhosted.org/packages/06/54/6faa39e9c1033ff6a3b6e76b5df536931cd30caf64988e112bbf91ef5ce5/ruff-0.15.6-py3-none-musllinux_1_2_i686.whl", hash = "sha256:85b042377c2a5561131767974617006f99f7e13c63c111b998f29fc1e58a4cfb", size = 10860583, upload-time = "2026-03-12T23:05:58.978Z" }, - { url = "https://files.pythonhosted.org/packages/cb/1e/509a201b843b4dfb0b32acdedf68d951d3377988cae43949ba4c4133a96a/ruff-0.15.6-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:cef49e30bc5a86a6a92098a7fbf6e467a234d90b63305d6f3ec01225a9d092e0", size = 11410976, upload-time = "2026-03-12T23:05:39.955Z" }, - { url = "https://files.pythonhosted.org/packages/6c/25/3fc9114abf979a41673ce877c08016f8e660ad6cf508c3957f537d2e9fa9/ruff-0.15.6-py3-none-win32.whl", hash = "sha256:bbf67d39832404812a2d23020dda68fee7f18ce15654e96fb1d3ad21a5fe436c", size = 10616872, upload-time = "2026-03-12T23:05:42.451Z" }, - { url = "https://files.pythonhosted.org/packages/89/7a/09ece68445ceac348df06e08bf75db72d0e8427765b96c9c0ffabc1be1d9/ruff-0.15.6-py3-none-win_amd64.whl", hash = "sha256:aee25bc84c2f1007ecb5037dff75cef00414fdf17c23f07dc13e577883dca406", size = 11787271, upload-time = "2026-03-12T23:05:20.168Z" }, - { url = "https://files.pythonhosted.org/packages/7f/d0/578c47dd68152ddddddf31cd7fc67dc30b7cdf639a86275fda821b0d9d98/ruff-0.15.6-py3-none-win_arm64.whl", hash = "sha256:c34de3dd0b0ba203be50ae70f5910b17188556630e2178fd7d79fc030eb0d837", size = 11060497, upload-time = "2026-03-12T23:05:25.968Z" }, + { url = "https://files.pythonhosted.org/packages/ce/8e/0c10ff1ea5d4360ab8bfca4cb2c9d979101a391f3e79d2616c9bf348cd26/ruff-0.14.3-py3-none-linux_armv6l.whl", hash = "sha256:876b21e6c824f519446715c1342b8e60f97f93264012de9d8d10314f8a79c371", size = 12535613, upload-time = "2025-10-31T00:25:44.302Z" }, + { url = "https://files.pythonhosted.org/packages/d3/c8/6724f4634c1daf52409fbf13fefda64aa9c8f81e44727a378b7b73dc590b/ruff-0.14.3-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:b6fd8c79b457bedd2abf2702b9b472147cd860ed7855c73a5247fa55c9117654", size = 12855812, upload-time = "2025-10-31T00:25:47.793Z" }, + { url = "https://files.pythonhosted.org/packages/de/03/db1bce591d55fd5f8a08bb02517fa0b5097b2ccabd4ea1ee29aa72b67d96/ruff-0.14.3-py3-none-macosx_11_0_arm64.whl", hash = "sha256:71ff6edca490c308f083156938c0c1a66907151263c4abdcb588602c6e696a14", size = 11944026, upload-time = "2025-10-31T00:25:49.657Z" }, + { url = "https://files.pythonhosted.org/packages/0b/75/4f8dbd48e03272715d12c87dc4fcaaf21b913f0affa5f12a4e9c6f8a0582/ruff-0.14.3-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:786ee3ce6139772ff9272aaf43296d975c0217ee1b97538a98171bf0d21f87ed", size = 12356818, upload-time = "2025-10-31T00:25:51.949Z" }, + { url = "https://files.pythonhosted.org/packages/ec/9b/506ec5b140c11d44a9a4f284ea7c14ebf6f8b01e6e8917734a3325bff787/ruff-0.14.3-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:cd6291d0061811c52b8e392f946889916757610d45d004e41140d81fb6cd5ddc", size = 12336745, upload-time = "2025-10-31T00:25:54.248Z" }, + { url = "https://files.pythonhosted.org/packages/c7/e1/c560d254048c147f35e7f8131d30bc1f63a008ac61595cf3078a3e93533d/ruff-0.14.3-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a497ec0c3d2c88561b6d90f9c29f5ae68221ac00d471f306fa21fa4264ce5fcd", size = 13101684, upload-time = "2025-10-31T00:25:56.253Z" }, + { url = "https://files.pythonhosted.org/packages/a5/32/e310133f8af5cd11f8cc30f52522a3ebccc5ea5bff4b492f94faceaca7a8/ruff-0.14.3-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:e231e1be58fc568950a04fbe6887c8e4b85310e7889727e2b81db205c45059eb", size = 14535000, upload-time = "2025-10-31T00:25:58.397Z" }, + { url = "https://files.pythonhosted.org/packages/a2/a1/7b0470a22158c6d8501eabc5e9b6043c99bede40fa1994cadf6b5c2a61c7/ruff-0.14.3-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:469e35872a09c0e45fecf48dd960bfbce056b5db2d5e6b50eca329b4f853ae20", size = 14156450, upload-time = "2025-10-31T00:26:00.889Z" }, + { url = "https://files.pythonhosted.org/packages/0a/96/24bfd9d1a7f532b560dcee1a87096332e461354d3882124219bcaff65c09/ruff-0.14.3-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3d6bc90307c469cb9d28b7cfad90aaa600b10d67c6e22026869f585e1e8a2db0", size = 13568414, upload-time = "2025-10-31T00:26:03.291Z" }, + { url = "https://files.pythonhosted.org/packages/a7/e7/138b883f0dfe4ad5b76b58bf4ae675f4d2176ac2b24bdd81b4d966b28c61/ruff-0.14.3-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0e2f8a0bbcffcfd895df39c9a4ecd59bb80dca03dc43f7fb63e647ed176b741e", size = 13315293, upload-time = "2025-10-31T00:26:05.708Z" }, + { url = "https://files.pythonhosted.org/packages/33/f4/c09bb898be97b2eb18476b7c950df8815ef14cf956074177e9fbd40b7719/ruff-0.14.3-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:678fdd7c7d2d94851597c23ee6336d25f9930b460b55f8598e011b57c74fd8c5", size = 13539444, upload-time = "2025-10-31T00:26:08.09Z" }, + { url = "https://files.pythonhosted.org/packages/9c/aa/b30a1db25fc6128b1dd6ff0741fa4abf969ded161599d07ca7edd0739cc0/ruff-0.14.3-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:1ec1ac071e7e37e0221d2f2dbaf90897a988c531a8592a6a5959f0603a1ecf5e", size = 12252581, upload-time = "2025-10-31T00:26:10.297Z" }, + { url = "https://files.pythonhosted.org/packages/da/13/21096308f384d796ffe3f2960b17054110a9c3828d223ca540c2b7cc670b/ruff-0.14.3-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:afcdc4b5335ef440d19e7df9e8ae2ad9f749352190e96d481dc501b753f0733e", size = 12307503, upload-time = "2025-10-31T00:26:12.646Z" }, + { url = "https://files.pythonhosted.org/packages/cb/cc/a350bac23f03b7dbcde3c81b154706e80c6f16b06ff1ce28ed07dc7b07b0/ruff-0.14.3-py3-none-musllinux_1_2_i686.whl", hash = "sha256:7bfc42f81862749a7136267a343990f865e71fe2f99cf8d2958f684d23ce3dfa", size = 12675457, upload-time = "2025-10-31T00:26:15.044Z" }, + { url = "https://files.pythonhosted.org/packages/cb/76/46346029fa2f2078826bc88ef7167e8c198e58fe3126636e52f77488cbba/ruff-0.14.3-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a65e448cfd7e9c59fae8cf37f9221585d3354febaad9a07f29158af1528e165f", size = 13403980, upload-time = "2025-10-31T00:26:17.81Z" }, + { url = "https://files.pythonhosted.org/packages/9f/a4/35f1ef68c4e7b236d4a5204e3669efdeefaef21f0ff6a456792b3d8be438/ruff-0.14.3-py3-none-win32.whl", hash = "sha256:f3d91857d023ba93e14ed2d462ab62c3428f9bbf2b4fbac50a03ca66d31991f7", size = 12500045, upload-time = "2025-10-31T00:26:20.503Z" }, + { url = "https://files.pythonhosted.org/packages/03/15/51960ae340823c9859fb60c63301d977308735403e2134e17d1d2858c7fb/ruff-0.14.3-py3-none-win_amd64.whl", hash = "sha256:d7b7006ac0756306db212fd37116cce2bd307e1e109375e1c6c106002df0ae5f", size = 13594005, upload-time = "2025-10-31T00:26:22.533Z" }, + { url = "https://files.pythonhosted.org/packages/b7/73/4de6579bac8e979fca0a77e54dec1f1e011a0d268165eb8a9bc0982a6564/ruff-0.14.3-py3-none-win_arm64.whl", hash = "sha256:26eb477ede6d399d898791d01961e16b86f02bc2486d0d1a7a9bb2379d055dc1", size = 12590017, upload-time = "2025-10-31T00:26:24.52Z" }, ] [[package]] @@ -981,7 +982,7 @@ dev = [ requires-dist = [ { name = "fsspec", extras = ["http"] }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" }, - { name = "ruff", marker = "extra == 'dev'" }, + { name = "ruff", marker = "extra == 'dev'", specifier = "==0.14.3" }, { name = "tqdm", specifier = ">=4.67.3" }, { name = "warcio", extras = ["s3"], specifier = "==1.8.1" }, ] From 5954a1627331b5719473474de22028085211cdf5 Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Mon, 12 Oct 2026 21:04:47 +0100 Subject: [PATCH 6/9] docs: cosmetics --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f2eeb85..a06d4db 100644 --- a/README.md +++ b/README.md @@ -407,7 +407,7 @@ head -1 manifest.csv | tr ',' '\n' | tr -d '"\r' | nl Filter by column name and value. Here some examples with `awk` and [miller](https://miller.readthedocs.io/en/latest/): -#### get all the records of a specific mime-type, e.g. pdf documents +#### Example 1: get all the records of a specific mime-type, e.g. pdf documents with `awk`: ```bash @@ -421,7 +421,7 @@ or with [miller](https://miller.readthedocs.io/en/latest/): mlr --csv filter '$detected_mime_type == "application/pdf"' manifest.csv > subset.csv ``` -#### get all records with http response code `200` from the column `http_status_code`: +#### Example 2: get all records with http response code `200` from the column `http_status_code`: with `awk`: ```bash From bb7e322b378f01f91a86cf2354c4997e0f3d9973 Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Sun, 18 Oct 2026 14:25:19 +0100 Subject: [PATCH 7/9] docs: loosen the instructions on the manifest filtering. --- README.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/README.md b/README.md index a06d4db..c0c469d 100644 --- a/README.md +++ b/README.md @@ -400,6 +400,15 @@ unzip -p meta.zip '*/manifest.csv' > manifest.csv ### Filter it with whatever you already use +> [!TIP] +> Feel free to use what you already know or use. warc2zip should work with CSV exported by any major tools. + +You can open the CSV in Excel, LibreOffice, Python, R, etc. and filter it there, then save it back to CSV. +Or use command-line tools like `awk`, `csvkit`, `miller`, `xsv`, `q`, etc. + + +#### Advanced filtering with bash and `awk` or [miller](https://miller.readthedocs.io/en/latest/) + List the column names: ```bash head -1 manifest.csv | tr ',' '\n' | tr -d '"\r' | nl From 9e5527d9f0e01b038578e33bb9f3d5e1f83b211a Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Sun, 18 Oct 2026 14:27:16 +0100 Subject: [PATCH 8/9] docs: cosmetics --- README.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index c0c469d..77a6d39 100644 --- a/README.md +++ b/README.md @@ -403,7 +403,10 @@ unzip -p meta.zip '*/manifest.csv' > manifest.csv > [!TIP] > Feel free to use what you already know or use. warc2zip should work with CSV exported by any major tools. -You can open the CSV in Excel, LibreOffice, Python, R, etc. and filter it there, then save it back to CSV. +You can open the CSV in Excel, LibreOffice, Python, R, etc. and filter it there, then save it back to CSV. +You can skip to the [next step](#convert-what-you-kept). + + Or use command-line tools like `awk`, `csvkit`, `miller`, `xsv`, `q`, etc. From a04941c8caf689ccd4ab925b0a89a85d479e5f5a Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Sun, 18 Oct 2026 14:30:48 +0100 Subject: [PATCH 9/9] docs: cosmetics --- README.md | 66 +++++++++++++++++++------------------------------------ 1 file changed, 23 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 77a6d39..a43c6c2 100644 --- a/README.md +++ b/README.md @@ -400,52 +400,32 @@ unzip -p meta.zip '*/manifest.csv' > manifest.csv ### Filter it with whatever you already use -> [!TIP] -> Feel free to use what you already know or use. warc2zip should work with CSV exported by any major tools. - You can open the CSV in Excel, LibreOffice, Python, R, etc. and filter it there, then save it back to CSV. You can skip to the [next step](#convert-what-you-kept). - -Or use command-line tools like `awk`, `csvkit`, `miller`, `xsv`, `q`, etc. - - -#### Advanced filtering with bash and `awk` or [miller](https://miller.readthedocs.io/en/latest/) - -List the column names: -```bash -head -1 manifest.csv | tr ',' '\n' | tr -d '"\r' | nl -``` - -Filter by column name and value. Here some examples with `awk` and [miller](https://miller.readthedocs.io/en/latest/): - -#### Example 1: get all the records of a specific mime-type, e.g. pdf documents - -with `awk`: -```bash -awk -F'","' -v col="detected_mime_type" -v val="application/pdf" ' -NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } -{ v=$c; gsub(/["\r]/,"",v) } v==val -' manifest.csv > subset.csv -``` -or with [miller](https://miller.readthedocs.io/en/latest/): -``` -mlr --csv filter '$detected_mime_type == "application/pdf"' manifest.csv > subset.csv -``` - -#### Example 2: get all records with http response code `200` from the column `http_status_code`: - -with `awk`: -```bash -awk -F'","' -v col="http_status_code" -v val="200" ' -NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } -{ v=$c; gsub(/["\r]/,"",v) } v==val -' manifest.csv > subset.csv -``` -or with [miller](https://miller.readthedocs.io/en/latest/): -``` -mlr --csv filter '$http_status_code == "200"' manifest.csv > subset.csv -``` +To use command-line tools like `awk`, `csvkit`, `miller`, `xsv`, `q`, etc, read the next few lines. + +For example to get all records with http response code `200` from the column `http_status_code`: + +1. List the column names: + + ```bash + head -1 manifest.csv | tr ',' '\n' | tr -d '"\r' | nl + ``` + +2. Filter by column name and value. Here some examples with `awk` and [miller](https://miller.readthedocs.io/en/latest/). + + with `awk`: + ```bash + awk -F'","' -v col="http_status_code" -v val="200" ' + NR==1 { for (i=1; i<=NF; i++) { h=$i; gsub(/["\r]/,"",h); if (h==col) c=i } print; next } + { v=$c; gsub(/["\r]/,"",v) } v==val + ' manifest.csv > subset.csv + ``` + or with [miller](https://miller.readthedocs.io/en/latest/): + ``` + mlr --csv filter '$http_status_code == "200"' manifest.csv > subset.csv + ``` ### Convert what you kept ```bash