104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
"""Fetch the latest (newest) NCMDR weekly alert row from the SFDA website."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from urllib.parse import urljoin
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from sfda_parser.sfda_scraper.http_client import SFDA_BASE, create_session
|
|
|
|
WEEKLY_ALERT_URL = f"{SFDA_BASE}/en/weekly-alert"
|
|
|
|
# TEMPORARY: how many recent weekly bulletins to import/scan. Set to 1 for latest-only.
|
|
WEEKLY_BULLETINS_TO_IMPORT = 1
|
|
|
|
|
|
@dataclass
|
|
class WeeklyAlertRow:
|
|
title: str
|
|
alert_date: date | None
|
|
pdf_url: str
|
|
|
|
|
|
def _parse_date(value: str) -> date | None:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
return None
|
|
try:
|
|
return date.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _parse_table_row(tr) -> WeeklyAlertRow | None:
|
|
title_el = tr.select_one(".views-field-title")
|
|
date_el = tr.select_one(".views-field-field-date")
|
|
file_el = tr.select_one(".views-field-field-file a[href]")
|
|
|
|
if not title_el or not file_el:
|
|
return None
|
|
|
|
title = title_el.get_text(strip=True)
|
|
date_str = date_el.get_text(strip=True) if date_el else ""
|
|
href = file_el.get("href")
|
|
if not title or not href:
|
|
return None
|
|
|
|
return WeeklyAlertRow(
|
|
title=title,
|
|
alert_date=_parse_date(date_str),
|
|
pdf_url=urljoin(SFDA_BASE, href),
|
|
)
|
|
|
|
|
|
def fetch_all_weekly_alerts(session=None) -> list[WeeklyAlertRow]:
|
|
"""Return every weekly bulletin row from the SFDA table (newest usually listed first)."""
|
|
session = session or create_session()
|
|
response = session.get(WEEKLY_ALERT_URL, timeout=60)
|
|
response.raise_for_status()
|
|
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
rows: list[WeeklyAlertRow] = []
|
|
|
|
for tr in soup.select("tr"):
|
|
parsed = _parse_table_row(tr)
|
|
if parsed:
|
|
rows.append(parsed)
|
|
|
|
if not rows:
|
|
raise ValueError("Could not find any weekly alert rows on SFDA page")
|
|
|
|
return rows
|
|
|
|
|
|
def _sort_weekly_alerts_newest_first(rows: list[WeeklyAlertRow]) -> list[WeeklyAlertRow]:
|
|
"""Newest bulletin first; top-of-page wins when dates tie."""
|
|
indexed = list(enumerate(rows))
|
|
indexed.sort(
|
|
key=lambda item: (item[1].alert_date or date.min, -item[0]),
|
|
reverse=True,
|
|
)
|
|
return [item[1] for item in indexed]
|
|
|
|
|
|
def fetch_recent_weekly_alerts(
|
|
session=None, count: int | None = None
|
|
) -> list[WeeklyAlertRow]:
|
|
"""
|
|
Return the N newest weekly bulletins (default: WEEKLY_BULLETINS_TO_IMPORT).
|
|
|
|
Order: newest first. Use reversed() when importing oldest-to-newest.
|
|
"""
|
|
if count is None:
|
|
count = WEEKLY_BULLETINS_TO_IMPORT
|
|
rows = _sort_weekly_alerts_newest_first(fetch_all_weekly_alerts(session=session))
|
|
return rows[: max(1, count)]
|
|
|
|
|
|
def fetch_latest_weekly_alert(session=None) -> WeeklyAlertRow:
|
|
"""Return the single newest weekly bulletin."""
|
|
return fetch_recent_weekly_alerts(session=session, count=1)[0]
|