75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
"""Parse Safety Alerts table from the NCMDR weekly update PDF."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import pdfplumber
|
|
|
|
NCMDR_REF_PATTERN = re.compile(r"^SA-\d", re.IGNORECASE)
|
|
|
|
|
|
@dataclass
|
|
class SafetyAlertRow:
|
|
ncmdr_ref: str
|
|
detail_url: str
|
|
|
|
|
|
def _collect_refs_from_page(page) -> list[str]:
|
|
refs: list[str] = []
|
|
for table in page.extract_tables() or []:
|
|
for row in table or []:
|
|
if not row:
|
|
continue
|
|
for cell in row:
|
|
if not cell:
|
|
continue
|
|
text = str(cell).strip()
|
|
if NCMDR_REF_PATTERN.match(text):
|
|
refs.append(text)
|
|
return refs
|
|
|
|
|
|
def _collect_links_from_page(page) -> list[str]:
|
|
hyperlinks = [
|
|
h
|
|
for h in (page.hyperlinks or [])
|
|
if h.get("uri") and "PublishDetails" in h["uri"]
|
|
]
|
|
hyperlinks.sort(key=lambda h: h.get("top", 0))
|
|
return [h["uri"] for h in hyperlinks]
|
|
|
|
|
|
def parse_safety_alerts_from_pdf(pdf_bytes: bytes) -> list[SafetyAlertRow]:
|
|
alerts: list[SafetyAlertRow] = []
|
|
|
|
with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf:
|
|
for page in pdf.pages:
|
|
refs = _collect_refs_from_page(page)
|
|
links = _collect_links_from_page(page)
|
|
if not refs:
|
|
continue
|
|
|
|
if len(refs) != len(links):
|
|
# Fallback: zip min length and log mismatch in importer
|
|
pair_count = min(len(refs), len(links))
|
|
refs = refs[:pair_count]
|
|
links = links[:pair_count]
|
|
|
|
for ncmdr_ref, detail_url in zip(refs, links):
|
|
alerts.append(SafetyAlertRow(ncmdr_ref=ncmdr_ref, detail_url=detail_url))
|
|
|
|
# Deduplicate by NCMDR ref while preserving order
|
|
seen: set[str] = set()
|
|
unique: list[SafetyAlertRow] = []
|
|
for alert in alerts:
|
|
key = alert.ncmdr_ref.strip().upper()
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
unique.append(alert)
|
|
|
|
return unique
|