Multi-source foundation: registry, shared fetch, and CLI#4
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the single-file Washington Attorney General breach scraper into a modular package structure, introducing a shared CLI, HTTP fetcher with backoff, multiple output formatters (JSON, CSV, Markdown), and a source registry, alongside comprehensive unit tests. The review feedback highlights opportunities to improve robustness and code quality, including handling non-numeric ranges in count normalization, validating candidate URLs in the HTTP fetcher, avoiding double carriage returns in CSV generation on Windows, and removing redundant strict-mode error checks in the CLI orchestration.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def _normalize_count(value: str) -> str: | ||
| cleaned = _clean_text(value) | ||
| if not cleaned: | ||
| return "" | ||
| digits = re.sub(r"[^\d]", "", cleaned) | ||
| if not digits: | ||
| return cleaned | ||
| return f"{int(digits):,}" |
There was a problem hiding this comment.
The current implementation of _normalize_count uses re.sub(r"[^\d]", "", cleaned) which strips all non-digits and concatenates them. If the input is a range like "1,200 - 1,500", it will be incorrectly normalized to "12,001,500". A safer approach is to check if the string (with commas removed) is purely numeric before formatting it, and otherwise return it as-is.
| def _normalize_count(value: str) -> str: | |
| cleaned = _clean_text(value) | |
| if not cleaned: | |
| return "" | |
| digits = re.sub(r"[^\d]", "", cleaned) | |
| if not digits: | |
| return cleaned | |
| return f"{int(digits):,}" | |
| def _normalize_count(value: str) -> str: | |
| cleaned = _clean_text(value) | |
| if not cleaned: | |
| return "" | |
| no_commas = cleaned.replace(",", "") | |
| if no_commas.isdigit(): | |
| return f"{int(no_commas):,}" | |
| return cleaned |
There was a problem hiding this comment.
Fixed in 14aeba0. _normalize_count now formats only when the comma-stripped value is all digits, otherwise it returns the original string. Covered by test_normalize_count_keeps_ranges_intact.
| def fetch_url( | ||
| candidates: list[str], | ||
| *, | ||
| timeout: int = 30, | ||
| retries: int = 3, | ||
| backoff: float = 0.5, | ||
| user_agent: str | None = None, | ||
| headers: dict[str, str] | None = None, | ||
| data: bytes | None = None, | ||
| ) -> str: | ||
| """Fetch the first candidate URL that succeeds. | ||
|
|
||
| Each candidate is tried in order; transient failures (timeouts, connection | ||
| errors, HTTP 5xx, protocol errors) are retried with exponential backoff | ||
| (capped at 30s). HTTP 403 raises an actionable error; other 4xx are not | ||
| retried and fall through to the next candidate. | ||
| """ | ||
| request_headers = {"User-Agent": user_agent or DEFAULT_USER_AGENT, **DEFAULT_HEADERS} |
There was a problem hiding this comment.
The fetch_url function does not validate if the candidates list is empty. If an empty list is passed, it will raise a generic RuntimeError: Failed to fetch source page: None which is not very descriptive. Adding an explicit check at the beginning of the function improves robustness.
def fetch_url(
candidates: list[str],
*,
timeout: int = 30,
retries: int = 3,
backoff: float = 0.5,
user_agent: str | None = None,
headers: dict[str, str] | None = None,
data: bytes | None = None,
) -> str:
"""Fetch the first candidate URL that succeeds.
Each candidate is tried in order; transient failures (timeouts, connection
errors, HTTP 5xx, protocol errors) are retried with exponential backoff
(capped at 30s). HTTP 403 raises an actionable error; other 4xx are not
retried and fall through to the next candidate.
"""
if not candidates:
raise ValueError("At least one candidate URL must be provided.")
request_headers = {"User-Agent": user_agent or DEFAULT_USER_AGENT, **DEFAULT_HEADERS}There was a problem hiding this comment.
Fixed in 14aeba0. fetch_url now raises ValueError on an empty candidates list. Covered by test_empty_candidates_raises_value_error.
| def _to_csv(records: list[dict[str, Any]]) -> str: | ||
| if not records: | ||
| return "" | ||
| columns: list[str] = [] | ||
| for row in records: | ||
| for key in row: | ||
| if key not in columns: | ||
| columns.append(key) | ||
| buffer = StringIO() | ||
| writer = csv.DictWriter(buffer, fieldnames=columns) | ||
| writer.writeheader() | ||
| writer.writerows(records) | ||
| return buffer.getvalue() |
There was a problem hiding this comment.
By default, csv.DictWriter uses \r\n as the line terminator. When the resulting string is written to a file using Path.write_text on Windows, Python's default text-mode translation will convert \n to \r\n, resulting in \r\r\n (double carriage returns). Specifying lineterminator="\n" ensures consistent \n line endings in the generated string, which plays nicely with standard text-mode file writing and printing.
| def _to_csv(records: list[dict[str, Any]]) -> str: | |
| if not records: | |
| return "" | |
| columns: list[str] = [] | |
| for row in records: | |
| for key in row: | |
| if key not in columns: | |
| columns.append(key) | |
| buffer = StringIO() | |
| writer = csv.DictWriter(buffer, fieldnames=columns) | |
| writer.writeheader() | |
| writer.writerows(records) | |
| return buffer.getvalue() | |
| def _to_csv(records: list[dict[str, Any]]) -> str: | |
| if not records: | |
| return "" | |
| columns: list[str] = [] | |
| for row in records: | |
| for key in row: | |
| if key not in columns: | |
| columns.append(key) | |
| buffer = StringIO() | |
| writer = csv.DictWriter(buffer, fieldnames=columns, lineterminator="\n") | |
| writer.writeheader() | |
| writer.writerows(records) | |
| return buffer.getvalue() |
There was a problem hiding this comment.
Fixed in 14aeba0. DictWriter now uses lineterminator='\n'. Covered by test_offline_csv_has_unix_line_endings.
| except Exception as exc: | ||
| message = f"source {source.key} failed: {exc}" | ||
| failures.append(message) | ||
| LOGGER.error(message) | ||
| if strict: | ||
| raise SourceRunError(message) from exc | ||
|
|
||
| if failures and not combined and strict: | ||
| raise SourceRunError("; ".join(failures)) |
There was a problem hiding this comment.
In scrape_all_sources, the check if failures and not combined and strict: at the end of the function is redundant and unreachable when strict is True. This is because any exception raised inside the loop already immediately raises SourceRunError(message) from exc when strict is True, terminating the function execution. Removing this dead code simplifies the function.
| except Exception as exc: | |
| message = f"source {source.key} failed: {exc}" | |
| failures.append(message) | |
| LOGGER.error(message) | |
| if strict: | |
| raise SourceRunError(message) from exc | |
| if failures and not combined and strict: | |
| raise SourceRunError("; ".join(failures)) | |
| except Exception as exc: | |
| message = f"source {source.key} failed: {exc}" | |
| failures.append(message) | |
| LOGGER.error(message) | |
| if strict: | |
| raise SourceRunError(message) from exc |
There was a problem hiding this comment.
Fixed in 14aeba0. Removed the unreachable end-of-function strict check and the now-unused failures list; strict mode still raises on the first source failure inside the loop.
1 participant
First stage of adopting the richer multi-source design (stdlib-only throughout).
Summary
registryof sources (SourceDefinition) and refactor Washington intobreach_scraper/sources/wa_atg.pywith richer normalization (dates, affected counts, field ordering).breach_scraper/http.pyfetch helper merging round-1 robustness (retries + exponential backoff capped at 30s, charset fallback, actionable HTTP 403) with candidate-URL fallback and optional POST support for later sources.breach_scraper/cli.py:--source/--list-sources,--start-date/--end-datedate filtering (default last 6 months),--strict,--verbose, plus merged round-1 options--input-html(offline),--user-agent,--retries.output.pyprovides json/csv/markdown (thereportandhtmlweb UI land in the next stage).breach-scraperconsole script atbreach_scraper.cli:main; remove the old single-source module.Test Plan
breach-scraper --list-sourcesshowswa_atgbreach-scraper --input-html tests/fixtures/wa_atg_sample.html --start-date 2024-01-01 --end-date 2024-12-31prints the 2 sample rows with source annotationPart of a staged adoption: (1) foundation [this PR], (2) more sources, (3) summary + web UI.