diff --git a/tests/v2/_differential_fixtures.py b/tests/v2/_differential_fixtures.py index 4a9b57ee..f7c8fae5 100644 --- a/tests/v2/_differential_fixtures.py +++ b/tests/v2/_differential_fixtures.py @@ -81,6 +81,29 @@ def read(self) -> bytes: return self.body +def own_cache(monkeypatch: pytest.MonkeyPatch, + tmp_path_factory: pytest.TempPathFactory) -> Path: + """Point the tools' cache at a fresh scratch folder and return it: + what each local-corpus test module's autouse fixture does, so that a + test that forgets to set a cache, or a guard that stops keeping a + test's corpus out of it, writes there rather than into the + developer's real cache.""" + path = tmp_path_factory.mktemp("cache") + monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(path)) + return path + + +def plant_corpus(monkeypatch: pytest.MonkeyPatch, corpora: ModuleType, source: str, + records: list[dict[str, str]], date: str, unit: str) -> None: + """`records` fetched as `source`'s local corpus into the cache, with no + network: through a Source carrying the real one's credit and nothing + else, so that none of the real source's renew or check applies.""" + monkeypatch.setitem(corpora.SOURCES, source, corpora.Source( + corpora.SOURCES[source].credit, unit, + lambda: corpora.Fetched(list(records), date, {}))) + corpora.fetch(source) + + def serve(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, body: bytes, *, requests: list[urllib.request.Request] | None = None, sleeps: list[float] | None = None) -> list[str]: diff --git a/tests/v2/test_local_agreement.py b/tests/v2/test_local_agreement.py index 60205c4d..b14f6d59 100644 --- a/tests/v2/test_local_agreement.py +++ b/tests/v2/test_local_agreement.py @@ -7,15 +7,17 @@ off, and the record's refusals: no record, a record that is not one, a record of another fetch or another scorer, and a corpus damaged under the stream. The names are invented. """ +import hashlib import json import sys +import unicodedata from pathlib import Path import pytest from nameparser import Parser -from ._differential_fixtures import load_tool +from ._differential_fixtures import load_tool, own_cache, plant_corpus agreement = load_tool("local_agreement") corpora = agreement.local_corpora @@ -26,10 +28,7 @@ @pytest.fixture(autouse=True) def _own_cache(monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.TempPathFactory) -> Path: - """Every test's cache is its own, as in test_local_corpora.py.""" - path = tmp_path_factory.mktemp("cache") - monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(path)) - return path + return own_cache(monkeypatch, tmp_path_factory) def _filed(title: str = "", forename: str = "", middle_name: str = "", surname: str = "", @@ -97,10 +96,7 @@ def test_a_title_box_holding_an_unlisted_word_the_parse_reads_otherwise_is_set_a def _corpus(monkeypatch: pytest.MonkeyPatch, *records: dict[str, str]) -> None: """A PSC corpus of `records` fetched into the cache, no network behind it.""" - monkeypatch.setitem(corpora.SOURCES, _PSC, corpora.Source( - corpora.SOURCES[_PSC].credit, "filed names", - lambda: corpora.Fetched(list(records), "2026-10-22", {}))) - corpora.fetch(_PSC) + plant_corpus(monkeypatch, corpora, _PSC, list(records), "2026-10-22", "filed names") _ALL = (_AGREES, _DOUBLE_SURNAME, _SUFFIX, _JOB_TITLE, _SHAPE_TITLE) @@ -217,6 +213,25 @@ def test_a_line_that_is_not_a_record_is_refused_as_damage( assert len(list(agreement.records(loaded))) == len(_ALL) # the control +def test_a_record_holding_a_value_that_is_not_text_is_refused_as_damage( + monkeypatch: pytest.MonkeyPatch, _own_cache: Path) -> None: + # With its sha256 rewritten to match, as a fetch that wrote a number + # into a box would leave it, only the stream's own check sees it, and + # the scorer would otherwise fold a number as a name's box. + _corpus(monkeypatch, *_ALL) + folder = _own_cache / "local" / _PSC + raw = folder / "raw.jsonl" + body = raw.read_bytes() + b'{"name": "Al Ng", "surname": 1}\n' + raw.write_bytes(body) + meta = json.loads((folder / "meta.json").read_text()) + meta["sha256"] = hashlib.sha256(body).hexdigest() + (folder / "meta.json").write_text(json.dumps(meta)) + loaded = corpora.load(_PSC) + assert loaded is not None + with pytest.raises(corpora.CorpusDamaged, match="not text.*fetch it again"): + list(agreement.records(loaded)) + + @pytest.mark.parametrize("body", ["", "{not json\n", '["a list"]\n', '{"raw_sha256": "x"}\n{"got": {}}\n'], ids=["empty", "header not JSON", "header not an object", @@ -248,6 +263,11 @@ def test_a_record_written_by_another_scorer_is_refused( patch.setattr(agreement.corpus_build, "jsonl_line", lambda obj, sort_keys=False: str(obj)) assert agreement.scorer_fingerprint() != before, "jsonl_line" + # fold() casefolds, so a Python whose Unicode database moves it reads + # names differently; its moves would be the database's, not the parser's. + with monkeypatch.context() as patch: + patch.setattr(unicodedata, "unidata_version", "0.0.0") + assert agreement.scorer_fingerprint() != before, "unidata_version" assert agreement.scorer_fingerprint() == before monkeypatch.setattr(agreement, "fold", lambda text: text.casefold()) with pytest.raises(SystemExit, match="another version of this scorer"): diff --git a/tests/v2/test_local_corpora.py b/tests/v2/test_local_corpora.py index f21913bd..e520fa49 100644 --- a/tests/v2/test_local_corpora.py +++ b/tests/v2/test_local_corpora.py @@ -31,7 +31,7 @@ import pytest -from ._differential_fixtures import _Response, load_tool +from ._differential_fixtures import _Response, load_tool, own_cache, plant_corpus local = load_tool("local_corpora") snapshot = load_tool("snapshot") @@ -42,12 +42,7 @@ @pytest.fixture(autouse=True) def _own_cache(monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.TempPathFactory) -> Path: - """Every test's cache is its own: a test that forgets to set one, or a - guard that stops keeping a test's corpus out of the cache, writes - into a scratch folder rather than the developer's real cache.""" - path = tmp_path_factory.mktemp("cache") - monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(path)) - return path + return own_cache(monkeypatch, tmp_path_factory) def _study(nct: str, *officials: str) -> dict[str, Any]: @@ -261,7 +256,7 @@ def test_the_last_attempt_raises_its_own_error( local.fetch(_SOURCE) assert raised.value.code == 503 raised.value.close() - assert sum("studies" in u for u in fetched) == local.CTGOV_RETRIES + assert sum("studies" in u for u in fetched) == local.RETRIES def test_a_source_detail_cannot_replace_what_load_checks( @@ -363,7 +358,7 @@ def read(self, n: int) -> bytes: def _fec_cache(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path: monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(tmp_path)) - monkeypatch.setattr(local, "FEC_CHUNK", 64) # several reads, not one + monkeypatch.setattr(local, "CHUNK", 64) # several reads, not one return tmp_path / "local" / _FEC / local.FEC_ZIP @@ -911,7 +906,7 @@ def test_a_tse_strong_etag_is_preferred_to_the_last_modified_date(cdn: _Cdn) -> def test_a_tse_request_cut_short_or_dropped_is_retried( cdn: _Cdn, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: - monkeypatch.setattr(local, "CTGOV_RETRIES", 5) # four failures on one range + monkeypatch.setattr(local, "RETRIES", 5) # four failures on one range cdn.short = 1 cdn.head_failures = [urllib.error.URLError("reset")] cdn.failures = [urllib.error.URLError("reset"), _http_error(503), _http_error(429)] @@ -1034,11 +1029,11 @@ def test_a_tse_rate_limit_waits_as_long_as_the_server_asks( def test_a_tse_range_failing_every_attempt_raises_the_last_error(cdn: _Cdn) -> None: - cdn.failures = [_http_error(503) for _ in range(local.CTGOV_RETRIES)] + cdn.failures = [_http_error(503) for _ in range(local.RETRIES)] with pytest.raises(urllib.error.HTTPError) as raised: local.fetch(_TSE) raised.value.close() - assert cdn.ranges == local.CTGOV_RETRIES + assert cdn.ranges == local.RETRIES def test_a_tse_server_that_states_no_accept_ranges_is_read(cdn: _Cdn) -> None: @@ -1556,7 +1551,7 @@ def test_a_listed_tw_report_that_keeps_failing_is_raised( with pytest.raises(urllib.error.HTTPError) as raised: local.fetch(_TW) raised.value.close() - assert len(_tw_serials(asked)) == local.CTGOV_RETRIES + assert len(_tw_serials(asked)) == local.RETRIES assert local.load(_TW) is None @@ -1691,6 +1686,51 @@ def _tolerate_a_small_fixture(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(local, "UKEC_MAX_NAMELESS", 1) +def _failing_once(monkeypatch: pytest.MonkeyPatch, url: str) -> list[str]: + """Make the stubbed urlopen raise a 503 the first time `url` is asked, + with no sleep between attempts; the URLs that failed.""" + served: Callable[..., object] = urllib.request.urlopen + failed: list[str] = [] + + def flaky(request: urllib.request.Request, timeout: int) -> object: + if request.full_url == url and not failed: + failed.append(url) + raise _http_error(503) + return served(request, timeout) + monkeypatch.setattr(urllib.request, "urlopen", flaky) + monkeypatch.setattr(time, "sleep", lambda seconds: None) + return failed + + +@pytest.mark.parametrize("which", ["UKEC_COUNT_URL", "UKEC_CSV_URL"]) +def test_a_ukec_request_failing_once_is_retried( + monkeypatch: pytest.MonkeyPatch, which: str) -> None: + # The register's search and export once went out bare: one 503 in + # either ended the fetch, where every other source retries. + _tolerate_a_small_fixture(monkeypatch) + _serve_ukec(monkeypatch, _ukec_csv()) + failed = _failing_once(monkeypatch, getattr(local, which)) + assert local.fetch(_UKEC).records + assert failed == [getattr(local, which)] + + +def test_a_ukec_search_answering_no_json_is_refused_as_a_value_error( + monkeypatch: pytest.MonkeyPatch) -> None: + # An error page served with HTTP 200 is retried as a body cut short, + # and once the retries run out the fetch must still fail as a + # ValueError, as the other refusals do, not as the retry's own class. + monkeypatch.setattr(time, "sleep", lambda seconds: None) + asked: list[str] = [] + + def urlopen(request: urllib.request.Request, timeout: int) -> _Response: + asked.append(request.full_url) + return _Response(b"Service Unavailable") + monkeypatch.setattr(urllib.request, "urlopen", urlopen) + with pytest.raises(ValueError, match="is not JSON"): + local.fetch(_UKEC) + assert asked == [local.UKEC_COUNT_URL] * local.RETRIES + + def test_the_ukec_count_asks_the_export_s_own_filter() -> None: # A count under another filter would vouch for an export it did not count. def query(url: str) -> list[tuple[str, str]]: @@ -1955,7 +1995,7 @@ def _serve_psc(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, linked) and each part file with its answer: bytes to serve, or an exception to raise. Returns the file names downloaded.""" monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(tmp_path)) - monkeypatch.setattr(local, "FEC_CHUNK", 64) # several reads, not one + monkeypatch.setattr(local, "CHUNK", 64) # several reads, not one monkeypatch.setattr(local, "PSC_SAMPLE", 1) # every record, unless a test samples page = _psc_page(*parts) if page is None else page asked: list[str] = [] @@ -1984,6 +2024,14 @@ def _psc_snapshot(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, for n, file in enumerate(_psc_files(of, day), 1)}) +def test_a_psc_listing_failing_once_is_retried( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + _psc_snapshot(monkeypatch, tmp_path) + failed = _failing_once(monkeypatch, local.PSC_PAGE) + day, parts = local._psc_listing() + assert (day, len(parts), failed) == (_PSC_DAY, len(_PSC_PARTS), [local.PSC_PAGE]) + + def _gone(file: str) -> urllib.error.HTTPError: return urllib.error.HTTPError(f"{local.PSC_PAGE}/{file}", 404, "Not Found", email.message.Message(), None) @@ -2290,10 +2338,7 @@ def _fetched(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, behind it; its folder.""" monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(tmp_path)) records = [{"name": n} for n in names or ("Ann Lee, MD", "Bo Wu, PhD")] - monkeypatch.setitem(local.SOURCES, _SOURCE, local.Source( - local.SOURCES[_SOURCE].credit, "officials", - lambda: local.Fetched(records, date, {}))) - local.fetch(_SOURCE) + plant_corpus(monkeypatch, local, _SOURCE, records, date, "officials") return _folder(tmp_path) @@ -2492,10 +2537,16 @@ def test_a_change_to_how_a_word_is_cut_or_folded_changes_the_key( assert local.fold_fingerprint() != before +@pytest.mark.parametrize("reader", ["read_records", "corpus_build.jsonl_lines"]) def test_a_change_to_how_raw_jsonl_is_cut_into_records_changes_the_key( - monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch: pytest.MonkeyPatch, reader: str) -> None: + # The index is built from Local.records, an attribute the fold's own + # functions read without calling what fills it, so the walk in + # test_the_fold_lists_everything_its_functions_call_and_read cannot + # see these two; they are hashed by name. before = local.fold_fingerprint() - monkeypatch.setattr(local.corpus_build, "jsonl_lines", lambda text: []) + owner = local.corpus_build if reader.startswith("corpus_build.") else local + monkeypatch.setattr(owner, reader.rsplit(".", 1)[-1], lambda *args, **kwargs: []) assert local.fold_fingerprint() != before @@ -2619,15 +2670,21 @@ def test_tally_over_damaged_records_exits_naming_the_fetch( local.main(["tally", _SOURCE]) -def test_a_record_without_a_name_fails_when_the_records_are_read( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: +@pytest.mark.parametrize(("raw", "match"), [ + (b'{"nom": "Ann Lee, MD"}\n', "record without a name"), + (b'{"name": "Ann Lee, MD"\n', "a line that is not JSON"), +], ids=["a record without a name", "a line that is not JSON"]) +def test_a_damaged_record_fails_when_the_records_are_read( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, raw: bytes, match: str) -> None: + # With the sha256 rewritten to match, only the record checks see it, + # and each must raise CorpusDamaged, which name_usage's per-row guard + # catches, not the bare ValueError a JSON error is. folder = _fetched(monkeypatch, tmp_path) - raw = b'{"nom": "Ann Lee, MD"}\n' (folder / "raw.jsonl").write_bytes(raw) _rewrite_meta(folder, sha256=hashlib.sha256(raw).hexdigest()) loaded = local.load(_SOURCE) assert loaded is not None - with pytest.raises(ValueError, match="record without a name"): + with pytest.raises(local.CorpusDamaged, match=match): loaded.records # noqa: B018 diff --git a/tests/v2/test_name_usage_tool.py b/tests/v2/test_name_usage_tool.py index 0db22f5d..ca38172e 100644 --- a/tests/v2/test_name_usage_tool.py +++ b/tests/v2/test_name_usage_tool.py @@ -31,6 +31,8 @@ import pytest +from ._differential_fixtures import own_cache, plant_corpus + _PATH = Path(__file__).parents[2] / "tools" / "name_usage.py" @@ -90,7 +92,7 @@ def _no_locals(monkeypatch: pytest.MonkeyPatch, tmp_path_factory: pytest.TempPathFactory) -> None: # The cache too: a local corpus's word index is written there, and a # test's must never land in the developer's. - monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(tmp_path_factory.mktemp("cache"))) + own_cache(monkeypatch, tmp_path_factory) monkeypatch.setattr(tool, "load_locals", lambda: []) @@ -2258,10 +2260,8 @@ def test_records_damaged_after_the_index_was_cached_fail_their_own_row( # word then reads them first in the lookup, outside its guard. monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(tmp_path)) module = tool._local_corpora() - monkeypatch.setitem(module.SOURCES, "ctgov-officials", module.Source( - module.SOURCES["ctgov-officials"].credit, "officials", - lambda: module.Fetched([{"name": "Ann Lee, Ph. D."}], "2026-10-21", {}))) - module.fetch("ctgov-officials") + plant_corpus(monkeypatch, module, "ctgov-officials", [{"name": "Ann Lee, Ph. D."}], + "2026-10-21", "officials") _LOAD_LOCALS() # builds and caches the index locals_ = [row for row in _LOAD_LOCALS() if row[0] == "ctgov-officials"] (tmp_path / "local" / "ctgov-officials" / "raw.jsonl").write_text("") @@ -2277,10 +2277,8 @@ def test_a_damaged_local_corpus_fails_its_row_not_the_run( # KeyError, which _attempt does not catch, and took every row down. monkeypatch.setenv("NAMEPARSER_TOOLS_CACHE", str(tmp_path)) module = tool._local_corpora() - monkeypatch.setitem(module.SOURCES, "ctgov-officials", module.Source( - module.SOURCES["ctgov-officials"].credit, "officials", - lambda: module.Fetched([{"name": "Ann Lee, MD"}], "2026-10-21", {}))) - module.fetch("ctgov-officials") + plant_corpus(monkeypatch, module, "ctgov-officials", [{"name": "Ann Lee, MD"}], + "2026-10-21", "officials") folder = tmp_path / "local" / "ctgov-officials" meta = json.loads((folder / "meta.json").read_text()) if damage == "meta key": diff --git a/tools/differential/README.md b/tools/differential/README.md index 78c585be..35e86d27 100644 --- a/tools/differential/README.md +++ b/tools/differential/README.md @@ -260,7 +260,7 @@ uv run python tools/differential/snapshot.py --local ctgov-officials | `ukec-individuals` | the distinct names of the individual donors in the UK Electoral Commission's register of donations (to parties, their members, third parties, permitted participants and regulated donees), from the search's CSV export under the site's own default filter, fetched whole on every fetch (9.5 MB on 2026-10-22, accepted dates from 2001 to 2026). The default leaves out donations reported only before a poll (`prePoll=false`: 374 rows and 21 names on 2026-10-22). A name as the donee's return gave it, mostly in its spoken order and mixed case, with the donor's title: of its 15,214 names on 2026-10-22, `parse()` read a title in 8,219 (its `title` field non-empty) (`Mr`, `Ms`, `Mrs`, `Lord`, `Dr`, `Sir`; 32 a run of two or more, `The Rt Hon`) and a suffix in 320 (`MP`, `OBE`, `CBE`, `QC`). The population for titled British names. The Commission publishes its documents under the Open Government Licence v3.0, but the register states no reuse terms of its own, which the credit line says | the family-comma shape (2 names hold a comma) and case (9 are in capitals); small donors, since a donation is reported only above a threshold that depends on the donee; names outside the UK; names in any script but Latin | | `psc-individuals` | the people with significant control of UK companies, and the beneficial owners of overseas entities, from Companies House's daily snapshot (33 zipped parts of JSON lines, 2.44 GB on 2026-10-22): each person's display name, the four boxes the filer typed it into (title, forename, middle name, surname) and the stated nationality, nothing else. A record is a distinct filed name, kept one in eight by a hash of the record (1,040,708 of 8,322,939 on 2026-10-22), a size the machinery can hold in memory. The one local corpus whose names carry their fields, scored by `local_agreement.py` | names nobody filed into boxes, so almost no comma (838 of 1,040,708 on 2026-10-22), no credential run and almost no suffix; a display name that is not its boxes joined, which the fetch drops (2 of 14,025,738 individuals on 2026-10-22) | -A live source has no fixed snapshot to pin, so the cached copy is the pin: `meta.json` holds the source's data date and the sha256 of `raw.jsonl` and `corpus.jsonl`, checked on every load, and every report, the `name_usage.py` row included, carries the source and that date. Quote a count with the date, and expect a later fetch to differ. A local snapshot records which fetch it was read from, and `snapshot.py --local` refuses to compare it with a corpus fetched since, which would report the names the source gained and lost as moves: after a `fetch --refresh`, record again from the base checkout. ClinicalTrials.gov's terms ask anyone redistributing its data to credit it, show NLM's processing date and say what was changed; keeping the names in the cache and citing counts with that header meets them. The FEC's bar on soliciting and commercial use is in the credit line of every `fec-individuals` report, so a count quoted from it carries the bar with it. A record there is a distinct name, not a contribution or a contributor: one donor gives many times (34,264,742 rows held 1,893,964 names in the file dated 2026-10-18), and counting contributions would weight a shape by how often its bearers give. `tools/name_usage.py` looks a word up in each corpus's word index, cached as `wordindex.json` beside it on first use and keyed to the records' sha256 and to the code that cuts and folds a word, so a refetch or a change to the fold rebuilds it; a single-word lookup it answers parses neither the records nor the names. Measured 2026-10-22 with both corpora fetched, a single-word `name_usage.py --no-wikidata` run took 2.5s with their indexes cached (12.5s before the cache, 14.8s on the run that builds it). A spaced word (`Ph. D.`) still scans every name, which took 8.6s, nearly all of it the FEC corpus's. `fetch --refresh` re-downloads the FEC's zip rather than re-reading the cached one, replacing it only once the new one is whole, and a corpus fetched for another cycle than `FEC_CYCLE` is refused on load. The `fec-individuals` fetch refuses a file whose lines stop matching the 21 columns it reads, or whose names stop holding a comma, rather than read it as a smaller or a different corpus. A `ukec-individuals` record is a distinct name too, not a donation or a `DonorId`: one name often holds several IDs, which seem to come with reports rather than with people (15,214 names held 24,473 on 2026-10-22, the IDs counted in `meta.json`'s detail; 2,663 names held several at one `RegulatedEntityId` alone, counted from the export and not kept, though some of those may be two people of one name), so a count of IDs would weight a name by how often it was reported. Its fetch reads the register's total from the JSON search under the same filter before and after the export, and refuses an export holding another number of rows, or a register that changed in between: the export ignores `rows=`, so the total is what tells a cut or capped export from the whole register. It refuses an export whose last row is cut short (Python's `csv` returns a body cut inside a row as a short last row, which the row count would otherwise take for a whole one), and a header other than the 29 columns `UKEC_COLUMNS` names, and counts into `meta.json`'s detail the rows it skips: rows not 29 fields, rows from a donor that is not an individual (the filter asks the server to leave those out), and individuals with no name. Past `UKEC_MAX_BAD` (1 in 10,000) of either of the first two, or `UKEC_MAX_NAMELESS` (1 in 1,000; 4 of 42,355 rows were nameless on 2026-10-22) of the third, it refuses the export. The data date is when the first search answered, in UTC: the register states no date of its own, so the counts above, dated by the day they were fetched (2026-10-22), carry the data date 2026-10-23T03:13:19Z in their header. `tw-donors` keeps one zip per report under `reports/`, whole, its masked ID, address and phone columns included, as the FEC's zip is kept: the platform's download answers a report that does not exist and a fault alike, with a 500 and an incident number, so the fetch reads which reports exist from the platform's listing and refuses a listed report it cannot download. It reads each file's columns by name, since a 2020 report lacks a column the newest ones write, and refuses a column it does not know, a category it does not know, a row whose serial column names another report (the zip is not the one asked for), a listing that changes during the fetch, and a report more than half of whose rows it reads past. It reads back the rows a stray quote took into one field (169 rows in 12 records of three reports on 2026-10-22, 8 of them two rows joined by two stray quotes), counts donor cells that hold no name (匿名, a dash) or a masked one, and caches a downloaded zip only once its rows are read, so a refused zip is asked for again. A zip the platform rebuilt under the name it had is not seen: delete `reports/` to download every report again. A layout check written from one surveyed file meets every other file only at fetch time: `tw-donors`' first three fetches were each refused partway, by a report with no income writing one row of dashes, by a 2020 report with 24 columns where the surveyed one had 25, and by a row with no category. Each refusal cost every download again until the source cached its zips, so a new source that downloads many files caches each one before its checks run. `tse-donors` caches nothing but the corpus: each election's national member (`receitas_candidatos__BRASIL.csv`) is read from the TSE's CDN in HTTP Range requests, 209 MB for the two elections where their zips are 1.82 GB (2026-10-22), because the TSE regenerates the zips every few days and a cached one would be stale within the week. Every range request carries `If-Range` with the ETag the HEAD gave if it is strong, else its `Last-Modified`, and every part must state the `Content-Range` asked for, so a zip replaced during the read is refused rather than read across two versions; a HEAD that states no length, neither a strong ETag nor a `Last-Modified`, or `Accept-Ranges` other than `bytes` is refused before the read. A fetch took 69s on 2026-10-22. It refuses a zip without the national member, a header that is not 60 columns or lacks its three columns, a `DS_ORIGEM_RECEITA` value it does not know (a renamed category, or the file no longer in Latin-1), malformed rows or lines folded into a field past `TSE_MAX_BAD`, an election with too few rows from individuals (`TSE_MIN_INDIVIDUALS`: a member cut to its header, or the donations moved under another category) too many without a name or registry name (`TSE_MAX_MISSING`, counting a placeholder such as `#NULO` or `***` as none: a masked column) or with a number in one (`TSE_MAX_NUMBERED`: a numeric column moved under a name), too few distinct filed or registry names (`TSE_MIN_DISTINCT`: a column masked with one word), and a corpus fetched for other elections than `TSE_ELECTIONS`. The CPF in each row is unmasked in the file, and like every other column it is read past, never stored or printed; a pair either of whose names holds more than 4 digits is dropped and counted, since campaigns sometimes typed a number, a CPF among them, into the name field. A record is a pair, so the counts in its reports (`tally`, the `name_usage.py` row) are of pairs, and a filed name standing beside two registry names counts twice. +A live source has no fixed snapshot to pin, so the cached copy is the pin: `meta.json` holds the source's data date and the sha256 of `raw.jsonl` and `corpus.jsonl`, checked on every load, and every report, the `name_usage.py` row included, carries the source and that date. Quote a count with the date, and expect a later fetch to differ. A local snapshot records which fetch it was read from, and `snapshot.py --local` refuses to compare it with a corpus fetched since, which would report the names the source gained and lost as moves: after a `fetch --refresh`, record again from the base checkout. ClinicalTrials.gov's terms ask anyone redistributing its data to credit it, show NLM's processing date and say what was changed; keeping the names in the cache and citing counts with that header meets them. The FEC's bar on soliciting and commercial use is in the credit line of every `fec-individuals` report, so a count quoted from it carries the bar with it. A record there is a distinct name, not a contribution or a contributor: one donor gives many times (34,264,742 rows held 1,893,964 names in the file dated 2026-10-18), and counting contributions would weight a shape by how often its bearers give. `tools/name_usage.py` looks a word up in each corpus's word index, cached as `wordindex.json` beside it on first use and keyed to the records' sha256 and to the code that cuts and folds a word, so a refetch or a change to the fold rebuilds it; a single-word lookup it answers parses neither the records nor the names. Measured 2026-10-22 with both corpora fetched, a single-word `name_usage.py --no-wikidata` run took 2.5s with their indexes cached (12.5s before the cache, 14.8s on the run that builds it). A spaced word (`Ph. D.`) still scans every name, which took 8.6s, nearly all of it the FEC corpus's. `fetch --refresh` re-downloads the FEC's zip rather than re-reading the cached one, replacing it only once the new one is whole, and a corpus fetched for another cycle than `FEC_CYCLE` is refused on load. The `fec-individuals` fetch refuses a file whose lines stop matching the 21 columns it reads, or whose names stop holding a comma, rather than read it as a smaller or a different corpus. A `ukec-individuals` record is a distinct name too, not a donation or a `DonorId`: one name often holds several IDs, which seem to come with reports rather than with people (15,214 names held 24,473 on 2026-10-22, the IDs counted in `meta.json`'s detail; 2,663 names held several at one `RegulatedEntityId` alone, counted from the export and not kept, though some of those may be two people of one name), so a count of IDs would weight a name by how often it was reported. Its fetch reads the register's total from the JSON search under the same filter before and after the export, and refuses an export holding another number of rows, or a register that changed in between: the export ignores `rows=`, so the total is what tells a cut or capped export from the whole register. It refuses an export whose last row is cut short (Python's `csv` returns a body cut inside a row as a short last row, which the row count would otherwise take for a whole one), and a header other than the 29 columns `UKEC_COLUMNS` names, and counts into `meta.json`'s detail the rows it skips: rows not 29 fields, rows from a donor that is not an individual (the filter asks the server to leave those out), and individuals with no name. Past `UKEC_MAX_BAD` of either of the first two, or `UKEC_MAX_NAMELESS` of the third (4 of 42,355 rows were nameless on 2026-10-22), it refuses the export. The data date is when the first search answered, in UTC: the register states no date of its own, so the counts above, dated by the day they were fetched (2026-10-22), carry the data date 2026-10-23T03:13:19Z in their header. `tw-donors` keeps one zip per report under `reports/`, whole, its masked ID, address and phone columns included, as the FEC's zip is kept: the platform's download answers a report that does not exist and a fault alike, with a 500 and an incident number, so the fetch reads which reports exist from the platform's listing and refuses a listed report it cannot download. It reads each file's columns by name, since a 2020 report lacks a column the newest ones write, and refuses a column it does not know, a category it does not know, a row whose serial column names another report (the zip is not the one asked for), a listing that changes during the fetch, and a report past `TW_MAX_REPORT_BAD` of whose rows it reads past. It reads back the rows a stray quote took into one field (169 rows in 12 records of three reports on 2026-10-22, 8 of them two rows joined by two stray quotes), counts donor cells that hold no name (匿名, a dash) or a masked one, and caches a downloaded zip only once its rows are read, so a refused zip is asked for again. A zip the platform rebuilt under the name it had is not seen: delete `reports/` to download every report again. A layout check written from one surveyed file meets every other file only at fetch time: `tw-donors`' first three fetches were each refused partway, by a report with no income writing one row of dashes, by a 2020 report with 24 columns where the surveyed one had 25, and by a row with no category. Each refusal cost every download again until the source cached its zips, so a new source that downloads many files caches each one before its checks run. `tse-donors` caches nothing but the corpus: each election's national member (`receitas_candidatos__BRASIL.csv`) is read from the TSE's CDN in HTTP Range requests, 209 MB for the two elections where their zips are 1.82 GB (2026-10-22), because the TSE regenerates the zips every few days and a cached one would be stale within the week. Every range request carries `If-Range` with the ETag the HEAD gave if it is strong, else its `Last-Modified`, and every part must state the `Content-Range` asked for, so a zip replaced during the read is refused rather than read across two versions; a HEAD that states no length, neither a strong ETag nor a `Last-Modified`, or `Accept-Ranges` other than `bytes` is refused before the read. A fetch took 69s on 2026-10-22. It refuses a zip without the national member, a header that is not 60 columns or lacks its three columns, a `DS_ORIGEM_RECEITA` value it does not know (a renamed category, or the file no longer in Latin-1), malformed rows or lines folded into a field past `TSE_MAX_BAD`, an election with too few rows from individuals (`TSE_MIN_INDIVIDUALS`: a member cut to its header, or the donations moved under another category) too many without a name or registry name (`TSE_MAX_MISSING`, counting a placeholder such as `#NULO` or `***` as none: a masked column) or with a number in one (`TSE_MAX_NUMBERED`: a numeric column moved under a name), too few distinct filed or registry names (`TSE_MIN_DISTINCT`: a column masked with one word), and a corpus fetched for other elections than `TSE_ELECTIONS`. The CPF in each row is unmasked in the file, and like every other column it is read past, never stored or printed; a pair either of whose names holds more than `TSE_MAX_DIGITS` digits is dropped and counted, since campaigns sometimes typed a number, a CPF among them, into the name field. A record is a pair, so the counts in its reports (`tally`, the `name_usage.py` row) are of pairs, and a filed name standing beside two registry names counts twice. `psc-individuals` caches an extract of each of the snapshot's 33 parts rather than the zips: 168 MB against 2.44 GB on 2026-10-22, each extract holding only the records and fields the corpus keeps. A download that fails costs one part, and running the fetch again resumes, keeping the parts already read for the same day; the set replaces the cached one only once every part is read, and a part that disappears mid-fetch (the files are overwritten each morning) is refused naming the refresh. The fetch refuses a snapshot where more than one individual in 10,000 has no `name_elements`, or a display name that is not its four boxes joined by spaces, or where fewer than half of the lines are individuals (87% were on 2026-10-22; a kind Companies House renamed would read as another kind), and counts what it skips into `meta.json`'s detail. It reads the cached extracts only as one whole snapshot of one date, and an extract written by another version of the extraction (its file name carries the version) is neither resumed from nor read. Companies House's data-products guidance says it imposes no rules on how the register is used and leaves data protection to the user; the dataset's data.gov.uk entry gives its licence as "TBD", and gov.uk's footer licenses its own page text under the Open Government Licence. The credit line says the names are personal data. diff --git a/tools/differential/local_agreement.py b/tools/differential/local_agreement.py index 3ca5e461..db37ae74 100644 --- a/tools/differential/local_agreement.py +++ b/tools/differential/local_agreement.py @@ -48,12 +48,11 @@ from __future__ import annotations import argparse -import hashlib import json import sys import types from collections import Counter -from collections.abc import Iterator +from collections.abc import Callable, Iterator from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any @@ -157,28 +156,11 @@ def group_of(record: dict[str, str], fielded: Fielded) -> str: def records(local: local_corpora.Local) -> Iterator[dict[str, str]]: - """raw.jsonl's records, streamed rather than held, checked against - its sha256 once the last is read, and each holding a name and - nothing but strings as it is read: CorpusDamaged either way, before any caller reports or - records what it read.""" - path = local_corpora.directory(local.source, create=False) / local_corpora.RAW - damaged = local_corpora.CorpusDamaged( - f"{path} does not match the sha256 in {local_corpora.META}; fetch it " - f"again: {local_corpora.fetch_command(local.source, refresh=True)}") - digest = hashlib.sha256() - with path.open("rb") as raw: - for line in raw: - digest.update(line) - try: - record = json.loads(line) - except ValueError: - raise damaged from None - if (not isinstance(record, dict) or not isinstance(record.get("name"), str) - or not all(isinstance(v, str) for v in record.values())): - raise damaged - yield record - if digest.hexdigest() != local.meta["sha256"]: - raise damaged + """raw.jsonl's records, every one read and checked before the first + is handed out (local_corpora.read_records), each nothing but + strings: CorpusDamaged before any caller reports or records what it + read.""" + return iter(local_corpora.read_records(local.source, local.meta, text_only=True)) @dataclass @@ -299,20 +281,19 @@ def read_record(path: Path, local: local_corpora.Local) -> dict[str, dict[str, A def scorer_fingerprint() -> str: - """A hash of the code that turns a parse into a record's entry and - decides which records get one (_SCORER_CODE), of - corpus_build.jsonl_line, which writes an entry's key, and of FIELDED, - read from the running code.""" + """local_corpora.code_fingerprint of the code that turns a parse into + a record's entry and decides which records get one (_SCORER_CODE, a + class by its methods), of corpus_build.jsonl_line, which writes an + entry's key, and of FIELDED.""" module = globals() - parts = [repr(sorted((k, repr(v)) for k, v in FIELDED.items()))] + functions: list[Callable[..., object]] = [] for name in _SCORER_CODE: target = module[name] - functions = ([v for v in vars(target).values() if isinstance(v, types.FunctionType)] - if isinstance(target, type) else [target]) - for fn in functions: - parts += local_corpora._code_digest(fn.__code__) - parts += local_corpora._code_digest(corpus_build.jsonl_line.__code__) - return hashlib.sha256("\0".join(parts).encode()).hexdigest() + functions += ([v for v in vars(target).values() if isinstance(v, types.FunctionType)] + if isinstance(target, type) else [target]) + return local_corpora.code_fingerprint( + [*functions, corpus_build.jsonl_line], + [repr(sorted((k, repr(v)) for k, v in FIELDED.items()))]) def _record_command(source: str) -> str: diff --git a/tools/differential/local_corpora.py b/tools/differential/local_corpora.py index f7dcdcd1..c9d11621 100644 --- a/tools/differential/local_corpora.py +++ b/tools/differential/local_corpora.py @@ -165,7 +165,7 @@ import zipfile import zlib from collections import Counter -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterable, Iterator from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, TypeVar @@ -273,7 +273,10 @@ def index(self) -> WordIndex: CTGOV_API = "https://clinicaltrials.gov/api/v2" CTGOV_PAGE = 1000 # the API's largest page CTGOV_FIELDS = "NCTId,OverallOfficialName,OverallOfficialRole" -CTGOV_RETRIES = 4 +#: Attempts per request, for every source that retries. +RETRIES = 4 +#: How much of a streamed download is read at a time. +CHUNK = 1 << 20 #: What a long run of requests meets and a retry survives: a refused or #: dropped connection or a timeout (OSError, which URLError and HTTPError #: are), a body cut short in the read (http.client's IncompleteRead), a @@ -285,7 +288,7 @@ def index(self) -> WordIndex: _T = TypeVar("_T") -class _Unparsed(Exception): +class _Unparsed(ValueError): """A response body that is not JSON: cut short in transit.""" @@ -299,33 +302,71 @@ def _get_json(url: str) -> dict[str, Any]: return data -def _retried(url: str, read: Callable[[bytes], _T], delay: float) -> _T: - """`url`'s body as `read` reads it, retried with a backoff and a line - on stderr for each retry; any ValueError from `read` is retried as a - body cut short in transit, and `delay` seconds of sleep follow each - request. A 4xx other than 429 is the request's fault and is raised, - as is the last attempt's error.""" - for attempt in range(CTGOV_RETRIES): +def _zip_stamp(info: zipfile.ZipInfo) -> str: + """A zip member's own timestamp, as a data date: the sources that + ship a zip state no other date for what is in it.""" + return "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(*info.date_time) + + +def _pinned(key: str, wanted: Callable[[], str], says: str + ) -> Callable[[dict[str, Any]], str]: + """A Source.check refusing a corpus whose meta.json detail `key` is + not `wanted()` now, read at check time so that a change to the + constant behind it refuses corpora fetched before; `says` words the + refusal, with {held} and {wanted}.""" + def check(meta: dict[str, Any]) -> str: + held, now = meta["detail"].get(key), wanted() + return "" if held == now else says.format(held=held, wanted=now) + return check + + +def _refuse_past(rows: Iterable[tuple[int, int, float, str]], why: str) -> None: + """The gate every source puts on what it reads past: for each + (count, of, share, what), refuse with "count of `of` what: why" once + count is more than `share` of `of`. A file whose layout moved under + a reader would otherwise read as a smaller corpus.""" + for count, of, share, what in rows: + if count > of * share: + raise ValueError(f"{count:,} of {of:,} {what}: {why}") + + +def _backed_off(exc: Exception, attempt: int, url: str) -> bool: + """The retry policy every source shares, after `exc` on try `attempt` + of `url`: False where the caller should raise it (a 4xx other than + 429 is the request's fault, and the last attempt's error is the + fetch's), else True once a backoff of 2**attempt seconds, or the + Retry-After a server asks for up to a minute, has been slept, with a + line on stderr.""" + if (isinstance(exc, urllib.error.HTTPError) and exc.code < 500 + and exc.code != 429 or attempt == RETRIES - 1): + return False + wait = 2 ** attempt + if isinstance(exc, urllib.error.HTTPError): + after = exc.headers.get("Retry-After", "") if exc.headers else "" + if after.isdigit(): + wait = max(wait, min(int(after), 60)) + exc.close() # an HTTPError holds the response open + print(f" retrying in {wait}s after {type(exc).__name__}: {url}", file=sys.stderr) + time.sleep(wait) + return True + + +def _retried(url: str, read: Callable[[bytes], _T], delay: float, *, + timeout: int = 60) -> _T: + """`url`'s body as `read` reads it, under _backed_off's policy; any + ValueError from `read` is retried as a body cut short in transit, and + `delay` seconds of sleep follow each request.""" + for attempt in range(RETRIES): try: body = corpus_build.fetch_bytes(url, user_agent=USER_AGENT, - timeout=60, delay=delay) + timeout=timeout, delay=delay) try: return read(body) except ValueError as exc: raise _Unparsed(f"{url}: {exc}") from None except (*_TRANSIENT, _Unparsed) as exc: - if (isinstance(exc, urllib.error.HTTPError) and exc.code < 500 - and exc.code != 429 or attempt == CTGOV_RETRIES - 1): + if not _backed_off(exc, attempt, url): raise - wait = 2 ** attempt - if isinstance(exc, urllib.error.HTTPError): - after = exc.headers.get("Retry-After", "") if exc.headers else "" - if after.isdigit(): - wait = max(wait, min(int(after), 60)) - exc.close() # an HTTPError holds the response open - print(f" retrying in {wait}s after {type(exc).__name__}: {url}", - file=sys.stderr) - time.sleep(wait) raise AssertionError("unreachable: the last attempt raises") @@ -405,7 +446,6 @@ def fetch_ctgov() -> Fetched: #: The file's columns (the FEC's contributions-by-individuals file #: description): 21, pipe-separated, ENTITY_TP the 7th and NAME the 8th. FEC_COLUMNS, FEC_ENTITY, FEC_NAME = 21, 6, 7 -FEC_CHUNK = 1 << 20 #: The file has no header row to check the columns against, so a fetch #: refuses one that no longer looks like the file the constants above #: were written for: more than this share of its lines malformed or not @@ -431,7 +471,7 @@ def _download(url: str, path: Path) -> None: with urllib.request.urlopen(request, timeout=60) as response, \ partial.open("wb") as out: stated = response.headers.get("Content-Length", "") - while chunk := response.read(FEC_CHUNK): + while chunk := response.read(CHUNK): out.write(chunk) got += len(chunk) if not stated.isdigit() or int(stated) != got: @@ -460,7 +500,7 @@ def _fec_read(path: Path) -> Fetched: if FEC_MEMBER not in archive.namelist(): raise ValueError(f"{path} holds no {FEC_MEMBER}") member = archive.getinfo(FEC_MEMBER) - stamp = "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(*member.date_time) + stamp = _zip_stamp(member) with archive.open(member) as lines: for line in lines: rows += 1 @@ -480,11 +520,9 @@ def _fec_read(path: Path) -> Fetched: names.add(name) else: nameless += 1 - for count, what in ((malformed, f"not {FEC_COLUMNS} columns"), - (latin1, "not UTF-8")): - if count > rows * FEC_MAX_BAD: - raise ValueError(f"{count:,} of {rows:,} lines of {path} are {what}: " - f"the file's layout has changed under FEC_COLUMNS") + _refuse_past([(malformed, rows, FEC_MAX_BAD, f"lines of {path} are not {FEC_COLUMNS} columns"), + (latin1, rows, FEC_MAX_BAD, f"lines of {path} are not UTF-8")], + "the file's layout has changed under FEC_COLUMNS") with_comma = sum("," in name for name in names) if names and with_comma < len(names) * FEC_MIN_COMMA: raise ValueError(f"{with_comma:,} of {len(names):,} names hold a comma: " @@ -529,10 +567,8 @@ def renew_fec() -> None: _fec_download(directory("fec-individuals") / FEC_ZIP) -def check_fec(meta: dict[str, Any]) -> str: - cycle = meta["detail"].get("cycle") - return "" if cycle == FEC_CYCLE else ( - f"holds the {cycle} cycle, and this tool reads the {FEC_CYCLE} cycle") +check_fec = _pinned("cycle", lambda: FEC_CYCLE, + "holds the {held} cycle, and this tool reads the {wanted} cycle") # -- TSE campaign donations (Brazil) ------------------------------------------- @@ -705,7 +741,7 @@ def _request(self, request: urllib.request.Request, want: int | None = None up to one byte past `want` so that a longer part shows; a shorter one is retried as a body cut short. The last attempt's error, or any 4xx but a 429, is kept in self.error and raised.""" - for attempt in range(CTGOV_RETRIES): + for attempt in range(RETRIES): try: with urllib.request.urlopen(request, timeout=60) as response: body = b"" @@ -715,20 +751,10 @@ def _request(self, request: urllib.request.Request, want: int | None = None raise http.client.IncompleteRead(body, want - len(body)) return response.status, response.headers, body except _TRANSIENT as exc: - if (isinstance(exc, urllib.error.HTTPError) and exc.code < 500 - and exc.code != 429 or attempt == CTGOV_RETRIES - 1): + if not _backed_off(exc, attempt, self.url): self.error = exc raise - wait = 2 ** attempt - if isinstance(exc, urllib.error.HTTPError): - after = exc.headers.get("Retry-After", "") if exc.headers else "" - if after.isdigit(): - wait = max(wait, min(int(after), 60)) - exc.close() # an HTTPError holds the response open - print(f" retrying in {wait}s after {type(exc).__name__}: " - f"{self.url}", file=sys.stderr) - time.sleep(wait) - raise AssertionError("unreachable") + raise AssertionError("unreachable: the last attempt raises") def _tse_read(year: str, archive: zipfile.ZipFile, @@ -739,7 +765,7 @@ def _tse_read(year: str, archive: zipfile.ZipFile, if member not in archive.namelist(): raise ValueError(f"{TSE_URL.format(year=year)} holds no {member}") info = archive.getinfo(member) - stamp = "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(*info.date_time) + stamp = _zip_stamp(info) counts: Counter[str] = Counter() before = len(pairs) distinct: set[tuple[str, str]] = set() @@ -802,14 +828,13 @@ def _tse_read(year: str, archive: zipfile.ZipFile, raise ValueError(f"{counts['individual']:,} of {counts['rows']:,} rows of " f"{member} are from individuals: the file is cut short or " f"its categories have moved") - for count, limit, what in ( - (counts["nameless"], TSE_MAX_MISSING, "no name"), - (counts["no_registry"], TSE_MAX_MISSING, "no registry name"), - (counts["numbered"], TSE_MAX_NUMBERED, "a number in a name")): - if count > counts["individual"] * limit: - raise ValueError(f"{count:,} of {counts['individual']:,} individuals' rows " - f"of {member} carry {what}: a name column may be " - f"masked or moved") + carry = f"individuals' rows of {member} carry" + _refuse_past([(counts["nameless"], counts["individual"], TSE_MAX_MISSING, f"{carry} no name"), + (counts["no_registry"], counts["individual"], TSE_MAX_MISSING, + f"{carry} no registry name"), + (counts["numbered"], counts["individual"], TSE_MAX_NUMBERED, + f"{carry} a number in a name")], + "a name column may be masked or moved") for column, index in ((TSE_NAME, 0), (TSE_REGISTRY, 1)): seen = len({pair[index] for pair in distinct}) if seen < counts["individual"] * TSE_MIN_DISTINCT: @@ -887,11 +912,8 @@ def fetch_tse() -> Fetched: for name, registry in sorted(pairs)], ", ".join(stamps), detail) -def check_tse(meta: dict[str, Any]) -> str: - elections = meta["detail"].get("elections") - wanted = " ".join(TSE_ELECTIONS) - return "" if elections == wanted else ( - f"holds the elections {elections}, and this tool reads {wanted}") +check_tse = _pinned("elections", lambda: " ".join(TSE_ELECTIONS), + "holds the elections {held}, and this tool reads {wanted}") # -- Taiwan political donations ------------------------------------------------ @@ -1022,7 +1044,7 @@ def _tw_incomes(body: bytes) -> tuple[bytes, str] | None: if TW_INCOMES not in archive.namelist(): return None member = archive.getinfo(TW_INCOMES) - stamp = "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}".format(*member.date_time) + stamp = _zip_stamp(member) return archive.read(member), stamp except (zipfile.BadZipFile, zlib.error, EOFError) as exc: start = body[:2] @@ -1037,11 +1059,6 @@ def _tw_incomes(body: bytes) -> tuple[bytes, str] | None: _TW_ZIP_NAME = re.compile(r"[\w-]+\.zip", re.ASCII) -def _tw_whole(body: bytes) -> bytes: - _tw_incomes(body) # raises ValueError where the zip is cut short - return body - - def _tw_cached(report: dict[str, Any]) -> Path: return directory("tw-donors") / TW_REPORTS / report["zipFileName"] @@ -1069,8 +1086,10 @@ def _tw_report_zip(report: dict[str, Any], except ValueError: print(f" downloading again a cached zip that is not whole: {path}", file=sys.stderr) - body = _retried(_tw_url(report), _tw_whole, TW_DELAY) - if (got := _tw_incomes(body)) is None: + # Read inside _retried, so that a zip cut short is retried, and once. + body, got = _retried(_tw_url(report), lambda body: (body, _tw_incomes(body)), + TW_DELAY) + if got is None: raise ValueError(f"{label}'s zip holds no {TW_INCOMES}: the reports' " f"layout has changed") return (*got, body) @@ -1234,16 +1253,16 @@ def fetch_tw() -> Fetched: _tw_read(incomes, report, label, names, counts) if downloaded is not None: _tw_cache_zip(report, downloaded) - for key, total, what in ( - ("rows_not_as_long_as_the_header", "rows", - "rows are not as long as their file's header"), - ("rows_without_a_category", "rows", "rows are without a category"), - ("rows_cut_by_a_stray_quote", "rows", "rows are cut by a stray quote"), - ("donations_naming_a_masked_donor", "individual_donations", - "individual donations name a masked donor")): - if counts[key] > counts[total] * TW_MAX_BAD: - raise ValueError(f"{counts[key]:,} of {counts[total]:,} {what}: the " - f"files' layout has changed") + rows, donations = counts["rows"], counts["individual_donations"] + _refuse_past([(counts["rows_not_as_long_as_the_header"], rows, TW_MAX_BAD, + "rows are not as long as their file's header"), + (counts["rows_without_a_category"], rows, TW_MAX_BAD, + "rows are without a category"), + (counts["rows_cut_by_a_stray_quote"], rows, TW_MAX_BAD, + "rows are cut by a stray quote"), + (counts["donations_naming_a_masked_donor"], donations, TW_MAX_BAD, + "individual donations name a masked donor")], + "the files' layout has changed") if _tw_versions(_tw_reports()) != _tw_versions(reports): raise ValueError("the platform published, replaced or withdrew a report " "during the fetch; fetch again") @@ -1309,9 +1328,8 @@ def fetch_tw() -> Fetched: def _ukec_total() -> tuple[int, str]: """The register's total for the filter, and the time the search answered, from the JSON search.""" - body = corpus_build.fetch_bytes(UKEC_COUNT_URL, user_agent=USER_AGENT, - timeout=60) - data = corpus_build.parse_json(body, UKEC_COUNT_URL) + data = _retried(UKEC_COUNT_URL, + lambda body: corpus_build.parse_json(body, UKEC_COUNT_URL), 0) if not isinstance(data, dict): raise ValueError(f"{UKEC_COUNT_URL} states no Total") total = data.get("Total") @@ -1373,13 +1391,12 @@ def _ukec_rows(rows: list[list[str]], text: str, total: int) -> Fetched: if count != total: raise ValueError(f"the export held {count:,} donations where the search " f"counts {total:,}: a capped or cut export; fetch again") - for skipped, share, what in ( - (malformed, UKEC_MAX_BAD, f"not {len(UKEC_COLUMNS)} fields"), - (other, UKEC_MAX_BAD, f"from a donor that is not {UKEC_INDIVIDUAL!r}"), - (nameless, UKEC_MAX_NAMELESS, "from an individual with no name")): - if skipped > count * share: - raise ValueError(f"{skipped:,} of {count:,} rows of the export are " - f"{what}: the export is not the one this reads") + are = "rows of the export are" + _refuse_past([(malformed, count, UKEC_MAX_BAD, f"{are} not {len(UKEC_COLUMNS)} fields"), + (other, count, UKEC_MAX_BAD, + f"{are} from a donor that is not {UKEC_INDIVIDUAL!r}"), + (nameless, count, UKEC_MAX_NAMELESS, f"{are} from an individual with no name")], + "the export is not the one this reads") return Fetched([{"name": name} for name in sorted(names)], "", { "donations": count, "donor_ids": len(donor_ids), "individuals_without_a_name": nameless, @@ -1407,7 +1424,7 @@ def fetch_ukec() -> Fetched: two, or UKEC_MAX_NAMELESS of the third, the export is refused.""" before, when = _ukec_total() print(f"downloading {before:,} donations from {UKEC_API} ...", file=sys.stderr) - body = corpus_build.fetch_bytes(UKEC_CSV_URL, user_agent=USER_AGENT, timeout=300) + body = _retried(UKEC_CSV_URL, lambda body: body, 0, timeout=300) after, _ = _ukec_total() if after != before: raise ValueError(f"the register changed during the fetch ({before:,} -> " @@ -1519,18 +1536,25 @@ def _psc_extract(path: Path, member: str) -> bytes: return gzip.compress("".join(f"{line}\n" for line in [header, *sorted(lines)]).encode()) +def _one_snapshot(dates: set[str], totals: set[int], numbers: list[int]) -> bool: + """Whether parts `numbers`, dated `dates` and each one of `totals`, + are every part of one snapshot.""" + return (len(dates) == 1 and len(totals) == 1 + and sorted(numbers) == list(range(1, max(totals) + 1))) + + def _psc_listing() -> tuple[str, list[tuple[int, str]]]: """The day's snapshot as the download page lists it: its date and each part's number and file, or ValueError where the page links no parts, or parts of more than one snapshot, or not every part. A part linked twice is one part.""" - page = corpus_build.fetch_bytes(PSC_PAGE, user_agent=USER_AGENT, timeout=60) + page = _retried(PSC_PAGE, lambda body: body, 0) links = PSC_PART.findall(page.decode("utf-8", "replace")) if not links: raise ValueError(f"{PSC_PAGE} links no snapshot parts: the page has changed") dates, totals = {d for _, d, _, _ in links}, {int(m) for *_, m in links} numbers = sorted(int(n) for _, _, n, _ in set(links)) - if len(dates) != 1 or len(totals) != 1 or numbers != list(range(1, max(totals) + 1)): + if not _one_snapshot(dates, totals, numbers): raise ValueError(f"{PSC_PAGE} links parts {numbers} of {sorted(totals)} " f"dated {sorted(dates)}, not every part of one snapshot; " f"fetch again") @@ -1626,8 +1650,7 @@ def _psc_read(parts: Path) -> Fetched: found[number] = path dates.add(named[1]) totals.add(int(named[3])) - if (len(dates) != 1 or len(totals) != 1 - or sorted(found) != list(range(1, max(totals) + 1))): + if not _one_snapshot(dates, totals, list(found)): raise ValueError(f"{parts} holds parts {sorted(found)} of {sorted(totals)} " f"dated {sorted(dates)}, not every part of one snapshot" f"{_again(PSC)}") @@ -1642,15 +1665,13 @@ def _psc_read(parts: Path) -> Fetched: raise ValueError(f"{individuals:,} of {counts['lines']:,} lines are " f"{' or '.join(PSC_KINDS)} records: a kind has been " f"renamed or moved under PSC_KINDS") - for key, of, what in ( - ("lines_not_json", counts["lines"], "lines are not JSON"), - ("individuals_without_name_elements", individuals, - "individuals have no name_elements"), - ("names_not_their_boxes_joined", individuals, - "individuals' display names are not their four boxes joined")): - if counts[key] > of * PSC_MAX_BAD: - raise ValueError(f"{counts[key]:,} of {of:,} {what}: the snapshot's " - f"layout has changed under PSC_ELEMENTS") + _refuse_past([(counts["lines_not_json"], counts["lines"], PSC_MAX_BAD, + "lines are not JSON"), + (counts["individuals_without_name_elements"], individuals, PSC_MAX_BAD, + "individuals have no name_elements"), + (counts["names_not_their_boxes_joined"], individuals, PSC_MAX_BAD, + "individuals' display names are not their four boxes joined")], + "the snapshot's layout has changed under PSC_ELEMENTS") records = [json.loads(line) for line in sorted(lines)] named_records = [record for record in records if "name" in record] if len(named_records) < len(records): @@ -1672,10 +1693,8 @@ def fetch_psc() -> Fetched: return _psc_read(parts) -def check_psc(meta: dict[str, Any]) -> str: - sample = meta["detail"].get("sample") - return "" if sample == f"1 in {PSC_SAMPLE}" else ( - f"keeps {sample} filed names, and this tool keeps 1 in {PSC_SAMPLE}") +check_psc = _pinned("sample", lambda: f"1 in {PSC_SAMPLE}", + "keeps {held} filed names, and this tool keeps {wanted}") SOURCES: dict[str, Source] = { @@ -1797,16 +1816,25 @@ def _verified(source: str, name: str, sha256: str) -> bytes: return body -def read_records(source: str, meta: dict[str, Any]) -> list[dict[str, str]]: - """raw.jsonl's records, checked against its sha256 again (load() - checked it, but the file is read later and may have changed), each - holding a name, or CorpusDamaged naming the way out.""" +def read_records(source: str, meta: dict[str, Any], *, + text_only: bool = False) -> list[dict[str, str]]: + """raw.jsonl's records, the file checked against its sha256 first + (load() checked it, but the file is read later and may have changed) + and cut by corpus_build.jsonl_lines, each holding a name, and with + `text_only` nothing but strings; CorpusDamaged naming the way out + otherwise. Read whole: streaming the lines cost more than holding + them (1.99s against 1.15s for fec-individuals, 2026-10-23).""" + path = directory(source, create=False) / RAW raw = _verified(source, RAW, meta["sha256"]) - records = [json.loads(line) for line in corpus_build.jsonl_lines(raw.decode())] - if any(not isinstance(r, dict) or not isinstance(r.get("name"), str) - for r in records): - raise CorpusDamaged(f"{directory(source, create=False) / RAW} holds a record " - f"without a name{_again(source)}") + try: + records = [json.loads(line) for line in corpus_build.jsonl_lines(raw.decode())] + except ValueError: + raise CorpusDamaged(f"{path} holds a line that is not JSON{_again(source)}") from None + if any(not isinstance(r, dict) or not isinstance(r.get("name"), str) for r in records): + raise CorpusDamaged(f"{path} holds a record without a name{_again(source)}") + if text_only and any(not isinstance(v, str) for r in records for v in r.values()): + raise CorpusDamaged(f"{path} holds a record with a value that is not " + f"text{_again(source)}") return records @@ -2021,11 +2049,12 @@ def pair_counts(local: Local, *, agreement: bool = False) -> PairCounts: filed, reference = record["name"], record["registry"] shape = pair_shape(filed, reference) shapes[shape] += 1 - a, b = _pair_words(filed), _pair_words(reference) - if shape == "words left out": - left_out.update((Counter(b) - Counter(a)).keys()) - elif shape == "words added": - added.update((Counter(a) - Counter(b)).keys()) + if shape in ("words left out", "words added"): + a, b = Counter(_pair_words(filed)), Counter(_pair_words(reference)) + if shape == "words left out": + left_out.update((b - a).keys()) + else: + added.update((a - b).keys()) if agreement and shape not in ("the same", "no reference name"): for name in (filed, reference): if name not in seen: @@ -2161,23 +2190,31 @@ def _code_digest(code: types.CodeType) -> list[str]: return parts -def fold_fingerprint() -> str: - """A hash of what turns the records into an index, read from the code - that is running rather than the source on disk, which may have been - edited since: _FOLD_CODE's functions and corpus_build.jsonl_lines - (which cuts raw.jsonl into records), _FOLD_CONSTANTS, WordIndex's - fields, and the Unicode database that casefold(), NFC and the - category test read, which a new Python may move. The bytecode is the - interpreter's own, so a new Python rebuilds the index too.""" - module = globals() - functions = [module[name] for name in _FOLD_CODE] + [corpus_build.jsonl_lines] +def code_fingerprint(functions: Iterable[Callable[..., object]], + extra: Iterable[str] = ()) -> str: + """A hash of `functions` as the interpreter will run them, read from + the running code rather than the source on disk, which may have been + edited since; of `extra`; and of the Unicode database that + casefold(), NFC and the category tests read and the interpreter whose + bytecode it is, which a new Python may move: the key for a result + cached from what that code made.""" parts = [part for fn in functions for part in _code_digest(inspect.unwrap(fn).__code__)] - parts += [_stable(module[name]) for name in _FOLD_CONSTANTS] - parts += [repr([f.name for f in dataclasses.fields(WordIndex)]), - unicodedata.unidata_version, sys.version] + parts += [*extra, unicodedata.unidata_version, sys.version] return hashlib.sha256("\0".join(parts).encode()).hexdigest() +def fold_fingerprint() -> str: + """code_fingerprint of what turns the records into an index: + _FOLD_CODE's functions, read_records and corpus_build.jsonl_lines + (which read raw.jsonl into the records Local.records holds, where + _names takes them), _FOLD_CONSTANTS and WordIndex's fields.""" + module = globals() + return code_fingerprint( + [*(module[name] for name in _FOLD_CODE), read_records, corpus_build.jsonl_lines], + [*(_stable(module[name]) for name in _FOLD_CONSTANTS), + repr([f.name for f in dataclasses.fields(WordIndex)])]) + + def _is_counts(value: object) -> bool: return isinstance(value, dict) and all( type(n) is int for n in value.values())