diff --git a/src/fortune_intel/cli.py b/src/fortune_intel/cli.py index 7a48943..ad70e40 100644 --- a/src/fortune_intel/cli.py +++ b/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( diff --git a/src/fortune_intel/importers/source_candidates.py b/src/fortune_intel/importers/source_candidates.py index 9a48ce8..0dc502d 100644 --- a/src/fortune_intel/importers/source_candidates.py +++ b/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 diff --git a/tests/test_source_candidate_importer.py b/tests/test_source_candidate_importer.py index 01ae2c3..a45780a 100644 --- a/tests/test_source_candidate_importer.py +++ b/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"