d726dca50e
All checks were successful
CI / lint (push) Successful in 40s
CI / notebooks-smoke (push) Successful in 1m37s
Deploy / notebooks (push) Has been skipped
Deploy / zotero (push) Has been skipped
Deploy / docs (push) Has been skipped
Deploy / api (push) Has been skipped
Deploy / llm (push) Has been skipped
Deploy / mc (push) Has been skipped
Infra CI / notebooks (push) Successful in 1m5s
Infra CI / zotero (push) Successful in 13s
Infra CI / docs (push) Successful in 1m22s
Infra CI / api (push) Successful in 1m15s
Infra CI / llm (push) Successful in 44s
Infra CI / mc (push) Successful in 15s
Deploy / report (push) Successful in 13s
CI / test (push) Successful in 17m19s
Notebooks Integration / notebooks-integration (push) Successful in 7m27s
Zotero Sync / zotero-sync (push) Successful in 1m10s
Package Supply Chain / pkg-supply-chain (push) Successful in 59s
195 lines
6.8 KiB
Python
195 lines
6.8 KiB
Python
"""Fetch Federal Register PDF and TXT for all rules in bib.sqlite.
|
|
|
|
For each rule item with a ``document_number``, queries the Federal
|
|
Register API for the PDF and raw text URLs, downloads both, and
|
|
attaches them to the item in the bib store.
|
|
|
|
The FR TXT response is HTML-wrapped (``<html><pre>...``); when it looks
|
|
that way, the raw bytes are saved alongside as ``<doc>.html`` and the
|
|
``.txt`` is normalized in place with ``rex.frtext.clean_fr_text`` before
|
|
being attached — the bib attachment always points at the cleaned
|
|
``.txt``. See ``dev/scripts/migrate_fr_txt.py`` for the one-off
|
|
migration over files downloaded before this normalization existed.
|
|
|
|
Usage::
|
|
|
|
uv run python dev/scripts/fetch_fr_attachments.py # fetch all
|
|
uv run python dev/scripts/fetch_fr_attachments.py --dry-run # preview
|
|
uv run python dev/scripts/fetch_fr_attachments.py --key KEY # single item
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from conf import connect
|
|
from rex.frtext import normalize_txt_file
|
|
|
|
FR_API = "https://www.federalregister.gov/api/v1/documents"
|
|
|
|
|
|
def _extract_doc_number(url: str) -> str | None:
|
|
"""Extract FR document number from a federalregister.gov URL."""
|
|
m = re.search(r"/documents/\d{4}/\d{2}/\d{2}/([^/]+)/", url)
|
|
return m.group(1) if m else None
|
|
|
|
|
|
def _fetch_fr_metadata(doc_number: str) -> dict | None:
|
|
"""Query the FR API for a document's metadata."""
|
|
try:
|
|
resp = httpx.get(
|
|
f"{FR_API}/{doc_number}.json",
|
|
timeout=30,
|
|
follow_redirects=True,
|
|
)
|
|
if resp.status_code != 200:
|
|
return None
|
|
data = resp.json()
|
|
# Verify this is actually a CMS rule (not a collision)
|
|
agencies = [a.get("id") for a in data.get("agencies", [])]
|
|
if "centers-for-medicare-medicaid-services" not in agencies and data.get(
|
|
"agency_names"
|
|
):
|
|
# Might be a doc number collision — skip
|
|
return None
|
|
return data
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _download_file(url: str, dest: Path) -> bool:
|
|
"""Download a file to a local path."""
|
|
try:
|
|
resp = httpx.get(url, timeout=120, follow_redirects=True)
|
|
if resp.status_code == 200 and len(resp.content) > 100:
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_bytes(resp.content)
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
parser.add_argument("--key", default="", help="Process a single item key")
|
|
args = parser.parse_args()
|
|
|
|
store = connect.bib()
|
|
rules = store.list_items(item_type="rule")
|
|
|
|
if args.key:
|
|
rules = [r for r in rules if r.key == args.key]
|
|
|
|
# Get existing attachment filenames to avoid re-downloading
|
|
con = store._con()
|
|
existing_atts = set()
|
|
for row in con.execute("SELECT a.filename FROM attachments a").fetchall():
|
|
existing_atts.add(row[0])
|
|
|
|
download_dir = Path("data/fr_downloads")
|
|
stats = {"checked": 0, "downloaded": 0, "skipped": 0, "errors": 0}
|
|
|
|
for rule in rules:
|
|
stats["checked"] += 1
|
|
|
|
# Get document number from the item or parse from URL
|
|
doc_num = rule.document_number or _extract_doc_number(rule.url or "")
|
|
if not doc_num:
|
|
print(f" SKIP {rule.key}: no document number")
|
|
stats["skipped"] += 1
|
|
continue
|
|
|
|
# Check if we already have PDF/TXT for this rule
|
|
pdf_name = f"{doc_num}.pdf"
|
|
txt_name = f"{doc_num}.txt"
|
|
has_pdf = pdf_name in existing_atts
|
|
has_txt = txt_name in existing_atts
|
|
|
|
if has_pdf and has_txt:
|
|
stats["skipped"] += 1
|
|
continue
|
|
|
|
print(f" {rule.key}: {rule.title[:60]} doc={doc_num}")
|
|
|
|
# Construct URLs directly (FR API can be slow/unreliable)
|
|
# govinfo.gov PDF: https://www.govinfo.gov/content/pkg/FR-YYYY-MM-DD/pdf/DOC.pdf
|
|
# We can also try the FR API for the exact URL
|
|
meta = _fetch_fr_metadata(doc_num)
|
|
time.sleep(0.5) # Rate limit
|
|
|
|
if meta is None:
|
|
# Try constructing PDF URL from the rule's FR URL
|
|
# The FR URL has the date: /documents/YYYY/MM/DD/doc_number/
|
|
date_match = re.search(
|
|
r"/documents/(\d{4})/(\d{2})/(\d{2})/", rule.url or ""
|
|
)
|
|
if date_match:
|
|
y, m, d = date_match.groups()
|
|
pdf_url = f"https://www.govinfo.gov/content/pkg/FR-{y}-{m}-{d}/pdf/{doc_num}.pdf"
|
|
txt_url = f"https://www.federalregister.gov/documents/full_text/text/{y}/{m}/{d}/{doc_num}.txt"
|
|
else:
|
|
print(" ERROR: can't determine date from URL")
|
|
stats["errors"] += 1
|
|
continue
|
|
else:
|
|
pdf_url = meta.get("pdf_url", "")
|
|
txt_url = meta.get("raw_text_url", "")
|
|
# Construct txt URL if missing
|
|
if not txt_url and meta.get("publication_date"):
|
|
parts = meta["publication_date"].split("-")
|
|
if len(parts) == 3:
|
|
txt_url = (
|
|
f"https://www.federalregister.gov/documents/full_text/text/"
|
|
f"{parts[0]}/{parts[1]}/{parts[2]}/{doc_num}.txt"
|
|
)
|
|
|
|
if args.dry_run:
|
|
if not has_pdf:
|
|
print(f" [dry-run] would download PDF: {pdf_url}")
|
|
if not has_txt:
|
|
print(f" [dry-run] would download TXT: {txt_url}")
|
|
continue
|
|
|
|
# Download PDF
|
|
if not has_pdf and pdf_url:
|
|
pdf_path = download_dir / pdf_name
|
|
if _download_file(pdf_url, pdf_path):
|
|
store.attach_file(rule.key, pdf_path, title=f"FR {doc_num} (PDF)")
|
|
print(f" PDF: {pdf_path.stat().st_size:,} bytes")
|
|
stats["downloaded"] += 1
|
|
else:
|
|
print(" PDF: download failed")
|
|
stats["errors"] += 1
|
|
time.sleep(1)
|
|
|
|
# Download TXT
|
|
if not has_txt and txt_url:
|
|
txt_path = download_dir / txt_name
|
|
if _download_file(txt_url, txt_path):
|
|
normalize_txt_file(txt_path)
|
|
store.attach_file(rule.key, txt_path, title=f"FR {doc_num} (TXT)")
|
|
print(f" TXT: {txt_path.stat().st_size:,} bytes")
|
|
stats["downloaded"] += 1
|
|
else:
|
|
print(" TXT: download failed")
|
|
stats["errors"] += 1
|
|
time.sleep(1)
|
|
|
|
print(
|
|
f"\nDone: checked={stats['checked']} downloaded={stats['downloaded']} "
|
|
f"skipped={stats['skipped']} errors={stats['errors']}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|