79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
EMAIL_RE = re.compile(r"^[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}$", re.IGNORECASE)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExpectedRecipients:
|
|
addresses: tuple[str, ...] = ()
|
|
invalid_entries: tuple[str, ...] = ()
|
|
|
|
|
|
def load_expected_recipients(
|
|
path: str | Path | None,
|
|
*,
|
|
csv_column: str | None = "email",
|
|
) -> ExpectedRecipients:
|
|
if not path:
|
|
return ExpectedRecipients()
|
|
|
|
recipient_path = Path(path)
|
|
if recipient_path.suffix.lower() == ".csv":
|
|
return _load_csv_recipients(recipient_path, csv_column=csv_column or "email")
|
|
return _load_line_recipients(recipient_path)
|
|
|
|
|
|
def normalize_email_address(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
normalized = value.strip().lower()
|
|
if not normalized:
|
|
return None
|
|
return normalized if EMAIL_RE.fullmatch(normalized) else None
|
|
|
|
|
|
@dataclass
|
|
class _RecipientCollector:
|
|
addresses: dict[str, None] = field(default_factory=dict)
|
|
invalid_entries: list[str] = field(default_factory=list)
|
|
|
|
def add(self, value: str | None, *, source: str) -> None:
|
|
normalized = normalize_email_address(value)
|
|
if normalized:
|
|
self.addresses[normalized] = None
|
|
return
|
|
if value and value.strip():
|
|
self.invalid_entries.append(f"{source}: {value.strip()}")
|
|
|
|
def result(self) -> ExpectedRecipients:
|
|
return ExpectedRecipients(
|
|
addresses=tuple(self.addresses.keys()),
|
|
invalid_entries=tuple(self.invalid_entries),
|
|
)
|
|
|
|
|
|
def _load_line_recipients(path: Path) -> ExpectedRecipients:
|
|
collector = _RecipientCollector()
|
|
for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
collector.add(line, source=f"{path}:{line_number}")
|
|
return collector.result()
|
|
|
|
|
|
def _load_csv_recipients(path: Path, *, csv_column: str) -> ExpectedRecipients:
|
|
collector = _RecipientCollector()
|
|
with path.open(newline="", encoding="utf-8") as fh:
|
|
reader = csv.DictReader(fh)
|
|
if reader.fieldnames is None:
|
|
return collector.result()
|
|
column = csv_column if csv_column in reader.fieldnames else reader.fieldnames[0]
|
|
for line_number, row in enumerate(reader, start=2):
|
|
collector.add(row.get(column), source=f"{path}:{line_number}:{column}")
|
|
return collector.result()
|