Import audited employer career portals safely

src/fortune_intel/cli.py

@@ -33,7 +33,10 @@
     SecSubmissionsWebsiteClient,
     import_sec_company_websites,
 )
-from fortune_intel.importers.source_candidates import import_reviewed_source_candidates
+from fortune_intel.importers.source_candidates import (
+    import_reviewed_source_candidates,
+    import_verified_employer_portals,
+)
 from fortune_intel.importers.sources import import_source_registry
 from fortune_intel.importers.website_leads import import_website_leads
 from fortune_intel.importers.websites import import_company_websites
@@ -155,6 +158,13 @@ def parser() -> argparse.ArgumentParser:
     )
     source_candidates.add_argument("csv")
 
+    verified_portals = commands.add_parser(
+        "import-verified-employer-portals",
+        help="Import audited current official employer portals as reviewable ATS candidates",
+    )
+    verified_portals.add_argument("csv")
+    verified_portals.add_argument("--actor", required=True)
+
     jobseek_registry = commands.add_parser(
         "import-jobseek-board-registry",
         help="Import attributed Jobseek board configuration as unverified passive leads",
@@ -417,6 +427,13 @@ def main() -> None:
             f"Imported {import_reviewed_source_candidates(repository, args.csv)} "
             "reviewed ATS candidates"
         )
+    elif args.command == "import-verified-employer-portals":
+        print(
+            json.dumps(
+                import_verified_employer_portals(repository, args.csv, actor=args.actor),
+                indent=2,
+            )
+        )
     elif args.command == "import-jobseek-board-registry":
         print(
             json.dumps(

src/fortune_intel/importers/source_candidates.py

@@ -4,7 +4,7 @@
 
 import csv
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from urllib.parse import urlsplit
 
@@ -133,3 +133,100 @@ def import_reviewed_source_candidates(repository: JobRepository, csv_path: str |
                 reviewed_at=seed.verified_at,
             )
     return len(resolved)
+
+
+def import_verified_employer_portals(
+    repository: JobRepository,
+    csv_path: str | Path,
+    *,
+    actor: str,
+) -> dict[str, int]:
+    """Import an audited employer-portal report without inventing a connector.
+
+    The report must identify a current official employer portal.  Only URLs
+    that match one of the platform's deterministic public connectors become
+    reviewable candidates; every other portal is deliberately skipped here.
+    """
+
+    if not actor.strip():
+        raise ValueError("actor is required")
+    report = {"rows_read": 0, "imported": 0, "existing": 0, "skipped": 0, "unmatched": 0}
+    with Path(csv_path).open(newline="", encoding="utf-8-sig") as handle:
+        rows = list(csv.DictReader(handle))
+    companies = {int(item["id"]): item for item in repository.list_companies()}
+    with repository.connect() as connection:
+        existing_candidates = {
+            (int(row["company_id"]), str(row["kind"]), str(row["base_url"]))
+            for row in connection.execute("SELECT company_id, kind, base_url FROM career_sources")
+        }
+        existing_candidates.update(
+            (int(row["company_id"]), str(row["kind"]), str(row["candidate_url"]))
+            for row in connection.execute(
+                "SELECT company_id, kind, candidate_url FROM career_source_candidates"
+            )
+        )
+    coverage_updates: dict[int, str] = {}
+    for row_number, row in enumerate(rows, start=2):
+        report["rows_read"] += 1
+        status = _value(row, "verification_status").casefold()
+        if status != "verified_current_official_url":
+            report["skipped"] += 1
+            continue
+        try:
+            company_id = int(_value(row, "company_id"))
+        except ValueError:
+            report["unmatched"] += 1
+            continue
+        company = companies.get(company_id)
+        if company is None or str(company["name"]) != _value(row, "company_name"):
+            report["unmatched"] += 1
+            continue
+        candidate_url = _value(row, "verified_job_url")
+        classified = classify_ats_url(candidate_url, origin="audited employer portal report")
+        if classified is None:
+            report["skipped"] += 1
+            continue
+        identity = (company_id, classified.connector_kind, classified.normalized_base_url)
+        if identity in existing_candidates:
+            report["existing"] += 1
+            continue
+        source_url = normalize_public_url(
+            _value(row, "verified_company_website"),
+            field=f"row {row_number} verified_company_website",
+        )
+        verified_date = _value(row, "verification_date")
+        try:
+            verified_at = datetime.fromisoformat(verified_date).replace(tzinfo=UTC).isoformat()
+        except ValueError as error:
+            raise ValueError(f"row {row_number}: invalid verification_date") from error
+        repository.upsert_source_candidate(
+            company_id,
+            candidate_url=classified.normalized_base_url,
+            kind=classified.connector_kind,
+            confidence=0.98,
+            evidence={
+                "review_method": "audited_verified_employer_portal",
+                "source_url": source_url,
+                "board_token": classified.board_token,
+                "classifier_evidence": list(classified.evidence),
+                "portal_scope": _value(row, "portal_scope"),
+                "verification_note": _value(row, "verification_note"),
+            },
+            robots_status="unknown",
+            terms_status="review_required",
+            discovered_at=verified_at,
+        )
+        coverage_updates[company_id] = verified_at
+        report["imported"] += 1
+        existing_candidates.add(identity)
+    if coverage_updates:
+        with repository.connect() as connection:
+            for company_id, reviewed_at in coverage_updates.items():
+                connection.execute(
+                    """UPDATE company_coverage SET disposition = 'candidate',
+                        reason = 'Audited current official employer portal uses a supported public ATS',
+                        last_reviewed_at = ?, reviewed_by = ?, updated_at = ?
+                    WHERE company_id = ? AND disposition NOT IN ('supported', 'stale')""",
+                    (reviewed_at, actor.strip(), reviewed_at, company_id),
+                )
+    return report

tests/test_source_candidate_importer.py

@@ -2,7 +2,10 @@
 
 import pytest
 
-from fortune_intel.importers.source_candidates import import_reviewed_source_candidates
+from fortune_intel.importers.source_candidates import (
+    import_reviewed_source_candidates,
+    import_verified_employer_portals,
+)
 from fortune_intel.storage import JobRepository
 
 
@@ -137,17 +140,75 @@ def test_imports_explicit_primary_source_structured_manifest(tmp_path):
     registry = tmp_path / "structured.csv"
     _write(
         registry,
-        [{
-            "company_name": "Example",
-            "candidate_url": "https://careers.example.com/job-sitemap.xml",
-            "kind": "official_structured",
-            "source_url": "https://example.com/careers",
-            "verified_at": "2026-08-13T12:00:00+00:00",
-            "actor": "reviewer@example.org",
-        }],
+        [
+            {
+                "company_name": "Example",
+                "candidate_url": "https://careers.example.com/job-sitemap.xml",
+                "kind": "official_structured",
+                "source_url": "https://example.com/careers",
+                "verified_at": "2026-08-13T12:00:00+00:00",
+                "actor": "reviewer@example.org",
+            }
+        ],
     )
 
     assert import_reviewed_source_candidates(repository, registry) == 1
     candidate = repository.get_source_candidate(1)
     assert candidate["kind"] == "official_structured"
     assert candidate["evidence"]["source_url"] == "https://example.com/careers"
+
+
+def test_imports_only_supported_urls_from_an_audited_employer_portal_report(tmp_path):
+    repository = JobRepository(tmp_path / "verified-portals.db")
+    repository.initialize()
+    company_id = repository.upsert_company("Example")
+    registry = tmp_path / "verified-portals.csv"
+    fields = (
+        "company_id",
+        "company_name",
+        "verified_company_website",
+        "verified_job_url",
+        "verification_status",
+        "portal_scope",
+        "verification_date",
+        "verification_note",
+    )
+    rows = [
+        {
+            "company_id": company_id,
+            "company_name": "Example",
+            "verified_company_website": "https://example.com/",
+            "verified_job_url": "https://job-boards.greenhouse.io/example",
+            "verification_status": "verified_current_official_url",
+            "portal_scope": "official_job_or_careers_portal",
+            "verification_date": "2026-08-14",
+            "verification_note": "Reviewed from the employer website.",
+        },
+        {
+            "company_id": company_id,
+            "company_name": "Example",
+            "verified_company_website": "https://example.com/",
+            "verified_job_url": "https://example.com/careers",
+            "verification_status": "verified_current_official_url",
+            "portal_scope": "official_job_or_careers_portal",
+            "verification_date": "2026-08-14",
+            "verification_note": "Custom portal retained outside the candidate importer.",
+        },
+    ]
+    with registry.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fields)
+        writer.writeheader()
+        writer.writerows(rows)
+
+    report = import_verified_employer_portals(repository, registry, actor="portal-auditor")
+
+    assert report == {
+        "rows_read": 2,
+        "imported": 1,
+        "existing": 0,
+        "skipped": 1,
+        "unmatched": 0,
+    }
+    candidate = repository.get_source_candidate(1)
+    assert candidate["kind"] == "greenhouse"
+    assert candidate["evidence"]["review_method"] == "audited_verified_employer_portal"