diff --git a/breach_scraper/registry.py b/breach_scraper/registry.py index 51409ef..3c24e75 100644 --- a/breach_scraper/registry.py +++ b/breach_scraper/registry.py @@ -5,7 +5,7 @@ from collections.abc import Callable from dataclasses import dataclass -from breach_scraper.sources import wa_atg +from breach_scraper.sources import ca_oag, hhs_ocr, maine_ag, or_doj, wa_atg @dataclass(frozen=True) @@ -25,6 +25,34 @@ class SourceDefinition: fetch_html=wa_atg.fetch_html, parse_html=wa_atg.parse_breach_table, ), + "ca_oag": SourceDefinition( + key="ca_oag", + display_name="California Attorney General", + default_url=ca_oag.DEFAULT_URL, + fetch_html=ca_oag.fetch_html, + parse_html=ca_oag.parse_breach_table, + ), + "or_doj": SourceDefinition( + key="or_doj", + display_name="Oregon Department of Justice", + default_url=or_doj.DEFAULT_URL, + fetch_html=or_doj.fetch_html, + parse_html=or_doj.parse_breach_table, + ), + "maine_ag": SourceDefinition( + key="maine_ag", + display_name="Maine Attorney General", + default_url=maine_ag.DEFAULT_URL, + fetch_html=maine_ag.fetch_html, + parse_html=maine_ag.parse_breach_table, + ), + "hhs_ocr": SourceDefinition( + key="hhs_ocr", + display_name="HHS OCR Breach Portal", + default_url=hhs_ocr.DEFAULT_URL, + fetch_html=hhs_ocr.fetch_html, + parse_html=hhs_ocr.parse_breach_table, + ), } diff --git a/breach_scraper/sources/ca_oag.py b/breach_scraper/sources/ca_oag.py new file mode 100644 index 0000000..7a5c806 --- /dev/null +++ b/breach_scraper/sources/ca_oag.py @@ -0,0 +1,514 @@ +"""California Attorney General breach notification connector.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from datetime import date, datetime +from html.parser import HTMLParser +from urllib.parse import urljoin + +from breach_scraper.http import fetch_url + +DEFAULT_URL = "https://oag.ca.gov/privacy/databreach/list" +DATE_FIELDS = ("date_of_breach", "date_reported") +DETAIL_URL_RE = re.compile(r"/ecrime/databreach/reports/") +PREFERRED_FIELD_ORDER = ( + "date_of_breach", + "date_reported", + "organization_name", + "organization_name_url", + "notice", + "notice_url", +) +DATE_TOKEN_RE = re.compile( + r"\d{4}-\d{2}-\d{2}" + r"|\d{1,2}/\d{1,2}/\d{2,4}" + r"|[A-Za-z]+\s+\d{1,2},\s+\d{4}" +) +ROW_VALUE_TOKEN_RE = re.compile( + r"n/a" + r"|\d{1,2}/\d{1,2}/\d{2,4}" + r"|[A-Za-z]+\s+\d{1,2},\s+\d{4}", + re.IGNORECASE, +) + + +def _clean_text(value: str) -> str: + return re.sub(r"\s+", " ", value or "").strip() + + +def _normalize_single_date(value: str) -> str: + cleaned = _clean_text(value) + if not cleaned: + return "" + + for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%m/%d/%y", "%B %d, %Y", "%b %d, %Y"): + try: + return datetime.strptime(cleaned, fmt).date().isoformat() + except ValueError: + continue + return cleaned + + +def _normalize_dateish(value: str) -> str: + cleaned = _clean_text(value) + if not cleaned: + return "" + if cleaned.lower() == "n/a": + return "n/a" + + tokens = DATE_TOKEN_RE.findall(cleaned) + if not tokens: + return cleaned + + normalized = [_normalize_single_date(token) for token in tokens] + if len(normalized) == 1 and cleaned == tokens[0]: + return normalized[0] + return ", ".join(normalized) + + +def _field_sort_key(item: tuple[str, str]) -> tuple[int, int | str]: + key, _ = item + if key in PREFERRED_FIELD_ORDER: + return (0, PREFERRED_FIELD_ORDER.index(key)) + return (1, key) + + +def _record_sort_key(record: dict[str, str]) -> tuple[int, str, str, str]: + date_of_breach = record.get("date_of_breach", "") + date_reported = record.get("date_reported", "") + primary_date = date_of_breach if date_of_breach and date_of_breach != "n/a" else date_reported + return ( + 1 if primary_date else 0, + primary_date, + date_reported, + record.get("organization_name", ""), + ) + + +def _record_reported_date(record: dict[str, str]) -> date | None: + normalized = normalize_record(record) + value = normalized.get("date_reported", "") + match = re.match(r"^\d{4}-\d{2}-\d{2}", value) + if match: + return date.fromisoformat(match.group(0)) + return None + + +def _record_in_date_range( + record: dict[str, str], start_date: date | None, end_date: date | None +) -> bool: + if start_date is None or end_date is None: + return True + reported_date = _record_reported_date(record) + if reported_date is None: + return False + return start_date <= reported_date <= end_date + + +def normalize_record(record: dict[str, str]) -> dict[str, str]: + normalized: dict[str, str] = {} + for key, value in record.items(): + cleaned = _clean_text(value) + if key in DATE_FIELDS: + cleaned = _normalize_dateish(cleaned) + normalized[key] = cleaned + return dict(sorted(normalized.items(), key=_field_sort_key)) + + +def normalize_records(records: list[dict[str, str]]) -> list[dict[str, str]]: + normalized = [normalize_record(record) for record in records] + return sorted(normalized, key=_record_sort_key, reverse=True) + + +def _fetch_url( + url: str, *, timeout: int = 30, user_agent: str | None = None, retries: int = 3 +) -> str: + return fetch_url( + [url], + timeout=timeout, + retries=retries, + user_agent=user_agent, + headers={"Referer": "https://oag.ca.gov/"}, + ) + + +@dataclass +class _Link: + href: str + text_chunks: list[str] = field(default_factory=list) + + @property + def text(self) -> str: + return _clean_text(" ".join(self.text_chunks)) + + +@dataclass +class _Block: + text_chunks: list[str] = field(default_factory=list) + links: list[_Link] = field(default_factory=list) + + @property + def text(self) -> str: + return _clean_text(" ".join(self.text_chunks)) + + +@dataclass +class _Cell: + text_chunks: list[str] = field(default_factory=list) + href: str | None = None + + def add_text(self, text: str) -> None: + if text: + self.text_chunks.append(text) + + @property + def text(self) -> str: + return _clean_text(" ".join(self.text_chunks)) + + +class _ListTableParser(HTMLParser): + """Extract the first table matching the California breach list columns.""" + + def __init__(self, base_url: str) -> None: + super().__init__() + self.base_url = base_url + self.in_table = False + self.table_depth = 0 + self.in_row = False + self.in_cell = False + self.current_cell_tag: str | None = None + self.current_row: list[tuple[str, _Cell]] = [] + self.current_cell: _Cell | None = None + self.tables: list[list[tuple[str, list[tuple[str, _Cell]]]]] = [] + self.current_table: list[tuple[str, list[tuple[str, _Cell]]]] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + + if tag == "table": + if not self.in_table: + self.in_table = True + self.current_table = [] + self.table_depth += 1 + return + + if not self.in_table: + return + + if tag == "tr": + self.in_row = True + self.current_row = [] + return + + if self.in_row and tag in {"th", "td"}: + self.in_cell = True + self.current_cell_tag = tag + self.current_cell = _Cell() + return + + if self.in_cell and tag == "a" and self.current_cell: + href = attrs_dict.get("href") + if href and not self.current_cell.href: + self.current_cell.href = urljoin(self.base_url, href) + + def handle_endtag(self, tag: str) -> None: + if not self.in_table: + return + + if tag == "table": + self.table_depth -= 1 + if self.table_depth == 0: + self.in_table = False + if self.current_table: + self.tables.append(self.current_table) + return + + if ( + tag in {"th", "td"} + and self.in_cell + and self.current_cell + and self.current_cell_tag == tag + ): + self.current_row.append((tag, self.current_cell)) + self.in_cell = False + self.current_cell_tag = None + self.current_cell = None + return + + if tag == "tr" and self.in_row: + self.in_row = False + if self.current_row: + row_type = ( + "header" + if all(cell_tag == "th" for cell_tag, _ in self.current_row) + else "data" + ) + self.current_table.append((row_type, self.current_row)) + + def handle_data(self, data: str) -> None: + if self.in_cell and self.current_cell: + self.current_cell.add_text(data) + + +def _map_header(header: str) -> str: + cleaned = _clean_text(header).lower() + if "organization" in cleaned: + return "organization_name" + if "date(s) of breach" in cleaned or "date of breach" in cleaned: + return "date_of_breach" + if "reported date" in cleaned: + return "date_reported" + return re.sub(r"[^a-z0-9]+", "_", cleaned).strip("_") or "column" + + +def _parse_list_table(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + parser = _ListTableParser(base_url=base_url) + parser.feed(html) + + for table in parser.tables: + headers: list[str] = [] + records: list[dict[str, str]] = [] + for row_type, row in table: + if row_type == "header" and not headers: + headers = [_map_header(cell.text) for _, cell in row] + continue + + if not headers: + continue + + values: dict[str, str] = {} + for idx, (_, cell) in enumerate(row): + if idx >= len(headers): + continue + key = headers[idx] + values[key] = cell.text + if cell.href: + values[f"{key}_url"] = cell.href + + if values.get("organization_name") and values.get("date_reported") is not None: + records.append(values) + + if records and {"organization_name", "date_of_breach", "date_reported"}.issubset( + records[0].keys() + ): + return records + + return [] + + +class _ListRowParser(HTMLParser): + BLOCK_TAGS = {"tr", "li", "div", "article", "section"} + + def __init__(self, base_url: str) -> None: + super().__init__() + self.base_url = base_url + self.blocks: list[_Block] = [] + self.candidates: list[_Block] = [] + self.current_link: _Link | None = None + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + if tag in self.BLOCK_TAGS: + self.blocks.append(_Block()) + return + + if tag == "a": + href = attrs_dict.get("href") + if href: + absolute = urljoin(self.base_url, href) + if DETAIL_URL_RE.search(absolute): + self.current_link = _Link(href=absolute) + + def handle_data(self, data: str) -> None: + if self.current_link: + self.current_link.text_chunks.append(data) + if self.blocks: + for block in self.blocks: + block.text_chunks.append(data) + + def handle_endtag(self, tag: str) -> None: + if tag == "a" and self.current_link: + for block in self.blocks: + block.links.append(self.current_link) + self.current_link = None + return + + if tag in self.BLOCK_TAGS and self.blocks: + block = self.blocks.pop() + if block.links: + self.candidates.append(block) + + +def _candidate_to_record(block: _Block) -> dict[str, str] | None: + if not block.links: + return None + + link = block.links[0] + organization_name = link.text + if not organization_name: + return None + + text = block.text + if organization_name in text: + remainder = _clean_text(text.split(organization_name, 1)[1]) + else: + remainder = text + + if not remainder: + return None + + tokens = [match.group(0) for match in ROW_VALUE_TOKEN_RE.finditer(remainder)] + if not tokens: + return None + + reported_date = tokens[-1] + reported_index = remainder.rfind(reported_date) + breach_part = _clean_text(remainder[:reported_index]) + if not breach_part: + breach_part = tokens[0] if len(tokens) == 1 else ", ".join(tokens[:-1]) + + return { + "organization_name": organization_name, + "organization_name_url": link.href, + "date_of_breach": breach_part or "n/a", + "date_reported": reported_date, + } + + +def _parse_list_rows(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + parser = _ListRowParser(base_url=base_url) + parser.feed(html) + + best_by_url: dict[str, tuple[int, dict[str, str]]] = {} + for block in parser.candidates: + record = _candidate_to_record(block) + if not record: + continue + + text_length = len(block.text) + existing = best_by_url.get(record["organization_name_url"]) + if existing is None or text_length < existing[0]: + best_by_url[record["organization_name_url"]] = (text_length, record) + + return [item[1] for item in best_by_url.values()] + + +def parse_list_table(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + records = _parse_list_table(html, base_url=base_url) + if records: + return records + return _parse_list_rows(html, base_url=base_url) + + +class _DetailPdfParser(HTMLParser): + def __init__(self, base_url: str) -> None: + super().__init__() + self.base_url = base_url + self.current_href: str | None = None + self.current_text: list[str] = [] + self.pdf_links: list[tuple[str, str]] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + if tag != "a": + return + href = dict(attrs).get("href") + if href: + self.current_href = urljoin(self.base_url, href) + self.current_text = [] + + def handle_data(self, data: str) -> None: + if self.current_href is not None: + self.current_text.append(data) + + def handle_endtag(self, tag: str) -> None: + if tag == "a" and self.current_href is not None: + text = _clean_text(" ".join(self.current_text)) + self.pdf_links.append((self.current_href, text)) + self.current_href = None + self.current_text = [] + + +def parse_detail_page(html: str, page_url: str) -> dict[str, str]: + parser = _DetailPdfParser(base_url=page_url) + parser.feed(html) + + for href, text in parser.pdf_links: + if href.lower().endswith(".pdf"): + return { + "organization_name_url": page_url, + "notice": text or "Sample Notification", + "notice_url": href, + } + + return {"organization_name_url": page_url} + + +def fetch_html( + url: str = DEFAULT_URL, + *, + timeout: int = 30, + limit: int = 0, + start_date: date | None = None, + end_date: date | None = None, + user_agent: str | None = None, + retries: int = 3, +) -> str: + list_html = _fetch_url(url, timeout=timeout, user_agent=user_agent, retries=retries) + records = parse_list_table(list_html, base_url=url) + if start_date is not None and end_date is not None: + records = [ + record for record in records if _record_in_date_range(record, start_date, end_date) + ] + if limit > 0: + records = records[:limit] + + detail_pages = [] + for record in records: + detail_url = record.get("organization_name_url") + if not detail_url: + continue + detail_pages.append( + { + "url": detail_url, + "html": _fetch_url( + detail_url, timeout=timeout, user_agent=user_agent, retries=retries + ), + } + ) + + return json.dumps({"list_url": url, "records": records, "detail_pages": detail_pages}) + + +def parse_breach_table(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + stripped = html.lstrip() + if stripped.startswith("{"): + payload = json.loads(stripped) + records = payload.get("records") + if records is None: + list_html = payload.get("list_html", "") + records = parse_list_table(list_html, base_url=payload.get("list_url", base_url)) + details_by_url = {} + for page in payload.get("detail_pages", []): + page_url = page.get("url", "") + if page_url: + details_by_url[page_url] = parse_detail_page(page.get("html", ""), page_url) + + merged = [] + for record in records: + detail = details_by_url.get(record.get("organization_name_url", ""), {}) + merged_record = dict(record) + merged_record.update({key: value for key, value in detail.items() if value}) + merged.append(merged_record) + return normalize_records(merged) + + records = parse_list_table(html, base_url=base_url) + if records: + return normalize_records(records) + + detail = parse_detail_page(html, page_url=base_url) + if any(detail.values()): + return normalize_records([detail]) + return [] diff --git a/breach_scraper/sources/hhs_ocr.py b/breach_scraper/sources/hhs_ocr.py new file mode 100644 index 0000000..f7288ae --- /dev/null +++ b/breach_scraper/sources/hhs_ocr.py @@ -0,0 +1,382 @@ +"""HHS OCR breach portal connector.""" + +from __future__ import annotations + +import csv +import io +import re +from datetime import date, datetime +from html.parser import HTMLParser +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import HTTPCookieProcessor, Request, build_opener + +DEFAULT_URL = "https://ocrportal.hhs.gov/ocr/breach/breach_report_hip.jsf" +ENTITY_TYPES = ( + "Healthcare Provider", + "Health Plan", + "Business Associate", + "Healthcare Clearing House", +) +BREACH_TYPES = ( + "Hacking/IT Incident", + "Unauthorized Access/Disclosure", + "Improper Disposal", + "Unauthorized Access", + "Theft", + "Loss", + "Other", +) +HEADER_TEXT = ( + "Expand AllName of Covered Entity State Covered Entity Type Individuals Affected " + "Breach Submission Date Type of Breach Location of Breached Information Business Associate Present Web Description" # noqa: E501 +) +CSV_ACTION_RE = re.compile( + r"onclick=\"mojarra\.jsfcljs\(document\.getElementById\('ocrForm'\),\{'(?PocrForm:j_idt\d+)':'[^']+'\},''\);return false\">" # noqa: E501 + r"]+alt=\"CSV\"", + re.IGNORECASE, +) +VIEWSTATE_RE = re.compile(r'name="javax\.faces\.ViewState"[^>]*value="([^"]+)"', re.IGNORECASE) + + +def _clean_text(value: str) -> str: + return re.sub(r"\s+", " ", value or "").strip() + + +def _normalize_date(value: str) -> str: + cleaned = _clean_text(value) + if not cleaned: + return "" + try: + return datetime.strptime(cleaned, "%m/%d/%Y").date().isoformat() + except ValueError: + return cleaned + + +def _normalize_count(value: str) -> str: + cleaned = _clean_text(value) + digits = re.sub(r"[^\d]", "", cleaned) + if not digits: + return cleaned + return f"{int(digits):,}" + + +def _record_sort_key(record: dict[str, str]) -> tuple[int, str, str]: + reported = record.get("date_reported", "") + return (1 if reported else 0, reported, record.get("organization_name", "")) + + +def normalize_record(record: dict[str, str]) -> dict[str, str]: + normalized = dict(record) + normalized["date_reported"] = _normalize_date(normalized.get("date_reported", "")) + normalized["individuals_affected"] = _normalize_count( + normalized.get("individuals_affected", "") + ) + return normalized + + +def normalize_records(records: list[dict[str, str]]) -> list[dict[str, str]]: + return sorted( + (normalize_record(record) for record in records), key=_record_sort_key, reverse=True + ) + + +class _TextExtractor(HTMLParser): + BREAK_TAGS = {"br", "p", "div", "li", "tr", "td", "th", "section", "article", "h1", "h2", "h3"} + + def __init__(self) -> None: + super().__init__() + self.parts: list[str] = [] + + def handle_starttag(self, tag: str, _attrs: list[tuple[str, str | None]]) -> None: + if tag == "br": + self.parts.append("\n") + + def handle_endtag(self, tag: str) -> None: + if tag in self.BREAK_TAGS: + self.parts.append("\n") + + def handle_data(self, data: str) -> None: + self.parts.append(data) + + def text(self) -> str: + return "".join(self.parts) + + +class _Cell: + def __init__(self) -> None: + self.parts: list[str] = [] + + def add_text(self, text: str) -> None: + if text: + self.parts.append(text) + + @property + def text(self) -> str: + return _clean_text("".join(self.parts)) + + +class _ResultsTableParser(HTMLParser): + def __init__(self) -> None: + super().__init__() + self.in_tbody = False + self.in_row = False + self.in_cell = False + self.current_row_valid = False + self.current_row: list[_Cell] = [] + self.current_cell: _Cell | None = None + self.rows: list[list[str]] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + if tag == "tbody" and attrs_dict.get("id") == "ocrForm:reportResultTable_data": + self.in_tbody = True + return + + if not self.in_tbody: + return + + if tag == "tr": + self.in_row = True + self.current_row = [] + self.current_row_valid = "data-ri" in attrs_dict + return + + if self.in_row and tag == "td": + self.in_cell = True + self.current_cell = _Cell() + return + + if self.in_cell and tag == "br" and self.current_cell: + self.current_cell.add_text(", ") + + def handle_endtag(self, tag: str) -> None: + if not self.in_tbody: + return + + if tag == "tbody": + self.in_tbody = False + return + + if tag == "td" and self.in_cell and self.current_cell: + self.current_row.append(self.current_cell) + self.current_cell = None + self.in_cell = False + return + + if tag == "tr" and self.in_row: + self.in_row = False + if self.current_row_valid and self.current_row: + self.rows.append([cell.text for cell in self.current_row]) + + def handle_data(self, data: str) -> None: + if self.in_cell and self.current_cell: + self.current_cell.add_text(data) + + +def _build_headers(user_agent: str | None = None) -> dict[str, str]: + return { + "User-Agent": user_agent + or ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36" + ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + "Connection": "keep-alive", + "Referer": "https://ocrportal.hhs.gov/", + } + + +def fetch_html( + url: str = DEFAULT_URL, + *, + timeout: int = 30, + start_date: date | None = None, + end_date: date | None = None, + user_agent: str | None = None, + retries: int = 3, +) -> str: + try: + opener = build_opener(HTTPCookieProcessor()) + request = Request(url, headers=_build_headers(user_agent)) + with opener.open(request, timeout=timeout) as response: # nosec B310 + charset = response.headers.get_content_charset() or "utf-8" + html: str = response.read().decode(charset, errors="replace") + + viewstate_match = VIEWSTATE_RE.search(html) + csv_action_match = CSV_ACTION_RE.search(html) + if not viewstate_match or not csv_action_match: + return html + + action = csv_action_match.group("action") + data = urlencode( + { + "ocrForm": "ocrForm", + action: action, + "javax.faces.ViewState": viewstate_match.group(1), + } + ).encode("utf-8") + export_request = Request( + url, + data=data, + headers={ + **_build_headers(user_agent), + "Content-Type": "application/x-www-form-urlencoded", + }, + ) + with opener.open(export_request, timeout=timeout) as response: # nosec B310 + charset = response.headers.get_content_charset() or "utf-8" + exported: str = response.read().decode(charset, errors="replace") + return exported or html + except (HTTPError, URLError, TimeoutError) as exc: + raise RuntimeError(f"Failed to fetch HHS OCR page: {exc}") from exc + + +def _parse_csv_export(text: str) -> list[dict[str, str]]: + cleaned = text.lstrip() + if cleaned.startswith("<") or "Covered Entity Type" not in cleaned: + return [] + + rows = list(csv.reader(io.StringIO(text))) + if len(rows) < 2: + return [] + + records: list[dict[str, str]] = [] + for row in rows[1:]: + if len(row) < 8: + continue + records.append( + { + "organization_name": row[0], + "state": row[1], + "covered_entity_type": row[2], + "individuals_affected": row[3], + "date_reported": row[4], + "type_of_breach": row[5], + "location_of_breached_information": row[6], + "business_associate_present": row[7], + "breach_description": row[8] if len(row) > 8 else "", + } + ) + + return records + + +def _parse_html_table(html: str) -> list[dict[str, str]]: + parser = _ResultsTableParser() + parser.feed(html) + + records: list[dict[str, str]] = [] + for row in parser.rows: + if len(row) < 8: + continue + records.append( + { + "organization_name": row[1] if len(row) > 1 else "", + "state": row[2] if len(row) > 2 else "", + "covered_entity_type": row[3] if len(row) > 3 else "", + "individuals_affected": row[4] if len(row) > 4 else "", + "date_reported": row[5] if len(row) > 5 else "", + "type_of_breach": row[6] if len(row) > 6 else "", + "location_of_breached_information": row[7] if len(row) > 7 else "", + "business_associate_present": row[8] if len(row) > 8 else "", + "breach_description": row[9] if len(row) > 9 else "", + } + ) + + return records + + +def _parse_row(line: str) -> dict[str, str] | None: + line = _clean_text(line) + if not line: + return None + + business_associate_present = None + for candidate in (" Yes", " No"): + if line.endswith(candidate): + business_associate_present = candidate.strip() + line = line[: -len(candidate)].rstrip() + break + if not business_associate_present: + return None + + breach_type = None + location = "" + for candidate in BREACH_TYPES: + marker = f" {candidate} " + if marker in line: + before, after = line.split(marker, 1) + breach_type = candidate + location = after.strip() + line = before.strip() + break + if not breach_type: + return None + + match = re.search(r"\s(\d+)\s+(\d{2}/\d{2}/\d{4})$", line) + if not match: + return None + individuals_affected = match.group(1) + submission_date = match.group(2) + line = line[: match.start()].rstrip() + + entity_type = None + for candidate in ENTITY_TYPES: + marker = f" {candidate}" + if line.endswith(marker): + entity_type = candidate + line = line[: -len(marker)].rstrip() + break + if not entity_type: + return None + + state_match = re.search(r"\s([A-Z]{2})$", line) + if not state_match: + return None + state = state_match.group(1) + organization_name = line[: state_match.start()].strip() + + return { + "organization_name": organization_name, + "state": state, + "covered_entity_type": entity_type, + "individuals_affected": individuals_affected, + "date_reported": submission_date, + "type_of_breach": breach_type, + "location_of_breached_information": location, + "business_associate_present": business_associate_present, + } + + +def _parse_flattened_text(html: str) -> list[dict[str, str]]: + """Fallback for older fixtures and any unexpected non-table renderings.""" + extractor = _TextExtractor() + extractor.feed(html) + lines = [_clean_text(line) for line in extractor.text().splitlines()] + lines = [line for line in lines if line] + + try: + start_index = lines.index(HEADER_TEXT) + 1 + except ValueError: + return [] + + records: list[dict[str, str]] = [] + for line in lines[start_index:]: + if line.startswith("(Displaying ") or line.startswith("We are generating the report"): + break + record = _parse_row(line) + if record: + records.append(record) + + return records + + +def parse_breach_table(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + records = _parse_csv_export(html) + if not records: + records = _parse_html_table(html) + if not records: + records = _parse_flattened_text(html) + return normalize_records(records) diff --git a/breach_scraper/sources/maine_ag.py b/breach_scraper/sources/maine_ag.py new file mode 100644 index 0000000..383980b --- /dev/null +++ b/breach_scraper/sources/maine_ag.py @@ -0,0 +1,518 @@ +"""Maine Attorney General breach notification connector.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from datetime import date, datetime +from html.parser import HTMLParser +from urllib.parse import urljoin + +from breach_scraper.http import fetch_url + +DEFAULT_URL = ( + "https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/list.html" +) +DATE_FIELDS = ( + "date_of_breach", + "date_breach_discovered", + "date_of_consumer_notification", + "date_reported", +) +COUNT_FIELDS = ("total_persons_affected", "number_of_maine_residents_affected") +PREFERRED_FIELD_ORDER = ( + "date_of_breach", + "date_reported", + "organization_name", + "organization_name_url", + "organization_type", + "number_of_maine_residents_affected", + "total_persons_affected", + "date_breach_discovered", + "date_of_consumer_notification", + "type_of_notification", + "information_compromised", + "breach_description", + "notice", + "notice_url", +) +LABEL_MAP = { + "Type of Organization": "organization_type", + "Entity Name": "organization_name", + "Total number of persons affected (including residents)": "total_persons_affected", + "Total number of Maine residents affected": "number_of_maine_residents_affected", + "Date(s) Breach Occured": "date_of_breach", + "Date Breach Discovered": "date_breach_discovered", + "Description of the Breach": "breach_description", + "Information Acquired - Name or other personal identifier in combination with": "information_compromised", # noqa: E501 + "Type of Notification": "type_of_notification", + "Date(s) of consumer notification": "date_of_consumer_notification", + "Copy of notice to affected Maine residents": "notice", +} +DATE_TOKEN_RE = re.compile( + r"\d{4}-\d{2}-\d{2}" + r"|\d{1,2}/\d{1,2}/\d{2,4}" + r"|\d{1,2}-\d{1,2}-\d{4}" + r"|[A-Za-z]+\s+\d{1,2},\s+\d{4}" +) +DETAIL_LINK_RE = re.compile( + r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.html$", re.IGNORECASE +) + + +def _clean_text(value: str) -> str: + return re.sub(r"\s+", " ", value or "").strip() + + +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_single_date(value: str) -> str: + cleaned = _clean_text(value) + if not cleaned: + return "" + + for fmt in ( + "%Y-%m-%d", + "%m/%d/%Y", + "%m/%d/%y", + "%m-%d-%Y", + "%B %d, %Y", + "%b %d, %Y", + ): + try: + return datetime.strptime(cleaned, fmt).date().isoformat() + except ValueError: + continue + return cleaned + + +def _repair_malformed_date_text(value: str) -> str: + repaired = _clean_text(value) + repaired = re.sub( + r"\b(\d{1,2})/(\d{1,2})/2-(\d{2})\b", + lambda match: f"{int(match.group(1)):02d}/{int(match.group(2)):02d}/20{match.group(3)}", + repaired, + ) + repaired = re.sub( + r"\b(\d{2})(\d{2})(20\d{2})\b", + lambda match: f"{match.group(1)}/{match.group(2)}/{match.group(3)}", + repaired, + ) + return repaired + + +def _normalize_dateish(value: str) -> str: + cleaned = _repair_malformed_date_text(value) + if not cleaned: + return "" + + tokens = DATE_TOKEN_RE.findall(cleaned) + if not tokens: + return cleaned + + normalized_tokens = [_normalize_single_date(token) for token in tokens] + if len(normalized_tokens) == 1 and cleaned == tokens[0]: + return normalized_tokens[0] + return " to ".join(normalized_tokens) + + +def _field_sort_key(item: tuple[str, str]) -> tuple[int, int | str]: + key, _ = item + if key in PREFERRED_FIELD_ORDER: + return (0, PREFERRED_FIELD_ORDER.index(key)) + return (1, key) + + +def _record_sort_key(record: dict[str, str]) -> tuple[int, str, str, str]: + date_of_breach = record.get("date_of_breach", "") + date_reported = record.get("date_reported", "") + primary_date = date_of_breach or date_reported + return ( + 1 if primary_date else 0, + primary_date, + date_reported, + record.get("organization_name", ""), + ) + + +def normalize_record(record: dict[str, str]) -> dict[str, str]: + normalized: dict[str, str] = {} + for key, value in record.items(): + cleaned = _clean_text(value) + if key in DATE_FIELDS: + cleaned = _normalize_dateish(cleaned) + elif key in COUNT_FIELDS: + cleaned = _normalize_count(cleaned) + normalized[key] = cleaned + + if not normalized.get("date_reported"): + normalized["date_reported"] = normalized.get( + "date_of_consumer_notification" + ) or normalized.get("date_breach_discovered", "") + return dict(sorted(normalized.items(), key=_field_sort_key)) + + +def normalize_records(records: list[dict[str, str]]) -> list[dict[str, str]]: + normalized = [normalize_record(record) for record in records] + return sorted(normalized, key=_record_sort_key, reverse=True) + + +def _fetch_url( + url: str, *, timeout: int = 30, user_agent: str | None = None, retries: int = 3 +) -> str: + return fetch_url( + [url], + timeout=timeout, + retries=retries, + user_agent=user_agent, + headers={"Referer": "https://www.maine.gov/"}, + ) + + +@dataclass +class _ListLink: + url: str + text_chunks: list[str] = field(default_factory=list) + date_reported: str = "" + + @property + def text(self) -> str: + return _clean_text(" ".join(self.text_chunks)) + + +class _ListCell: + def __init__(self) -> None: + self.text_chunks: list[str] = [] + self.href: str | None = None + + def add_text(self, text: str) -> None: + if text: + self.text_chunks.append(text) + + @property + def text(self) -> str: + return _clean_text("".join(self.text_chunks)) + + +class _MaineAGListParser(HTMLParser): + def __init__(self, base_url: str) -> None: + super().__init__() + self.base_url = base_url + self.in_table = False + self.in_row = False + self.in_cell = False + self.current_tag: str | None = None + self.current_row: list[_ListCell] = [] + self.current_cell: _ListCell | None = None + self.entries: list[_ListLink] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + if tag == "table" and "breachTable" in (attrs_dict.get("class") or ""): + self.in_table = True + return + + if not self.in_table: + return + + if tag == "tr": + self.in_row = True + self.current_row = [] + return + + if self.in_row and tag in {"td", "th"}: + self.in_cell = True + self.current_tag = tag + self.current_cell = _ListCell() + return + + if tag == "a" and self.in_cell and self.current_cell: + href = attrs_dict.get("href") + if not href: + return + absolute = urljoin(self.base_url, href) + if absolute.endswith("list.html") or not DETAIL_LINK_RE.search(absolute): + return + self.current_cell.href = absolute + + def handle_data(self, data: str) -> None: + if self.in_cell and self.current_cell: + self.current_cell.add_text(data) + + def handle_endtag(self, tag: str) -> None: + if not self.in_table: + return + + if tag == "table": + self.in_table = False + return + + if tag in {"td", "th"} and self.in_cell and self.current_cell and self.current_tag == tag: + self.current_row.append(self.current_cell) + self.in_cell = False + self.current_tag = None + self.current_cell = None + return + + if tag == "tr" and self.in_row: + self.in_row = False + if len(self.current_row) >= 2 and self.current_row[1].href: + self.entries.append( + _ListLink( + url=self.current_row[1].href or "", + text_chunks=self.current_row[1].text_chunks, + date_reported=self.current_row[0].text, + ) + ) + + +def _entry_in_date_range(entry: _ListLink, start_date: date | None, end_date: date | None) -> bool: + if start_date is None or end_date is None: + return True + normalized = _normalize_dateish(entry.date_reported) + match = re.match(r"^\d{4}-\d{2}-\d{2}", normalized) + if not match: + return False + reported_date = date.fromisoformat(match.group(0)) + return start_date <= reported_date <= end_date + + +class _SimpleLinkParser(HTMLParser): + def __init__(self, base_url: str) -> None: + super().__init__() + self.base_url = base_url + self.current_link: _ListLink | None = None + self.links: list[_ListLink] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + if tag != "a": + return + + href = dict(attrs).get("href") + if not href: + return + + absolute = urljoin(self.base_url, href) + if absolute.endswith("list.html") or not DETAIL_LINK_RE.search(absolute): + return + + self.current_link = _ListLink(url=absolute) + + def handle_data(self, data: str) -> None: + if self.current_link: + self.current_link.text_chunks.append(data) + + def handle_endtag(self, tag: str) -> None: + if tag == "a" and self.current_link: + self.links.append(self.current_link) + self.current_link = None + + +def extract_detail_links(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + entries = extract_list_entries(html, base_url=base_url) + if entries: + return [{"url": entry["url"], "text": entry["text"]} for entry in entries] + + seen: set[str] = set() + links: list[dict[str, str]] = [] + parser = _SimpleLinkParser(base_url=base_url) + parser.feed(html) + for link in parser.links: + if link.url in seen: + continue + seen.add(link.url) + links.append({"url": link.url, "text": link.text}) + return links + + +def extract_list_entries(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + parser = _MaineAGListParser(base_url=base_url) + parser.feed(html) + + seen: set[str] = set() + entries: list[dict[str, str]] = [] + for link in parser.entries: + if link.url in seen: + continue + seen.add(link.url) + entries.append({"url": link.url, "text": link.text, "date_reported": link.date_reported}) + return entries + + +@dataclass +class _DetailItem: + text_chunks: list[str] = field(default_factory=list) + href: str | None = None + + def add_child_text(self, text: str) -> None: + if not text: + return + if self.text_chunks and self.text_chunks[-1].rstrip().endswith(":"): + self.text_chunks.append(" " + text) + else: + self.text_chunks.append("; " + text) + + @property + def text(self) -> str: + return _clean_text(" ".join(self.text_chunks)) + + +class _MaineAGDetailParser(HTMLParser): + def __init__(self, base_url: str) -> None: + super().__init__() + self.base_url = base_url + self.current_section = "" + self.in_heading = False + self.heading_chunks: list[str] = [] + self.item_stack: list[_DetailItem] = [] + self.items: list[tuple[str, str, str | None]] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + if tag in {"h1", "h2", "h3", "h4"}: + self.in_heading = True + self.heading_chunks = [] + return + + if tag == "li": + self.item_stack.append(_DetailItem()) + return + + if tag == "a" and self.item_stack: + href = attrs_dict.get("href") + if href and not self.item_stack[-1].href: + self.item_stack[-1].href = urljoin(self.base_url, href) + + def handle_data(self, data: str) -> None: + if self.in_heading: + self.heading_chunks.append(data) + elif self.item_stack: + self.item_stack[-1].text_chunks.append(data) + + def handle_endtag(self, tag: str) -> None: + if tag in {"h1", "h2", "h3", "h4"} and self.in_heading: + self.in_heading = False + self.current_section = _clean_text(" ".join(self.heading_chunks)) + self.heading_chunks = [] + return + + if tag == "li" and self.item_stack: + item = self.item_stack.pop() + text = item.text + if self.item_stack: + self.item_stack[-1].add_child_text(text) + if item.href and not self.item_stack[-1].href: + self.item_stack[-1].href = item.href + elif text: + self.items.append((self.current_section, text, item.href)) + + +def _split_label_value(text: str) -> tuple[str, str]: + if ":" not in text: + return _clean_text(text), "" + label, value = text.split(":", 1) + return _clean_text(label), _clean_text(value) + + +def parse_detail_page(html: str, page_url: str) -> dict[str, str]: + parser = _MaineAGDetailParser(base_url=page_url) + parser.feed(html) + + record: dict[str, str] = {"organization_name_url": page_url} + for _section, text, href in parser.items: + label, value = _split_label_value(text) + key = LABEL_MAP.get(label) + if not key: + continue + + if key == "notice": + record[key] = value or "View notice" + if href: + record["notice_url"] = href + else: + record[key] = value + if href: + record[f"{key}_url"] = href + + return record + + +def fetch_html( + url: str = DEFAULT_URL, + *, + timeout: int = 30, + limit: int = 0, + start_date: date | None = None, + end_date: date | None = None, + user_agent: str | None = None, + retries: int = 3, +) -> str: + list_html = _fetch_url(url, timeout=timeout, user_agent=user_agent, retries=retries) + detail_links = extract_list_entries(list_html, base_url=url) + if start_date is not None and end_date is not None: + detail_links = [ + link + for link in detail_links + if _entry_in_date_range( + _ListLink( + url=link["url"], + text_chunks=[link["text"]], + date_reported=link.get("date_reported", ""), + ), + start_date, + end_date, + ) + ] + if limit > 0: + detail_links = detail_links[:limit] + detail_pages = [ + { + "url": link["url"], + "html": _fetch_url( + link["url"], timeout=timeout, user_agent=user_agent, retries=retries + ), + "list_text": link["text"], + } + for link in detail_links + ] + return json.dumps({"list_url": url, "detail_pages": detail_pages}) + + +def parse_breach_table(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + stripped = html.lstrip() + if stripped.startswith("{"): + payload = json.loads(stripped) + detail_pages = payload.get("detail_pages", []) + records = [] + for page in detail_pages: + record = parse_detail_page(page.get("html", ""), page.get("url", base_url)) + if page.get("list_text") and not record.get("organization_name"): + record["organization_name"] = page["list_text"] + records.append(record) + return normalize_records([record for record in records if any(record.values())]) + + detail_links = extract_detail_links(html, base_url=base_url) + if detail_links: + records = [] + for link in detail_links: + record = { + "organization_name": link["text"] or "View detail", + "organization_name_url": link["url"], + } + records.append(record) + return normalize_records(records) + + record = parse_detail_page(html, page_url=base_url) + if any(record.values()): + return normalize_records([record]) + return [] diff --git a/breach_scraper/sources/or_doj.py b/breach_scraper/sources/or_doj.py new file mode 100644 index 0000000..4be3ab4 --- /dev/null +++ b/breach_scraper/sources/or_doj.py @@ -0,0 +1,314 @@ +"""Oregon Department of Justice breach notification connector.""" + +from __future__ import annotations + +import re +from datetime import date, datetime +from html.parser import HTMLParser + +from breach_scraper.http import fetch_url + +DEFAULT_URL = "https://justice.oregon.gov/consumer/databreach/" +HEADER_TEXT = ( + "Organization Reported Date Dates of Breach Dates of Discovery Date Notice Sent Number Affected" +) +SENTINEL_DATES = {"1/1/0001", "01/01/0001", "1/1/2001", "01/01/2001"} +COUNT_FIELDS = ("number_affected",) +DATE_FIELDS = ("date_reported", "date_of_breach", "date_of_discovery", "date_notice_sent") +HEADER_MAP = { + "organization": "organization_name", + "reported_date": "date_reported", + "dates_of_breach": "date_of_breach", + "dates_of_discovery": "date_of_discovery", + "date_notice_sent": "date_notice_sent", + "number_affected": "number_affected", +} +ROW_START_RE = re.compile( + r"^(?P.+?)\s+" + r"(?P\d{1,2}/\d{1,2}/\d{4})\s+" + r"(?P.+?)\s+" + r"(?P(?:\d{1,2}/\d{1,2}/\d{4}|1/1/0001|1/1/2001)(?:\s*-\s*(?:\d{1,2}/\d{1,2}/\d{4}|1/1/0001|1/1/2001))?)$" +) +DATE_TOKEN_RE = re.compile(r"\d{1,2}/\d{1,2}/\d{4}|1/1/0001|1/1/2001", re.IGNORECASE) + + +def _clean_text(value: str) -> str: + return re.sub(r"\s+", " ", value or "").strip() + + +def _normalize_count(value: str) -> str: + cleaned = _clean_text(value) + digits = re.sub(r"[^\d]", "", cleaned) + if not digits: + return cleaned + return f"{int(digits):,}" + + +def _normalize_single_date(value: str) -> str: + cleaned = _clean_text(value) + if not cleaned or cleaned in SENTINEL_DATES: + return "" + try: + return datetime.strptime(cleaned, "%m/%d/%Y").date().isoformat() + except ValueError: + return cleaned + + +def _normalize_dateish(value: str) -> str: + cleaned = _clean_text(value) + if not cleaned: + return "" + + normalized_parts: list[str] = [] + for part in [segment.strip() for segment in cleaned.split(",")]: + if not part: + continue + + if "-" in part: + range_parts = [_normalize_single_date(item) for item in re.split(r"\s*-\s*", part)] + range_parts = [item for item in range_parts if item] + if range_parts: + normalized_parts.append(" to ".join(range_parts)) + continue + + normalized = _normalize_single_date(part) + if normalized: + normalized_parts.append(normalized) + + return ", ".join(normalized_parts) + + +def _record_sort_key(record: dict[str, str]) -> tuple[int, str, str]: + reported_date = record.get("date_reported", "") + primary_date = reported_date or record.get("date_of_breach", "") + return (1 if primary_date else 0, primary_date, record.get("organization_name", "")) + + +def normalize_record(record: dict[str, str]) -> dict[str, str]: + normalized: dict[str, str] = {} + for key, value in record.items(): + cleaned = _clean_text(value) + if key in DATE_FIELDS: + cleaned = _normalize_dateish(cleaned) + elif key in COUNT_FIELDS: + cleaned = _normalize_count(cleaned) + normalized[key] = cleaned + return normalized + + +def normalize_records(records: list[dict[str, str]]) -> list[dict[str, str]]: + return sorted( + (normalize_record(record) for record in records), key=_record_sort_key, reverse=True + ) + + +class _TextExtractor(HTMLParser): + BREAK_TAGS = {"br", "p", "div", "li", "tr", "td", "th", "section", "article", "h1", "h2", "h3"} + + def __init__(self) -> None: + super().__init__() + self.parts: list[str] = [] + + def handle_starttag(self, tag: str, _attrs: list[tuple[str, str | None]]) -> None: + if tag == "br": + self.parts.append("\n") + + def handle_endtag(self, tag: str) -> None: + if tag in self.BREAK_TAGS: + self.parts.append("\n") + + def handle_data(self, data: str) -> None: + self.parts.append(data) + + def text(self) -> str: + return "".join(self.parts) + + +class _Cell: + def __init__(self) -> None: + self.parts: list[str] = [] + + def add_text(self, text: str) -> None: + if text: + self.parts.append(text) + + @property + def text(self) -> str: + return _clean_text("".join(self.parts)) + + +class _TableParser(HTMLParser): + def __init__(self) -> None: + super().__init__() + self.in_table = False + self.in_row = False + self.in_cell = False + self.current_tag: str | None = None + self.current_row: list[tuple[str, _Cell]] = [] + self.current_cell: _Cell | None = None + self.rows: list[tuple[str, list[_Cell]]] = [] + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + if tag == "table" and attrs_dict.get("id") == "grid": + self.in_table = True + return + + if not self.in_table: + return + + if tag == "tr": + self.in_row = True + self.current_row = [] + return + + if self.in_row and tag in {"th", "td"}: + self.in_cell = True + self.current_tag = tag + self.current_cell = _Cell() + return + + if self.in_cell and tag == "br" and self.current_cell: + self.current_cell.add_text(", ") + + def handle_endtag(self, tag: str) -> None: + if not self.in_table: + return + + if tag == "table": + self.in_table = False + return + + if tag in {"th", "td"} and self.in_cell and self.current_cell and self.current_tag == tag: + self.current_row.append((tag, self.current_cell)) + self.in_cell = False + self.current_tag = None + self.current_cell = None + return + + if tag == "tr" and self.in_row: + self.in_row = False + if self.current_row: + row_type = ( + "header" + if all(cell_tag == "th" for cell_tag, _ in self.current_row) + else "data" + ) + self.rows.append((row_type, [cell for _, cell in self.current_row])) + + def handle_data(self, data: str) -> None: + if self.in_cell and self.current_cell: + self.current_cell.add_text(data) + + +def fetch_html( + url: str = DEFAULT_URL, + *, + timeout: int = 30, + start_date: date | None = None, + end_date: date | None = None, + user_agent: str | None = None, + retries: int = 3, +) -> str: + return fetch_url( + [url], + timeout=timeout, + retries=retries, + user_agent=user_agent, + headers={"Referer": "https://justice.oregon.gov/"}, + ) + + +def _header_key(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "_", _clean_text(value).lower()).strip("_") + + +def _parse_html_table(html: str) -> list[dict[str, str]]: + parser = _TableParser() + parser.feed(html) + + headers: list[str] = [] + records: list[dict[str, str]] = [] + + for row_type, row in parser.rows: + if row_type == "header" and not headers: + headers = [ + HEADER_MAP.get(_header_key(cell.text), _header_key(cell.text)) for cell in row + ] + continue + + if row_type != "data" or not headers: + continue + + values: dict[str, str] = {} + for idx, cell in enumerate(row): + if idx >= len(headers): + continue + values[headers[idx]] = cell.text + + if values.get("organization_name"): + records.append(values) + + return records + + +def _parse_flattened_text(html: str) -> list[dict[str, str]]: + """Fallback for older fixtures and any unexpected non-table renderings.""" + extractor = _TextExtractor() + extractor.feed(html) + lines = [_clean_text(line) for line in extractor.text().splitlines()] + lines = [line for line in lines if line] + + try: + start_index = lines.index(HEADER_TEXT) + 1 + except ValueError: + return [] + + records: list[dict[str, str]] = [] + current: dict[str, str] | None = None + notice_dates: list[str] = [] + + for line in lines[start_index:]: + if line.startswith("Page ") or line.startswith("Next") or line.startswith("Previous"): + break + + match = ROW_START_RE.match(line) + if match: + if current: + current["date_notice_sent"] = ", ".join(notice_dates) + records.append(current) + current = { + "organization_name": match.group("organization"), + "date_reported": match.group("reported"), + "date_of_breach": match.group("breach"), + "date_of_discovery": match.group("discovery"), + } + notice_dates = [] + continue + + if current is None: + continue + + if re.fullmatch(r"\d[\d,]*", line): + current["number_affected"] = line + current["date_notice_sent"] = ", ".join(notice_dates) + records.append(current) + current = None + notice_dates = [] + continue + + if DATE_TOKEN_RE.search(line): + notice_dates.append(line) + + if current: + current["date_notice_sent"] = ", ".join(notice_dates) + records.append(current) + + return records + + +def parse_breach_table(html: str, base_url: str = DEFAULT_URL) -> list[dict[str, str]]: + records = _parse_html_table(html) + if not records: + records = _parse_flattened_text(html) + return normalize_records(records) diff --git a/pyproject.toml b/pyproject.toml index 1ccb125..5799e31 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,10 @@ target-version = "py310" [tool.ruff.lint] select = ["E", "F", "I", "UP", "B", "SIM", "C4"] +[tool.ruff.lint.per-file-ignores] +# Test fixtures embed long inline HTML/CSV sample strings. +"tests/**" = ["E501"] + [tool.mypy] python_version = "3.10" strict = true diff --git a/tests/test_ca_oag.py b/tests/test_ca_oag.py new file mode 100644 index 0000000..eb5b6a2 --- /dev/null +++ b/tests/test_ca_oag.py @@ -0,0 +1,125 @@ +import json +import unittest + +from breach_scraper.sources.ca_oag import parse_breach_table, parse_list_table + +LIST_HTML = """ + + + + + + + + + + + + + + + + + + + + + + +
Organization NameDate(s) of BreachReported Date
OSI Systems, Inc.12/23/202503/11/2026
SitusAMC Holdings Corporation11/13/2025, 11/21/202503/10/2026
+ + +""" + +DIV_LIST_HTML = """ + + +
+
+ OSI Systems, Inc. +
+
12/23/2025
+
03/11/2026
+
+
+
+ SitusAMC Holdings Corporation +
+
11/13/2025, 11/21/2025
+
03/10/2026
+
+ + +""" + + +DETAIL_HTML = """ + + +

Submitted Breach Notification Sample

+

Sample of Notice:

+ OSI Systems Inc. - Sample Notice.pdf +

Organization Name:

+

OSI Systems, Inc.

+

Date(s) of Breach (if known):

+

Tuesday, December 23, 2025

+ + +""" + + +class TestCaliforniaOagScraper(unittest.TestCase): + def test_parse_list_table_extracts_rows(self) -> None: + records = parse_list_table(LIST_HTML, base_url="https://oag.ca.gov/privacy/databreach/list") + + self.assertEqual(len(records), 2) + self.assertEqual(records[0]["organization_name"], "OSI Systems, Inc.") + self.assertEqual( + records[0]["organization_name_url"], + "https://oag.ca.gov/ecrime/databreach/reports/sb24-619986", + ) + self.assertEqual(records[0]["date_of_breach"], "12/23/2025") + self.assertEqual(records[0]["date_reported"], "03/11/2026") + + def test_parse_list_table_extracts_rows_from_div_layout(self) -> None: + records = parse_list_table( + DIV_LIST_HTML, base_url="https://oag.ca.gov/privacy/databreach/list" + ) + + self.assertEqual(len(records), 2) + self.assertEqual(records[0]["organization_name"], "OSI Systems, Inc.") + self.assertEqual(records[0]["date_of_breach"], "12/23/2025") + self.assertEqual(records[0]["date_reported"], "03/11/2026") + self.assertEqual(records[1]["date_of_breach"], "11/13/2025, 11/21/2025") + + def test_parse_breach_table_merges_notice_link_from_detail_pages(self) -> None: + payload = json.dumps( + { + "list_url": "https://oag.ca.gov/privacy/databreach/list", + "records": parse_list_table( + DIV_LIST_HTML, base_url="https://oag.ca.gov/privacy/databreach/list" + ), + "detail_pages": [ + { + "url": "https://oag.ca.gov/ecrime/databreach/reports/sb24-619986", + "html": DETAIL_HTML, + } + ], + } + ) + + records = parse_breach_table(payload, base_url="https://oag.ca.gov/privacy/databreach/list") + + self.assertEqual(len(records), 2) + self.assertEqual(records[0]["organization_name"], "OSI Systems, Inc.") + self.assertEqual(records[0]["date_of_breach"], "2025-12-23") + self.assertEqual(records[0]["date_reported"], "2026-03-11") + self.assertEqual( + records[0]["notice_url"], + "https://oag.ca.gov/system/files/attachments/press-docs/osi_notice.pdf", + ) + self.assertEqual(records[0]["notice"], "OSI Systems Inc. - Sample Notice.pdf") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cli.py b/tests/test_cli.py index 4abf875..e289ddc 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -39,7 +39,9 @@ def read(self) -> bytes: class TestCliOffline(unittest.TestCase): def test_offline_json_annotates_source(self) -> None: - rc, out, _ = run(["--input-html", str(FIXTURE), "--output", "json", *RANGE]) + rc, out, _ = run( + ["--input-html", str(FIXTURE), "--source", "wa_atg", "--output", "json", *RANGE] + ) self.assertEqual(rc, 0) data = json.loads(out) self.assertEqual(len(data), 2) @@ -50,18 +52,22 @@ def test_offline_json_annotates_source(self) -> None: ) def test_offline_markdown_header(self) -> None: - rc, out, _ = run(["--input-html", str(FIXTURE), "--output", "markdown", *RANGE]) + rc, out, _ = run( + ["--input-html", str(FIXTURE), "--source", "wa_atg", "--output", "markdown", *RANGE] + ) self.assertEqual(rc, 0) self.assertIn("| Source | Date Reported | Organization Name", out) def test_default_window_filters_old_records(self) -> None: # Default range is the last 6 months; the 2024 fixture rows fall outside it. - rc, out, _ = run(["--input-html", str(FIXTURE), "--output", "json"]) + rc, out, _ = run(["--input-html", str(FIXTURE), "--source", "wa_atg", "--output", "json"]) self.assertEqual(rc, 0) self.assertEqual(json.loads(out), []) def test_offline_csv_has_unix_line_endings(self) -> None: - rc, out, _ = run(["--input-html", str(FIXTURE), "--output", "csv", *RANGE]) + rc, out, _ = run( + ["--input-html", str(FIXTURE), "--source", "wa_atg", "--output", "csv", *RANGE] + ) self.assertEqual(rc, 0) self.assertNotIn("\r", out) self.assertIn("date_reported,organization_name", out) @@ -71,7 +77,7 @@ class TestCliOnline(unittest.TestCase): @mock.patch("breach_scraper.http.urlopen") def test_online_pipeline(self, mock_urlopen: mock.Mock) -> None: mock_urlopen.return_value = _FakeResponse(FIXTURE.read_bytes()) - rc, out, _ = run(["--output", "json", *RANGE]) + rc, out, _ = run(["--source", "wa_atg", "--output", "json", *RANGE]) self.assertEqual(rc, 0) self.assertEqual(len(json.loads(out)), 2) @@ -80,7 +86,8 @@ class TestCliMisc(unittest.TestCase): def test_list_sources(self) -> None: rc, out, _ = run(["--list-sources"]) self.assertEqual(rc, 0) - self.assertIn("wa_atg", out) + for key in ("wa_atg", "ca_oag", "or_doj", "maine_ag", "hhs_ocr"): + self.assertIn(key, out) def test_unknown_source_errors(self) -> None: rc, _, err = run(["--source", "nope", *RANGE]) diff --git a/tests/test_hhs_ocr.py b/tests/test_hhs_ocr.py new file mode 100644 index 0000000..93d9d2a --- /dev/null +++ b/tests/test_hhs_ocr.py @@ -0,0 +1,71 @@ +import unittest + +from breach_scraper.sources.hhs_ocr import _parse_csv_export, parse_breach_table + +HHS_HTML = """ + + +
Breach Report Results
+
Expand AllName of Covered Entity State Covered Entity Type Individuals Affected Breach Submission Date Type of Breach Location of Breached Information Business Associate Present Web Description
+
Manhattan Retirement Foundation d/b/a Meadowlark Hills KS Healthcare Provider 14442 02/26/2026 Hacking/IT Incident Network Server No
+
Commonwealth Care Alliance MA Health Plan 634 02/25/2026 Unauthorized Access/Disclosure Paper/Films No
+
44North MI Business Associate 2158 02/16/2026 Hacking/IT Incident Desktop Computer Yes
+
(Displaying 1 - 100 of 728)
+ + +""" + +HHS_CSV = """"javax.faces.component.UIPanel@22cd6ba5","State","Covered Entity Type","Individuals Affected","Breach Submission Date","Type of Breach","Location of Breached Information","javax.faces.component.UIPanel@7ef7c58e","Web Description" +"Manhattan Retirement Foundation d/b/a Meadowlark Hills","KS","Healthcare Provider","14442","02/26/2026","Hacking/IT Incident","Network Server","No","" +"Commonwealth Care Alliance","MA","Health Plan","634","02/25/2026","Unauthorized Access/Disclosure","Paper/Films","No","" +"44North","MI","Business Associate","2158","02/16/2026","Hacking/IT Incident","Desktop Computer","Yes","" +""" + + +class TestHhsOcrScraper(unittest.TestCase): + def test_parse_breach_table_extracts_rows(self) -> None: + records = parse_breach_table(HHS_HTML) + + self.assertEqual(len(records), 3) + self.assertEqual( + records[0]["organization_name"], + "Manhattan Retirement Foundation d/b/a Meadowlark Hills", + ) + self.assertEqual(records[0]["state"], "KS") + self.assertEqual(records[0]["covered_entity_type"], "Healthcare Provider") + self.assertEqual(records[0]["individuals_affected"], "14,442") + self.assertEqual(records[0]["date_reported"], "2026-02-26") + self.assertEqual(records[0]["type_of_breach"], "Hacking/IT Incident") + self.assertEqual(records[0]["location_of_breached_information"], "Network Server") + self.assertEqual(records[0]["business_associate_present"], "No") + + def test_parse_breach_table_handles_business_associate_rows(self) -> None: + records = parse_breach_table(HHS_HTML) + + business_associate = next( + record for record in records if record["organization_name"] == "44North" + ) + self.assertEqual(business_associate["covered_entity_type"], "Business Associate") + self.assertEqual(business_associate["business_associate_present"], "Yes") + + def test_parse_breach_table_handles_csv_export(self) -> None: + records = parse_breach_table(HHS_CSV) + + self.assertEqual(len(records), 3) + self.assertEqual( + records[0]["organization_name"], + "Manhattan Retirement Foundation d/b/a Meadowlark Hills", + ) + self.assertEqual(records[0]["individuals_affected"], "14,442") + + +class TestHhsCsvDetection(unittest.TestCase): + def test_html_is_not_treated_as_csv(self) -> None: + self.assertEqual(_parse_csv_export(" nope"), []) + + def test_csv_without_known_header_is_rejected(self) -> None: + self.assertEqual(_parse_csv_export('"a","b"\n"1","2"\n'), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_maine_ag.py b/tests/test_maine_ag.py new file mode 100644 index 0000000..179405f --- /dev/null +++ b/tests/test_maine_ag.py @@ -0,0 +1,150 @@ +import json +import unittest + +from breach_scraper.sources.maine_ag import extract_detail_links, parse_breach_table + +LIST_HTML = """ + + + + + +""" + + +DETAIL_HTML = """ + + +

Data Breach Notifications

+

Entity Information

+
    +
  • Type of Organization: Healthcare
    • +
  • Entity Name: Example Hospital
    • +
+

Breach Information

+
    +
  • Total number of persons affected (including residents): 2068
    • +
  • Total number of Maine residents affected: 1733
    • +
  • Date(s) Breach Occured: 11/22/2024
    • +
  • Date Breach Discovered: 07-01-2025
    • +
  • Description of the Breach: +
      +
    • External system breach (hacking)
      • +
    +
    • +
  • Information Acquired - Name or other personal identifier in combination with: +
      +
    • Social Security Number
      • +
    • Medical Information
      • +
    +
    • +
+

Notification and Protection Services

+
    +
  • Type of Notification: Written
    • +
  • Date(s) of consumer notification: 08/08/2025
    • +
  • Copy of notice to affected Maine residents: + Example Notice +
    • +
+ + +""" + +MALFORMED_DATE_DETAIL_HTML = """ + + +

Data Breach Notifications

+

Entity Information

+
    +
  • Type of Organization: Financial Services
    • +
  • Entity Name: Example CPA
    • +
+

Notification and Protection Services

+
    +
  • Date(s) of consumer notification: 03/17/2-26
    • +
+

Breach Information

+
    +
  • Date(s) Breach Occured: 03132026
    • +
+ + +""" + + +class TestMaineAgScraper(unittest.TestCase): + def test_extract_detail_links_finds_uuid_detail_pages(self) -> None: + links = extract_detail_links( + LIST_HTML, + base_url="https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/list.html", + ) + + self.assertEqual(len(links), 2) + self.assertEqual(links[0]["text"], "Example Hospital") + self.assertTrue(links[0]["url"].endswith("11111111-1111-1111-1111-111111111111.html")) + + def test_parse_breach_table_parses_detail_bundle(self) -> None: + payload = json.dumps( + { + "list_url": "https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/list.html", + "detail_pages": [ + { + "url": "https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/11111111-1111-1111-1111-111111111111.html", + "html": DETAIL_HTML, + } + ], + } + ) + + records = parse_breach_table( + payload, + base_url="https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/list.html", + ) + + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["organization_name"], "Example Hospital") + self.assertEqual(records[0]["date_of_breach"], "2024-11-22") + self.assertEqual(records[0]["date_reported"], "2025-08-08") + self.assertEqual(records[0]["number_of_maine_residents_affected"], "1,733") + self.assertEqual(records[0]["total_persons_affected"], "2,068") + self.assertEqual( + records[0]["notice_url"], + "https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/docs/example-notice.pdf", + ) + self.assertIn("Social Security Number", records[0]["information_compromised"]) + + def test_parse_breach_table_can_fall_back_to_list_page_links(self) -> None: + records = parse_breach_table( + LIST_HTML, + base_url="https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/list.html", + ) + + self.assertEqual(len(records), 2) + self.assertEqual(records[0]["organization_name"], "Example Hospital") + self.assertTrue(records[0]["organization_name_url"].endswith(".html")) + + def test_parse_breach_table_repairs_malformed_live_dates(self) -> None: + payload = json.dumps( + { + "list_url": "https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/list.html", + "detail_pages": [ + { + "url": "https://www.maine.gov/agviewer/content/ag/985235c7-cb95-4be2-8792-a1252b4f8318/33333333-3333-3333-3333-333333333333.html", + "html": MALFORMED_DATE_DETAIL_HTML, + } + ], + } + ) + + records = parse_breach_table(payload) + + self.assertEqual(records[0]["date_of_breach"], "2026-03-13") + self.assertEqual(records[0]["date_reported"], "2026-03-17") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_or_doj.py b/tests/test_or_doj.py new file mode 100644 index 0000000..b04dcd3 --- /dev/null +++ b/tests/test_or_doj.py @@ -0,0 +1,112 @@ +import unittest + +from breach_scraper.sources.or_doj import _normalize_dateish, parse_breach_table + +OREGON_HTML = """ + + +

Search Data Breaches

+
Organization Reported Date Dates of Breach Dates of Discovery Date Notice Sent Number Affected
+
Pyramid Advisors Limited Partnership d/b/a Pyramid Global Hospitality 02/25/2026 8/13/2025 - 8/14/2025 8/14/2025
+
10/29/2025
+
2/25/2026
+
139899
+
Adapt Oregon Health Care 02/24/2026 1/1/2001 1/1/2001
+
1/1/2001
+
2908
+
Decisely Insurance Services 12/30/2025 12/15/2024 - 12/17/2024 12/17/2024
+
1/1/0001
+
6/13/2025
+
7/15/2025
+
261155
+ + +""" + +OREGON_TABLE_HTML = """ + + + + + + + + + + + + + + + + + + + + + + + +
OrganizationReported DateDates of BreachDates of DiscoveryDate Notice SentNumber Affected
Pyramid Advisors Limited Partnership d/b/a Pyramid Global Hospitality02/25/20268/13/2025 - 8/14/20258/14/2025
10/29/2025
2/25/2026
139899
+ + +""" + + +class TestOregonDojScraper(unittest.TestCase): + def test_parse_breach_table_extracts_rows_and_normalizes_dates(self) -> None: + records = parse_breach_table(OREGON_HTML) + + self.assertEqual(len(records), 3) + self.assertEqual( + records[0]["organization_name"], + "Pyramid Advisors Limited Partnership d/b/a Pyramid Global Hospitality", + ) + self.assertEqual(records[0]["date_reported"], "2026-02-25") + self.assertEqual(records[0]["date_of_breach"], "2025-08-13 to 2025-08-14") + self.assertEqual(records[0]["date_of_discovery"], "2025-08-14") + self.assertEqual(records[0]["date_notice_sent"], "2025-10-29, 2026-02-25") + self.assertEqual(records[0]["number_affected"], "139,899") + + def test_parse_breach_table_drops_placeholder_dates(self) -> None: + records = parse_breach_table(OREGON_HTML) + + adapt = next( + record + for record in records + if record["organization_name"] == "Adapt Oregon Health Care" + ) + self.assertEqual(adapt["date_of_breach"], "") + self.assertEqual(adapt["date_of_discovery"], "") + self.assertEqual(adapt["date_notice_sent"], "") + self.assertEqual(adapt["number_affected"], "2,908") + + decisely = next( + record + for record in records + if record["organization_name"] == "Decisely Insurance Services" + ) + self.assertEqual(decisely["date_notice_sent"], "2025-06-13, 2025-07-15") + + def test_parse_breach_table_handles_live_table_markup(self) -> None: + records = parse_breach_table(OREGON_TABLE_HTML) + + self.assertEqual(len(records), 1) + self.assertEqual( + records[0]["organization_name"], + "Pyramid Advisors Limited Partnership d/b/a Pyramid Global Hospitality", + ) + self.assertEqual(records[0]["date_notice_sent"], "2025-10-29, 2026-02-25") + self.assertEqual(records[0]["number_affected"], "139,899") + + +class TestOrDojDateNormalization(unittest.TestCase): + def test_tight_range_is_split(self) -> None: + self.assertEqual(_normalize_dateish("8/13/2025-8/14/2025"), "2025-08-13 to 2025-08-14") + + def test_invalid_date_returns_original(self) -> None: + self.assertEqual(_normalize_dateish("13/45/2025"), "13/45/2025") + + +if __name__ == "__main__": + unittest.main()