"""Parse Affected Device List tables from ADE detail PDFs.""" from __future__ import annotations import io import re from dataclasses import dataclass import pdfplumber # Order matters: more specific aliases are checked first within each field. HEADER_ALIASES: dict[str, tuple[str, ...]] = { "catalog_number": ( "procedure pack code", "product code / part number", "product code", "manufacturer's product number/catalog number", "product number/catalog number", "catalogue number", "catalog number", "model number (ref)", "model number", "model no", "model #", "product number", "part number", "item number", "list number", "system code", "unit model", "article no", "ref. number", "ref#", "ref no", "catalog no", "catalogue no", "material n° (upn)", "material n°", "material id", "upn", "ref", ), "material_description": ( "material description", "product description", "model description", "material desc", "description", "product code", ), "serial_no": ( "unit serial number", "unit serial no", "system serial number", "system serial no", "affected serial numbers", "lot/serial number", "serial numbers", "serial number", "serial no.", "serial no", "s/n", "sn", "lot numbers", "lot number", "batch lot", "lot id", "lot no.", "lot no", "lot", ), "udi": ( "udi number", "udi code", "udi impacted", "unique device identifier", "basic udi di", "basic udi-di", "udi di", "udi-di", "udi - di", "udi", ), "material": ( "affected product name", "commercial name", "device name", "material", "product name", "model name", "product/trade name", "trade name", "brand name", "system model", "unit model", ), "gstn": ( "gtin number", "gtin", "gstn", "gtn", ), "batch": ( "affected lot numbers", "batch / lot number", "batch/lot number", "lot / batch n°", "batch", ), } # Map fields in specificity order so e.g. "model number" maps to catalog_number, not material. FIELD_PRIORITY = ( "catalog_number", "material_description", "serial_no", "udi", "material", "gstn", "batch", ) PRODUCT_SECTION_KEYWORDS = ( "affected device", "affected devices", "impacted product", "affected product", "affected products", "product table", "table 1", "affected product information", "affected product details", "product affected", "product details", "list of affected", "affected device list", "device list", "3. affected products", "how to identify them", "affected product codes and batches", "procedure pack code", "detail on affected devices", ) FORM_TABLE_KEYWORDS = ( "signature", "phone number", "fax", "e-mail", "email address", "printed name", "facility name", "purchased directly", "end user with", "quantity for return", "customer contact", "account/organisation", "qty sent", "qty to return", "quantity distributed", "quantity in inventory", "quantity being returned", "customer po", "po no.", "job-no.", "distributor", "distribution dates", "actions to be", "recalibration needed", "pressure circ-value", "software name", "previous range", "updated range", "previous target", "updated target", ) # Section titles / narrative blocks mis-detected as product tables. NON_PRODUCT_HEADER_KEYWORDS = ( "affected product information table", "please follow your local", "correction letter dated", "figure ", "this issue may potentially result", "product(s) distributed", "name/title", "analyte", # NOTE: do not include generic table columns like expiry/manufacturing dates here. # They can appear in real affected-product tables. ) def _is_strong_product_header(header_text: str) -> bool: # Accept product tables even if they include operational columns like distribution/expiry dates. strong_terms = ( "procedure pack", "udi", "gtin", "catalog", "catalogue", "batch", "lot", "serial", "product description", "material description", "device name", "product name", "trade name", "system model", "unit model", ) if sum(1 for t in strong_terms if t in header_text) >= 2: return True # Installed-base / distribution list tables (e.g. System Model + Unit serial number). if "system model" in header_text and "serial" in header_text: return True if "unit model" in header_text and "serial" in header_text: return True return False @dataclass class DeviceRow: material: str material_description: str catalog_number: str udi: str serial_no: str gstn: str batch: str def as_dict(self) -> dict[str, str]: return { "material": self.material, "material_description": self.material_description, "catalog_number": self.catalog_number, "udi": self.udi, "serial_no": self.serial_no, "gstn": self.gstn, "batch": self.batch, } # Multi-value identifier fields in one cell → one ERP row per value. _MULTI_VALUE_DEVICE_FIELDS = ("serial_no", "catalog_number", "batch", "udi", "gstn") _NUMERIC_ID_TOKEN_RE = re.compile(r"^\d{8,14}$") _LOT_CODE_TOKEN_RE = re.compile(r"^\d{2}[A-Z]\d{4}$", re.IGNORECASE) _SRN_ID_TOKEN_RE = re.compile(r"^[A-Z]{2}-MF-\d{9}$", re.IGNORECASE) _ALPHANUM_CODE_TOKEN_RE = re.compile(r"^[A-Z0-9][A-Z0-9\-\/\.]+$", re.IGNORECASE) _DATE_VALUE_RES = ( re.compile(r"^\d{1,2}[-/\s][A-Z]{3,9}[-/\s]\d{2,4}$", re.IGNORECASE), re.compile(r"^\d{1,2}[-/]\d{1,2}[-/]\d{2,4}$"), re.compile(r"^\d{4}[-/]\d{1,2}[-/]\d{1,2}$"), re.compile(r"^\d{1,2}\s+[A-Z]{3,9}\s+\d{4}$", re.IGNORECASE), ) _PRODUCT_CONCERNED_PREFIX_RE = re.compile( r"^(?:(?:name\s+)?of\s+the\s+product\s+concerned\s*:\s*)+", re.IGNORECASE, ) def _is_likely_date_value(value: str) -> bool: candidate = (value or "").strip() if not candidate: return False return any(pattern.fullmatch(candidate) for pattern in _DATE_VALUE_RES) _PROSE_PREFIXES = ( "dear ", "the ", "manufacturer", "please ", "required ", "potential ", "transmission", "contact ", "action", "[enter", "incorrectly", "manufactured:", "alternatively", "you can", "this notice", "and notify", "in the image", "consequently", "ensure effectiveness", "radiological", "examination is", "the diagnosis", "the correct", "the dimensions", "the synthetic", "the x-ray", ) def _is_likely_prose(value: str) -> bool: """Reject narrative sentences mis-parsed as serial/UDI identifiers.""" candidate = (value or "").strip() if not candidate: return True parts = [part.strip() for part in re.split(r"[,;]+", candidate) if part.strip()] if len(parts) > 1: valid_parts = sum( 1 for part in parts if not _is_likely_date_value(part) and ( _is_splittable_token(part) or bool(re.fullmatch(r"\d{6,20}", part)) ) ) if valid_parts >= 2: return False if len(candidate) > 45 and len(parts) <= 1: return True lower = candidate.lower() if any(lower.startswith(prefix) for prefix in _PROSE_PREFIXES): return True if " " in candidate: words = candidate.split() if len(words) >= 3: return True if lower.endswith(":") and len(words) >= 2: return True if re.search(r"[.!?]", candidate) and len(candidate) > 20: return True return False def _is_valid_device_identifier(value: str) -> bool: candidate = (value or "").strip() if not candidate or _is_likely_date_value(candidate) or _is_likely_prose(candidate): return False if _is_splittable_token(candidate): return True return bool(re.fullmatch(r"\d{6,20}", candidate)) def _split_identifier_list(value: str) -> list[str]: value = (value or "").strip() if not value: return [] if re.search(r"[,;]", value): parts = [ part.strip() for part in re.split(r"[,;]+", value) if part.strip() and _is_valid_device_identifier(part.strip()) ] if parts: return parts tokens = value.split() if len(tokens) >= 2 and all(_is_valid_device_identifier(token) for token in tokens): return tokens if _is_valid_device_identifier(value): return [value] return [] def _paired_rows_from_summary(summary: dict[str, str]) -> list[DeviceRow]: """Zip parallel UDI-DI and SN lists from summary labels into one row per device.""" udi_list = _split_identifier_list(summary.get("udi", "")) sn_list = _split_identifier_list(summary.get("serial_no", "")) if not udi_list or not sn_list: return [] count = min(len(udi_list), len(sn_list)) if count < 2: return [] material = _clean_material_value(summary.get("material", "")) material_description = summary.get("material_description", "") return [ DeviceRow( material=material, material_description=material_description, catalog_number="", udi=udi_list[index], serial_no=sn_list[index], gstn="", batch="", ) for index in range(count) ] def _is_splittable_token(token: str) -> bool: """True when a whitespace-separated token looks like an ID/code, not prose.""" token = (token or "").strip() if not token or " " in token or _is_likely_date_value(token): return False if _NUMERIC_ID_TOKEN_RE.fullmatch(token): return True if _LOT_CODE_TOKEN_RE.fullmatch(token): return True if _SRN_ID_TOKEN_RE.fullmatch(token): return True return ( len(token) <= 25 and _ALPHANUM_CODE_TOKEN_RE.fullmatch(token) is not None and re.search(r"\d", token) is not None ) def _split_multi_values(value: str) -> list[str]: value = re.sub(r"[\n\r]+", ", ", (value or "").strip()) if not value: return [""] if re.search(r"[,;]", value): parts = [ part.strip() for part in re.split(r"[,;]+", value) if part.strip() and not _is_likely_date_value(part.strip()) ] if len(parts) > 1: return parts # Some ADE PDFs list UDIs/serials as space-separated 10-digit numbers in one cell. tokens = value.split() if len(tokens) >= 2 and all(_is_splittable_token(token) for token in tokens): return [token for token in tokens if not _is_likely_date_value(token)] return [value] def _expand_device_row(row: DeviceRow) -> list[DeviceRow]: """Split one parsed row into many when serial, catalog, or batch lists several values.""" field_parts = { field: _split_multi_values(getattr(row, field)) for field in _MULTI_VALUE_DEVICE_FIELDS } multi_fields = [field for field, parts in field_parts.items() if len(parts) > 1] if not multi_fields: return [row] if len(multi_fields) == 1: field = multi_fields[0] return [ DeviceRow(**{**row.as_dict(), field: part}) for part in field_parts[field] ] lengths = {field: len(field_parts[field]) for field in multi_fields} if len(set(lengths.values())) == 1: count = next(iter(lengths.values())) return [ DeviceRow( **{ **row.as_dict(), **{field: field_parts[field][index] for field in multi_fields}, } ) for index in range(count) ] primary_field = max(multi_fields, key=lambda field: len(field_parts[field])) expanded: list[DeviceRow] = [] for part in field_parts[primary_field]: row_data = row.as_dict() row_data[primary_field] = part for field in multi_fields: if field != primary_field: row_data[field] = field_parts[field][0] expanded.append(DeviceRow(**row_data)) return expanded def _expand_device_rows(devices: list[DeviceRow]) -> list[DeviceRow]: expanded: list[DeviceRow] = [] for device in devices: expanded.extend(_expand_device_row(device)) return expanded def _normalize_header(cell: str | None) -> str: if not cell: return "" return re.sub(r"\s+", " ", str(cell).strip().lower()) _DATE_COLUMN_TERMS = ( "date", "manufacturing date", "mfg date", "prod date", "production date", "expiry date", "expiration date", ) def _is_date_column_header(header: str) -> bool: normalized = _normalize_header(header) if not normalized: return False if normalized in ("date", "dates"): return True if "update" in normalized: return False return any(term in normalized for term in _DATE_COLUMN_TERMS) def _clean_material_value(value: str) -> str: cleaned = (value or "").strip() if not cleaned: return "" concerned_match = re.search( r"product\s+concerned\s*:\s*(.+)$", cleaned, re.IGNORECASE, ) if concerned_match: return concerned_match.group(1).strip() while True: stripped = _PRODUCT_CONCERNED_PREFIX_RE.sub("", cleaned).strip() if stripped == cleaned: break cleaned = stripped return cleaned def _clean_serial_value(value: str) -> str: cleaned = re.sub(r"[\n\r]+", ", ", (value or "").strip()) if not cleaned or _is_likely_date_value(cleaned): return "" if re.search(r"[,;]", cleaned): parts = [ part.strip() for part in re.split(r"[,;]+", cleaned) if part.strip() and not _is_likely_date_value(part.strip()) ] valid_parts = [ part for part in parts if not _is_likely_prose(part) or _is_valid_device_identifier(part) ] if len(valid_parts) > 1: return ", ".join(valid_parts) if len(valid_parts) == 1: return valid_parts[0] return "" if _is_likely_prose(cleaned): return "" return cleaned def _sanitize_device_row(row: DeviceRow) -> DeviceRow: data = row.as_dict() data["material"] = _clean_material_value(data["material"]) data["serial_no"] = _clean_serial_value(data["serial_no"]) return DeviceRow(**data) def _header_matches(header: str, alias: str) -> bool: if not header or not alias: return False if header == alias: return True if len(header) <= 3 or len(alias) <= 3: return False # e.g. header "product" must not match alias "product code" if len(header) < len(alias) and header in alias: return False return alias in header or header in alias def _map_columns_base(header_row: list) -> dict[str, int]: normalized = [_normalize_header(c) for c in header_row] mapping: dict[str, int] = {} used_indices: set[int] = set() for field in FIELD_PRIORITY: aliases = sorted(HEADER_ALIASES[field], key=len, reverse=True) for idx, header in enumerate(normalized): if idx in used_indices or not header: continue if field == "serial_no" and _is_date_column_header(header): continue if any(_header_matches(header, alias) for alias in aliases): mapping[field] = idx used_indices.add(idx) break return mapping def _apply_column_heuristics(header_row: list, mapping: dict[str, int]) -> dict[str, int]: """GE-style tables: product name in column 0 with an empty header cell.""" normalized = [_normalize_header(c) for c in header_row] used_indices = set(mapping.values()) if ( "material" not in mapping and len(normalized) > 0 and not normalized[0] and 0 not in used_indices ): mapping["material"] = 0 # When both system and unit serial columns exist, prefer unit serial number. for idx, header in enumerate(normalized): if "unit serial" in header: mapping["serial_no"] = idx break return mapping def _merge_header_rows(row0: list, row1: list) -> list[str]: max_len = max(len(row0 or []), len(row1 or [])) merged: list[str] = [] for idx in range(max_len): parts: list[str] = [] for row in (row0, row1): if row and idx < len(row) and row[idx]: part = str(row[idx]).strip() if part: parts.append(part) merged.append(" ".join(parts).strip()) return merged _LOT_BATCH_TOKEN_RE = re.compile(r"\b\d{2}[A-Z]\d{4}\b") def _is_likely_header_continuation_row(row: list) -> bool: """Detect sub-header rows like ['Code', '', 'return'] under 'Procedure Pack'.""" if not row: return False cells = [str(cell).strip() for cell in row if cell is not None and str(cell).strip()] if not cells: return False for cell in cells: if _LOT_BATCH_TOKEN_RE.search(cell): return False if len(cell) > 40: return False return True def _resolve_table_header(table: list) -> tuple[list, int]: """Return (header_row, data_start_index), merging split headers when needed.""" if not table: return [], 0 mapping = _map_columns(table[0]) if mapping and len(mapping) >= 2: return table[0], 1 if len(table) >= 3 and _is_likely_header_continuation_row(table[1]): merged = _merge_header_rows(table[0], table[1]) merged_mapping = _map_columns(merged) if merged_mapping and len(merged_mapping) >= 2: return merged, 2 return table[0], 1 def _map_columns(header_row: list) -> dict[str, int] | None: mapping = _apply_column_heuristics(header_row, _map_columns_base(header_row)) if len(mapping) < 2: return None return mapping def _get_cell_value(cells: list, idx: int | None) -> str: """Read cell; try adjacent columns when pdfplumber shifts values off header.""" if idx is None or idx >= len(cells): return "" for try_idx in (idx, idx - 1, idx + 1): if 0 <= try_idx < len(cells) and cells[try_idx] is not None: val = re.sub(r"[\n\r]+", ", ", str(cells[try_idx]).strip()) val = re.sub(r"\s+", " ", val).strip() if val: return val return "" def _cell_text(cells: list, idx: int) -> str: if idx < 0 or idx >= len(cells) or cells[idx] is None: return "" return re.sub(r"[\n\r]+", ", ", str(cells[idx]).strip()) def _best_cell_for_field(cells: list, field: str, used_indices: set[int]) -> str: """Scan row cells when sparse tables shift values away from header indices.""" best_value = "" best_score = -1 for idx, raw in enumerate(cells): if idx in used_indices or raw is None: continue text = _cell_text(cells, idx) if not text: continue score = 0 if field in ("serial_no", "batch"): parts = [p.strip() for p in re.split(r"[,;]+", text) if p.strip()] identifier_parts = sum(1 for part in parts if _is_valid_device_identifier(part)) if identifier_parts >= 2: score = 100 + identifier_parts elif identifier_parts == 1 and len(parts) == 1: score = 40 elif field in ("gstn", "udi"): tokens = _UDI_TOKEN_RE.findall(text) if tokens: score = 80 + len(tokens) elif field == "catalog_number": if text and not _UDI_TOKEN_RE.fullmatch(text) and len(text) <= 40: score = 30 elif field == "material": if text and not _UDI_TOKEN_RE.fullmatch(text): score = 20 if score > best_score: best_score = score best_value = text return best_value if best_score > 0 else "" def _resolve_sparse_row_values(cells: list, mapping: dict[str, int]) -> dict[str, str]: """Fill empty mapped fields from other cells in misaligned sparse tables.""" used_indices: set[int] = set() values: dict[str, str] = {} for field in FIELD_PRIORITY: idx = mapping.get(field) value = _get_cell_value(cells, idx) if value: values[field] = value if idx is not None: used_indices.add(idx) for offset in (-1, 1): neighbor = idx + offset if ( 0 <= neighbor < len(cells) and _cell_text(cells, neighbor) == value ): used_indices.add(neighbor) for field in ("serial_no", "gstn", "udi", "catalog_number", "batch", "material"): if values.get(field): continue fallback = _best_cell_for_field(cells, field, used_indices) if fallback: values[field] = fallback return values def _row_from_cells(cells: list, mapping: dict[str, int]) -> DeviceRow | None: resolved = _resolve_sparse_row_values(cells, mapping) row = DeviceRow( material=resolved.get("material", ""), material_description=resolved.get("material_description", ""), catalog_number=resolved.get("catalog_number", ""), udi=resolved.get("udi", ""), serial_no=resolved.get("serial_no", ""), gstn=resolved.get("gstn", ""), batch=resolved.get("batch", ""), ) if not any(row.as_dict().values()): return None return _sanitize_device_row(row) def _is_form_table(header_row: list) -> bool: text = " ".join(_normalize_header(c) for c in header_row) if not any(keyword in text for keyword in FORM_TABLE_KEYWORDS): return False # If it's clearly an affected-product table, don't treat it as a response/form table. return not _is_strong_product_header(text) def _is_product_table( header_row: list, page_text: str, mapping: dict[str, int] | None = None ) -> bool: header_text = " ".join(_normalize_header(c) for c in header_row) page_lower = page_text.lower() if _is_form_table(header_row): return False # Section numbering only (e.g. "3. affected products") — not a column header row. normalized_headers = [_normalize_header(c) for c in header_row if _normalize_header(c)] if normalized_headers and all( re.match(r"^\d+\.\s", h) or h in ("affected products", "actions to be taken") for h in normalized_headers ): return False if mapping is None: mapping = _map_columns(header_row) # If headers map cleanly and look product-like, accept even if they contain unrelated columns. if mapping and len(mapping) >= 2 and _is_strong_product_header(header_text): return True if any(keyword in header_text for keyword in NON_PRODUCT_HEADER_KEYWORDS): return False # Appendix-style tables (system code + commercial name) without section keywords. if mapping and len(mapping) >= 2: return True product_header_terms = ( "catalog", "model", "product", "udi", "gtin", "serial", "lot", "material", "ref", "part", "device", "batch", "system", "commercial", ) product_term_count = sum(1 for term in product_header_terms if term in header_text) if any(keyword in page_lower for keyword in PRODUCT_SECTION_KEYWORDS): return product_term_count >= 1 return product_term_count >= 2 def parse_affected_device_list(pdf_bytes: bytes) -> list[DeviceRow]: table_devices: list[DeviceRow] = [] all_lines: list[str] = [] with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf: for page in pdf.pages: text = page.extract_text() or "" all_lines.extend(_normalize_lines(text)) for table in page.extract_tables() or []: if not table or len(table) < 2: continue header, data_start = _resolve_table_header(table) mapping = _map_columns(header) if not mapping: continue if not _is_product_table(header, text, mapping=mapping): continue for row in table[data_start:]: device = _row_from_cells(row, mapping) if device: table_devices.append(device) summary, text_rows = _parse_text_content(all_lines) paired_rows = _paired_rows_from_summary(summary) if paired_rows and not table_devices: devices = paired_rows elif table_devices: devices = _enrich_table_devices_from_text(table_devices, summary, text_rows) else: devices = _devices_from_text_only(summary, text_rows) devices = [_sanitize_device_row(device) for device in devices] return _expand_device_rows(devices) _LABEL_RE = re.compile( r"^\s*(?P