From 2eaf18243f6f89ebded0612e84f8dfb11ba208ce Mon Sep 17 00:00:00 2001 From: David Huser <4357648+davidhuser@users.noreply.github.com> Date: Wed, 15 Apr 2026 19:46:43 +0200 Subject: [PATCH] - acc mod --- .gitignore | 3 - README.md | 4 +- pyproject.toml | 4 - src/mail_municipalities/accuracy/README.md | 331 ------- src/mail_municipalities/accuracy/__init__.py | 1 - src/mail_municipalities/accuracy/check.py | 203 ----- src/mail_municipalities/accuracy/collector.py | 187 ---- src/mail_municipalities/accuracy/config.py | 56 -- src/mail_municipalities/accuracy/metrics.py | 135 --- src/mail_municipalities/accuracy/models.py | 126 --- .../accuracy/ndr_parser.py | 414 --------- src/mail_municipalities/accuracy/report.py | 138 --- src/mail_municipalities/accuracy/sampler.py | 148 --- src/mail_municipalities/accuracy/sender.py | 156 ---- src/mail_municipalities/accuracy/state.py | 303 ------- src/mail_municipalities/cli.py | 440 --------- tests/test_accuracy.py | 854 ------------------ uv.lock | 94 -- 18 files changed, 1 insertion(+), 3596 deletions(-) delete mode 100644 src/mail_municipalities/accuracy/README.md delete mode 100644 src/mail_municipalities/accuracy/__init__.py delete mode 100644 src/mail_municipalities/accuracy/check.py delete mode 100644 src/mail_municipalities/accuracy/collector.py delete mode 100644 src/mail_municipalities/accuracy/config.py delete mode 100644 src/mail_municipalities/accuracy/metrics.py delete mode 100644 src/mail_municipalities/accuracy/models.py delete mode 100644 src/mail_municipalities/accuracy/ndr_parser.py delete mode 100644 src/mail_municipalities/accuracy/report.py delete mode 100644 src/mail_municipalities/accuracy/sampler.py delete mode 100644 src/mail_municipalities/accuracy/sender.py delete mode 100644 src/mail_municipalities/accuracy/state.py delete mode 100644 tests/test_accuracy.py diff --git a/.gitignore b/.gitignore index 96a0424..8bfcba2 100644 --- a/.gitignore +++ b/.gitignore @@ -2,9 +2,6 @@ data/*/cache.db data/*/cache.db-wal data/*/cache.db-shm -output/accuracy/*.db -output/accuracy/*.db-wal -output/accuracy/*.db-shm #output/export.xlsx output/~$export.xlsx output/domains/resolve_xx.log diff --git a/README.md b/README.md index 2503968..c664341 100644 --- a/README.md +++ b/README.md @@ -44,9 +44,7 @@ uv run scan ch -v # verbose (streams Docker output) ## Validation -The classifier is validated via bounce probing — see [`src/mail_municipalities/accuracy/README.md`](src/mail_municipalities/accuracy/README.md) for the full protocol. -This is a separate, manual process (not part of the pipeline above) that sends probe emails to a stratified sample of municipalities -and parses the resulting NDRs to identify the actual backend MTA. +The classifier is validated via bounce probing — a separate, manual process (not part of the pipeline above) that sends probe emails to a stratified sample of municipalities and parses the resulting NDRs to identify the actual backend MTA. The validation tooling is maintained in a separate repository (available on request). ## Output files diff --git a/pyproject.toml b/pyproject.toml index 722423a..3150b5d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,6 @@ description = "Domain resolution, provider classification, and security analysis readme = "README.md" requires-python = ">=3.13" dependencies = [ - "aiosmtplib>=3.0", "aiosqlite>=0.20.0", "dnspython>=2.8.0", "playwright>=1.50", @@ -13,14 +12,12 @@ dependencies = [ "httpx[http2]>=0.28.1", "loguru>=0.7.3", "pydantic>=2.12.5", - "pydantic-settings>=2.0", "python-slugify>=8.0.4", "stamina>=25.2.0", "tldextract>=5.3.1", "typer>=0.24.1", "pandas>=2.2", "openpyxl>=3.1", - "scikit-learn>=1.6", "seaborn>=0.13.2", "scipy>=1.17.1", ] @@ -47,7 +44,6 @@ resolve = "mail_municipalities.cli:resolve" classify = "mail_municipalities.cli:classify" analyze = "mail_municipalities.cli:analyze" scan = "mail_municipalities.cli:scan" -accuracy = "mail_municipalities.cli:accuracy" [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/src/mail_municipalities/accuracy/README.md b/src/mail_municipalities/accuracy/README.md deleted file mode 100644 index a09ef70..0000000 --- a/src/mail_municipalities/accuracy/README.md +++ /dev/null @@ -1,331 +0,0 @@ -# Bounce-Probe Accuracy Validation - -Validates the provider classifier by sending a single email to a provably -non-existent address at each sampled municipality domain, collecting the NDR -(bounce), and parsing its headers to identify the actual backend MTA. - -## Prerequisites - -1. **Gmail account** with an App Password (not your regular password - 2FA must be enabled): - - Go to https://myaccount.google.com/apppasswords - - Create an app password for "Mail" - - Enable IMAP: Gmail Settings > Forwarding and POP/IMAP > Enable IMAP - -2. **Provider classification output** must exist: - ``` - output/providers/providers_ch.json - output/providers/providers_de.json - output/providers/providers_at.json - ``` - -3. **Dependencies** installed: - ```bash - uv sync - ``` - -## Step 1: Configure credentials - -Create a `.env` file in the **project root** (it is already gitignored): - -```bash -cp .env.example .env # or create from scratch -``` - -Put your Gmail address and the App Password you just generated: - -```env -ACCURACY_SMTP_USER=your-account@gmail.com -ACCURACY_SMTP_PASSWORD=abcd efgh ijkl mnop -ACCURACY_IMAP_USER=your-account@gmail.com -ACCURACY_IMAP_PASSWORD=abcd efgh ijkl mnop -``` - -The SMTP and IMAP passwords are the same App Password (the 16-character code -Google gave you, with or without spaces). Both SMTP (sending) and IMAP -(collecting bounces) use the same Gmail account. - -All settings use the `ACCURACY_` prefix. See `config.py` for the full list. -Key defaults: - -| Variable | Default | Description | -|-----------------------------------|-------------------|--------------------------------| -| `ACCURACY_SMTP_HOST` | `smtp.gmail.com` | SMTP server | -| `ACCURACY_SMTP_PORT` | `587` | SMTP port (STARTTLS) | -| `ACCURACY_IMAP_HOST` | `imap.gmail.com` | IMAP server | -| `ACCURACY_IMAP_PORT` | `993` | IMAP port (SSL) | -| `ACCURACY_SEND_RATE_PER_SECOND` | `1.0` | Max emails/sec | -| `ACCURACY_SEND_BATCH_SIZE` | `25` | Emails per batch | -| `ACCURACY_SEND_BATCH_PAUSE_SECONDS` | `30.0` | Pause between batches (sec) | -| `ACCURACY_MAX_PROBES_PER_RUN` | `100` | Hard cap per invocation | -| `ACCURACY_NDR_MAX_WAIT_HOURS` | `24.0` | Max time to wait for bounces | - -## Quick smoke test (1-2 domains) - -Before running a real sample, verify the full send/collect cycle works with -just 2 probes: - -```bash -# 1. Sample only 2 municipalities from CH -uv run accuracy sample ch --size 2 --min-per-class 1 - -# 2. Check what would be sent -uv run accuracy send --dry-run - -# 3. Actually send (only 2 emails) -uv run accuracy send --no-dry-run --max-probes 2 - -# 4. Wait 1-2 minutes for bounces to arrive, then collect -uv run accuracy collect --poll-once - -# 5. Check status — you should see ndr_received > 0 -uv run accuracy status - -# 6. If NDRs were collected, view the report -uv run accuracy report -``` - -If `status` still shows `sent: 2, ndr_received: 0` after a few minutes, wait -a bit longer and run `collect --poll-once` again. Some mail servers take -5-10 minutes to generate the bounce. - -Once this works, reset the state to start a real run: - -```bash -rm output/accuracy/state.db -``` - -## Step 2: Create a sample - -Pick a country and sample size. This creates probe records in a local SQLite -database (`output/accuracy/state.db`) but sends nothing. - -```bash -# Sample 50 Swiss municipalities (stratified by provider class) -uv run accuracy sample ch --size 50 - -# Sample across all three countries -uv run accuracy sample --all --size 200 --min-per-class 10 - -# Adjust minimum per provider class (default 5) -uv run accuracy sample ch --size 100 --min-per-class 8 -``` - -Review the summary table. Re-running `sample` for the same country is safe -- -domains already in the database are skipped. - -## Step 3: Verify with a dry run - -By default, `send` is in dry-run mode. It prints what would be sent without -touching the network: - -```bash -uv run accuracy send --dry-run -``` - -Check the probe list, domains, and recipient addresses. - -## Step 4: Send probes - -Disable dry-run explicitly. The tool will ask you to type `YES` before -sending: - -```bash -# Interactive confirmation (recommended for first run) -uv run accuracy send --no-dry-run - -# Skip the interactive prompt (e.g. on a remote VM) -uv run accuracy send --no-dry-run --confirm - -# Limit to 10 probes for an initial test -uv run accuracy send --no-dry-run --max-probes 10 - -# Override rate limit (default 1/sec) -uv run accuracy send --no-dry-run --rate 0.5 -``` - -The tool rate-limits at 1 email/second and pauses 30 seconds every 25 emails. -Gmail's own limit (~500/day on a free account) acts as an additional backstop. - -Each probe sends to `validation-probe-@municipality-domain` -- a -provably non-existent address. The email includes `Auto-Submitted: -auto-generated` (RFC 3834) to prevent auto-reply loops. - -## Step 5: Check status - -Monitor the probe lifecycle at any time: - -```bash -uv run accuracy status -``` - -Shows counts per status (pending / sent / ndr_received / no_ndr / failed) and -per country. - -## Step 6: Collect NDRs - -Wait some time for bounces to arrive (minutes to hours), then poll your -Gmail inbox for NDR messages: - -```bash -# Single check, then exit -uv run accuracy collect --poll-once - -# Keep polling every 5 minutes for up to 12 hours -uv run accuracy collect --wait-hours 12 - -# Custom poll interval (seconds) -uv run accuracy collect --poll-interval 600 --wait-hours 24 -``` - -The collector matches NDRs to probes via Message-ID, In-Reply-To, and the -UUID embedded in the probe recipient address. Unmatched NDRs are skipped. -Probes without a bounce after `--wait-hours` are marked `no_ndr`. - -## Step 7: Generate the accuracy report - -```bash -# Console report with confusion matrix and per-class metrics -uv run accuracy report - -# Also export a LaTeX table for the paper -uv run accuracy report --latex -``` - -Output files are written to `output/accuracy/`: -- `accuracy_report.json` -- full metrics as JSON -- `accuracy_report.tex` -- LaTeX table (with `--latex`) - -The report shows: -- Overall accuracy -- Response rate (NDRs received / probes sent) -- Per-class precision, recall, F1, support -- Confusion matrix (predicted vs actual) - -## Production run protocol - -A clean, reproducible validation run. Gmail free accounts allow ~500 emails -per day, so a 400-probe run fits in a single day. Target: 400 probes, ~300 -responses (~75% response rate), giving +/-5% margin of error at 95% CI for -the combined population of 15,331 municipalities. - -### Prerequisites - -- Fresh Gmail account with App Password and IMAP enabled -- `.env` configured with the account credentials -- No prior state: `rm -f output/accuracy/state.db` - -### Run steps - -```bash -# 0. Clean slate -rm -f output/accuracy/* - -# 1. Create stratified sample (400 probes across all countries) -# min-per-class=22 ensures census of all AWS (22) and Google (12) -# municipalities; rest is proportional (domestic ~259, microsoft ~77) -# Target: ~300 responses at 75% response rate -> +/-5.6% margin at 95% CI -# Fits within Gmail free account daily limit of 500. -uv run accuracy sample --all --size 400 --min-per-class 22 - -# 2. Verify sample distribution -uv run accuracy status - -# 3. Dry run — review probe list -uv run accuracy send --dry-run - -# 4. Send all 400 probes (4 batches of 100) -uv run accuracy send --no-dry-run --max-probes 100 --confirm -uv run accuracy send --no-dry-run --max-probes 100 --confirm -uv run accuracy send --no-dry-run --max-probes 100 --confirm -uv run accuracy send --no-dry-run --max-probes 100 --confirm - -# 5. Verify all sent (pending should be 0, failed should be 0) -uv run accuracy status - -# 6. Wait 10-15 minutes for bounces to arrive - -# 7. IMPORTANT: Gmail may route some NDRs to Spam. -# Open Gmail in a browser, go to Spam folder, select all NDR messages -# (from "Mail Delivery Subsystem" / "mailer-daemon"), and click -# "Not spam". Move them to Inbox. The IMAP collector only searches -# INBOX by default. - -# 8. Collect NDRs (first pass) -uv run accuracy collect --poll-once - -# 9. Check response rate -uv run accuracy status - -# 10. Wait another 10-15 minutes, check Spam again, collect stragglers -uv run accuracy collect --poll-once - -# 11. Final report -uv run accuracy report --latex -``` - -### Verifying sample validity - -After collection, check that the response distribution matches the sample -stratification. If the Gmail daily limit was hit mid-run, some strata will -have unsent probes (`pending` or `failed` status), invalidating the sample. - -```bash -# Check for unsent/failed probes — must all be 0 -sqlite3 output/accuracy/state.db \ - "SELECT predicted_provider, status, COUNT(*) FROM probes GROUP BY predicted_provider, status" -``` - -If any probes are `failed`, reset them and send from another account: - -```bash -sqlite3 output/accuracy/state.db "UPDATE probes SET status='pending' WHERE status='failed';" -vim .env # switch to second Gmail account -uv run accuracy send --no-dry-run --confirm -# Then collect from this account: -uv run accuracy collect --poll-once -# Switch .env back to original account and collect its NDRs too: -vim .env # switch back -uv run accuracy collect --poll-once -``` - -## Resumability - -The SQLite state database tracks every probe's lifecycle. If the process is -interrupted: -- `sample` skips domains already in the database -- `send` picks up probes still in `pending` status -- `collect` only processes new IMAP messages and skips already-matched probes - -You can safely re-run any command. - -## Known limitations - -### AWS SES inbound relay - -Municipalities classified as AWS typically use `inbound-smtp.eu-west-1.amazonaws.com` -as their MX. The classifier correctly identifies AWS from DNS (MX, ASN, TXT -verification). However, bounce probes reveal that these domains use AWS SES -purely as an **inbound relay** that forwards to on-premises Exchange (e.g. -`rzpex0x.public.hi-ag.ch` at Hosting Informatik AG). The NDR is generated by -the backend Exchange server, not by AWS. - -This architecture layering -- cloud relay in front of on-prem mailbox hosting -- -is invisible to DNS-based classification. It is a methodological boundary of -the bounce-probe approach rather than a classifier error. AWS accounts for -~1% of municipalities, so the impact on overall accuracy is small. - -When reporting accuracy for the paper, exclude AWS and report per-class metrics -for the three dominant classes (Microsoft, self-hosted, Google) which cover -97% of all municipalities. - -## Safety controls - -| Control | Default | Override | -|--------------------------------|---------|-----------------------------------| -| Dry run | On | `--no-dry-run` | -| Interactive confirmation | On | `--confirm` | -| Max probes per run | 100 | `--max-probes N` | -| Rate limit | 1/sec | `--rate N` | -| Batch pause | 30s/25 | `ACCURACY_SEND_BATCH_*` env vars | -| Gmail daily limit | ~500 | (external, not overridable) | -| RFC 3834 Auto-Submitted header | Always | (not overridable) | diff --git a/src/mail_municipalities/accuracy/__init__.py b/src/mail_municipalities/accuracy/__init__.py deleted file mode 100644 index f2a5a55..0000000 --- a/src/mail_municipalities/accuracy/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Bounce-probe accuracy validation for provider classification.""" diff --git a/src/mail_municipalities/accuracy/check.py b/src/mail_municipalities/accuracy/check.py deleted file mode 100644 index e9ea055..0000000 --- a/src/mail_municipalities/accuracy/check.py +++ /dev/null @@ -1,203 +0,0 @@ -"""Spot-check provider classification for specific domains. - -Creates probes in state.db so that ``accuracy send`` / ``accuracy collect`` -can verify the classification via NDR. Running ``accuracy check`` again -after collection shows the verified result. -""" - -from __future__ import annotations - -import json -import uuid -from dataclasses import dataclass -from pathlib import Path - -from loguru import logger -from rich.console import Console -from rich.table import Table - -from mail_municipalities.accuracy.models import ( - CLASSIFIER_TO_EVAL, - NDR_TO_CLASSIFIER, - Probe, - ProbeStatus, -) -from mail_municipalities.accuracy.state import StateDB - -console = Console() - -_COUNTRIES = ("de", "at", "ch") - - -@dataclass(frozen=True) -class CheckResult: - domain: str - name: str | None - country: str | None - provider: str | None - confidence: float | None - status: str # "not_found" | probe status value | "new" - actual: str | None - match: bool | None - - -async def check_domains( - domains: list[str], - providers_dir: Path, - state: StateDB, -) -> list[CheckResult]: - """Look up domains, create probes for new ones, show status for existing.""" - # Build domain → (entry, cc) from all provider files. - catalog: dict[str, tuple[dict, str]] = {} - for cc in _COUNTRIES: - path = providers_dir / f"providers_{cc}.json" - if not path.exists(): - continue - data = json.loads(path.read_text()) - for entry in data["municipalities"]: - d = entry.get("domain") - if d: - catalog[d] = (entry, cc) - - # Load existing probes and NDRs. - all_probes = await state.get_all_probes() - all_ndrs = await state.get_all_ndrs() - probes_by_domain: dict[str, Probe] = {} - for p in all_probes: - probes_by_domain[p.domain] = p - ndr_by_probe = {ndr.probe_id: ndr for ndr in all_ndrs} - - results: list[CheckResult] = [] - new_probes: list[Probe] = [] - - for domain in domains: - if domain not in catalog: - results.append( - CheckResult( - domain=domain, - name=None, - country=None, - provider=None, - confidence=None, - status="not_found", - actual=None, - match=None, - ) - ) - continue - - entry, cc = catalog[domain] - provider = entry.get("provider") - confidence = entry.get("classification_confidence") - - existing = probes_by_domain.get(domain) - if existing is not None: - # Already probed — show current status and NDR result if available. - actual = None - match = None - ndr = ndr_by_probe.get(existing.probe_id) - if ndr is not None: - actual = NDR_TO_CLASSIFIER.get(ndr.ndr_provider.value, "unknown") - pred_eval = CLASSIFIER_TO_EVAL.get(provider or "", "unknown") - match = pred_eval == actual - results.append( - CheckResult( - domain=domain, - name=entry.get("name"), - country=cc.upper(), - provider=provider, - confidence=confidence, - status=existing.status.value, - actual=actual, - match=match, - ) - ) - else: - # New — create a probe. - probe = _make_probe(entry, cc) - new_probes.append(probe) - results.append( - CheckResult( - domain=domain, - name=entry.get("name"), - country=cc.upper(), - provider=provider, - confidence=confidence, - status="new", - actual=None, - match=None, - ) - ) - - if new_probes: - inserted = await state.insert_probes(new_probes) - logger.info("Created {} probe(s) for check domains", inserted) - - return results - - -def _make_probe(entry: dict, country: str) -> Probe: - """Create a Probe from a municipality entry.""" - short_uuid = uuid.uuid4().hex[:12] - domain = entry["domain"] - return Probe( - probe_id=uuid.uuid4().hex, - domain=domain, - municipality_code=str(entry["code"]), - municipality_name=entry["name"], - country=country, - recipient=f"validation-probe-{short_uuid}@{domain}", - predicted_provider=entry["provider"], - predicted_confidence=entry.get("classification_confidence", 0.0), - gateway=entry.get("gateway"), - status=ProbeStatus.PENDING, - ) - - -def print_check_table(results: list[CheckResult]) -> None: - """Print check results as a Rich table.""" - table = Table(title="Domain Classification Check", show_lines=True) - table.add_column("Domain", style="bold") - table.add_column("City") - table.add_column("Country") - table.add_column("Provider") - table.add_column("Confidence", justify="right") - table.add_column("Status") - table.add_column("Actual") - table.add_column("Match") - - for r in results: - if r.status == "not_found": - table.add_row(r.domain, "[dim]not found[/dim]", "", "", "", "", "", "") - continue - - match_str = "" - if r.match is True: - match_str = "[green]yes[/green]" - elif r.match is False: - match_str = "[red]no[/red]" - - status_str = r.status - if r.status == "new": - status_str = "[cyan]pending[/cyan]" - elif r.status == "ndr_received": - status_str = "[green]ndr_received[/green]" - - table.add_row( - r.domain, - r.name or "", - r.country or "", - r.provider or "", - f"{r.confidence:.0f}" if r.confidence is not None else "", - status_str, - r.actual or "", - match_str, - ) - - console.print(table) - - new_count = sum(1 for r in results if r.status == "new") - if new_count: - console.print( - f"\n[cyan]{new_count} new probe(s) created.[/cyan] Run [bold]accuracy send --no-dry-run[/bold] then [bold]accuracy collect[/bold] to verify." - ) diff --git a/src/mail_municipalities/accuracy/collector.py b/src/mail_municipalities/accuracy/collector.py deleted file mode 100644 index 899e2b8..0000000 --- a/src/mail_municipalities/accuracy/collector.py +++ /dev/null @@ -1,187 +0,0 @@ -"""IMAP NDR collector — polls a mailbox for bounce messages and matches them to probes.""" - -from __future__ import annotations - -import asyncio -import email -import email.policy -import imaplib -import re -from datetime import datetime, timedelta, timezone -from email.message import EmailMessage - -from loguru import logger -from rich.console import Console - -from mail_municipalities.accuracy.config import AccuracyConfig -from mail_municipalities.accuracy.models import NdrResult, ProbeStatus -from mail_municipalities.accuracy.ndr_parser import parse_ndr -from mail_municipalities.accuracy.state import StateDB - -console = Console() - -# UUID pattern used in probe recipient addresses (12 hex chars). -_UUID_RE = re.compile(r"validation-probe-([0-9a-f]{12})") - - -async def collect_ndrs( - state: StateDB, - config: AccuracyConfig, - *, - poll_once: bool = False, - wait_hours: float | None = None, - poll_interval: int | None = None, -) -> None: - """Poll IMAP for NDRs, match to probes, and update state.""" - max_wait = wait_hours or config.ndr_max_wait_hours - interval = poll_interval or config.ndr_poll_interval_seconds - deadline = datetime.now(tz=timezone.utc) + timedelta(hours=max_wait) - - if not config.imap_user or not config.imap_password.get_secret_value(): - console.print("[red]IMAP credentials not configured. Set ACCURACY_IMAP_USER and ACCURACY_IMAP_PASSWORD.[/red]") - return - - round_num = 0 - while True: - round_num += 1 - logger.info("Collection round {} ...", round_num) - matched, skipped = await _collect_one_round(state, config) - console.print(f" Round {round_num}: {matched} matched, {skipped} skipped") - - # Mark timed-out probes. - cutoff = datetime.now(tz=timezone.utc) - timedelta(hours=max_wait) - timed_out = await state.mark_timed_out(cutoff) - if timed_out: - logger.info("Marked {} probes as timed out (no NDR)", timed_out) - - if poll_once: - break - if datetime.now(tz=timezone.utc) >= deadline: - logger.info("Reached wait deadline ({:.1f}h). Stopping.", max_wait) - break - - logger.info("Sleeping {}s until next poll...", interval) - await asyncio.sleep(interval) - - -async def _collect_one_round(state: StateDB, config: AccuracyConfig) -> tuple[int, int]: - """Fetch NDR messages via IMAP, parse, and match. Returns (matched, skipped).""" - - def _imap_fetch() -> list[EmailMessage]: - with imaplib.IMAP4_SSL(config.imap_host, config.imap_port) as imap: - imap.login(config.imap_user, config.imap_password.get_secret_value()) - imap.select(config.imap_folder, readonly=True) - # Search for bounce-like messages. - _status, data = imap.search(None, '(OR FROM "mailer-daemon" FROM "postmaster")') - msg_nums = data[0].split() if data[0] else [] - messages: list[EmailMessage] = [] - for num in msg_nums: - _status, msg_data = imap.fetch(num, "(RFC822)") - if msg_data and msg_data[0] and isinstance(msg_data[0], tuple): - raw = msg_data[0][1] - if isinstance(raw, bytes): - parsed = email.message_from_bytes(raw, policy=email.policy.default) - if isinstance(parsed, EmailMessage): - messages.append(parsed) - return messages - - messages = await asyncio.to_thread(_imap_fetch) - logger.debug("Fetched {} candidate NDR messages", len(messages)) - - matched = 0 - skipped = 0 - - for msg in messages: - probe = await _match_ndr_to_probe(msg, state) - if probe is None: - skipped += 1 - continue - - # Check if we already have an NDR for this probe. - if await state.has_ndr_for_probe(probe.probe_id): - skipped += 1 - continue - - provider, confidence, generating_mta, evidence = parse_ndr(msg) - - ndr = NdrResult( - probe_id=probe.probe_id, - received_at=datetime.now(tz=timezone.utc), - ndr_from=(msg.get("From") or ""), - ndr_provider=provider, - generating_mta=generating_mta, - confidence=confidence, - evidence=evidence, - raw_headers=_extract_headers(msg), - ) - await state.insert_ndr(ndr) - await state.update_probe_status(probe.probe_id, ProbeStatus.NDR_RECEIVED) - matched += 1 - logger.debug("Matched NDR to probe {} ({}): {}", probe.probe_id[:8], probe.domain, provider.value) - - return matched, skipped - - -async def _match_ndr_to_probe(msg: EmailMessage, state: StateDB): - """Try to match an NDR message to a probe via multiple strategies.""" - # Strategy 1: In-Reply-To / References contain the original Message-ID. - for hdr in ("In-Reply-To", "References"): - ref = msg.get(hdr) - if ref: - # May contain multiple message-ids; try each. - for mid in re.findall(r"<[^>]+>", ref): - probe = await state.find_probe_by_message_id(mid) - if probe: - return probe - - # Strategy 2: Extract recipient UUID from DSN Final-Recipient or body. - body_text = _get_text(msg) - for m in _UUID_RE.finditer(body_text): - uuid_part = m.group(1) - probe = await state.find_probe_by_recipient_substring(uuid_part) - if probe: - return probe - - # Strategy 3: Look for probe recipient in any header or body text. - headers_text = "\n".join(f"{k}: {v}" for k, v in msg.items()) - for m in _UUID_RE.finditer(headers_text): - uuid_part = m.group(1) - probe = await state.find_probe_by_recipient_substring(uuid_part) - if probe: - return probe - - return None - - -def _get_text(msg: EmailMessage) -> str: - """Extract all text content from a message.""" - parts: list[str] = [] - if msg.is_multipart(): - for part in msg.walk(): - payload = part.get_payload(decode=True) - if isinstance(payload, bytes): - parts.append(payload.decode("utf-8", errors="replace")) - else: - payload = msg.get_payload(decode=True) - if isinstance(payload, bytes): - parts.append(payload.decode("utf-8", errors="replace")) - return "\n".join(parts) - - -def _extract_headers(msg: EmailMessage) -> str: - """Return the outer headers plus DSN fields for debugging.""" - lines = [f"{k}: {v}" for k, v in msg.items()] - # Append DSN delivery-status content so it's visible in the DB. - if msg.is_multipart(): - for part in msg.walk(): - ct = part.get_content_type() - if ct in ("message/delivery-status", "text/delivery-status"): - lines.append("\n--- DSN delivery-status ---") - payload = part.get_payload(decode=True) - if isinstance(payload, bytes): - lines.append(payload.decode("utf-8", errors="replace")) - elif isinstance(payload, list): - lines.append("\n".join(str(p) for p in payload)) - elif isinstance(payload, str): - lines.append(payload) - return "\n".join(lines) diff --git a/src/mail_municipalities/accuracy/config.py b/src/mail_municipalities/accuracy/config.py deleted file mode 100644 index 59794bd..0000000 --- a/src/mail_municipalities/accuracy/config.py +++ /dev/null @@ -1,56 +0,0 @@ -"""Configuration for bounce-probe accuracy validation.""" - -from __future__ import annotations - -from pathlib import Path - -from pydantic import SecretStr -from pydantic_settings import BaseSettings, SettingsConfigDict - - -class AccuracyConfig(BaseSettings): - """Settings loaded from environment variables with ``ACCURACY_`` prefix. - - Credentials can be set via a ``.env`` file or exported in the shell. - """ - - model_config = SettingsConfigDict(env_prefix="ACCURACY_", env_file=".env", env_file_encoding="utf-8") - - # ── SMTP (Gmail relay) ───────────────────────────────────────── - smtp_host: str = "smtp.gmail.com" - smtp_port: int = 587 - smtp_user: str = "" - smtp_password: SecretStr = SecretStr("") - smtp_from: str = "" # envelope sender; defaults to smtp_user if empty - - # ── IMAP (Gmail) ────────────────────────────────────────────── - imap_host: str = "imap.gmail.com" - imap_port: int = 993 - imap_user: str = "" - imap_password: SecretStr = SecretStr("") - imap_folder: str = "INBOX" - - # ── Rate limiting ───────────────────────────────────────────── - send_rate_per_second: float = 1.0 - send_batch_size: int = 25 - send_batch_pause_seconds: float = 30.0 - - # ── Safety ──────────────────────────────────────────────────── - max_probes_per_run: int = 100 - dry_run: bool = True # must explicitly disable - - # ── NDR collection ──────────────────────────────────────────── - ndr_poll_interval_seconds: int = 300 - ndr_max_wait_hours: float = 24.0 - - # ── Paths ───────────────────────────────────────────────────── - output_dir: Path = Path("output/accuracy") - providers_dir: Path = Path("output/providers") - - @property - def sender_address(self) -> str: - return self.smtp_from or self.smtp_user - - @property - def state_db_path(self) -> Path: - return self.output_dir / "state.db" diff --git a/src/mail_municipalities/accuracy/metrics.py b/src/mail_municipalities/accuracy/metrics.py deleted file mode 100644 index 3fa8bc1..0000000 --- a/src/mail_municipalities/accuracy/metrics.py +++ /dev/null @@ -1,135 +0,0 @@ -"""Accuracy metrics via scikit-learn.""" - -from __future__ import annotations - -from datetime import datetime, timezone - -from sklearn.metrics import accuracy_score, confusion_matrix, f1_score, precision_recall_fscore_support - -from mail_municipalities.accuracy.models import ( - CLASSIFIER_TO_EVAL, - NDR_TO_CLASSIFIER, - AccuracyReport, - ClassMetrics, - NdrResult, - ProbeStatus, -) -from mail_municipalities.accuracy.state import StateDB - -# Label set used for evaluation (after mapping both sides). -EVAL_LABELS = ("microsoft", "google", "aws", "self-hosted", "unknown") - -# Labels included in the weighted F1 headline metric. -# AWS is excluded: DNS probes see the inbound relay (AWS SES) while the -# bounce reveals the on-prem backend — a methodological boundary, not a -# classifier error. "unknown" is excluded because it has 0 support. -WEIGHTED_F1_LABELS = ("microsoft", "google", "self-hosted") - - -async def compute_accuracy(state: StateDB) -> AccuracyReport: - """Compute accuracy metrics from matched probes and NDRs.""" - all_probes = await state.get_all_probes() - all_ndrs = await state.get_all_ndrs() - - ndr_by_probe: dict[str, NdrResult] = {ndr.probe_id: ndr for ndr in all_ndrs} - - total_probes = len(all_probes) - total_sent = sum( - 1 for p in all_probes if p.status in (ProbeStatus.SENT, ProbeStatus.NDR_RECEIVED, ProbeStatus.NO_NDR) - ) - total_ndrs = len(all_ndrs) - - # Build parallel label lists from matched pairs. - y_true: list[str] = [] - y_pred: list[str] = [] - for probe in all_probes: - if probe.probe_id not in ndr_by_probe: - continue - ndr = ndr_by_probe[probe.probe_id] - y_pred.append(CLASSIFIER_TO_EVAL.get(probe.predicted_provider, "unknown")) - y_true.append(NDR_TO_CLASSIFIER.get(ndr.ndr_provider.value, "unknown")) - - response_rate = total_ndrs / total_sent if total_sent > 0 else 0.0 - - if not y_true: - return _empty_report(total_probes, total_sent, total_ndrs, response_rate) - - # Overall accuracy across all labels. - overall_acc = float(accuracy_score(y_true, y_pred)) - - # Per-class metrics across all EVAL_LABELS. - present_labels = [label for label in EVAL_LABELS if label in set(y_true) | set(y_pred)] - prec_arr, rec_arr, f1_arr, sup_arr = precision_recall_fscore_support( - y_true, - y_pred, - labels=present_labels, - zero_division="warn", # type requires str - ) - per_class: dict[str, ClassMetrics] = {} - for i, label in enumerate(present_labels): - per_class[label] = ClassMetrics( - precision=float(prec_arr[i]), # type: ignore[index] - recall=float(rec_arr[i]), # type: ignore[index] - f1=float(f1_arr[i]), # type: ignore[index] - support=int(sup_arr[i]), # type: ignore[index] - ) - # Fill missing labels with zeros. - for label in EVAL_LABELS: - if label not in per_class: - per_class[label] = ClassMetrics(precision=0.0, recall=0.0, f1=0.0, support=0) - - # Weighted F1 over the dominant classes only. - # Filter to pairs where BOTH predicted and actual are in WEIGHTED_F1_LABELS, - # so excluded classes (AWS) don't affect precision/recall of included ones. - wf1_true = [t for t, p in zip(y_true, y_pred) if t in WEIGHTED_F1_LABELS and p in WEIGHTED_F1_LABELS] - wf1_pred = [p for t, p in zip(y_true, y_pred) if t in WEIGHTED_F1_LABELS and p in WEIGHTED_F1_LABELS] - if wf1_true: - weighted_f1 = float( - f1_score(wf1_true, wf1_pred, labels=list(WEIGHTED_F1_LABELS), average="weighted", zero_division="warn") - ) - else: - weighted_f1 = 0.0 - - # Confusion matrix as nested dict (rows=predicted, cols=actual). - # sklearn returns rows=true, cols=predicted — transpose via indexing. - cm_labels = [label for label in EVAL_LABELS if label in set(y_true) | set(y_pred)] - cm_array = confusion_matrix(y_true, y_pred, labels=cm_labels) - cm: dict[str, dict[str, int]] = {} - for i, pred in enumerate(cm_labels): - cm[pred] = {actual: int(cm_array[j][i]) for j, actual in enumerate(cm_labels)} - # Fill missing labels. - for label in EVAL_LABELS: - if label not in cm: - cm[label] = {a: 0 for a in EVAL_LABELS} - for a in EVAL_LABELS: - cm[label].setdefault(a, 0) - - return AccuracyReport( - generated=datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), - total_probes=total_probes, - total_sent=total_sent, - total_ndrs=total_ndrs, - response_rate=round(response_rate, 4), - overall_accuracy=round(overall_acc, 4), - weighted_f1=round(weighted_f1, 4), - weighted_f1_labels=list(WEIGHTED_F1_LABELS), - per_class=per_class, - confusion_matrix=cm, - ) - - -def _empty_report(total_probes: int, total_sent: int, total_ndrs: int, response_rate: float) -> AccuracyReport: - per_class = {label: ClassMetrics(precision=0.0, recall=0.0, f1=0.0, support=0) for label in EVAL_LABELS} - cm = {p: {a: 0 for a in EVAL_LABELS} for p in EVAL_LABELS} - return AccuracyReport( - generated=datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), - total_probes=total_probes, - total_sent=total_sent, - total_ndrs=total_ndrs, - response_rate=round(response_rate, 4), - overall_accuracy=0.0, - weighted_f1=0.0, - weighted_f1_labels=list(WEIGHTED_F1_LABELS), - per_class=per_class, - confusion_matrix=cm, - ) diff --git a/src/mail_municipalities/accuracy/models.py b/src/mail_municipalities/accuracy/models.py deleted file mode 100644 index 704bf34..0000000 --- a/src/mail_municipalities/accuracy/models.py +++ /dev/null @@ -1,126 +0,0 @@ -"""Pydantic models for bounce-probe accuracy validation.""" - -from __future__ import annotations - -import enum -from datetime import datetime - -from pydantic import BaseModel, ConfigDict, Field - - -class ProbeStatus(str, enum.Enum): - PENDING = "pending" - SENT = "sent" - SEND_FAILED = "failed" - NDR_RECEIVED = "ndr_received" - NO_NDR = "no_ndr" - - -class NdrProvider(str, enum.Enum): - """Provider identified from NDR headers/body.""" - - MICROSOFT = "microsoft" - GOOGLE = "google" - AWS = "aws" - POSTFIX = "postfix" - EXIM = "exim" - EXCHANGE_ONPREM = "exchange_onprem" - OTHER = "other" - UNKNOWN = "unknown" - - -# Map NDR-detected provider to classifier output labels. -# NDR cannot distinguish domestic vs foreign — both use Postfix/Exim. -# For evaluation we merge domestic+foreign into "self-hosted". -NDR_TO_CLASSIFIER: dict[str, str] = { - NdrProvider.MICROSOFT: "microsoft", - NdrProvider.GOOGLE: "google", - NdrProvider.AWS: "aws", - NdrProvider.POSTFIX: "self-hosted", - NdrProvider.EXIM: "self-hosted", - NdrProvider.EXCHANGE_ONPREM: "self-hosted", - NdrProvider.OTHER: "unknown", - NdrProvider.UNKNOWN: "unknown", -} - -# Map classifier output labels to the common evaluation label set. -CLASSIFIER_TO_EVAL: dict[str, str] = { - "microsoft": "microsoft", - "google": "google", - "aws": "aws", - "domestic": "self-hosted", - "foreign": "self-hosted", - "unknown": "unknown", -} - - -class Probe(BaseModel): - """A single probe email targeting a municipality domain.""" - - model_config = ConfigDict(frozen=True) - - probe_id: str - domain: str - municipality_code: str - municipality_name: str - country: str - recipient: str - predicted_provider: str - predicted_confidence: float - gateway: str | None = None - status: ProbeStatus = ProbeStatus.PENDING - sent_at: datetime | None = None - message_id: str | None = None - smtp_response: str | None = None - - -class NdrEvidence(BaseModel): - """A single piece of evidence extracted from an NDR.""" - - model_config = ConfigDict(frozen=True) - - pattern: str - matched_value: str - - -class NdrResult(BaseModel): - """Parsed NDR matched to a probe.""" - - model_config = ConfigDict(frozen=True) - - probe_id: str - received_at: datetime - ndr_from: str - ndr_provider: NdrProvider - generating_mta: str - confidence: float = Field(ge=0.0, le=1.0) - evidence: list[NdrEvidence] = [] - raw_headers: str = "" - - -class ClassMetrics(BaseModel): - """Per-class precision / recall / F1.""" - - model_config = ConfigDict(frozen=True) - - precision: float - recall: float - f1: float - support: int - - -class AccuracyReport(BaseModel): - """Aggregate accuracy metrics.""" - - model_config = ConfigDict(frozen=True) - - generated: str - total_probes: int - total_sent: int - total_ndrs: int - response_rate: float - overall_accuracy: float - weighted_f1: float - weighted_f1_labels: list[str] - per_class: dict[str, ClassMetrics] - confusion_matrix: dict[str, dict[str, int]] diff --git a/src/mail_municipalities/accuracy/ndr_parser.py b/src/mail_municipalities/accuracy/ndr_parser.py deleted file mode 100644 index 835eeac..0000000 --- a/src/mail_municipalities/accuracy/ndr_parser.py +++ /dev/null @@ -1,414 +0,0 @@ -"""Parse NDR (bounce) emails to identify the generating MTA's provider. - -IMPORTANT: when NDRs are collected via Gmail IMAP, every message carries -Google relay headers (X-Gm-*, Received from *.google.com). The parser must -ignore these and focus on the DSN content (Reporting-MTA, Remote-MTA, -Diagnostic-Code) and the bounce body text, which originate from the actual -target MTA. -""" - -from __future__ import annotations - -import re -from email.message import EmailMessage - -from mail_municipalities.accuracy.models import NdrEvidence, NdrProvider - -# Patterns that indicate a Received header belongs to our Gmail relay -# (not the target MTA). These are filtered out before analysis. -_RELAY_RECEIVED_RE = re.compile( - r"google\.com|googlemail\.com|gmail-smtp|smtp\.gmail\.com", - re.I, -) - - -def parse_ndr(msg: EmailMessage) -> tuple[NdrProvider, float, str, list[NdrEvidence]]: - """Analyse an NDR email and identify the backend provider. - - Returns ``(provider, confidence, generating_mta, evidence)``. - """ - evidence: list[NdrEvidence] = [] - scores: dict[NdrProvider, float] = {} - - # ── Collect raw material ────────────────────────────────────── - body = _body_text(msg) - dsn_fields = _extract_dsn_fields(msg) - from_addr = (msg.get("From") or "").lower() - subject = (msg.get("Subject") or "").lower() - - # DSN fields — these are authoritative (from the actual bouncing MTA). - reporting_mta = dsn_fields.get("reporting-mta", "") - remote_mta = dsn_fields.get("remote-mta", "") - diagnostic_code = dsn_fields.get("diagnostic-code", "") - - # Filter Received chain: remove hops from our Gmail relay. - all_received = [v for v in msg.get_all("Received", []) if isinstance(v, str)] - received_chain = [r for r in all_received if not _RELAY_RECEIVED_RE.search(r)] - - # Collect non-relay X-headers (skip X-Gm-* and X-Google-* from Gmail). - target_headers: dict[str, str] = {} - for key in msg.keys(): - kl = key.lower() - if kl.startswith("x-gm-") or kl.startswith("x-google-"): - continue # Gmail relay artifact - if kl.startswith("x-"): - target_headers[kl] = str(msg[key]) - - # Combined text for Diagnostic-Code + Remote-MTA scanning. - dsn_text = f"{reporting_mta} {remote_mta} {diagnostic_code}".lower() - - # ── Microsoft 365 / Exchange Online ─────────────────────────── - ms_patterns: list[tuple[str, str]] = [] - - # X-MS-Exchange-* headers (these survive Gmail relay). - if any(k.startswith("x-ms-exchange") for k in target_headers): - ms_patterns.append(("x-ms-exchange header", "present")) - - if re.search(r"postmaster@.*outlook\.com", from_addr): - ms_patterns.append(("from postmaster@outlook", from_addr)) - if re.search(r"\.onmicrosoft\.com", from_addr): - ms_patterns.append(("from *.onmicrosoft.com", from_addr)) - - if re.search(r"protection\.outlook\.com|\.outlook\.com", reporting_mta): - ms_patterns.append(("reporting-mta outlook", reporting_mta)) - - if re.search(r"protection\.outlook\.com|\.outlook\.com", dsn_text): - ms_patterns.append(("dsn mentions outlook", dsn_text[:120])) - - if any(re.search(r"\.outlook\.com|protection\.outlook\.com", r, re.I) for r in received_chain): - ms_patterns.append(("received via outlook", "received chain")) - - if "delivery has failed" in body.lower() and "recipients or groups" in body.lower(): - ms_patterns.append(("exchange dsn body", "delivery has failed to recipients")) - - if ms_patterns: - is_online = any( - "outlook.com" in p[1].lower() or "protection.outlook" in p[1].lower() or "onmicrosoft.com" in p[1].lower() - for p in ms_patterns - if p[0] - in ( - "reporting-mta outlook", - "received via outlook", - "from postmaster@outlook", - "from *.onmicrosoft.com", - "dsn mentions outlook", - ) - ) - provider = NdrProvider.MICROSOFT if is_online else NdrProvider.EXCHANGE_ONPREM - - # Definitive: Microsoft's own header distinguishes hosted vs hybrid. - entity_header = target_headers.get("x-ms-exchange-crosstenant-fromentityheader", "").lower() - if entity_header == "hosted": - provider = NdrProvider.MICROSOFT - evidence.append(NdrEvidence(pattern="crosstenant-fromentityheader", matched_value="Hosted")) - elif entity_header == "hybridonprem": - provider = NdrProvider.EXCHANGE_ONPREM - evidence.append(NdrEvidence(pattern="crosstenant-fromentityheader", matched_value="HybridOnPrem")) - - conf = min(0.3 + 0.15 * len(ms_patterns), 1.0) - for pat, val in ms_patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[provider] = conf - - # ── Gmail relay bounce handling ───────────────────────────────── - # When we send FROM Gmail and Gmail generates the NDR, the bounce is - # ALWAYS a relay bounce — the target is whatever Remote-MTA says, - # never Google Workspace itself. (If the target were Google, the - # email would be delivered to Google and Google would bounce as the - # destination, not as the relay.) - is_gmail_relay_bounce = bool(re.search(r"mailer-daemon@google(mail)?\.com", from_addr)) - - if is_gmail_relay_bounce: - # This is Gmail reporting a remote rejection. Identify the real - # target from Remote-MTA / Diagnostic-Code. - relay_target = _detect_relay_target(remote_mta, diagnostic_code) - if relay_target is not None: - relay_patterns = _relay_target_evidence(relay_target, remote_mta, diagnostic_code) - for pat, val in relay_patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[relay_target] = min(0.3 + 0.15 * len(relay_patterns), 1.0) - elif remote_mta: - # Remote-MTA exists but doesn't match any cloud provider. - # The target is a self-hosted MTA (Postfix, Exim, etc.). - evidence.append(NdrEvidence(pattern="remote-mta (non-cloud relay)", matched_value=remote_mta)) - scores[NdrProvider.POSTFIX] = 0.45 # generic self-hosted - # Do NOT score for Google — this is our relay, not the target. - else: - # ── Google Workspace (non-relay) ────────────────────────── - # The NDR was sent by Google as the *destination* MTA. - goog_patterns: list[tuple[str, str]] = [] - - if re.search(r"google\.com|googlemail\.com", reporting_mta): - goog_patterns.append(("reporting-mta google", reporting_mta)) - - if "delivery to the following recipient failed" in body.lower(): - goog_patterns.append(("google dsn body", "delivery failed permanently")) - - if goog_patterns: - conf = min(0.3 + 0.2 * len(goog_patterns), 1.0) - for pat, val in goog_patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[NdrProvider.GOOGLE] = conf - - # ── AWS SES ─────────────────────────────────────────────────── - aws_patterns: list[tuple[str, str]] = [] - - if "amazonses.com" in from_addr: - aws_patterns.append(("from amazonses", from_addr)) - - if "x-ses-outgoing" in target_headers: - aws_patterns.append(("x-ses-outgoing header", "present")) - - if re.search(r"amazonses\.com|amazonaws\.com", dsn_text): - aws_patterns.append(("dsn mentions aws", dsn_text[:120])) - - if any(re.search(r"amazonses\.com|amazonaws\.com", r, re.I) for r in received_chain): - aws_patterns.append(("received via aws", "received chain")) - - if aws_patterns: - conf = min(0.3 + 0.2 * len(aws_patterns), 1.0) - for pat, val in aws_patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[NdrProvider.AWS] = conf - - # ── Postfix ─────────────────────────────────────────────────── - postfix_patterns: list[tuple[str, str]] = [] - - if "this is the mail system at host" in body.lower(): - postfix_patterns.append(("postfix dsn body", "this is the mail system at host")) - - if any(k.startswith("x-postfix") for k in target_headers): - postfix_patterns.append(("x-postfix header", "present")) - - if any("postfix" in r.lower() for r in received_chain): - postfix_patterns.append(("received via postfix", "received chain")) - - if "postfix" in reporting_mta.lower(): - postfix_patterns.append(("reporting-mta postfix", reporting_mta)) - - if postfix_patterns: - conf = min(0.3 + 0.2 * len(postfix_patterns), 1.0) - for pat, val in postfix_patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[NdrProvider.POSTFIX] = conf - - # ── Exim ────────────────────────────────────────────────────── - exim_patterns: list[tuple[str, str]] = [] - - if "mail delivery failed" in subject: - exim_patterns.append(("exim subject", subject)) - - if "a message that you sent could not be delivered" in body.lower(): - exim_patterns.append(("exim dsn body", "could not be delivered")) - - if any(re.search(r"exim\s+\d+", r, re.I) for r in received_chain): - exim_patterns.append(("received via exim", "received chain")) - - if exim_patterns: - conf = min(0.3 + 0.2 * len(exim_patterns), 1.0) - for pat, val in exim_patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[NdrProvider.EXIM] = conf - - # ── Fallback: check DSN fields for provider hints ───────────── - # If no provider scored yet, try to identify from Reporting-MTA, - # Remote-MTA, or Diagnostic-Code alone. - if not scores: - dsn_provider = _detect_from_dsn(reporting_mta, remote_mta, diagnostic_code, body) - if dsn_provider is not None: - patterns = _relay_target_evidence(dsn_provider, remote_mta, diagnostic_code) - if reporting_mta: - patterns.append(("reporting-mta", reporting_mta)) - for pat, val in patterns: - evidence.append(NdrEvidence(pattern=pat, matched_value=val)) - scores[dsn_provider] = min(0.3 + 0.15 * len(patterns), 1.0) - - # ── Winner selection ────────────────────────────────────────── - if not scores: - generating_mta = reporting_mta or _guess_mta_from_received(all_received) - return NdrProvider.UNKNOWN, 0.0, generating_mta, evidence - - winner = max(scores, key=lambda p: scores[p]) - generating_mta = reporting_mta or _guess_mta_from_received(all_received) - return winner, scores[winner], generating_mta, evidence - - -# ── Relay bounce detection ──────────────────────────────────────── - - -def _detect_relay_target(remote_mta: str, diagnostic_code: str) -> NdrProvider | None: - """Detect the actual target provider from DSN Remote-MTA / Diagnostic-Code. - - When Gmail generates a bounce on behalf of a remote rejection, the real - target's identity is encoded in these fields. - """ - combined = f"{remote_mta} {diagnostic_code}".lower() - return _match_provider_in_text(combined) - - -def _detect_from_dsn(reporting_mta: str, remote_mta: str, diagnostic_code: str, body: str) -> NdrProvider | None: - """Try to identify provider purely from DSN fields and body text.""" - combined = f"{reporting_mta} {remote_mta} {diagnostic_code}".lower() - result = _match_provider_in_text(combined) - if result: - return result - # Try body text for MTA-specific templates. - bl = body.lower() - if "this is the mail system at host" in bl: - return NdrProvider.POSTFIX - if "a message that you sent could not be delivered" in bl: - return NdrProvider.EXIM - return None - - -def _match_provider_in_text(text: str) -> NdrProvider | None: - """Match provider patterns in a combined DSN text string.""" - if re.search(r"protection\.outlook\.com|\.outlook\.com|\.onmicrosoft\.com|microsoft", text): - return NdrProvider.MICROSOFT - if re.search(r"amazonses\.com|amazonaws\.com|\.awsapps\.com", text): - return NdrProvider.AWS - if re.search(r"\.google\.com|googlemail\.com|aspmx\.l\.google\.com", text): - return NdrProvider.GOOGLE - return None - - -def _relay_target_evidence(provider: NdrProvider, remote_mta: str, diagnostic_code: str) -> list[tuple[str, str]]: - """Build evidence entries for a relay-detected target.""" - patterns: list[tuple[str, str]] = [] - if remote_mta: - patterns.append(("remote-mta (relay bounce)", remote_mta)) - if diagnostic_code: - patterns.append(("diagnostic-code (relay bounce)", diagnostic_code[:200])) - return patterns - - -# ── Text extraction helpers ─────────────────────────────────────── - - -def _body_text(msg: EmailMessage) -> str: - """Extract the plain-text body from the message (first text/plain part).""" - if msg.is_multipart(): - for part in msg.walk(): - ct = part.get_content_type() - if ct == "text/plain": - payload = part.get_payload(decode=True) - if isinstance(payload, bytes): - return payload.decode("utf-8", errors="replace") - # Fall back to the first delivery-status part text. - for part in msg.walk(): - ct = part.get_content_type() - if ct == "message/delivery-status": - payload = part.get_payload(decode=True) - if isinstance(payload, bytes): - return payload.decode("utf-8", errors="replace") - if isinstance(payload, list): - return "\n".join(str(p) for p in payload) - return str(payload) if payload else "" - return "" - payload = msg.get_payload(decode=True) - return payload.decode("utf-8", errors="replace") if isinstance(payload, bytes) else "" - - -def _extract_dsn_fields(msg: EmailMessage) -> dict[str, str]: - """Extract DSN fields (Reporting-MTA, Remote-MTA, Final-Recipient, etc.). - - Handles both ``message/delivery-status`` (RFC 3464) and the common - ``text/delivery-status`` variant. - - For ``message/delivery-status``, Python's email library parses the - DSN field groups as sub-messages whose *headers* carry the DSN fields - (Reporting-MTA, Remote-MTA, Diagnostic-Code, etc.). For - ``text/delivery-status``, the fields appear as plain text lines. - """ - fields: dict[str, str] = {} - if not msg.is_multipart(): - _parse_dsn_text(msg.get_payload(decode=True), fields) - return fields - - for part in msg.walk(): - ct = part.get_content_type() - - if ct == "message/delivery-status": - # RFC 3464: payload is a list of sub-messages. - # Sub-message 0 = per-message fields (Reporting-MTA). - # Sub-message 1+ = per-recipient fields (Remote-MTA, Diagnostic-Code). - raw = part.get_payload() - if isinstance(raw, list): - for sub in raw: - if isinstance(sub, EmailMessage): - for key, value in sub.items(): - _store_dsn_field(key, str(value), fields) - - # Some libraries also allow bytes access. - if not fields: - decoded = part.get_payload(decode=True) - if decoded: - _parse_dsn_text(decoded, fields) - - elif ct == "text/delivery-status": - # Non-standard but common: DSN as plain text. - _parse_dsn_text(part.get_payload(decode=True), fields) - if not fields: - raw = part.get_payload() - if isinstance(raw, str): - _parse_dsn_text(raw, fields) - - # Fallback: scan all text/plain parts for DSN-like fields. - if not fields: - for part in msg.walk(): - if part.get_content_type() == "text/plain": - _parse_dsn_text(part.get_payload(decode=True), fields) - if fields: - break - - return fields - - -_DSN_KEYS = { - "reporting-mta", - "remote-mta", - "diagnostic-code", - "final-recipient", - "original-envelope-id", - "action", - "status", - "last-attempt-date", -} - - -def _store_dsn_field(key: str, value: str, fields: dict[str, str]) -> None: - """Normalise and store a single DSN key-value pair.""" - key = key.strip().lower() - if key not in _DSN_KEYS: - return - value = value.strip() - # Strip DSN type prefix (e.g. "dns;" or "rfc822;"). - if ";" in value: - value = value.split(";", 1)[1].strip() - fields[key] = value - - -def _parse_dsn_text(payload: object, fields: dict[str, str]) -> None: - """Parse DSN key-value lines from raw payload into *fields*.""" - text = "" - if isinstance(payload, bytes): - text = payload.decode("utf-8", errors="replace") - elif isinstance(payload, str): - text = payload - else: - return - - for line in text.splitlines(): - if ":" not in line: - continue - key, _, value = line.partition(":") - _store_dsn_field(key, value, fields) - - -def _guess_mta_from_received(received_chain: list[str]) -> str: - """Best-effort extraction of the innermost MTA hostname from Received headers.""" - if not received_chain: - return "" - last = received_chain[-1] - m = re.search(r"\bby\s+([\w.\-]+)", last) - return m.group(1) if m else last[:120] diff --git a/src/mail_municipalities/accuracy/report.py b/src/mail_municipalities/accuracy/report.py deleted file mode 100644 index abf6a5b..0000000 --- a/src/mail_municipalities/accuracy/report.py +++ /dev/null @@ -1,138 +0,0 @@ -"""Console and LaTeX reporting for accuracy metrics.""" - -from __future__ import annotations - -import json -from pathlib import Path - -from rich.console import Console -from rich.table import Table - -from mail_municipalities.accuracy.metrics import EVAL_LABELS -from mail_municipalities.accuracy.models import AccuracyReport, ProbeStatus -from mail_municipalities.accuracy.state import StateDB - -console = Console() - - -async def print_status(state: StateDB) -> None: - """Print current probe lifecycle status.""" - counts = await state.status_counts() - country_counts = await state.country_counts() - - table = Table(title="Probe Status", show_lines=True) - table.add_column("Status", style="bold") - table.add_column("Count", justify="right") - total = 0 - for status in ProbeStatus: - n = counts.get(status.value, 0) - total += n - table.add_row(status.value, str(n)) - table.add_row("Total", str(total), style="bold") - console.print(table) - - if country_counts: - ct = Table(title="By Country", show_lines=True) - ct.add_column("Country", style="bold") - ct.add_column("Count", justify="right") - for cc in sorted(country_counts): - ct.add_row(cc.upper(), str(country_counts[cc])) - console.print(ct) - - -def print_report(report: AccuracyReport) -> None: - """Print accuracy metrics to the console using Rich tables.""" - console.print() - console.print("[bold]Accuracy Report[/bold]") - console.print(f" Generated: {report.generated}") - console.print(f" Total probes: {report.total_probes}") - console.print(f" Total sent: {report.total_sent}") - console.print(f" NDRs received: {report.total_ndrs}") - console.print(f" Response rate: {report.response_rate:.1%}") - console.print(f" Overall accuracy:{report.overall_accuracy:.1%}") - labels_str = ", ".join(report.weighted_f1_labels) - console.print(f" Weighted F1: {report.weighted_f1:.3f} ({labels_str})") - console.print() - - # ── Per-class metrics ───────────────────────────────────────── - mt = Table(title="Per-Class Metrics", show_lines=True) - mt.add_column("Provider", style="bold") - mt.add_column("Precision", justify="right") - mt.add_column("Recall", justify="right") - mt.add_column("F1", justify="right") - mt.add_column("Support", justify="right") - - for label in EVAL_LABELS: - if label not in report.per_class: - continue - m = report.per_class[label] - mt.add_row( - label, - f"{m.precision:.3f}", - f"{m.recall:.3f}", - f"{m.f1:.3f}", - str(m.support), - ) - console.print(mt) - - # ── Confusion matrix ────────────────────────────────────────── - labels = [ - label - for label in EVAL_LABELS - if label in report.confusion_matrix or any(label in row for row in report.confusion_matrix.values()) - ] - if labels: - cm = Table(title="Confusion Matrix (rows=predicted, cols=actual)", show_lines=True) - cm.add_column("Predicted \\ Actual", style="bold") - for label in labels: - cm.add_column(label, justify="right") - - for pred in labels: - row_data = report.confusion_matrix.get(pred, {}) - cells = [str(row_data.get(actual, 0)) for actual in labels] - cm.add_row(pred, *cells) - console.print(cm) - - -def export_report_json(report: AccuracyReport, output_dir: Path) -> Path: - """Write the accuracy report as JSON.""" - output_dir.mkdir(parents=True, exist_ok=True) - path = output_dir / "accuracy_report.json" - path.write_text(json.dumps(report.model_dump(), indent=2, default=str)) - console.print(f"[green]Report written to {path}[/green]") - return path - - -def export_report_latex(report: AccuracyReport, output_dir: Path) -> Path: - """Export accuracy metrics as a LaTeX table.""" - output_dir.mkdir(parents=True, exist_ok=True) - path = output_dir / "accuracy_report.tex" - - lines: list[str] = [] - lines.append(r"\begin{table}[t]") - lines.append(r"\centering") - lines.append(r"\caption{Provider classification accuracy (bounce-probe validation)}") - lines.append(r"\label{tab:accuracy}") - lines.append(r"\begin{tabular}{lrrrr}") - lines.append(r"\toprule") - lines.append(r"Provider & Precision & Recall & F1 & Support \\") - lines.append(r"\midrule") - - for label in EVAL_LABELS: - if label not in report.per_class: - continue - m = report.per_class[label] - name = label.replace("-", " ").title() - lines.append(f"{name} & {m.precision:.3f} & {m.recall:.3f} & {m.f1:.3f} & {m.support} \\\\") - - lines.append(r"\midrule") - lines.append(f"Weighted F1 & \\multicolumn{{4}}{{r}}{{{report.weighted_f1:.3f}}} \\\\") - lines.append(f"Overall accuracy & \\multicolumn{{4}}{{r}}{{{report.overall_accuracy:.1%}}} \\\\") - lines.append(f"Response rate & \\multicolumn{{4}}{{r}}{{{report.response_rate:.1%}}} \\\\") - lines.append(r"\bottomrule") - lines.append(r"\end{tabular}") - lines.append(r"\end{table}") - - path.write_text("\n".join(lines) + "\n") - console.print(f"[green]LaTeX table written to {path}[/green]") - return path diff --git a/src/mail_municipalities/accuracy/sampler.py b/src/mail_municipalities/accuracy/sampler.py deleted file mode 100644 index 45dae31..0000000 --- a/src/mail_municipalities/accuracy/sampler.py +++ /dev/null @@ -1,148 +0,0 @@ -"""Stratified sampling from provider classification output.""" - -from __future__ import annotations - -import json -import random -import uuid -from collections import defaultdict -from pathlib import Path - -from loguru import logger -from rich.console import Console -from rich.table import Table - -from mail_municipalities.accuracy.models import Probe, ProbeStatus -from mail_municipalities.accuracy.state import StateDB - -console = Console() - -# Providers we can validate via NDR (exclude "unknown" — nothing to compare against). -_VALIDATABLE_PROVIDERS = {"microsoft", "google", "aws", "domestic", "foreign"} - - -def _load_municipalities(providers_path: Path) -> list[dict]: - """Load municipality entries from a providers JSON file.""" - data = json.loads(providers_path.read_text()) - return data["municipalities"] - - -def _stratified_sample( - entries: list[dict], - total_size: int, - min_per_class: int, -) -> list[dict]: - """Return a stratified random sample, grouped by provider.""" - by_provider: dict[str, list[dict]] = defaultdict(list) - for e in entries: - provider = e.get("provider", "unknown") - if provider in _VALIDATABLE_PROVIDERS and e.get("domain"): - by_provider[provider].append(e) - - if not by_provider: - return [] - - # Guarantee minimum per class, then distribute remainder proportionally. - sampled: list[dict] = [] - remaining_budget = total_size - - for provider, pool in by_provider.items(): - n = min(min_per_class, len(pool)) - picked = random.sample(pool, n) - sampled.extend(picked) - remaining_budget -= n - # Remove picked from pool for proportional phase. - picked_set = {id(e) for e in picked} - by_provider[provider] = [e for e in pool if id(e) not in picked_set] - - if remaining_budget > 0: - # Proportional allocation of remaining budget. - total_remaining = sum(len(pool) for pool in by_provider.values()) - if total_remaining > 0: - for provider, pool in by_provider.items(): - share = int(remaining_budget * len(pool) / total_remaining) - n = min(share, len(pool)) - if n > 0: - sampled.extend(random.sample(pool, n)) - - return sampled - - -def _make_probe(entry: dict, country: str) -> Probe: - """Create a Probe from a municipality entry.""" - short_uuid = uuid.uuid4().hex[:12] - domain = entry["domain"] - return Probe( - probe_id=uuid.uuid4().hex, - domain=domain, - municipality_code=str(entry["code"]), - municipality_name=entry["name"], - country=country, - recipient=f"validation-probe-{short_uuid}@{domain}", - predicted_provider=entry["provider"], - predicted_confidence=entry.get("classification_confidence", 0.0), - gateway=entry.get("gateway"), - status=ProbeStatus.PENDING, - ) - - -async def create_sample( - countries: list[str], - total_size: int, - min_per_class: int, - providers_dir: Path, - state: StateDB, -) -> list[Probe]: - """Sample municipalities and insert probes into the state DB. - - Skips domains already present in the DB (for resumability). - Returns the list of newly created probes. - """ - existing_domains = await state.get_existing_domains() - - all_entries: list[tuple[dict, str]] = [] - for cc in countries: - path = providers_dir / f"providers_{cc}.json" - if not path.exists(): - logger.warning("Providers file not found: {}", path) - continue - for entry in _load_municipalities(path): - if entry.get("domain") and entry["domain"] not in existing_domains: - all_entries.append((entry, cc)) - - # Flatten for sampling, then restore country. - flat = [e for e, _ in all_entries] - cc_map = {id(e): cc for e, cc in all_entries} - - sampled = _stratified_sample(flat, total_size, min_per_class) - probes = [_make_probe(e, cc_map[id(e)]) for e in sampled] - - inserted = await state.insert_probes(probes) - logger.info("Created {} probes ({} skipped as duplicates)", inserted, len(probes) - inserted) - - _print_summary(probes) - return probes - - -def _print_summary(probes: list[Probe]) -> None: - """Print a Rich summary of the sample.""" - by_provider: dict[str, int] = defaultdict(int) - by_country: dict[str, int] = defaultdict(int) - for p in probes: - by_provider[p.predicted_provider] += 1 - by_country[p.country] += 1 - - table = Table(title="Sample Summary", show_lines=True) - table.add_column("Provider", style="bold") - table.add_column("Count", justify="right") - for provider in sorted(by_provider): - table.add_row(provider, str(by_provider[provider])) - table.add_row("Total", str(len(probes)), style="bold") - console.print(table) - - country_table = Table(title="By Country", show_lines=True) - country_table.add_column("Country", style="bold") - country_table.add_column("Count", justify="right") - for cc in sorted(by_country): - country_table.add_row(cc.upper(), str(by_country[cc])) - console.print(country_table) diff --git a/src/mail_municipalities/accuracy/sender.py b/src/mail_municipalities/accuracy/sender.py deleted file mode 100644 index 6909691..0000000 --- a/src/mail_municipalities/accuracy/sender.py +++ /dev/null @@ -1,156 +0,0 @@ -"""SMTP probe sender with rate limiting and dry-run support.""" - -from __future__ import annotations - -import asyncio -import smtplib -from datetime import datetime, timezone -from email.message import EmailMessage -from email.utils import formatdate, make_msgid - -from loguru import logger -from rich.console import Console -from rich.table import Table - -from mail_municipalities.accuracy.config import AccuracyConfig -from mail_municipalities.accuracy.models import Probe, ProbeStatus -from mail_municipalities.accuracy.state import StateDB - -console = Console() - -_SUBJECT = "Automated delivery validation probe \u2014 please disregard" - -_BODY_TEMPLATE = """\ -This is an automated email delivery validation probe sent as part of -academic research on email infrastructure of public administrations. - -This message was sent to a randomly generated, non-existent address -and requires no action on your part. It will likely generate an -automatic bounce/NDR, which is the intended outcome. - -If you have questions, please contact: {sender} - -Probe-ID: {probe_id} -""" - - -async def send_probes( - state: StateDB, - config: AccuracyConfig, - *, - max_probes: int | None = None, - dry_run: bool = True, - confirm: bool = False, - batch_size: int | None = None, - rate: float | None = None, -) -> None: - """Send pending probes via SMTP. - - In *dry_run* mode (default), only prints what would be sent. - """ - cap = max_probes or config.max_probes_per_run - probes = await state.get_probes_by_status(ProbeStatus.PENDING, limit=cap) - - if not probes: - console.print("[yellow]No pending probes to send.[/yellow]") - return - - _print_send_plan(probes) - - if dry_run: - console.print(f"\n[bold cyan]DRY RUN[/bold cyan]: would send {len(probes)} probe(s). No emails sent.") - return - - # Live mode — require confirmation. - if not confirm: - answer = input(f"\nAbout to send {len(probes)} real emails. Type YES to proceed: ") - if answer.strip() != "YES": - console.print("[red]Aborted.[/red]") - return - - # Validate config. - if not config.smtp_user or not config.smtp_password.get_secret_value(): - console.print("[red]SMTP credentials not configured. Set ACCURACY_SMTP_USER and ACCURACY_SMTP_PASSWORD.[/red]") - return - - effective_rate = rate or config.send_rate_per_second - effective_batch = batch_size or config.send_batch_size - delay = 1.0 / effective_rate if effective_rate > 0 else 1.0 - - sent = 0 - failed = 0 - start = datetime.now(tz=timezone.utc) - - for i, probe in enumerate(probes): - try: - message_id = await _send_one(probe, config) - await state.update_probe_status( - probe.probe_id, - ProbeStatus.SENT, - sent_at=datetime.now(tz=timezone.utc), - message_id=message_id, - ) - sent += 1 - logger.debug("Sent probe {} to {}", probe.probe_id[:8], probe.recipient) - except Exception as exc: - await state.update_probe_status( - probe.probe_id, - ProbeStatus.SEND_FAILED, - smtp_response=str(exc)[:500], - ) - failed += 1 - logger.warning("Failed to send probe {} to {}: {}", probe.probe_id[:8], probe.recipient, exc) - - # Rate limiting. - if i + 1 < len(probes): - await asyncio.sleep(delay) - # Batch pause. - if (i + 1) % effective_batch == 0: - logger.info("Batch pause ({} sent so far)...", sent) - await asyncio.sleep(config.send_batch_pause_seconds) - - elapsed = (datetime.now(tz=timezone.utc) - start).total_seconds() - console.print(f"\n[bold green]Done.[/bold green] Sent: {sent}, Failed: {failed}, Elapsed: {elapsed:.1f}s") - - -async def _send_one(probe: Probe, config: AccuracyConfig) -> str: - """Send a single probe email. Returns the Message-ID.""" - msg = EmailMessage() - msg["From"] = config.sender_address - msg["To"] = probe.recipient - msg["Subject"] = _SUBJECT - msg["Date"] = formatdate(localtime=True) - msg_id = make_msgid(domain=config.sender_address.split("@")[-1] if "@" in config.sender_address else "probe.local") - msg["Message-ID"] = msg_id - msg["Auto-Submitted"] = "auto-generated" # RFC 3834 - msg["Precedence"] = "junk" - msg["X-Probe-ID"] = probe.probe_id - - body = _BODY_TEMPLATE.format(sender=config.sender_address, probe_id=probe.probe_id) - msg.set_content(body) - - # Send via smtplib in a thread to keep async compat. - def _smtp_send() -> None: - with smtplib.SMTP(config.smtp_host, config.smtp_port, timeout=30) as smtp: - smtp.starttls() - smtp.login(config.smtp_user, config.smtp_password.get_secret_value()) - smtp.send_message(msg) - - await asyncio.to_thread(_smtp_send) - return msg_id - - -def _print_send_plan(probes: list[Probe]) -> None: - """Print a summary of probes about to be sent.""" - table = Table(title=f"Probes to Send ({len(probes)})", show_lines=False) - table.add_column("#", justify="right", width=4) - table.add_column("Domain", min_width=25) - table.add_column("Recipient", min_width=35) - table.add_column("Predicted", min_width=12) - table.add_column("Country", width=4) - - for i, p in enumerate(probes[:20], 1): - table.add_row(str(i), p.domain, p.recipient, p.predicted_provider, p.country.upper()) - if len(probes) > 20: - table.add_row("...", f"({len(probes) - 20} more)", "", "", "") - console.print(table) diff --git a/src/mail_municipalities/accuracy/state.py b/src/mail_municipalities/accuracy/state.py deleted file mode 100644 index e900ad2..0000000 --- a/src/mail_municipalities/accuracy/state.py +++ /dev/null @@ -1,303 +0,0 @@ -"""SQLite state management for probe lifecycle tracking.""" - -from __future__ import annotations - -import json -from datetime import datetime, timezone -from pathlib import Path -from typing import Any - -import aiosqlite - -from mail_municipalities.accuracy.models import NdrEvidence, NdrProvider, NdrResult, Probe, ProbeStatus - -_SCHEMA = """ -CREATE TABLE IF NOT EXISTS probes ( - probe_id TEXT PRIMARY KEY, - domain TEXT NOT NULL, - municipality_code TEXT NOT NULL, - municipality_name TEXT NOT NULL, - country TEXT NOT NULL, - recipient TEXT NOT NULL, - predicted_provider TEXT NOT NULL, - predicted_confidence REAL NOT NULL, - gateway TEXT, - status TEXT NOT NULL DEFAULT 'pending', - sent_at TEXT, - message_id TEXT, - smtp_response TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL -); - -CREATE TABLE IF NOT EXISTS ndrs ( - ndr_id INTEGER PRIMARY KEY AUTOINCREMENT, - probe_id TEXT NOT NULL REFERENCES probes(probe_id), - received_at TEXT NOT NULL, - ndr_from TEXT NOT NULL, - ndr_provider TEXT NOT NULL, - generating_mta TEXT NOT NULL, - confidence REAL NOT NULL, - evidence_json TEXT NOT NULL DEFAULT '[]', - raw_headers TEXT NOT NULL DEFAULT '', - created_at TEXT NOT NULL -); - -CREATE TABLE IF NOT EXISTS runs ( - run_id INTEGER PRIMARY KEY AUTOINCREMENT, - started_at TEXT NOT NULL, - command TEXT NOT NULL, - config_json TEXT NOT NULL DEFAULT '{}', - completed_at TEXT -); - -CREATE INDEX IF NOT EXISTS idx_probes_status ON probes(status); -CREATE INDEX IF NOT EXISTS idx_probes_domain ON probes(domain); -CREATE INDEX IF NOT EXISTS idx_probes_message_id ON probes(message_id); -CREATE INDEX IF NOT EXISTS idx_ndrs_probe_id ON ndrs(probe_id); -""" - - -def _now_utc() -> str: - return datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - -class StateDB: - """Async SQLite state store for probe lifecycle.""" - - def __init__(self, path: Path) -> None: - self._path = path - self._db: aiosqlite.Connection | None = None - - async def __aenter__(self) -> StateDB: - self._path.parent.mkdir(parents=True, exist_ok=True) - self._db = await aiosqlite.connect(self._path) - self._db.row_factory = aiosqlite.Row - await self._db.execute("PRAGMA journal_mode=WAL") - await self._db.execute("PRAGMA synchronous=NORMAL") - await self._db.executescript(_SCHEMA) - await self._db.commit() - return self - - async def __aexit__(self, *_exc: object) -> None: - if self._db is not None: - await self._db.close() - self._db = None - - # ── Runs ────────────────────────────────────────────────────── - - async def start_run(self, command: str, config_json: str = "{}") -> int: - assert self._db is not None - now = _now_utc() - cur = await self._db.execute( - "INSERT INTO runs (started_at, command, config_json) VALUES (?, ?, ?)", - (now, command, config_json), - ) - await self._db.commit() - assert cur.lastrowid is not None - return cur.lastrowid - - async def finish_run(self, run_id: int) -> None: - assert self._db is not None - await self._db.execute("UPDATE runs SET completed_at = ? WHERE run_id = ?", (_now_utc(), run_id)) - await self._db.commit() - - # ── Probes ──────────────────────────────────────────────────── - - async def insert_probes(self, probes: list[Probe]) -> int: - """Insert probes, skipping domains that already exist. Returns count inserted.""" - assert self._db is not None - now = _now_utc() - inserted = 0 - for p in probes: - try: - await self._db.execute( - "INSERT INTO probes " - "(probe_id, domain, municipality_code, municipality_name, country, " - "recipient, predicted_provider, predicted_confidence, gateway, " - "status, sent_at, message_id, smtp_response, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ( - p.probe_id, - p.domain, - p.municipality_code, - p.municipality_name, - p.country, - p.recipient, - p.predicted_provider, - p.predicted_confidence, - p.gateway, - p.status.value, - p.sent_at.isoformat() if p.sent_at else None, - p.message_id, - p.smtp_response, - now, - now, - ), - ) - inserted += 1 - except aiosqlite.IntegrityError: - pass # duplicate probe_id — skip - await self._db.commit() - return inserted - - async def get_probes_by_status(self, status: ProbeStatus, limit: int | None = None) -> list[Probe]: - assert self._db is not None - sql = "SELECT * FROM probes WHERE status = ? ORDER BY created_at" - params: list[object] = [status.value] - if limit is not None: - sql += " LIMIT ?" - params.append(limit) - rows = await self._db.execute_fetchall(sql, params) - return [_row_to_probe(r) for r in rows] - - async def get_all_probes(self) -> list[Probe]: - assert self._db is not None - rows = await self._db.execute_fetchall("SELECT * FROM probes ORDER BY created_at") - return [_row_to_probe(r) for r in rows] - - async def get_existing_domains(self) -> set[str]: - """Return domains that already have probes.""" - assert self._db is not None - rows = await self._db.execute_fetchall("SELECT DISTINCT domain FROM probes") - return {r[0] for r in rows} - - async def update_probe_status( - self, - probe_id: str, - status: ProbeStatus, - *, - sent_at: datetime | None = None, - message_id: str | None = None, - smtp_response: str | None = None, - ) -> None: - assert self._db is not None - now = _now_utc() - await self._db.execute( - "UPDATE probes SET status = ?, sent_at = COALESCE(?, sent_at), " - "message_id = COALESCE(?, message_id), smtp_response = COALESCE(?, smtp_response), " - "updated_at = ? WHERE probe_id = ?", - ( - status.value, - sent_at.isoformat() if sent_at else None, - message_id, - smtp_response, - now, - probe_id, - ), - ) - await self._db.commit() - - async def mark_timed_out(self, cutoff: datetime) -> int: - """Mark SENT probes older than *cutoff* as NO_NDR. Returns count updated.""" - assert self._db is not None - now = _now_utc() - cur = await self._db.execute( - "UPDATE probes SET status = ?, updated_at = ? WHERE status = ? AND sent_at < ?", - (ProbeStatus.NO_NDR.value, now, ProbeStatus.SENT.value, cutoff.isoformat()), - ) - await self._db.commit() - return cur.rowcount - - async def find_probe_by_message_id(self, message_id: str) -> Probe | None: - assert self._db is not None - rows: list[Any] = list( - await self._db.execute_fetchall("SELECT * FROM probes WHERE message_id = ?", (message_id,)) - ) - return _row_to_probe(rows[0]) if rows else None - - async def find_probe_by_recipient(self, recipient: str) -> Probe | None: - assert self._db is not None - rows: list[Any] = list( - await self._db.execute_fetchall("SELECT * FROM probes WHERE recipient = ?", (recipient,)) - ) - return _row_to_probe(rows[0]) if rows else None - - async def find_probe_by_recipient_substring(self, substring: str) -> Probe | None: - """Find a probe whose recipient contains *substring* (e.g. the UUID portion).""" - assert self._db is not None - rows: list[Any] = list( - await self._db.execute_fetchall("SELECT * FROM probes WHERE recipient LIKE ? LIMIT 1", (f"%{substring}%",)) - ) - return _row_to_probe(rows[0]) if rows else None - - # ── NDRs ────────────────────────────────────────────────────── - - async def insert_ndr(self, ndr: NdrResult) -> None: - assert self._db is not None - now = _now_utc() - evidence_json = json.dumps([{"pattern": e.pattern, "matched_value": e.matched_value} for e in ndr.evidence]) - await self._db.execute( - "INSERT INTO ndrs " - "(probe_id, received_at, ndr_from, ndr_provider, generating_mta, " - "confidence, evidence_json, raw_headers, created_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - ( - ndr.probe_id, - ndr.received_at.isoformat(), - ndr.ndr_from, - ndr.ndr_provider.value, - ndr.generating_mta, - ndr.confidence, - evidence_json, - ndr.raw_headers, - now, - ), - ) - await self._db.commit() - - async def get_all_ndrs(self) -> list[NdrResult]: - assert self._db is not None - rows = await self._db.execute_fetchall("SELECT * FROM ndrs ORDER BY created_at") - return [_row_to_ndr(r) for r in rows] - - async def has_ndr_for_probe(self, probe_id: str) -> bool: - assert self._db is not None - rows: list[Any] = list( - await self._db.execute_fetchall("SELECT 1 FROM ndrs WHERE probe_id = ? LIMIT 1", (probe_id,)) - ) - return len(rows) > 0 - - # ── Status summary ──────────────────────────────────────────── - - async def status_counts(self) -> dict[str, int]: - assert self._db is not None - rows = await self._db.execute_fetchall("SELECT status, COUNT(*) FROM probes GROUP BY status") - return {r[0]: r[1] for r in rows} - - async def country_counts(self) -> dict[str, int]: - assert self._db is not None - rows = await self._db.execute_fetchall("SELECT country, COUNT(*) FROM probes GROUP BY country") - return {r[0]: r[1] for r in rows} - - -def _row_to_probe(row: aiosqlite.Row) -> Probe: - return Probe( - probe_id=row["probe_id"], - domain=row["domain"], - municipality_code=row["municipality_code"], - municipality_name=row["municipality_name"], - country=row["country"], - recipient=row["recipient"], - predicted_provider=row["predicted_provider"], - predicted_confidence=row["predicted_confidence"], - gateway=row["gateway"], - status=ProbeStatus(row["status"]), - sent_at=datetime.fromisoformat(row["sent_at"]) if row["sent_at"] else None, - message_id=row["message_id"], - smtp_response=row["smtp_response"], - ) - - -def _row_to_ndr(row: aiosqlite.Row) -> NdrResult: - evidence_raw = json.loads(row["evidence_json"]) if row["evidence_json"] else [] - return NdrResult( - probe_id=row["probe_id"], - received_at=datetime.fromisoformat(row["received_at"]), - ndr_from=row["ndr_from"], - ndr_provider=NdrProvider(row["ndr_provider"]), - generating_mta=row["generating_mta"], - confidence=row["confidence"], - evidence=[NdrEvidence(pattern=e["pattern"], matched_value=e["matched_value"]) for e in evidence_raw], - raw_headers=row["raw_headers"], - ) diff --git a/src/mail_municipalities/cli.py b/src/mail_municipalities/cli.py index 89abcfb..08d3049 100644 --- a/src/mail_municipalities/cli.py +++ b/src/mail_municipalities/cli.py @@ -450,441 +450,6 @@ def _scan_main( run(domains_path, output_path, cc=country, verbose=verbose) -# ── Accuracy subcommands ──────────────────────────────────────────── - -_accuracy_app = typer.Typer(add_completion=False, help="Bounce-probe accuracy validation for provider classification.") -app.add_typer(_accuracy_app, name="accuracy") - - -def _accuracy_config(output: Path | None = None): - """Lazily load accuracy config with optional output dir override.""" - from mail_municipalities.accuracy.config import AccuracyConfig - - cfg = AccuracyConfig() - if output is not None: - cfg = cfg.model_copy(update={"output_dir": output}) - return cfg - - -@_accuracy_app.command("sample") -def accuracy_sample_cmd( - country: Annotated[ - Optional[str], - typer.Argument(help="Country code: ch, de, at"), - ] = None, - all_countries: Annotated[ - bool, - typer.Option("--all", help="Sample from all countries"), - ] = False, - size: Annotated[ - int, - typer.Option("--size", help="Total sample size"), - ] = 50, - min_per_class: Annotated[ - int, - typer.Option("--min-per-class", help="Minimum samples per provider class"), - ] = 5, - providers_dir: Annotated[ - Path, - typer.Option("--providers-dir", help="Directory with provider classification output"), - ] = Path("output/providers"), - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Create a stratified sample and prepare probes (no sending).""" - if not country and not all_countries: - typer.echo("Provide a country code (ch, de, at) or use --all", err=True) - raise typer.Exit(code=1) - - cfg = _accuracy_config(output) - setup_logging(verbose, log_path=cfg.output_dir / "accuracy.log") - assert all_countries or country is not None - countries: list[str] = ["de", "at", "ch"] if all_countries else [country] # type: ignore[list-item] - - from mail_municipalities.accuracy.sampler import create_sample - from mail_municipalities.accuracy.state import StateDB - - async def _run() -> None: - async with StateDB(cfg.state_db_path) as state: - await create_sample(countries, size, min_per_class, providers_dir, state) - - asyncio.run(_run()) - - -@_accuracy_app.command("send") -def accuracy_send_cmd( - dry_run: Annotated[ - bool, - typer.Option("--dry-run/--no-dry-run", help="Dry run (default: enabled)"), - ] = True, - confirm: Annotated[ - bool, - typer.Option("--confirm", help="Skip interactive confirmation prompt"), - ] = False, - batch_size: Annotated[ - Optional[int], - typer.Option("--batch-size", help="Emails per batch before pause"), - ] = None, - max_probes: Annotated[ - Optional[int], - typer.Option("--max-probes", help="Maximum probes to send this run"), - ] = None, - rate: Annotated[ - Optional[float], - typer.Option("--rate", help="Emails per second"), - ] = None, - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Send probe emails to sampled municipality domains.""" - cfg = _accuracy_config(output) - setup_logging(verbose, log_path=cfg.output_dir / "accuracy.log") - - from mail_municipalities.accuracy.sender import send_probes - from mail_municipalities.accuracy.state import StateDB - - async def _run() -> None: - async with StateDB(cfg.state_db_path) as state: - await send_probes( - state, - cfg, - max_probes=max_probes, - dry_run=dry_run, - confirm=confirm, - batch_size=batch_size, - rate=rate, - ) - - asyncio.run(_run()) - - -@_accuracy_app.command("collect") -def accuracy_collect_cmd( - poll_once: Annotated[ - bool, - typer.Option("--poll-once", help="Single IMAP check then exit"), - ] = False, - wait_hours: Annotated[ - Optional[float], - typer.Option("--wait-hours", help="Max hours to keep polling"), - ] = None, - poll_interval: Annotated[ - Optional[int], - typer.Option("--poll-interval", help="Seconds between IMAP polls"), - ] = None, - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Poll IMAP for NDRs and match to sent probes.""" - cfg = _accuracy_config(output) - setup_logging(verbose, log_path=cfg.output_dir / "accuracy.log") - - from mail_municipalities.accuracy.collector import collect_ndrs - from mail_municipalities.accuracy.state import StateDB - - async def _run() -> None: - async with StateDB(cfg.state_db_path) as state: - await collect_ndrs( - state, - cfg, - poll_once=poll_once, - wait_hours=wait_hours, - poll_interval=poll_interval, - ) - - asyncio.run(_run()) - - -@_accuracy_app.command("report") -def accuracy_report_cmd( - latex: Annotated[ - bool, - typer.Option("--latex", help="Export LaTeX tables for the paper"), - ] = False, - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Compute and display accuracy metrics from collected NDRs.""" - cfg = _accuracy_config(output) - setup_logging(verbose, log_path=cfg.output_dir / "accuracy.log") - - from mail_municipalities.accuracy.metrics import compute_accuracy - from mail_municipalities.accuracy.report import export_report_json, export_report_latex, print_report - from mail_municipalities.accuracy.state import StateDB - - async def _run() -> None: - async with StateDB(cfg.state_db_path) as state: - report = await compute_accuracy(state) - print_report(report) - export_report_json(report, cfg.output_dir) - if latex: - export_report_latex(report, cfg.output_dir) - - asyncio.run(_run()) - - -@_accuracy_app.command("status") -def accuracy_status_cmd( - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Show current probe lifecycle state.""" - cfg = _accuracy_config(output) - setup_logging(verbose, log_path=cfg.output_dir / "accuracy.log") - - from mail_municipalities.accuracy.report import print_status - from mail_municipalities.accuracy.state import StateDB - - async def _run() -> None: - async with StateDB(cfg.state_db_path) as state: - await print_status(state) - - asyncio.run(_run()) - - -@_accuracy_app.command("check") -def accuracy_check_cmd( - domains: Annotated[ - list[str], - typer.Argument(help="One or more email domains to look up"), - ], - providers_dir: Annotated[ - Path, - typer.Option("--providers-dir", help="Directory with provider classification output"), - ] = Path("output/providers"), - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Spot-check provider classification for specific domains.""" - cfg = _accuracy_config(output) - setup_logging(verbose, log_path=cfg.output_dir / "accuracy.log") - - from mail_municipalities.accuracy.check import check_domains, print_check_table - from mail_municipalities.accuracy.state import StateDB - - async def _run() -> None: - async with StateDB(cfg.state_db_path) as state: - results = await check_domains(domains, providers_dir, state) - print_check_table(results) - - asyncio.run(_run()) - - -_accuracy_standalone_app = typer.Typer(add_completion=False) - - -@_accuracy_standalone_app.command("sample") -def _accuracy_sample_main( - country: Annotated[ - Optional[str], - typer.Argument(help="Country code: ch, de, at"), - ] = None, - all_countries: Annotated[ - bool, - typer.Option("--all", help="Sample from all countries"), - ] = False, - size: Annotated[ - int, - typer.Option("--size", help="Total sample size"), - ] = 50, - min_per_class: Annotated[ - int, - typer.Option("--min-per-class", help="Minimum samples per provider class"), - ] = 5, - providers_dir: Annotated[ - Path, - typer.Option("--providers-dir", help="Directory with provider classification output"), - ] = Path("output/providers"), - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Create a stratified sample and prepare probes (no sending).""" - accuracy_sample_cmd( - country=country, - all_countries=all_countries, - size=size, - min_per_class=min_per_class, - providers_dir=providers_dir, - verbose=verbose, - output=output, - ) - - -@_accuracy_standalone_app.command("send") -def _accuracy_send_main( - dry_run: Annotated[ - bool, - typer.Option("--dry-run/--no-dry-run", help="Dry run (default: enabled)"), - ] = True, - confirm: Annotated[ - bool, - typer.Option("--confirm", help="Skip interactive confirmation prompt"), - ] = False, - batch_size: Annotated[ - Optional[int], - typer.Option("--batch-size", help="Emails per batch before pause"), - ] = None, - max_probes: Annotated[ - Optional[int], - typer.Option("--max-probes", help="Maximum probes to send this run"), - ] = None, - rate: Annotated[ - Optional[float], - typer.Option("--rate", help="Emails per second"), - ] = None, - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Send probe emails to sampled municipality domains.""" - accuracy_send_cmd( - dry_run=dry_run, - confirm=confirm, - batch_size=batch_size, - max_probes=max_probes, - rate=rate, - verbose=verbose, - output=output, - ) - - -@_accuracy_standalone_app.command("collect") -def _accuracy_collect_main( - poll_once: Annotated[ - bool, - typer.Option("--poll-once", help="Single IMAP check then exit"), - ] = False, - wait_hours: Annotated[ - Optional[float], - typer.Option("--wait-hours", help="Max hours to keep polling"), - ] = None, - poll_interval: Annotated[ - Optional[int], - typer.Option("--poll-interval", help="Seconds between IMAP polls"), - ] = None, - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Poll IMAP for NDRs and match to sent probes.""" - accuracy_collect_cmd( - poll_once=poll_once, - wait_hours=wait_hours, - poll_interval=poll_interval, - verbose=verbose, - output=output, - ) - - -@_accuracy_standalone_app.command("report") -def _accuracy_report_main( - latex: Annotated[ - bool, - typer.Option("--latex", help="Export LaTeX tables for the paper"), - ] = False, - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Compute and display accuracy metrics from collected NDRs.""" - accuracy_report_cmd(latex=latex, verbose=verbose, output=output) - - -@_accuracy_standalone_app.command("status") -def _accuracy_status_main( - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Show current probe lifecycle state.""" - accuracy_status_cmd(verbose=verbose, output=output) - - -@_accuracy_standalone_app.command("check") -def _accuracy_check_main( - domains: Annotated[ - list[str], - typer.Argument(help="One or more email domains to look up"), - ], - providers_dir: Annotated[ - Path, - typer.Option("--providers-dir", help="Directory with provider classification output"), - ] = Path("output/providers"), - verbose: Annotated[ - bool, - typer.Option("-v", "--verbose", help="Enable debug logging"), - ] = False, - output: Annotated[ - Optional[Path], - typer.Option("-o", "--output", help="Custom output directory"), - ] = None, -) -> None: - """Spot-check provider classification for specific domains.""" - accuracy_check_cmd(domains=domains, providers_dir=providers_dir, verbose=verbose, output=output) - - def resolve() -> None: """Entry point for 'resolve' script.""" _resolve_app() @@ -903,8 +468,3 @@ def analyze() -> None: def scan() -> None: """Entry point for 'scan' script.""" _scan_app() - - -def accuracy() -> None: - """Entry point for 'accuracy' script.""" - _accuracy_standalone_app() diff --git a/tests/test_accuracy.py b/tests/test_accuracy.py deleted file mode 100644 index d06de63..0000000 --- a/tests/test_accuracy.py +++ /dev/null @@ -1,854 +0,0 @@ -"""Tests for the accuracy validation package.""" - -from __future__ import annotations - -import json -import uuid -from datetime import datetime, timezone -from email.message import EmailMessage -from pathlib import Path - -import pytest - -from mail_municipalities.accuracy.metrics import compute_accuracy -from mail_municipalities.accuracy.models import ( - CLASSIFIER_TO_EVAL, - NDR_TO_CLASSIFIER, - NdrEvidence, - NdrProvider, - NdrResult, - Probe, - ProbeStatus, -) -from mail_municipalities.accuracy.ndr_parser import parse_ndr -from mail_municipalities.accuracy.sampler import _make_probe, _stratified_sample -from mail_municipalities.accuracy.state import StateDB - - -# ── Model tests ─────────────────────────────────────────────────── - - -class TestModels: - def test_probe_status_values(self): - assert ProbeStatus.PENDING.value == "pending" - assert ProbeStatus.SENT.value == "sent" - assert ProbeStatus.NDR_RECEIVED.value == "ndr_received" - - def test_ndr_provider_values(self): - assert NdrProvider.MICROSOFT.value == "microsoft" - assert NdrProvider.POSTFIX.value == "postfix" - - def test_ndr_to_classifier_mapping(self): - assert NDR_TO_CLASSIFIER[NdrProvider.MICROSOFT] == "microsoft" - assert NDR_TO_CLASSIFIER[NdrProvider.POSTFIX] == "self-hosted" - assert NDR_TO_CLASSIFIER[NdrProvider.EXIM] == "self-hosted" - assert NDR_TO_CLASSIFIER[NdrProvider.EXCHANGE_ONPREM] == "self-hosted" - assert NDR_TO_CLASSIFIER[NdrProvider.UNKNOWN] == "unknown" - - def test_classifier_to_eval_mapping(self): - assert CLASSIFIER_TO_EVAL["domestic"] == "self-hosted" - assert CLASSIFIER_TO_EVAL["foreign"] == "self-hosted" - assert CLASSIFIER_TO_EVAL["microsoft"] == "microsoft" - - def test_probe_creation(self): - p = Probe( - probe_id="abc123", - domain="example.ch", - municipality_code="42", - municipality_name="Testingen", - country="ch", - recipient="validation-probe-abc@example.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - assert p.status == ProbeStatus.PENDING - assert p.sent_at is None - assert p.gateway is None - - -# ── Sampler tests ───────────────────────────────────────────────── - - -class TestSampler: - @pytest.fixture - def entries(self): - """Fake municipality entries for sampling.""" - result = [] - providers = ["microsoft"] * 50 + ["google"] * 5 + ["aws"] * 10 + ["domestic"] * 30 + ["foreign"] * 5 - for i, prov in enumerate(providers): - result.append( - { - "code": str(i), - "name": f"Municipality {i}", - "domain": f"m{i}.example.ch", - "provider": prov, - "classification_confidence": 80.0, - } - ) - return result - - def test_stratified_sample_respects_size(self, entries): - sample = _stratified_sample(entries, total_size=20, min_per_class=2) - assert len(sample) <= 20 - - def test_stratified_sample_minimum_per_class(self, entries): - sample = _stratified_sample(entries, total_size=20, min_per_class=3) - by_provider = {} - for e in sample: - p = e["provider"] - by_provider[p] = by_provider.get(p, 0) + 1 - # Each validatable class should have at least min_per_class (or all if fewer exist). - for provider in ("microsoft", "google", "aws", "domestic", "foreign"): - assert by_provider.get(provider, 0) >= min(3, sum(1 for e in entries if e["provider"] == provider)) - - def test_excludes_unknown_provider(self, entries): - entries.append({"code": "999", "name": "Unknown", "domain": "unk.ch", "provider": "unknown"}) - sample = _stratified_sample(entries, total_size=200, min_per_class=1) - providers = {e["provider"] for e in sample} - assert "unknown" not in providers - - def test_excludes_empty_domain(self, entries): - entries.append({"code": "998", "name": "NoDomain", "domain": "", "provider": "microsoft"}) - sample = _stratified_sample(entries, total_size=200, min_per_class=1) - assert not any(e["domain"] == "" for e in sample) - - def test_make_probe(self): - entry = { - "code": "42", - "name": "Test", - "domain": "test.ch", - "provider": "microsoft", - "classification_confidence": 90.0, - "gateway": "seppmail", - } - probe = _make_probe(entry, "ch") - assert probe.domain == "test.ch" - assert probe.country == "ch" - assert probe.predicted_provider == "microsoft" - assert probe.gateway == "seppmail" - assert "validation-probe-" in probe.recipient - assert probe.recipient.endswith("@test.ch") - assert probe.status == ProbeStatus.PENDING - - -# ── NDR parser tests ────────────────────────────────────────────── - - -def _make_ndr_email(**kwargs) -> EmailMessage: - """Build a minimal NDR EmailMessage for testing.""" - msg = EmailMessage() - msg["From"] = kwargs.get("from_addr", "mailer-daemon@example.com") - msg["Subject"] = kwargs.get("subject", "Delivery Status Notification") - for hdr, val in kwargs.get("extra_headers", {}).items(): - msg[hdr] = val - if "received" in kwargs: - for r in kwargs["received"]: - msg["Received"] = r - msg.set_content(kwargs.get("body", "")) - return msg - - -class TestNdrParser: - def test_microsoft_exchange_online(self): - msg = _make_ndr_email( - from_addr="postmaster@outlook.com", - extra_headers={ - "X-MS-Exchange-Message-Sent-Representing-Type": "1", - }, - received=["from mail-eopbgr70045.outbound.protection.outlook.com by mx.example.com"], - body="Delivery has failed to these recipients or groups.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.MICROSOFT - assert confidence > 0.3 - assert len(evidence) >= 2 - - def test_google_workspace_relay_bounce(self): - """Gmail relay bounce where the target IS Google Workspace. - - The Remote-MTA in the DSN points to Google, confirming the - target is actually Google (not just our relay). - """ - from email.mime.multipart import MIMEMultipart - from email.mime.text import MIMEText - - outer = MIMEMultipart("report", report_type="delivery-status") - outer["From"] = "mailer-daemon@googlemail.com" - outer["Received"] = "from mail-wr1-f54.google.com by mx.google.com" - - text_part = MIMEText("Delivery to the following recipient failed permanently.") - outer.attach(text_part) - - dsn_part = MIMEText( - "Reporting-MTA: dns; googlemail.com\n" - "Remote-MTA: dns; aspmx.l.google.com\n" - "Diagnostic-Code: smtp; 550 5.1.1 not exist\n", - "delivery-status", - ) - outer.attach(dsn_part) - - import email - import email.policy - - parsed = email.message_from_bytes(outer.as_bytes(), policy=email.policy.default) - assert isinstance(parsed, EmailMessage) - provider, confidence, mta, evidence = parse_ndr(parsed) - assert provider == NdrProvider.GOOGLE - - def test_google_workspace_direct_ndr(self): - """NDR sent directly by Google (not via our Gmail relay).""" - msg = _make_ndr_email( - from_addr="postmaster@google.com", - received=["from mail-wr1-f54.google.com by mx.example.com"], - body="Delivery to the following recipient failed permanently.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.GOOGLE - assert confidence > 0.3 - - def test_aws_ses(self): - msg = _make_ndr_email( - from_addr="MAILER-DAEMON@amazonses.com", - extra_headers={"X-SES-Outgoing": "2024.01.01-54.240.0.1"}, - received=["from a48-93.smtp-out.amazonses.com by mx.example.com"], - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.AWS - assert confidence > 0.3 - - def test_postfix(self): - msg = _make_ndr_email( - from_addr="MAILER-DAEMON@mail.example.ch", - received=["from mail.example.ch (Postfix) by mail.example.ch"], - body="This is the mail system at host mail.example.ch.\n\nI'm sorry to have to inform you...", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.POSTFIX - assert confidence > 0.3 - - def test_exim(self): - msg = _make_ndr_email( - from_addr="MAILER-DAEMON@mail.example.de", - subject="Mail delivery failed: returning message to sender", - received=["from mail.example.de (Exim 4.96) by mail.example.de"], - body="A message that you sent could not be delivered to one or more of its recipients.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.EXIM - assert confidence > 0.3 - - def test_exchange_onprem(self): - msg = _make_ndr_email( - from_addr="postmaster@internal.example.ch", - extra_headers={ - "X-MS-Exchange-Organization-SCL": "-1", - }, - received=["from mail.internal.example.ch by mail.internal.example.ch"], - body="Delivery has failed to these recipients.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - # Has Exchange headers but no outlook.com in the chain -> on-premises. - assert provider == NdrProvider.EXCHANGE_ONPREM - - def test_gmail_relay_bounce_to_microsoft(self): - """Gmail generates the NDR but the actual target is MS365. - - This is the key scenario: our Gmail relay reports a rejection from - a Microsoft MTA. The parser must see through Gmail's headers. - """ - from email.mime.multipart import MIMEMultipart - from email.mime.text import MIMEText - - outer = MIMEMultipart("report", report_type="delivery-status") - outer["From"] = "mailer-daemon@googlemail.com" - outer["X-Gm-Message-State"] = "some-state" - outer["Received"] = "from mail-wr1-f54.google.com by mx.google.com" - - text_part = MIMEText("Delivery to the following recipient failed permanently.") - outer.attach(text_part) - - dsn_part = MIMEText( - "Reporting-MTA: dns; googlemail.com\n" - "Remote-MTA: dns; municipality.mail.protection.outlook.com\n" - "Diagnostic-Code: smtp; 550 5.1.1 The email account does not exist\n", - "delivery-status", - ) - outer.attach(dsn_part) - - # Parse via email.message.EmailMessage - import email - import email.policy - - parsed = email.message_from_bytes(outer.as_bytes(), policy=email.policy.default) - assert isinstance(parsed, EmailMessage) - provider, confidence, mta, evidence = parse_ndr(parsed) - assert provider == NdrProvider.MICROSOFT, f"Expected MICROSOFT, got {provider}" - - def test_gmail_relay_bounce_to_aws(self): - """Gmail NDR where the actual target is AWS SES.""" - from email.mime.multipart import MIMEMultipart - from email.mime.text import MIMEText - - outer = MIMEMultipart("report", report_type="delivery-status") - outer["From"] = "mailer-daemon@googlemail.com" - outer["X-Gm-Message-State"] = "some-state" - outer["Received"] = "from mail-wr1-f54.google.com by mx.google.com" - - text_part = MIMEText("Delivery to the following recipient failed permanently.") - outer.attach(text_part) - - dsn_part = MIMEText( - "Reporting-MTA: dns; googlemail.com\n" - "Remote-MTA: dns; inbound-smtp.eu-west-1.amazonaws.com\n" - "Diagnostic-Code: smtp; 550 5.1.1 unknown user\n", - "delivery-status", - ) - outer.attach(dsn_part) - - import email - import email.policy - - parsed = email.message_from_bytes(outer.as_bytes(), policy=email.policy.default) - assert isinstance(parsed, EmailMessage) - provider, confidence, mta, evidence = parse_ndr(parsed) - assert provider == NdrProvider.AWS, f"Expected AWS, got {provider}" - - def test_gmail_relay_bounce_to_selfhosted(self): - """Gmail relay bounce to a generic self-hosted MTA (not a cloud provider).""" - from email.mime.multipart import MIMEMultipart - from email.mime.text import MIMEText - - outer = MIMEMultipart("report", report_type="delivery-status") - outer["From"] = "mailer-daemon@googlemail.com" - outer["X-Gm-Message-State"] = "some-state" - outer["Received"] = "from mail-wr1-f54.google.com by mx.google.com" - - text_part = MIMEText("Delivery to the following recipient failed permanently.") - outer.attach(text_part) - - dsn_part = MIMEText( - "Reporting-MTA: dns; googlemail.com\n" - "Remote-MTA: dns; mail.gemeinde-insul.de\n" - "Diagnostic-Code: smtp; 550 5.1.1 user unknown\n", - "delivery-status", - ) - outer.attach(dsn_part) - - import email - import email.policy - - parsed = email.message_from_bytes(outer.as_bytes(), policy=email.policy.default) - assert isinstance(parsed, EmailMessage) - provider, confidence, mta, evidence = parse_ndr(parsed) - # Remote-MTA is a generic hostname — should be self-hosted, NOT Google. - assert provider == NdrProvider.POSTFIX, f"Expected POSTFIX (self-hosted), got {provider}" - - def test_gmail_relay_headers_ignored(self): - """X-Gm-* and Received from google.com should not trigger Google detection.""" - msg = _make_ndr_email( - from_addr="postmaster@mx.example.ch", - extra_headers={ - "X-Gm-Message-State": "some-state", - "X-MS-Exchange-Organization-SCL": "-1", - }, - received=[ - "from mail-wr1-f54.google.com by mx.google.com", - "from mx.example.ch by mx.example.ch", - ], - body="Delivery has failed to these recipients or groups.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - # Should detect Exchange, not Google. - assert provider in (NdrProvider.MICROSOFT, NdrProvider.EXCHANGE_ONPREM) - - def test_exchange_online_onmicrosoft_from(self): - """NDR from postmaster@*.onmicrosoft.com with Hosted entity header → MICROSOFT.""" - msg = _make_ndr_email( - from_addr="postmaster@stadtxyz.onmicrosoft.com", - extra_headers={ - "X-MS-Exchange-Message-Is-Ndr": "", - "X-Ms-Exchange-Crosstenant-Fromentityheader": "Hosted", - }, - received=[ - "from GVAP278CU002.outbound.protection.outlook.com by slusv0255.example.ch", - "from ZR3P278MB1195.CHEP278.PROD.OUTLOOK.COM by ZR3P278MB1195", - ], - body="Delivery has failed to these recipients or groups.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.MICROSOFT - - def test_exchange_hybrid_onprem(self): - """NDR from postmaster@domain.ch with HybridOnPrem entity header → EXCHANGE_ONPREM.""" - msg = _make_ndr_email( - from_addr="postmaster@bern.ch", - extra_headers={ - "X-MS-Exchange-Message-Is-Ndr": "", - "X-Ms-Exchange-Crosstenant-Fromentityheader": "HybridOnPrem", - }, - received=[ - "from GVAP278CU002.outbound.protection.outlook.com by mx.google.com", - "from AutoDiscover.bgov.ch by ZRH2EPF00000151.mail.protection.outlook.com", - ], - body="Delivery has failed to these recipients or groups.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.EXCHANGE_ONPREM - - def test_unknown_ndr(self): - msg = _make_ndr_email( - from_addr="postmaster@somegateway.net", - body="Your message could not be delivered.", - ) - provider, confidence, mta, evidence = parse_ndr(msg) - assert provider == NdrProvider.UNKNOWN - assert confidence == 0.0 - - -# ── State DB tests ──────────────────────────────────────────────── - - -class TestStateDB: - @pytest.fixture - async def state(self, tmp_path): - db_path = tmp_path / "test_state.db" - async with StateDB(db_path) as s: - yield s - - async def test_insert_and_retrieve_probes(self, state: StateDB): - probe = Probe( - probe_id=uuid.uuid4().hex, - domain="test.ch", - municipality_code="1", - municipality_name="Test", - country="ch", - recipient="validation-probe-abc@test.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - inserted = await state.insert_probes([probe]) - assert inserted == 1 - - probes = await state.get_probes_by_status(ProbeStatus.PENDING) - assert len(probes) == 1 - assert probes[0].domain == "test.ch" - - async def test_duplicate_probe_skipped(self, state: StateDB): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain="test.ch", - municipality_code="1", - municipality_name="Test", - country="ch", - recipient="validation-probe-abc@test.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - inserted = await state.insert_probes([probe]) - assert inserted == 0 - - async def test_update_probe_status(self, state: StateDB): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain="test.ch", - municipality_code="1", - municipality_name="Test", - country="ch", - recipient="validation-probe-abc@test.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - now = datetime.now(tz=timezone.utc) - await state.update_probe_status(pid, ProbeStatus.SENT, sent_at=now, message_id="") - - probes = await state.get_probes_by_status(ProbeStatus.SENT) - assert len(probes) == 1 - assert probes[0].message_id == "" - - async def test_find_by_message_id(self, state: StateDB): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain="test.ch", - municipality_code="1", - municipality_name="Test", - country="ch", - recipient="validation-probe-abc@test.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - await state.update_probe_status(pid, ProbeStatus.SENT, message_id="") - - found = await state.find_probe_by_message_id("") - assert found is not None - assert found.probe_id == pid - - async def test_find_by_recipient_substring(self, state: StateDB): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain="test.ch", - municipality_code="1", - municipality_name="Test", - country="ch", - recipient="validation-probe-abc123def456@test.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - - found = await state.find_probe_by_recipient_substring("abc123def456") - assert found is not None - assert found.probe_id == pid - - async def test_insert_and_retrieve_ndr(self, state: StateDB): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain="test.ch", - municipality_code="1", - municipality_name="Test", - country="ch", - recipient="validation-probe-abc@test.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - - ndr = NdrResult( - probe_id=pid, - received_at=datetime.now(tz=timezone.utc), - ndr_from="mailer-daemon@googlemail.com", - ndr_provider=NdrProvider.GOOGLE, - generating_mta="mail-wr1-f54.google.com", - confidence=0.9, - evidence=[NdrEvidence(pattern="from mailer-daemon@google", matched_value="mailer-daemon@googlemail.com")], - ) - await state.insert_ndr(ndr) - - ndrs = await state.get_all_ndrs() - assert len(ndrs) == 1 - assert ndrs[0].ndr_provider == NdrProvider.GOOGLE - assert len(ndrs[0].evidence) == 1 - - async def test_status_counts(self, state: StateDB): - for i in range(3): - probe = Probe( - probe_id=uuid.uuid4().hex, - domain=f"test{i}.ch", - municipality_code=str(i), - municipality_name=f"Test {i}", - country="ch", - recipient=f"validation-probe-{i}@test{i}.ch", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - - counts = await state.status_counts() - assert counts["pending"] == 3 - - async def test_existing_domains(self, state: StateDB): - for domain in ("a.ch", "b.ch", "a.ch"): - probe = Probe( - probe_id=uuid.uuid4().hex, - domain=domain, - municipality_code="1", - municipality_name="Test", - country="ch", - recipient=f"validation-probe-{uuid.uuid4().hex[:6]}@{domain}", - predicted_provider="microsoft", - predicted_confidence=90.0, - ) - await state.insert_probes([probe]) - - domains = await state.get_existing_domains() - assert domains == {"a.ch", "b.ch"} - - -# ── Metrics tests ───────────────────────────────────────────────── - - -class TestMetrics: - @pytest.fixture - async def state_with_data(self, tmp_path): - """State DB with matched probes and NDRs for metrics testing.""" - db_path = tmp_path / "metrics_test.db" - async with StateDB(db_path) as state: - # Create probes: 3 microsoft, 2 domestic, 1 google. - test_data = [ - ("microsoft", NdrProvider.MICROSOFT), # correct - ("microsoft", NdrProvider.MICROSOFT), # correct - ("microsoft", NdrProvider.GOOGLE), # wrong - ("domestic", NdrProvider.POSTFIX), # correct (domestic -> self-hosted, postfix -> self-hosted) - ("domestic", NdrProvider.MICROSOFT), # wrong - ("google", NdrProvider.GOOGLE), # correct - ] - - for i, (predicted, actual_ndr) in enumerate(test_data): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain=f"test{i}.ch", - municipality_code=str(i), - municipality_name=f"Test {i}", - country="ch", - recipient=f"probe-{i}@test{i}.ch", - predicted_provider=predicted, - predicted_confidence=80.0, - status=ProbeStatus.NDR_RECEIVED, - ) - await state.insert_probes([probe]) - await state.update_probe_status(pid, ProbeStatus.NDR_RECEIVED, sent_at=datetime.now(tz=timezone.utc)) - - ndr = NdrResult( - probe_id=pid, - received_at=datetime.now(tz=timezone.utc), - ndr_from="mailer-daemon@example.com", - ndr_provider=actual_ndr, - generating_mta="mta.example.com", - confidence=0.8, - ) - await state.insert_ndr(ndr) - - yield state - - async def test_overall_accuracy(self, state_with_data: StateDB): - report = await compute_accuracy(state_with_data) - # 4 correct out of 6 total. - assert abs(report.overall_accuracy - 4 / 6) < 0.01 - - async def test_per_class_metrics(self, state_with_data: StateDB): - report = await compute_accuracy(state_with_data) - # Microsoft: TP=2, FP=1 (domestic predicted microsoft actual), FN=1 (microsoft predicted google actual). - ms = report.per_class["microsoft"] - assert ms.support == 3 # 3 actual microsoft - assert ms.precision == pytest.approx(2 / 3, abs=0.01) - assert ms.recall == pytest.approx(2 / 3, abs=0.01) - - async def test_confusion_matrix(self, state_with_data: StateDB): - report = await compute_accuracy(state_with_data) - cm = report.confusion_matrix - assert cm["microsoft"]["microsoft"] == 2 - assert cm["microsoft"]["google"] == 1 - assert cm["self-hosted"]["microsoft"] == 1 - assert cm["self-hosted"]["self-hosted"] == 1 - - async def test_response_rate(self, state_with_data: StateDB): - report = await compute_accuracy(state_with_data) - assert report.total_ndrs == 6 - assert report.total_probes == 6 - - async def test_weighted_f1_no_aws(self, state_with_data: StateDB): - """Weighted F1 over microsoft, google, self-hosted (no AWS in fixture).""" - report = await compute_accuracy(state_with_data) - assert report.weighted_f1_labels == ["microsoft", "google", "self-hosted"] - # Manually compute: weight each class F1 by its support. - ms = report.per_class["microsoft"] - go = report.per_class["google"] - sh = report.per_class["self-hosted"] - expected = (ms.f1 * ms.support + go.f1 * go.support + sh.f1 * sh.support) / ( - ms.support + go.support + sh.support - ) - assert report.weighted_f1 == pytest.approx(expected, abs=0.001) - - async def test_weighted_f1_excludes_aws(self, tmp_path): - """AWS probes must not affect the weighted F1 score.""" - db_path = tmp_path / "wf1_test.db" - async with StateDB(db_path) as state: - # 2 correct microsoft, 2 correct domestic, 1 aws->self-hosted (wrong but excluded) - test_data = [ - ("microsoft", NdrProvider.MICROSOFT), - ("microsoft", NdrProvider.MICROSOFT), - ("domestic", NdrProvider.POSTFIX), - ("domestic", NdrProvider.POSTFIX), - ("aws", NdrProvider.POSTFIX), # AWS misclassified — should NOT affect weighted F1 - ] - for i, (predicted, actual_ndr) in enumerate(test_data): - pid = uuid.uuid4().hex - probe = Probe( - probe_id=pid, - domain=f"test{i}.ch", - municipality_code=str(i), - municipality_name=f"Test {i}", - country="ch", - recipient=f"probe-{i}@test{i}.ch", - predicted_provider=predicted, - predicted_confidence=80.0, - status=ProbeStatus.NDR_RECEIVED, - ) - await state.insert_probes([probe]) - await state.update_probe_status(pid, ProbeStatus.NDR_RECEIVED, sent_at=datetime.now(tz=timezone.utc)) - ndr = NdrResult( - probe_id=pid, - received_at=datetime.now(tz=timezone.utc), - ndr_from="mailer-daemon@example.com", - ndr_provider=actual_ndr, - generating_mta="mta.example.com", - confidence=0.8, - ) - await state.insert_ndr(ndr) - - report = await compute_accuracy(state) - - # Overall accuracy includes AWS miss: 4/5 = 0.80 - assert report.overall_accuracy == pytest.approx(0.8, abs=0.01) - # Weighted F1 excludes AWS: microsoft and self-hosted both perfect -> 1.0 - assert report.weighted_f1 == pytest.approx(1.0, abs=0.001) - assert "aws" not in report.weighted_f1_labels - - -# ── Config tests ────────────────────────────────────────────────── - - -class TestConfig: - def test_defaults(self): - from mail_municipalities.accuracy.config import AccuracyConfig - - cfg = AccuracyConfig( - _env_file=None, # type: ignore[call-arg] - ) - assert cfg.smtp_host == "smtp.gmail.com" - assert cfg.dry_run is True - assert cfg.max_probes_per_run == 100 - assert cfg.send_rate_per_second == 1.0 - - def test_sender_address_fallback(self): - from mail_municipalities.accuracy.config import AccuracyConfig - - cfg = AccuracyConfig( - smtp_user="user@gmail.com", - _env_file=None, # type: ignore[call-arg] - ) - assert cfg.sender_address == "user@gmail.com" - - def test_sender_address_explicit(self): - from mail_municipalities.accuracy.config import AccuracyConfig - - cfg = AccuracyConfig( - smtp_user="user@gmail.com", - smtp_from="sender@custom.com", - _env_file=None, # type: ignore[call-arg] - ) - assert cfg.sender_address == "sender@custom.com" - - -# ── Check tests ────────────────────────────────────────────────── - - -class TestCheck: - @pytest.fixture - def providers_dir(self, tmp_path: Path) -> Path: - """Create temp providers JSON files.""" - for cc, entries in [ - ( - "de", - [ - { - "code": "1", - "name": "Berlin", - "domain": "berlin.de", - "provider": "domestic", - "classification_confidence": 80.0, - } - ], - ), - ( - "ch", - [ - { - "code": "2", - "name": "Zürich", - "domain": "zurich.ch", - "provider": "microsoft", - "classification_confidence": 100.0, - } - ], - ), - ( - "at", - [ - { - "code": "3", - "name": "Wien", - "domain": "wien.gv.at", - "provider": "domestic", - "classification_confidence": 80.0, - } - ], - ), - ]: - (tmp_path / f"providers_{cc}.json").write_text(json.dumps({"municipalities": entries})) - return tmp_path - - @pytest.mark.asyncio - async def test_check_creates_probes(self, providers_dir: Path, tmp_path: Path): - from mail_municipalities.accuracy.check import check_domains - - db_path = tmp_path / "state.db" - async with StateDB(db_path) as state: - results = await check_domains(["berlin.de", "zurich.ch"], providers_dir, state) - # Probes should have been created. - existing = await state.get_existing_domains() - - assert len(results) == 2 - assert results[0].name == "Berlin" - assert results[0].provider == "domestic" - assert results[0].status == "new" - assert results[1].provider == "microsoft" - assert "berlin.de" in existing - assert "zurich.ch" in existing - - @pytest.mark.asyncio - async def test_check_domains_not_found(self, providers_dir: Path, tmp_path: Path): - from mail_municipalities.accuracy.check import check_domains - - db_path = tmp_path / "state.db" - async with StateDB(db_path) as state: - results = await check_domains(["nonexistent.de"], providers_dir, state) - - assert len(results) == 1 - assert results[0].name is None - assert results[0].status == "not_found" - - @pytest.mark.asyncio - async def test_check_shows_ndr_result(self, providers_dir: Path, tmp_path: Path): - from mail_municipalities.accuracy.check import check_domains - - db_path = tmp_path / "state.db" - async with StateDB(db_path) as state: - probe = Probe( - probe_id="p1", - domain="berlin.de", - municipality_code="1", - municipality_name="Berlin", - country="de", - recipient="validation-probe-abc@berlin.de", - predicted_provider="domestic", - predicted_confidence=80.0, - ) - await state.insert_probes([probe]) - ndr = NdrResult( - probe_id="p1", - received_at=datetime.now(tz=timezone.utc), - ndr_from="mailer-daemon@berlin.de", - ndr_provider=NdrProvider.POSTFIX, - generating_mta="mail.berlin.de", - confidence=0.9, - evidence=[NdrEvidence(pattern="postfix_body", matched_value="Postfix")], - raw_headers="", - ) - await state.insert_ndr(ndr) - await state.update_probe_status("p1", ProbeStatus.NDR_RECEIVED) - - results = await check_domains(["berlin.de"], providers_dir, state) - - assert len(results) == 1 - assert results[0].status == "ndr_received" - assert results[0].actual == "self-hosted" - assert results[0].match is True diff --git a/uv.lock b/uv.lock index ce79b45..1f1e91a 100644 --- a/uv.lock +++ b/uv.lock @@ -10,15 +10,6 @@ resolution-markers = [ "python_full_version < '3.14' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] -[[package]] -name = "aiosmtplib" -version = "5.1.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/e7/ad/240a7ce4e50713b111dff8b781a898d8d4770e5d6ad4899103f84c86005c/aiosmtplib-5.1.0.tar.gz", hash = "sha256:2504a23b2b63c9de6bc4ea719559a38996dba68f73f6af4eb97be20ee4c5e6c4", size = 66176, upload-time = "2026-01-25T01:51:11.408Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/37/82/70f2c452acd7ed18c558c8ace9a8cf4fdcc70eae9a41749b5bdc53eb6f45/aiosmtplib-5.1.0-py3-none-any.whl", hash = "sha256:368029440645b486b69db7029208a7a78c6691b90d24a5332ddba35d9109d55b", size = 27778, upload-time = "2026-01-25T01:51:10.026Z" }, -] - [[package]] name = "aiosqlite" version = "0.22.1" @@ -476,15 +467,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] -[[package]] -name = "joblib" -version = "1.5.3" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/41/f2/d34e8b3a08a9cc79a50b2208a93dce981fe615b64d5a4d4abee421d898df/joblib-1.5.3.tar.gz", hash = "sha256:8561a3269e6801106863fd0d6d84bb737be9e7631e33aaed3fb9ce5953688da3", size = 331603, upload-time = "2025-12-15T08:41:46.427Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/7b/91/984aca2ec129e2757d1e4e3c81c3fcda9d0f85b74670a094cc443d9ee949/joblib-1.5.3-py3-none-any.whl", hash = "sha256:5fc3c5039fc5ca8c0276333a188bbd59d6b7ab37fe6632daa76bc7f9ec18e713", size = 309071, upload-time = "2025-12-15T08:41:44.973Z" }, -] - [[package]] name = "kiwisolver" version = "1.5.0" @@ -570,7 +552,6 @@ name = "mail-municipalities" version = "0.2.0" source = { editable = "." } dependencies = [ - { name = "aiosmtplib" }, { name = "aiosqlite" }, { name = "dnspython" }, { name = "email-validator" }, @@ -580,9 +561,7 @@ dependencies = [ { name = "pandas" }, { name = "playwright" }, { name = "pydantic" }, - { name = "pydantic-settings" }, { name = "python-slugify" }, - { name = "scikit-learn" }, { name = "scipy" }, { name = "seaborn" }, { name = "stamina" }, @@ -602,7 +581,6 @@ dev = [ [package.metadata] requires-dist = [ - { name = "aiosmtplib", specifier = ">=3.0" }, { name = "aiosqlite", specifier = ">=0.20.0" }, { name = "dnspython", specifier = ">=2.8.0" }, { name = "email-validator", specifier = ">=2.3.0" }, @@ -612,9 +590,7 @@ requires-dist = [ { name = "pandas", specifier = ">=2.2" }, { name = "playwright", specifier = ">=1.50" }, { name = "pydantic", specifier = ">=2.12.5" }, - { name = "pydantic-settings", specifier = ">=2.0" }, { name = "python-slugify", specifier = ">=8.0.4" }, - { name = "scikit-learn", specifier = ">=1.6" }, { name = "scipy", specifier = ">=1.17.1" }, { name = "seaborn", specifier = ">=0.13.2" }, { name = "stamina", specifier = ">=25.2.0" }, @@ -978,20 +954,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9f/ed/068e41660b832bb0b1aa5b58011dea2a3fe0ba7861ff38c4d4904c1c1a99/pydantic_core-2.41.5-cp314-cp314t-win_arm64.whl", hash = "sha256:35b44f37a3199f771c3eaa53051bc8a70cd7b54f333531c59e29fd4db5d15008", size = 1974769, upload-time = "2025-11-04T13:42:01.186Z" }, ] -[[package]] -name = "pydantic-settings" -version = "2.13.1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "pydantic" }, - { name = "python-dotenv" }, - { name = "typing-inspection" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/52/6d/fffca34caecc4a3f97bda81b2098da5e8ab7efc9a66e819074a11955d87e/pydantic_settings-2.13.1.tar.gz", hash = "sha256:b4c11847b15237fb0171e1462bf540e294affb9b86db4d9aa5c01730bdbe4025", size = 223826, upload-time = "2026-02-19T13:45:08.055Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/00/4b/ccc026168948fec4f7555b9164c724cf4125eac006e176541483d2c959be/pydantic_settings-2.13.1-py3-none-any.whl", hash = "sha256:d56fd801823dbeae7f0975e1f8c8e25c258eb75d278ea7abb5d9cebb01b56237", size = 58929, upload-time = "2026-02-19T13:45:06.034Z" }, -] - [[package]] name = "pyee" version = "13.0.1" @@ -1089,15 +1051,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, ] -[[package]] -name = "python-dotenv" -version = "1.2.2" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/82/ed/0301aeeac3e5353ef3d94b6ec08bbcabd04a72018415dcb29e588514bba8/python_dotenv-1.2.2.tar.gz", hash = "sha256:2c371a91fbd7ba082c2c1dc1f8bf89ca22564a087c2c287cd9b662adde799cf3", size = 50135, upload-time = "2026-03-01T16:00:26.196Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/0b/d7/1959b9648791274998a9c3526f6d0ec8fd2233e4d4acce81bbae76b44b2a/python_dotenv-1.2.2-py3-none-any.whl", hash = "sha256:1d8214789a24de455a8b8bd8ae6fe3c6b69a5e3d64aa8a8e5d68e694bbcb285a", size = 22101, upload-time = "2026-03-01T16:00:25.09Z" }, -] - [[package]] name = "python-slugify" version = "8.0.4" @@ -1187,44 +1140,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/15/e2/77be4fff062fa78d9b2a4dea85d14785dac5f1d0c1fb58ed52331f0ebe28/ruff-0.15.8-py3-none-win_arm64.whl", hash = "sha256:cf891fa8e3bb430c0e7fac93851a5978fc99c8fa2c053b57b118972866f8e5f2", size = 11048175, upload-time = "2026-03-26T18:40:01.06Z" }, ] -[[package]] -name = "scikit-learn" -version = "1.8.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "joblib" }, - { name = "numpy" }, - { name = "scipy" }, - { name = "threadpoolctl" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/0e/d4/40988bf3b8e34feec1d0e6a051446b1f66225f8529b9309becaeef62b6c4/scikit_learn-1.8.0.tar.gz", hash = "sha256:9bccbb3b40e3de10351f8f5068e105d0f4083b1a65fa07b6634fbc401a6287fd", size = 7335585, upload-time = "2025-12-10T07:08:53.618Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/03/aa/e22e0768512ce9255eba34775be2e85c2048da73da1193e841707f8f039c/scikit_learn-1.8.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:0d6ae97234d5d7079dc0040990a6f7aeb97cb7fa7e8945f1999a429b23569e0a", size = 8513770, upload-time = "2025-12-10T07:08:03.251Z" }, - { url = "https://files.pythonhosted.org/packages/58/37/31b83b2594105f61a381fc74ca19e8780ee923be2d496fcd8d2e1147bd99/scikit_learn-1.8.0-cp313-cp313-macosx_12_0_arm64.whl", hash = "sha256:edec98c5e7c128328124a029bceb09eda2d526997780fef8d65e9a69eead963e", size = 8044458, upload-time = "2025-12-10T07:08:05.336Z" }, - { url = "https://files.pythonhosted.org/packages/2d/5a/3f1caed8765f33eabb723596666da4ebbf43d11e96550fb18bdec42b467b/scikit_learn-1.8.0-cp313-cp313-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:74b66d8689d52ed04c271e1329f0c61635bcaf5b926db9b12d58914cdc01fe57", size = 8610341, upload-time = "2025-12-10T07:08:07.732Z" }, - { url = "https://files.pythonhosted.org/packages/38/cf/06896db3f71c75902a8e9943b444a56e727418f6b4b4a90c98c934f51ed4/scikit_learn-1.8.0-cp313-cp313-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8fdf95767f989b0cfedb85f7ed8ca215d4be728031f56ff5a519ee1e3276dc2e", size = 8900022, upload-time = "2025-12-10T07:08:09.862Z" }, - { url = "https://files.pythonhosted.org/packages/1c/f9/9b7563caf3ec8873e17a31401858efab6b39a882daf6c1bfa88879c0aa11/scikit_learn-1.8.0-cp313-cp313-win_amd64.whl", hash = "sha256:2de443b9373b3b615aec1bb57f9baa6bb3a9bd093f1269ba95c17d870422b271", size = 7989409, upload-time = "2025-12-10T07:08:12.028Z" }, - { url = "https://files.pythonhosted.org/packages/49/bd/1f4001503650e72c4f6009ac0c4413cb17d2d601cef6f71c0453da2732fc/scikit_learn-1.8.0-cp313-cp313-win_arm64.whl", hash = "sha256:eddde82a035681427cbedded4e6eff5e57fa59216c2e3e90b10b19ab1d0a65c3", size = 7619760, upload-time = "2025-12-10T07:08:13.688Z" }, - { url = "https://files.pythonhosted.org/packages/d2/7d/a630359fc9dcc95496588c8d8e3245cc8fd81980251079bc09c70d41d951/scikit_learn-1.8.0-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:7cc267b6108f0a1499a734167282c00c4ebf61328566b55ef262d48e9849c735", size = 8826045, upload-time = "2025-12-10T07:08:15.215Z" }, - { url = "https://files.pythonhosted.org/packages/cc/56/a0c86f6930cfcd1c7054a2bc417e26960bb88d32444fe7f71d5c2cfae891/scikit_learn-1.8.0-cp313-cp313t-macosx_12_0_arm64.whl", hash = "sha256:fe1c011a640a9f0791146011dfd3c7d9669785f9fed2b2a5f9e207536cf5c2fd", size = 8420324, upload-time = "2025-12-10T07:08:17.561Z" }, - { url = "https://files.pythonhosted.org/packages/46/1e/05962ea1cebc1cf3876667ecb14c283ef755bf409993c5946ade3b77e303/scikit_learn-1.8.0-cp313-cp313t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:72358cce49465d140cc4e7792015bb1f0296a9742d5622c67e31399b75468b9e", size = 8680651, upload-time = "2025-12-10T07:08:19.952Z" }, - { url = "https://files.pythonhosted.org/packages/fe/56/a85473cd75f200c9759e3a5f0bcab2d116c92a8a02ee08ccd73b870f8bb4/scikit_learn-1.8.0-cp313-cp313t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:80832434a6cc114f5219211eec13dcbc16c2bac0e31ef64c6d346cde3cf054cb", size = 8925045, upload-time = "2025-12-10T07:08:22.11Z" }, - { url = "https://files.pythonhosted.org/packages/cc/b7/64d8cfa896c64435ae57f4917a548d7ac7a44762ff9802f75a79b77cb633/scikit_learn-1.8.0-cp313-cp313t-win_amd64.whl", hash = "sha256:ee787491dbfe082d9c3013f01f5991658b0f38aa8177e4cd4bf434c58f551702", size = 8507994, upload-time = "2025-12-10T07:08:23.943Z" }, - { url = "https://files.pythonhosted.org/packages/5e/37/e192ea709551799379958b4c4771ec507347027bb7c942662c7fbeba31cb/scikit_learn-1.8.0-cp313-cp313t-win_arm64.whl", hash = "sha256:bf97c10a3f5a7543f9b88cbf488d33d175e9146115a451ae34568597ba33dcde", size = 7869518, upload-time = "2025-12-10T07:08:25.71Z" }, - { url = "https://files.pythonhosted.org/packages/24/05/1af2c186174cc92dcab2233f327336058c077d38f6fe2aceb08e6ab4d509/scikit_learn-1.8.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:c22a2da7a198c28dd1a6e1136f19c830beab7fdca5b3e5c8bba8394f8a5c45b3", size = 8528667, upload-time = "2025-12-10T07:08:27.541Z" }, - { url = "https://files.pythonhosted.org/packages/a8/25/01c0af38fe969473fb292bba9dc2b8f9b451f3112ff242c647fee3d0dfe7/scikit_learn-1.8.0-cp314-cp314-macosx_12_0_arm64.whl", hash = "sha256:6b595b07a03069a2b1740dc08c2299993850ea81cce4fe19b2421e0c970de6b7", size = 8066524, upload-time = "2025-12-10T07:08:29.822Z" }, - { url = "https://files.pythonhosted.org/packages/be/ce/a0623350aa0b68647333940ee46fe45086c6060ec604874e38e9ab7d8e6c/scikit_learn-1.8.0-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:29ffc74089f3d5e87dfca4c2c8450f88bdc61b0fc6ed5d267f3988f19a1309f6", size = 8657133, upload-time = "2025-12-10T07:08:31.865Z" }, - { url = "https://files.pythonhosted.org/packages/b8/cb/861b41341d6f1245e6ca80b1c1a8c4dfce43255b03df034429089ca2a2c5/scikit_learn-1.8.0-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fb65db5d7531bccf3a4f6bec3462223bea71384e2cda41da0f10b7c292b9e7c4", size = 8923223, upload-time = "2025-12-10T07:08:34.166Z" }, - { url = "https://files.pythonhosted.org/packages/76/18/a8def8f91b18cd1ba6e05dbe02540168cb24d47e8dcf69e8d00b7da42a08/scikit_learn-1.8.0-cp314-cp314-win_amd64.whl", hash = "sha256:56079a99c20d230e873ea40753102102734c5953366972a71d5cb39a32bc40c6", size = 8096518, upload-time = "2025-12-10T07:08:36.339Z" }, - { url = "https://files.pythonhosted.org/packages/d1/77/482076a678458307f0deb44e29891d6022617b2a64c840c725495bee343f/scikit_learn-1.8.0-cp314-cp314-win_arm64.whl", hash = "sha256:3bad7565bc9cf37ce19a7c0d107742b320c1285df7aab1a6e2d28780df167242", size = 7754546, upload-time = "2025-12-10T07:08:38.128Z" }, - { url = "https://files.pythonhosted.org/packages/2d/d1/ef294ca754826daa043b2a104e59960abfab4cf653891037d19dd5b6f3cf/scikit_learn-1.8.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:4511be56637e46c25721e83d1a9cea9614e7badc7040c4d573d75fbe257d6fd7", size = 8848305, upload-time = "2025-12-10T07:08:41.013Z" }, - { url = "https://files.pythonhosted.org/packages/5b/e2/b1f8b05138ee813b8e1a4149f2f0d289547e60851fd1bb268886915adbda/scikit_learn-1.8.0-cp314-cp314t-macosx_12_0_arm64.whl", hash = "sha256:a69525355a641bf8ef136a7fa447672fb54fe8d60cab5538d9eb7c6438543fb9", size = 8432257, upload-time = "2025-12-10T07:08:42.873Z" }, - { url = "https://files.pythonhosted.org/packages/26/11/c32b2138a85dcb0c99f6afd13a70a951bfdff8a6ab42d8160522542fb647/scikit_learn-1.8.0-cp314-cp314t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c2656924ec73e5939c76ac4c8b026fc203b83d8900362eb2599d8aee80e4880f", size = 8678673, upload-time = "2025-12-10T07:08:45.362Z" }, - { url = "https://files.pythonhosted.org/packages/c7/57/51f2384575bdec454f4fe4e7a919d696c9ebce914590abf3e52d47607ab8/scikit_learn-1.8.0-cp314-cp314t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:15fc3b5d19cc2be65404786857f2e13c70c83dd4782676dd6814e3b89dc8f5b9", size = 8922467, upload-time = "2025-12-10T07:08:47.408Z" }, - { url = "https://files.pythonhosted.org/packages/35/4d/748c9e2872637a57981a04adc038dacaa16ba8ca887b23e34953f0b3f742/scikit_learn-1.8.0-cp314-cp314t-win_amd64.whl", hash = "sha256:00d6f1d66fbcf4eba6e356e1420d33cc06c70a45bb1363cd6f6a8e4ebbbdece2", size = 8774395, upload-time = "2025-12-10T07:08:49.337Z" }, - { url = "https://files.pythonhosted.org/packages/60/22/d7b2ebe4704a5e50790ba089d5c2ae308ab6bb852719e6c3bd4f04c3a363/scikit_learn-1.8.0-cp314-cp314t-win_arm64.whl", hash = "sha256:f28dd15c6bb0b66ba09728cf09fd8736c304be29409bd8445a080c1280619e8c", size = 8002647, upload-time = "2025-12-10T07:08:51.601Z" }, -] - [[package]] name = "scipy" version = "1.17.1" @@ -1338,15 +1253,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a6/a5/c0b6468d3824fe3fde30dbb5e1f687b291608f9473681bbf7dabbf5a87d7/text_unidecode-1.3-py2.py3-none-any.whl", hash = "sha256:1311f10e8b895935241623731c2ba64f4c455287888b18189350b67134a822e8", size = 78154, upload-time = "2019-08-30T21:37:03.543Z" }, ] -[[package]] -name = "threadpoolctl" -version = "3.6.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/b7/4d/08c89e34946fce2aec4fbb45c9016efd5f4d7f24af8e5d93296e935631d8/threadpoolctl-3.6.0.tar.gz", hash = "sha256:8ab8b4aa3491d812b623328249fab5302a68d2d71745c8a4c719a2fcaba9f44e", size = 21274, upload-time = "2025-03-13T13:49:23.031Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/32/d5/f9a850d79b0851d1d4ef6456097579a9005b31fea68726a4ae5f2d82ddd9/threadpoolctl-3.6.0-py3-none-any.whl", hash = "sha256:43a0b8fd5a2928500110039e43a5eed8480b918967083ea48dc3ab9f13c4a7fb", size = 18638, upload-time = "2025-03-13T13:49:21.846Z" }, -] - [[package]] name = "tldextract" version = "5.3.1"