hermes/jobhunt-platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(W1): packages/connectors - AF API, generic URL, dedupe, normalizer

Browse source 
- Connector protocol with fetch(query: SearchQuery) -> list[RawPosting]
- ArbetsformedlingenConnector: official Platsbanken API, field mapping
  (headline->title, employer.name->company, webpage_url->url,
  description.text->description, id->external_id), region mapping,
  polite User-Agent
- GenericUrlConnector: readability extraction (prefers main/article,
  strips nav/footer/script/style), title/company guess, Cloudflare
  challenge detection -> UnsupportedSite
- dedupe helper: (source, url) + external_id keying
- normalizer: RawPosting -> JobPosting (job_posting table shape)
- 37 tests, all passing, recorded fixtures (no live network)
- README with usage and mock examples
This commit is contained in:
hermes 2026-07-30 18:23:24 +00:00
parent d1753bb70a
commit 0e13ee5d51
18 changed files with 1515 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

135
packages/connectors/README.md Normal file
View file

@ -0,0 +1,135 @@
# connectors
Job source adapters that fetch postings from external sources and normalize
them to a common `JobPosting` shape matching the `job_posting` database table.
## Connectors
### Arbetsformedlingen (Platsbanken)
Uses the official Swedish Public Employment Service API. Free, no API key
required.
- Endpoint: `https://jobsearch.api.jobtechdev.se/search`
- Method: GET, header `Accept: application/json`
- Field mapping: `headline -> title`, `employer.name -> company`,
`webpage_url -> url`, `description.text -> description`, `id -> external_id`
- Region mapping: common names (e.g. "malmo", "skane") mapped to API filter values
- Polite User-Agent header
### Generic URL
Fetches a single job posting URL and extracts clean text with simple
readability extraction:
- Prefers `<main>` or `<article>` containers
- Strips `nav`, `footer`, `script`, `style`, `aside`, `header`, `noscript` tags
- Extracts title from `<h1>` (falls back to `<title>`)
- Attempts company name from `og:site_name` meta tag, then text heuristics
- **Does NOT follow Cloudflare challenge pages** -- raises `UnsupportedSite`
## Usage
### Arbetsformedlingen
```python
from connectors import ArbetsformedlingenConnector, SearchQuery
connector = ArbetsformedlingenConnector()
query = SearchQuery(query="python developer", region="malmo", limit=20)
raw_postings = connector.fetch(query)
# Normalize to JobPosting shape
from connectors import normalize
job_postings = [normalize(p) for p in raw_postings]
```
### Generic URL
```python
from connectors import GenericUrlConnector, SearchQuery
connector = GenericUrlConnector()
query = SearchQuery(query="https://example.com/jobs/123")
raw_postings = connector.fetch(query)
# Returns a single-element list
```
### Deduplication
```python
from connectors import dedupe
# Remove duplicates by (source, url) and (source, external_id)
unique_postings = dedupe(all_raw_postings)
```
### Normalization
```python
from connectors import normalize, RawPosting
raw = RawPosting(
source="manual",
external_id=None,
url="https://example.com/job",
company="Corp",
title="Developer",
description="A great job.",
)
job = normalize(raw)
# job.source, job.url, job.company, job.title, job.description, ...
```
## Error handling
```python
from connectors import UnsupportedSite
try:
connector.fetch(SearchQuery(query="https://cloudflare-protected.com/job/1"))
except UnsupportedSite as e:
print(f"Cannot scrape {e.url}: {e.reason}")
```
## Data models
### SearchQuery
| Field | Type | Description |
|----------|----------------|--------------------------------------|
| query | `str` | Search keywords or URL |
| region | `str \| None` | Optional region filter |
| limit | `int` | Max results (default 20) |
### RawPosting
| Field | Type | Description |
|--------------|------------------|--------------------------------------|
| source | `str` | Source identifier |
| external_id | `str \| None` | Source-native ID |
| url | `str` | Posting URL |
| company | `str` | Company name |
| title | `str` | Job title |
| location | `str \| None` | Job location |
| description | `str` | Job description text |
| raw | `dict` | Original source payload |
### JobPosting
Same fields as `RawPosting`. Matches the `job_posting` table shape:
`source`, `external_id`, `url`, `company`, `title`, `location`,
`description`, `raw`.
## Development
```bash
cd packages/connectors
uv venv
. .venv/bin/activate
uv pip install -e ".[dev]"
pytest -q
```
Tests use recorded JSON/HTML fixtures (no live network calls). Fixtures live
in `tests/fixtures/`.

28
packages/connectors/pyproject.toml Normal file
View file

@ -0,0 +1,28 @@
[project]
name = "connectors"
version = "0.1.0"
description = "Job source connectors: Arbetsformedlingen API, generic URL extractor, dedupe, normalizer."
requires-python = ">=3.13"
dependencies = [
"httpx>=0.27",
"beautifulsoup4>=4.12",
"lxml>=5.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.24",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/connectors"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
asyncio_mode = "auto"

36
packages/connectors/src/connectors/__init__.py Normal file
View file

@ -0,0 +1,36 @@
"""Connectors package.
Job source adapters that fetch postings from external sources and normalize
them to a common JobPosting shape.
Public API:
Connector: protocol every adapter implements.
SearchQuery: query parameters for connector.fetch.
RawPosting: raw posting returned by a connector before normalization.
JobPosting: normalized posting matching the job_posting table shape.
ArbetsformedlingenConnector: Swedish Platsbanken API adapter.
GenericUrlConnector: fetch a single posting URL with readability extraction.
UnsupportedSite: raised when a site cannot be scraped (Cloudflare, etc.).
dedupe: remove duplicate postings by (source, url) + external_id.
normalize: convert a RawPosting into a JobPosting.
"""
from connectors.models import JobPosting, RawPosting, SearchQuery
from connectors.protocol import Connector
from connectors.exceptions import UnsupportedSite
from connectors.arbetsformedlingen import ArbetsformedlingenConnector
from connectors.generic_url import GenericUrlConnector
from connectors.dedupe import dedupe
from connectors.normalizer import normalize
__all__ = [
"Connector",
"SearchQuery",
"RawPosting",
"JobPosting",
"ArbetsformedlingenConnector",
"GenericUrlConnector",
"UnsupportedSite",
"dedupe",
"normalize",
]

123
packages/connectors/src/connectors/arbetsformedlingen.py Normal file
View file

@ -0,0 +1,123 @@
"""Arbetsformedlingen (Platsbanken) job search connector.
Uses the official Swedish Public Employment Service API:
https://jobsearch.api.jobtechdev.se/search
Free, no API key required. GET requests with Accept: application/json header.
"""
from __future__ import annotations
import httpx
from connectors.models import RawPosting, SearchQuery
API_URL = "https://jobsearch.api.jobtechdev.se/search"
USER_AGENT = "jobhunt-platform/0.1 (contact: dev@jobhunt.local)"
HEADERS = {
"Accept": "application/json",
"User-Agent": USER_AGENT,
}
# Region name -> API filter value mapping for Swedish lan.
# The AF API accepts region as a free-text filter that matches against
# the region field. We map common user-facing names to the API's format.
REGION_MAP: dict[str, str] = {
"skane": "Skane lan",
"skane lan": "Skane lan",
"malmo": "Skane lan",
"stockholm": "Stockholms lan",
"stockholms lan": "Stockholms lan",
"goteborg": "Vastra Gotalands lan",
"vastra gotaland": "Vastra Gotalands lan",
"vastra gotalands lan": "Vastra Gotalands lan",
}
class ArbetsformedlingenConnector:
"""Connector for the Arbetsformedlingen Platsbanken API.
The fetch method issues a GET request to the official API and maps
the response to RawPosting objects. Field mapping:
- headline -> title
-employer.name -> company
- webpage_url -> url
- description.text -> description
- id -> external_id
- workplace_address.city -> location (when available)
"""
def __init__(
self,
client: httpx.Client | None = None,
base_url: str = API_URL,
) -> None:
"""Initialize the connector.
Args:
client: Optional pre-configured httpx.Client (e.g. for testing
with a fixture-loaded transport). If None, a new client
is created on each fetch call.
base_url: API endpoint URL (override for testing).
"""
self._client = client
self._base_url = base_url
def fetch(self, query: SearchQuery) -> list[RawPosting]:
"""Fetch postings from the Arbetsformedlingen API.
Args:
query: Search parameters. ``query.query`` maps to the ``q``
parameter, ``query.region`` is mapped via REGION_MAP to
a region filter, ``query.limit`` maps to ``limit``.
Returns:
List of RawPosting objects with source='arbetsformedlingen'.
"""
params: dict[str, str | int] = {"q": query.query, "limit": query.limit}
if query.region:
mapped = REGION_MAP.get(query.region.lower().strip(), query.region)
params["region"] = mapped
if self._client is not None:
return self._do_fetch(self._client, params)
with httpx.Client(headers=HEADERS, timeout=30.0) as client:
return self._do_fetch(client, params)
def _do_fetch(
self,
client: httpx.Client,
params: dict[str, str | int],
) -> list[RawPosting]:
response = client.get(self._base_url, params=params)
response.raise_for_status()
data = response.json()
hits = data.get("hits", [])
return [self._map_hit(hit) for hit in hits]
@staticmethod
def _map_hit(hit: dict) -> RawPosting:
"""Map a single API hit to a RawPosting."""
description = ""
desc_obj = hit.get("description", {})
if isinstance(desc_obj, dict):
description = desc_obj.get("text", "") or ""
elif isinstance(desc_obj, str):
description = desc_obj
location = None
workplace = hit.get("workplace_address", {})
if isinstance(workplace, dict):
location = workplace.get("city") or workplace.get("municipality")
return RawPosting(
source="arbetsformedlingen",
external_id=str(hit.get("id", "")) or None,
url=hit.get("webpage_url", "") or "",
company=hit.get("employer", {}).get("name", "") or "",
title=hit.get("headline", "") or "",
location=location,
description=description,
raw=hit,
)

62
packages/connectors/src/connectors/dedupe.py Normal file
View file

@ -0,0 +1,62 @@
"""Deduplication helper for job postings.
Dedupes by (source, url) first, then by external_id when available.
Postings encountered first are kept; later duplicates are dropped.
"""
from __future__ import annotations
from connectors.models import JobPosting, RawPosting
def dedupe(postings: list[RawPosting]) -> list[RawPosting]:
"""Remove duplicate RawPosting entries.
Deduplication keys (in priority order):
1. (source, url) -- always checked.
2. (source, external_id) -- checked when external_id is not None.
Args:
postings: List of RawPosting objects, potentially with duplicates.
Returns:
Deduplicated list preserving first-occurrence order.
"""
seen_urls: set[tuple[str, str]] = set()
seen_ext_ids: set[tuple[str, str]] = set()
result: list[RawPosting] = []
for p in postings:
url_key = (p.source, p.url)
if url_key in seen_urls:
continue
if p.external_id is not None:
ext_key = (p.source, p.external_id)
if ext_key in seen_ext_ids:
continue
seen_ext_ids.add(ext_key)
seen_urls.add(url_key)
result.append(p)
return result
def dedupe_job_postings(postings: list[JobPosting]) -> list[JobPosting]:
"""Remove duplicate JobPosting entries (same logic as dedupe)."""
seen_urls: set[tuple[str, str]] = set()
seen_ext_ids: set[tuple[str, str]] = set()
result: list[JobPosting] = []
for p in postings:
url_key = (p.source, p.url)
if url_key in seen_urls:
continue
if p.external_id is not None:
ext_key = (p.source, p.external_id)
if ext_key in seen_ext_ids:
continue
seen_ext_ids.add(ext_key)
seen_urls.add(url_key)
result.append(p)
return result

18
packages/connectors/src/connectors/exceptions.py Normal file
View file

@ -0,0 +1,18 @@
"""Connector exceptions."""
from __future__ import annotations
class UnsupportedSite(Exception):
"""Raised when a site cannot be scraped.
Common causes:
- Cloudflare challenge / interstitial page detected.
- Page returns no parseable content.
- Site explicitly blocks automated access.
"""
def __init__(self, url: str, reason: str = "") -> None:
self.url = url
self.reason = reason
super().__init__(f"Unsupported site {url}: {reason}" if reason else f"Unsupported site {url}")

207
packages/connectors/src/connectors/generic_url.py Normal file
View file

@ -0,0 +1,207 @@
"""Generic URL connector: fetch a single posting URL and extract text.
Implements simple readability extraction:
- Prefers content inside <main> or <article> tags.
- Falls back to <body> if neither is present.
- Removes nav, footer, script, style, aside, header, noscript tags.
- Extracts <title> or first <h1> as the posting title.
- Attempts company name guess from structured data or meta tags.
- Detects Cloudflare challenge pages and raises UnsupportedSite.
"""
from __future__ import annotations
import re
import httpx
from bs4 import BeautifulSoup
from connectors.exceptions import UnsupportedSite
from connectors.models import RawPosting, SearchQuery
USER_AGENT = (
"Mozilla/5.0 (compatible; jobhunt-platform/0.1; "
"+https://github.com/jobhunt-platform)"
)
HEADERS = {
"User-Agent": USER_AGENT,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en,sv;q=0.9",
}
# Patterns that indicate a Cloudflare challenge / interstitial page.
CLOUDFLARE_INDICATORS = [
re.compile(r"cloudflare", re.IGNORECASE),
re.compile(r"cf-challenge", re.IGNORECASE),
re.compile(r"just a moment", re.IGNORECASE),
re.compile(r"cf-browser-verification", re.IGNORECASE),
re.compile(r"challenge-platform", re.IGNORECASE),
re.compile(r"ray id", re.IGNORECASE),
]
# Tags to strip during readability extraction.
STRIP_TAGS = frozenset({
"nav", "footer", "script", "style", "aside", "header",
"noscript", "iframe", "form", "svg",
})
class GenericUrlConnector:
"""Connector that fetches a single job posting from a URL.
The connector does simple readability extraction and does NOT follow
Cloudflare challenge pages. When a challenge is detected, it raises
UnsupportedSite.
"""
def __init__(self, client: httpx.Client | None = None) -> None:
"""Initialize the connector.
Args:
client: Optional pre-configured httpx.Client for testing.
"""
self._client = client
def fetch(self, query: SearchQuery) -> list[RawPosting]:
"""Fetch a posting from the URL in query.query.
Args:
query: SearchQuery whose ``query`` field is the URL to fetch.
Returns:
A single-element list containing the extracted RawPosting.
Raises:
UnsupportedSite: If the page is a Cloudflare challenge or
cannot be parsed.
"""
url = query.query
if self._client is not None:
return self._fetch_with_client(self._client, url)
with httpx.Client(headers=HEADERS, timeout=30.0, follow_redirects=True) as client:
return self._fetch_with_client(client, url)
def _fetch_with_client(
self, client: httpx.Client, url: str,
) -> list[RawPosting]:
response = client.get(url)
response.raise_for_status()
html = response.text
if self._is_cloudflare_challenge(html, response.headers):
raise UnsupportedSite(url, "Cloudflare challenge page detected")
return [self._extract(url, html, response)]
@staticmethod
def _is_cloudflare_challenge(html: str, headers: httpx.Headers) -> bool:
"""Detect Cloudflare challenge/interstitial pages."""
# Check response headers
server = headers.get("server", "")
cf_ray = headers.get("cf-ray", "")
if "cloudflare" in server.lower() and cf_ray:
# Could be a normal CF-proxied site. Check body for challenge markers.
body_lower = html[:5000].lower()
if any(p.search(body_lower) for p in CLOUDFLARE_INDICATORS):
if "just a moment" in body_lower or "challenge-platform" in body_lower:
return True
# If the body is very short and has cf markers, likely a challenge
if len(html.strip()) < 2000:
if any(p.search(html) for p in CLOUDFLARE_INDICATORS):
return True
# Also check for challenge pages without server header (some setups)
body_lower = html[:5000].lower()
if "cf-browser-verification" in body_lower:
return True
if "challenge-platform" in body_lower and "just a moment" in body_lower:
return True
return False
@staticmethod
def _extract(
url: str, html: str, response: httpx.Response,
) -> RawPosting:
"""Extract a RawPosting from HTML content."""
soup = BeautifulSoup(html, "lxml")
# Strip unwanted tags
for tag_name in STRIP_TAGS:
for tag in soup.find_all(tag_name):
tag.decompose()
# Find main content container
container = soup.find("main")
if container is None:
container = soup.find("article")
if container is None:
container = soup.find("body") or soup
# Extract title
title = ""
title_tag = soup.find("title")
h1_tag = container.find("h1") if container else None
if h1_tag and h1_tag.get_text(strip=True):
title = h1_tag.get_text(strip=True)
elif title_tag and title_tag.get_text(strip=True):
title = title_tag.get_text(strip=True)
# Remove common site suffixes from title tag
for sep in [" | ", " - ", " _ "]:
if sep in title:
title = title.split(sep)[0].strip()
break
# Extract company guess
company = ""
# Try meta tags
og_site = soup.find("meta", property="og:site_name")
if og_site and og_site.get("content"):
company = og_site["content"].strip()
if not company:
# Try JSON-LD
for script in soup.find_all("script", type="application/ld+json"):
# Scripts should already be decomposed, but check anyway
pass
if not company:
# Try looking for common company patterns in the text
text = container.get_text(" ", strip=True) if container else ""
# Look for "at CompanyName" or "Company: Name" patterns
at_match = re.search(r"\bat\s+([A-Z][A-Za-z0-9&\s]+?)(?:\.|,|$)", text)
if at_match:
company = at_match.group(1).strip()
if not company:
# Fall back to domain name
company = httpx.URL(url).host or ""
# Extract clean description text
description = container.get_text("\n", strip=True) if container else ""
# Collapse excessive whitespace
description = re.sub(r"\n{3,}", "\n\n", description)
description = description.strip()
# Extract location if present (simple heuristic)
location = None
loc_match = re.search(
r"(?:location|stad|city|ort)\s*[:\-]\s*(.+?)(?:\n|$)",
description,
re.IGNORECASE,
)
if loc_match:
location = loc_match.group(1).strip()
return RawPosting(
source="generic_url",
external_id=None,
url=url,
company=company,
title=title,
location=location,
description=description,
raw={
"status_code": response.status_code,
"content_length": len(html),
},
)

56
packages/connectors/src/connectors/models.py Normal file
View file

@ -0,0 +1,56 @@
"""Data models for connectors."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
@dataclass
class SearchQuery:
"""Query parameters passed to Connector.fetch.
Attributes:
query: Free-text search string (e.g. \"python developer\").
region: Optional region filter (e.g. \"Skane lan\").
limit: Maximum number of results to return.
"""
query: str = ""
region: str | None = None
limit: int = 20
@dataclass
class RawPosting:
"""Raw posting as returned by a connector before normalization.
The ``raw`` dict holds the original source payload for audit/debugging.
"""
source: str
external_id: str | None
url: str
company: str
title: str
location: str | None = None
description: str = ""
raw: dict[str, Any] = field(default_factory=dict)
@dataclass
class JobPosting:
"""Normalized posting matching the ``job_posting`` table shape.
Fields map 1:1 to columns in the database schema:
source, external_id, url, company, title, location, description, raw.
"""
source: str
external_id: str | None
url: str
company: str
title: str
location: str | None = None
description: str = ""
raw: dict[str, Any] = field(default_factory=dict)

31
packages/connectors/src/connectors/normalizer.py Normal file
View file

@ -0,0 +1,31 @@
"""Normalizer: convert RawPosting to JobPosting."""
from __future__ import annotations
from connectors.models import JobPosting, RawPosting
def normalize(posting: RawPosting) -> JobPosting:
"""Normalize a RawPosting into a JobPosting.
The current mapping is 1:1 because RawPosting already carries the
fields needed by the job_posting table. This function exists as a
single point of transformation so that future field remapping,
cleaning, or enrichment is centralized.
Args:
posting: A RawPosting from any connector.
Returns:
A JobPosting with fields matching the database schema.
"""
return JobPosting(
source=posting.source,
external_id=posting.external_id,
url=posting.url,
company=posting.company,
title=posting.title,
location=posting.location,
description=posting.description,
raw=posting.raw,
)

32
packages/connectors/src/connectors/protocol.py Normal file
View file

@ -0,0 +1,32 @@
"""Connector protocol definition."""
from __future__ import annotations
from typing import Protocol, runtime_checkable
from connectors.models import RawPosting, SearchQuery
@runtime_checkable
class Connector(Protocol):
"""Protocol every job-source adapter implements.
Implementations fetch postings from an external source and return
a list of RawPosting objects. Normalization to JobPosting is done
separately via :func:`connectors.normalize`.
"""
def fetch(self, query: SearchQuery) -> list[RawPosting]:
"""Fetch raw postings matching the given query.
Args:
query: Search parameters (keywords, region, limit).
Returns:
List of RawPosting objects.
Raises:
UnsupportedSite: If the source site cannot be scraped.
httpx.HTTPError: On network errors (live calls only).
"""
...

1
packages/connectors/tests/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Test package init."""

75
packages/connectors/tests/fixtures/af_search_response.json vendored Normal file
View file

@ -0,0 +1,75 @@
{
"hits": [
{
"id": "12345678901",
"headline": "Senior Python Developer",
"employer": {
"name": "Tech Innovators AB",
"workplace": "Skane"
},
"webpage_url": "https://arbetsformedlingen.se/platsbanken/annonser/12345678901",
"description": {
"text": "We are looking for a Senior Python Developer to join our team in Malmo. You will work on backend services, APIs, and data pipelines. Experience with FastAPI, PostgreSQL, and Docker is required."
},
"workplace_address": {
"city": "Malmo",
"municipality": "Malmo",
"country": "Sverige"
},
"occupation": {
"label": "Systemutvecklare",
"concept_id": "abc123"
},
"publication_date": "2026-07-28T10:00:00+02:00",
"application_deadline": "2026-08-15T23:59:59+02:00"
},
{
"id": "12345678902",
"headline": "Fullstack Utvecklare",
"employer": {
"name": "Nordic Solutions AB"
},
"webpage_url": "https://arbetsformedlingen.se/platsbanken/annonser/12345678902",
"description": {
"text": "Vi soker en fullstack utvecklare med erfarenhet av Vue.js och Python. Du kommer att arbeta med vara webbapplikationer i en agil miljo."
},
"workplace_address": {
"city": "Lund",
"municipality": "Lund",
"country": "Sverige"
},
"occupation": {
"label": "Webbutvecklare",
"concept_id": "def456"
},
"publication_date": "2026-07-29T08:00:00+02:00",
"application_deadline": "2026-08-20T23:59:59+02:00"
},
{
"id": "12345678903",
"headline": "DevOps Engineer",
"employer": {
"name": "Cloud Services Nordic"
},
"webpage_url": "https://arbetsformedlingen.se/platsbanken/annonser/12345678903",
"description": {
"text": "DevOps Engineer with experience in Kubernetes, Terraform, and CI/CD pipelines. You will manage our cloud infrastructure."
},
"workplace_address": {
"city": "Helsingborg",
"municipality": "Helsingborg",
"country": "Sverige"
},
"occupation": {
"label": "DevOps-ingenjor",
"concept_id": "ghi789"
},
"publication_date": "2026-07-30T09:00:00+02:00",
"application_deadline": "2026-08-30T23:59:59+02:00"
}
],
"total": {
"value": 3,
"relation": "eq"
}
}

28
packages/connectors/tests/fixtures/cloudflare_challenge.html vendored Normal file
View file

@ -0,0 +1,28 @@
<!DOCTYPE html>
<html lang="en">
<head>
<title>Just a moment...</title>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<style>
body { font-family: sans-serif; }
.cf-spinner { display: block; margin: 50px auto; }
</style>
</head>
<body>
<div id="cf-challenge-running">
<div class="cf-spinner"></div>
<h1>Just a moment...</h1>
<p>Don't refresh this page.</p>
</div>
<script src="/cdn-cgi/challenge-platform/h/b/cv/result/0" type="text/javascript"></script>
<script>
(function(){
var a = document.getElementById('cf-challenge-running');
a.style.display = 'none';
// challenge-platform code
window._cf_chl_opt = {cfRay: 'abc123-xyz789'};
})();
</script>
</body>
</html>

64
packages/connectors/tests/fixtures/generic_job_page.html vendored Normal file
View file

@ -0,0 +1,64 @@
<!DOCTYPE html>
<html lang="en">
<head>
<title>Senior Backend Engineer at Acme Corp | JobBoard</title>
<meta property="og:site_name" content="Acme Corp">
<meta name="description" content="We are hiring a Senior Backend Engineer">
<link rel="stylesheet" href="/styles.css">
<script src="/analytics.js"></script>
<nav class="navbar">
<a href="/">Home</a>
<a href="/jobs">Jobs</a>
<a href="/about">About</a>
<a href="/contact">Contact</a>
</nav>
</head>
<body>
<header class="site-header">
<div class="logo">JobBoard</div>
<div class="user-menu">Login | Register</div>
</header>
<nav class="breadcrumb">
<a href="/">Home</a> > <a href="/jobs">Jobs</a> > Senior Backend Engineer
</nav>
<main>
<article class="job-posting">
<h1>Senior Backend Engineer</h1>
<div class="job-meta">
<span class="company">Acme Corp</span>
<span class="location">Malmo, Sweden</span>
</div>
<div class="job-description">
<p>We are looking for a Senior Backend Engineer to join our growing team in Malmo. You will be responsible for designing and building scalable backend services using Python, FastAPI, and PostgreSQL.</p>
<h2>Requirements</h2>
<ul>
<li>5+ years of Python experience</li>
<li>Strong knowledge of REST API design</li>
<li>Experience with PostgreSQL and database optimization</li>
<li>Familiarity with Docker and Kubernetes</li>
</ul>
<h2>What we offer</h2>
<ul>
<li>Competitive salary</li>
<li>Flexible working hours</li>
<li>Remote-first culture</li>
<li>Health insurance</li>
</ul>
<p>Location: Malmo, Sweden</p>
</div>
</article>
</main>
<footer class="site-footer">
<p>&copy; 2026 JobBoard. All rights reserved.</p>
<a href="/privacy">Privacy Policy</a>
<a href="/terms">Terms of Service</a>
<script>
(function() {
var analytics = window.analytics = window.analytics || [];
analytics.track('job_view', {id: '123'});
})();
</script>
</footer>
<script src="/main.js"></script>
</body>
</html>

191
packages/connectors/tests/test_arbetsformedlingen.py Normal file
View file

@ -0,0 +1,191 @@
"""Tests for the Arbetsformedlingen connector.
Uses a recorded JSON fixture (no live network). The fixture is loaded
into an httpx.MockTransport that returns the recorded response.
"""
from __future__ import annotations
import json
from pathlib import Path
import httpx
import pytest
from connectors.arbetsformedlingen import ArbetsformedlingenConnector
from connectors.models import SearchQuery
FIXTURES_DIR = Path(__file__).parent / "fixtures"
def _load_fixture(name: str) -> str:
"""Load a fixture file as raw text."""
return (FIXTURES_DIR / name).read_text(encoding="utf-8")
def _make_client(response_body: str, status_code: int = 200) -> httpx.Client:
"""Create an httpx.Client with a mock transport returning the fixture."""
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(status_code, content=response_body, headers={
"content-type": "application/json",
})
transport = httpx.MockTransport(handler)
return httpx.Client(transport=transport)
class TestArbetsformedlingenMapping:
"""Test field mapping from AF API response to RawPosting."""
def test_basic_mapping(self):
"""Test that API fields map correctly to RawPosting fields."""
fixture = _load_fixture("af_search_response.json")
client = _make_client(fixture)
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="python", limit=3))
assert len(results) == 3
# First hit
first = results[0]
assert first.source == "arbetsformedlingen"
assert first.external_id == "12345678901"
assert first.title == "Senior Python Developer"
assert first.company == "Tech Innovators AB"
assert first.url == "https://arbetsformedlingen.se/platsbanken/annonser/12345678901"
assert "FastAPI" in first.description
assert first.location == "Malmo"
# raw should contain the original hit
assert first.raw["id"] == "12345678901"
def test_swedish_text_preserved(self):
"""Test that Swedish characters (a, a, o) are preserved."""
fixture = _load_fixture("af_search_response.json")
client = _make_client(fixture)
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="utvecklare", limit=3))
second = results[1]
assert second.title == "Fullstack Utvecklare"
assert "Nordic Solutions AB" in second.company
assert "soker" in second.description
assert "miljo" in second.description
def test_missing_description_field(self):
"""Test graceful handling when description is missing."""
fixture_data = {
"hits": [
{
"id": "999",
"headline": "No Description Job",
"employer": {"name": "Empty Corp"},
"webpage_url": "https://example.com/job/999",
}
]
}
client = _make_client(json.dumps(fixture_data))
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="test", limit=1))
assert len(results) == 1
assert results[0].description == ""
assert results[0].title == "No Description Job"
assert results[0].company == "Empty Corp"
def test_missing_employer_name(self):
"""Test graceful handling when employer name is missing."""
fixture_data = {
"hits": [
{
"id": "888",
"headline": "Mystery Job",
"webpage_url": "https://example.com/job/888",
"description": {"text": "A job with no employer name."},
}
]
}
client = _make_client(json.dumps(fixture_data))
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="test", limit=1))
assert len(results) == 1
assert results[0].company == ""
def test_empty_hits(self):
"""Test handling of an empty results list."""
fixture_data = {"hits": [], "total": {"value": 0, "relation": "eq"}}
client = _make_client(json.dumps(fixture_data))
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="nonexistent", limit=10))
assert results == []
def test_region_param_passed(self):
"""Test that region parameter is mapped and sent to the API."""
fixture = _load_fixture("af_search_response.json")
captured_params: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured_params.update(dict(request.url.params))
return httpx.Response(200, content=fixture, headers={
"content-type": "application/json",
})
transport = httpx.MockTransport(handler)
client = httpx.Client(transport=transport)
connector = ArbetsformedlingenConnector(client=client)
connector.fetch(SearchQuery(query="dev", region="malmo"))
assert captured_params.get("q") == "dev"
assert captured_params.get("region") == "Skane lan"
def test_limit_param_passed(self):
"""Test that limit parameter is sent to the API."""
fixture = _load_fixture("af_search_response.json")
captured_params: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured_params.update(dict(request.url.params))
return httpx.Response(200, content=fixture, headers={
"content-type": "application/json",
})
transport = httpx.MockTransport(handler)
client = httpx.Client(transport=transport)
connector = ArbetsformedlingenConnector(client=client)
connector.fetch(SearchQuery(query="dev", limit=50))
assert captured_params.get("limit") == "50"
def test_workplace_address_municipality_fallback(self):
"""Test that municipality is used when city is missing."""
fixture_data = {
"hits": [
{
"id": "777",
"headline": "Rural Job",
"employer": {"name": "Rural Corp"},
"webpage_url": "https://example.com/job/777",
"description": {"text": "Work in the countryside."},
"workplace_address": {
"municipality": "Helsingborg",
"country": "Sverige",
},
}
]
}
client = _make_client(json.dumps(fixture_data))
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="test", limit=1))
assert results[0].location == "Helsingborg"
def test_raw_preserves_original_hit(self):
"""Test that the raw field preserves the full original hit data."""
fixture = _load_fixture("af_search_response.json")
client = _make_client(fixture)
connector = ArbetsformedlingenConnector(client=client)
results = connector.fetch(SearchQuery(query="dev", limit=3))
third = results[2]
assert third.raw["occupation"]["label"] == "DevOps-ingenjor"
assert third.raw["publication_date"] == "2026-07-30T09:00:00+02:00"

119
packages/connectors/tests/test_dedupe.py Normal file
View file

@ -0,0 +1,119 @@
"""Tests for the deduplication helper."""
from __future__ import annotations
from connectors.dedupe import dedupe, dedupe_job_postings
from connectors.models import JobPosting, RawPosting
def _make_raw(
source: str = "arbetsformedlingen",
external_id: str | None = "1",
url: str = "https://example.com/1",
company: str = "Corp",
title: str = "Dev",
) -> RawPosting:
return RawPosting(
source=source,
external_id=external_id,
url=url,
company=company,
title=title,
)
class TestDedupeRawPostings:
"""Test dedupe function with RawPosting objects."""
def test_no_duplicates_unchanged(self):
"""Test that a list with no duplicates is unchanged."""
postings = [
_make_raw(url="https://a.com/1", external_id="1"),
_make_raw(url="https://a.com/2", external_id="2"),
_make_raw(url="https://a.com/3", external_id="3"),
]
result = dedupe(postings)
assert len(result) == 3
def test_dedupe_by_url(self):
"""Test deduplication by (source, url)."""
postings = [
_make_raw(url="https://a.com/1", external_id="1"),
_make_raw(url="https://a.com/1", external_id="2"), # same URL, diff ext_id
_make_raw(url="https://a.com/2", external_id="3"),
]
result = dedupe(postings)
assert len(result) == 2
assert result[0].external_id == "1" # first occurrence kept
assert result[1].external_id == "3"
def test_dedupe_by_external_id(self):
"""Test deduplication by (source, external_id) when URL differs."""
postings = [
_make_raw(url="https://a.com/1", external_id="100"),
_make_raw(url="https://a.com/2", external_id="100"), # same ext_id
_make_raw(url="https://a.com/3", external_id="200"),
]
result = dedupe(postings)
assert len(result) == 2
assert result[0].url == "https://a.com/1"
assert result[1].url == "https://a.com/3"
def test_different_sources_same_url_not_deduped(self):
"""Test that same URL from different sources are NOT deduped."""
postings = [
_make_raw(source="arbetsformedlingen", url="https://a.com/1", external_id="1"),
_make_raw(source="generic_url", url="https://a.com/1", external_id=None),
]
result = dedupe(postings)
assert len(result) == 2
def test_none_external_id_ignored(self):
"""Test that None external_id does not cause dedup by ext_id."""
postings = [
_make_raw(url="https://a.com/1", external_id=None),
_make_raw(url="https://a.com/2", external_id=None),
_make_raw(url="https://a.com/3", external_id=None),
]
result = dedupe(postings)
assert len(result) == 3
def test_empty_list(self):
"""Test that an empty list returns empty."""
assert dedupe([]) == []
def test_single_item(self):
"""Test that a single item list is unchanged."""
postings = [_make_raw()]
result = dedupe(postings)
assert len(result) == 1
def test_order_preserved(self):
"""Test that first-occurrence order is preserved."""
postings = [
_make_raw(url="https://a.com/3", external_id="3", title="Third"),
_make_raw(url="https://a.com/1", external_id="1", title="First"),
_make_raw(url="https://a.com/3", external_id="3", title="Third-Dup"),
_make_raw(url="https://a.com/2", external_id="2", title="Second"),
]
result = dedupe(postings)
assert len(result) == 3
assert result[0].title == "Third"
assert result[1].title == "First"
assert result[2].title == "Second"
class TestDedupeJobPostings:
"""Test dedupe_job_postings function with JobPosting objects."""
def test_dedupe_job_postings_by_url(self):
"""Test deduplication of JobPosting objects."""
postings = [
JobPosting(source="af", external_id="1", url="https://a.com/1",
company="C", title="T"),
JobPosting(source="af", external_id="2", url="https://a.com/1",
company="C", title="T"),
]
result = dedupe_job_postings(postings)
assert len(result) == 1
assert result[0].external_id == "1"

222
packages/connectors/tests/test_generic_url.py Normal file
View file

@ -0,0 +1,222 @@
"""Tests for the GenericUrlConnector.
Uses recorded HTML fixtures (no live network). Fixtures are loaded into
an httpx.MockTransport.
"""
from __future__ import annotations
from pathlib import Path
import httpx
import pytest
from connectors.exceptions import UnsupportedSite
from connectors.generic_url import GenericUrlConnector
from connectors.models import SearchQuery
FIXTURES_DIR = Path(__file__).parent / "fixtures"
def _load_fixture(name: str) -> str:
"""Load a fixture file as raw text."""
return (FIXTURES_DIR / name).read_text(encoding="utf-8")
def _make_client(
response_body: str,
status_code: int = 200,
headers: dict | None = None,
) -> httpx.Client:
"""Create an httpx.Client with a mock transport returning the fixture."""
default_headers = {"content-type": "text/html; charset=utf-8"}
if headers:
default_headers.update(headers)
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(status_code, content=response_body, headers=default_headers)
transport = httpx.MockTransport(handler)
return httpx.Client(transport=transport)
class TestGenericUrlExtraction:
"""Test readability extraction from a messy HTML page."""
def test_extracts_title_from_h1(self):
"""Test that the h1 tag is used as the title."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
assert len(results) == 1
posting = results[0]
assert posting.title == "Senior Backend Engineer"
def test_extracts_company_from_meta(self):
"""Test that og:site_name meta tag is used as company name."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
assert results[0].company == "Acme Corp"
def test_strips_nav_and_footer(self):
"""Test that nav and footer content is removed from description."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
description = results[0].description
# Nav links should not appear
assert "Home" not in description or "Requirements" in description
assert "Privacy Policy" not in description
assert "Terms of Service" not in description
assert "Login | Register" not in description
# Footer copyright should not appear
assert "2026 JobBoard" not in description
# Job content should be present
assert "Python" in description
assert "FastAPI" in description
assert "PostgreSQL" in description
def test_strips_script_and_style(self):
"""Test that script and style tags are removed."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
description = results[0].description
assert "analytics" not in description.lower()
assert "var " not in description
assert "function()" not in description
def test_source_is_generic_url(self):
"""Test that source is set to 'generic_url'."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
assert results[0].source == "generic_url"
def test_url_preserved(self):
"""Test that the original URL is preserved in the result."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
url = "https://example.com/job/123"
results = connector.fetch(SearchQuery(query=url))
assert results[0].url == url
def test_external_id_is_none(self):
"""Test that external_id is None for generic URL postings."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
assert results[0].external_id is None
def test_description_is_clean(self):
"""Test that description text is clean and readable."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
description = results[0].description
# Should not have excessive blank lines
assert "\n\n\n" not in description
# Should start with the job title or job content
assert len(description) > 50
def test_location_extracted_from_text(self):
"""Test that location is extracted from description text."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
# The fixture contains "Location: Malmo, Sweden"
assert results[0].location is not None
assert "Malmo" in results[0].location
def test_raw_contains_metadata(self):
"""Test that the raw field contains response metadata."""
html = _load_fixture("generic_job_page.html")
client = _make_client(html)
connector = GenericUrlConnector(client=client)
results = connector.fetch(SearchQuery(query="https://example.com/job/123"))
raw = results[0].raw
assert raw["status_code"] == 200
assert raw["content_length"] > 0
class TestCloudflareDetection:
"""Test Cloudflare challenge page detection."""
def test_raises_on_cloudflare_challenge(self):
"""Test that UnsupportedSite is raised for Cloudflare challenge pages."""
html = _load_fixture("cloudflare_challenge.html")
client = _make_client(
html,
headers={
"server": "cloudflare",
"cf-ray": "abc123-xyz789",
},
)
connector = GenericUrlConnector(client=client)
with pytest.raises(UnsupportedSite) as exc_info:
connector.fetch(SearchQuery(query="https://protected-site.com/job/1"))
assert "protected-site.com" in str(exc_info.value)
assert "Cloudflare" in str(exc_info.value)
def test_raises_on_challenge_without_server_header(self):
"""Test detection of challenge pages without cloudflare server header."""
# A page with cf-browser-verification but no CF server header
html = '<html><body><div id="cf-browser-verification">Loading...</div></body></html>'
client = _make_client(html)
connector = GenericUrlConnector(client=client)
with pytest.raises(UnsupportedSite):
connector.fetch(SearchQuery(query="https://sneaky-site.com/job/1"))
def test_normal_cf_proxied_site_is_ok(self):
"""Test that a normal site behind CF (with content) is NOT flagged."""
html = _load_fixture("generic_job_page.html")
client = _make_client(
html,
headers={
"server": "cloudflare",
"cf-ray": "abc123-xyz789",
},
)
connector = GenericUrlConnector(client=client)
# Should NOT raise -- this is a normal page that happens to be behind CF
results = connector.fetch(SearchQuery(query="https://cf-proxied-site.com/job/1"))
assert len(results) == 1
assert results[0].title == "Senior Backend Engineer"
def test_unsupported_site_exception_has_url(self):
"""Test that UnsupportedSite exception carries the URL."""
html = _load_fixture("cloudflare_challenge.html")
client = _make_client(
html,
headers={"server": "cloudflare", "cf-ray": "xyz"},
)
connector = GenericUrlConnector(client=client)
with pytest.raises(UnsupportedSite) as exc_info:
connector.fetch(SearchQuery(query="https://example.org/protected"))
assert exc_info.value.url == "https://example.org/protected"

87
packages/connectors/tests/test_normalizer.py Normal file
View file

@ -0,0 +1,87 @@
"""Tests for the normalizer and connector protocol."""
from __future__ import annotations
from connectors import (
Connector,
GenericUrlConnector,
JobPosting,
RawPosting,
normalize,
)
from connectors.arbetsformedlingen import ArbetsformedlingenConnector
class TestNormalize:
"""Test the normalize function."""
def test_basic_normalization(self):
"""Test that normalize maps all fields correctly."""
raw = RawPosting(
source="arbetsformedlingen",
external_id="12345",
url="https://example.com/job/12345",
company="Test Corp",
title="Test Job",
location="Malmo",
description="A test job description.",
raw={"id": "12345", "extra": "data"},
)
job = normalize(raw)
assert job.source == "arbetsformedlingen"
assert job.external_id == "12345"
assert job.url == "https://example.com/job/12345"
assert job.company == "Test Corp"
assert job.title == "Test Job"
assert job.location == "Malmo"
assert job.description == "A test job description."
assert job.raw == {"id": "12345", "extra": "data"}
def test_none_fields_preserved(self):
"""Test that None fields are preserved."""
raw = RawPosting(
source="generic_url",
external_id=None,
url="https://example.com/page",
company="",
title="Unknown",
)
job = normalize(raw)
assert job.external_id is None
assert job.location is None
assert job.description == ""
assert job.raw == {}
def test_swedish_chars_preserved(self):
"""Test that Swedish characters are preserved through normalization."""
raw = RawPosting(
source="arbetsformedlingen",
external_id="1",
url="https://example.com/1",
company="Nordic Solutions AB",
title="Utvecklare",
location="Lund",
description="Vi soker en utvecklare.",
)
job = normalize(raw)
assert "Solutions" in job.company
assert "Utvecklare" in job.title
assert "soker" in job.description
assert "Lund" in (job.location or "")
class TestConnectorProtocol:
"""Test that connectors satisfy the Connector protocol."""
def test_af_connector_is_connector(self):
"""Test that ArbetsformedlingenConnector satisfies the Connector protocol."""
connector = ArbetsformedlingenConnector()
assert isinstance(connector, Connector)
def test_generic_url_connector_is_connector(self):
"""Test that GenericUrlConnector satisfies the Connector protocol."""
connector = GenericUrlConnector()
assert isinstance(connector, Connector)