# SPDX-License-Identifier: Apache-2.0 # Copyright 2026 Aaron K. Clark """User-editable discovery criteria for the poller. The .NET API persists this data in ``search_profiles``; the poller reads it through :class:`applytrack.db.PollRepo`. """ from __future__ import annotations import re from dataclasses import asdict, dataclass, field from urllib.parse import urlsplit LANES = ("dotnet", "devrel", "ai") # Source ids the poller knows how to fetch without per-source configuration. BUILTIN_SOURCES = ( "remotive", "remoteok", "arbeitnow", "jobicy", "weworkremotely", "remotefirstjobs", "workanywhere", "hn_whoishiring", # Greenhouse's candidate portal, signed in with the tenant's own account (#218): # per tenant, never shared, off by default. "mygreenhouse", # LinkedIn, signed in with the tenant's own account (#233): the employer's posting # behind each offsite-apply listing, Easy Apply skipped. Per tenant, off by default. "linkedin", # Handshake, signed in with the tenant's own account (#267): campus recruiting, the # employer's own posting behind each externally-applied listing. Per tenant, off by # default. "handshake", ) # ATS providers the board adder understands (public boards, no auth). ATS_PROVIDERS = ("greenhouse", "lever", "paylocity") # A Paylocity board is keyed by the company's recruiting GUID rather than a name # slug, and users paste the whole board URL. Mirrors AtsBoard.PaylocityGuid on the # .NET side, which normalizes the same way on the write path. _PAYLOCITY_GUID_RE = re.compile(r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}") # A Greenhouse / Lever board token. The fetchers paste it into a URL path on the # provider's host, so anything else (``/``, ``..``, ``?``, ``#``) could rewrite that # path or query (#346). Mirrors AtsBoard.IsBoardSlug on the .NET side. _BOARD_SLUG_RE = re.compile(r"[A-Za-z0-9_-]{1,100}") def is_board_slug(slug: str) -> bool: """True when ``slug`` is safe to use as a Greenhouse / Lever board token.""" return _BOARD_SLUG_RE.fullmatch(slug) is not None # The original per-lane keyword lists, flattened (order-preserving, de-duped) into # one flat match list — the default the UI's keyword box starts from. DEFAULT_KEYWORDS: tuple[str, ...] = ( ".net", "dotnet", "c#", "csharp", "asp.net", "blazor", "entity framework", "ef core", "f#", "backend engineer", "back-end engineer", "backend developer", "web api", "microservices", "developer advocate", "developer relations", "devrel", "developer experience", "technical writer", "technical writing", "documentation engineer", "community manager", "developer evangelist", "evangelist", "dx engineer", "content engineer", "developer educator", "ai engineer", "agentic", "llm", "large language model", "machine learning", "ml engineer", "applied ai", "prompt engineer", "rag", "langchain", "generative ai", "genai", "ai/ml", "ai agent", "mlops", ) # Only Remotive + RemoteOK are on by default, matching the original poller. New # sources are opt-in (the user enables them in the Criteria panel). DEFAULT_SOURCES: dict[str, bool] = {s: s in ("remotive", "remoteok") for s in BUILTIN_SOURCES} MIN_SCORE_FLOOR = 0 MIN_SCORE_CEIL = 100 # How many custom feeds one tenant may follow; mirrors InputLimits.RssFeeds on the # .NET side, which is what actually rejects an over-long list at the API edge. MAX_RSS_FEEDS = 25 @dataclass class AtsBoard: """A company's public ATS board to scan (provider + company slug).""" provider: str slug: str @classmethod def from_dict(cls, data: dict[str, object]) -> AtsBoard | None: provider = str(data.get("provider", "")).strip().lower() slug = str(data.get("slug", "")).strip() if provider == "paylocity": # Users paste the board URL; keep only the GUID it contains. found = _PAYLOCITY_GUID_RE.search(slug) slug = found.group(0).lower() if found else "" elif not is_board_slug(slug): return None if provider not in ATS_PROVIDERS or not slug: return None return cls(provider=provider, slug=slug) def _clean_list(values: object) -> list[str]: """Coerce arbitrary JSON into a de-duped, stripped, order-preserving str list.""" if not isinstance(values, (list, tuple)): return [] out: list[str] = [] seen: set[str] = set() for v in values: s = str(v).strip() key = s.lower() if s and key not in seen: seen.add(key) out.append(s) return out def _clean_feeds(values: object) -> list[str]: """Keep only absolute http(s) feed URLs, de-duped case-insensitively. The URLs are user-supplied and fetched server-side, so anything that isn't a plain http(s) URL on a default port is dropped here rather than handed to the fetcher; :func:`applytrack.linkcheck.fetch_public` is the second gate (it refuses private/link-local addresses at connect time). The .NET ``Criteria.CleanFeeds`` applies the same rules on the write path. """ out: list[str] = [] for raw in _clean_list(values): # already stripped and de-duped parts = urlsplit(raw) if parts.scheme not in ("http", "https") or not parts.hostname: continue try: if parts.port is not None: continue # explicit port: not a default-port http(s) URL except ValueError: continue # unparseable port out.append(raw) if len(out) >= MAX_RSS_FEEDS: break return out @dataclass class Criteria: """Everything the poller needs to decide what to fetch and what to stage.""" keywords: list[str] = field(default_factory=lambda: list(DEFAULT_KEYWORDS)) default_lane: str = "ai" min_fit_score: int = 55 remote_only: bool = False exclude_locations: list[str] = field(default_factory=list) sources: dict[str, bool] = field(default_factory=lambda: dict(DEFAULT_SOURCES)) ats_boards: list[AtsBoard] = field(default_factory=list) rss_feeds: list[str] = field(default_factory=list) # -- (de)serialization -------------------------------------------------- @classmethod def from_dict(cls, data: dict[str, object]) -> Criteria: """Build a normalized Criteria from loose JSON, ignoring junk keys.""" lane = str(data.get("default_lane", "ai")).strip().lower() raw_score = data.get("min_fit_score", 55) try: score = int(raw_score) if isinstance(raw_score, (int, float, str)) else 55 except (TypeError, ValueError): score = 55 raw_sources = data.get("sources", {}) sources = dict(DEFAULT_SOURCES) if isinstance(raw_sources, dict): for key, on in raw_sources.items(): if key in BUILTIN_SOURCES: sources[key] = bool(on) raw_boards = data.get("ats_boards", []) boards: list[AtsBoard] = [] if isinstance(raw_boards, (list, tuple)): seen: set[tuple[str, str]] = set() for entry in raw_boards: if not isinstance(entry, dict): continue board = AtsBoard.from_dict(entry) if board and (board.provider, board.slug.lower()) not in seen: seen.add((board.provider, board.slug.lower())) boards.append(board) keywords = _clean_list(data.get("keywords")) return cls( keywords=keywords if keywords else list(DEFAULT_KEYWORDS), default_lane=lane if lane in LANES else "ai", min_fit_score=max(MIN_SCORE_FLOOR, min(MIN_SCORE_CEIL, score)), remote_only=bool(data.get("remote_only", False)), exclude_locations=_clean_list(data.get("exclude_locations")), sources=sources, ats_boards=boards, rss_feeds=_clean_feeds(data.get("rss_feeds")), ) def to_dict(self) -> dict[str, object]: return { "keywords": list(self.keywords), "default_lane": self.default_lane, "min_fit_score": self.min_fit_score, "remote_only": self.remote_only, "exclude_locations": list(self.exclude_locations), "sources": dict(self.sources), "ats_boards": [asdict(b) for b in self.ats_boards], "rss_feeds": list(self.rss_feeds), } # Defaults used when a tenant has no persisted profile row. DEFAULT_CRITERIA = Criteria()