From cdae23e303f7e035b44cd78c47061145a4ecdd4c Mon Sep 17 00:00:00 2001 From: Antoine Date: Wed, 3 Jun 2026 21:46:48 +0200 Subject: [PATCH] fix: distinguish Apec search failures from empty results --- src/job_research/apec/adapter.py | 39 +++++++++++++----- tests/apec/test_adapter.py | 69 +++++++++++++++++++++++++++++--- 2 files changed, 92 insertions(+), 16 deletions(-) diff --git a/src/job_research/apec/adapter.py b/src/job_research/apec/adapter.py index b9fbbff..bf72f78 100644 --- a/src/job_research/apec/adapter.py +++ b/src/job_research/apec/adapter.py @@ -12,6 +12,7 @@ from playwright.sync_api import sync_playwright _SEARCH_URL = "https://www.apec.fr/candidat/recherche-emploi.html/emploi" _FRANCE_LOCATION_ID = "799" _CDI_CONTRACT_ID = "101888" +_SEARCH_INPUT_SELECTOR = 'input[name="keywords"]' _RESULT_LINK_SELECTOR = "a[href*='/candidat/recherche-emploi.html/emploi/detail-offre/']" _DETAIL_JOB_ID_PATTERN = re.compile(r"/detail-offre/([^/?#]+)") _APEC_HOSTS = {"apec.fr", "www.apec.fr"} @@ -29,6 +30,10 @@ class ApecSearchFilters: contract_type: str | None = None +class ApecSearchError(RuntimeError): + pass + + @contextmanager def _open_public_page(): with sync_playwright() as playwright: @@ -96,6 +101,7 @@ class ApecAdapter: def search(self, queries: list[str], search_filters: ApecSearchFilters) -> list[ApecSearchResult]: results: list[ApecSearchResult] = [] seen_keys: set[str] = set() + usable_search_page_seen = False with _open_public_page() as page: for query in queries: @@ -109,31 +115,47 @@ class ApecAdapter: _accept_cookies_if_present(page) page.wait_for_load_state("domcontentloaded") + try: + page.wait_for_selector(_SEARCH_INPUT_SELECTOR, timeout=5_000) + except PlaywrightTimeoutError: + continue + + usable_search_page_seen = True + try: page.wait_for_selector(_RESULT_LINK_SELECTOR, timeout=5_000) except PlaywrightTimeoutError: continue result_page_url = page.url + seen_page_urls: set[str] = {result_page_url} page_number = 0 - no_progress_pages = 0 while len(results) < self.max_listings: if page_number > 0: - page.goto(_search_results_url(result_page_url, page_number), wait_until="domcontentloaded") + next_page_url = _search_results_url(result_page_url, page_number) + if next_page_url in seen_page_urls: + break + + page.goto(next_page_url, wait_until="domcontentloaded") try: page.wait_for_selector(_RESULT_LINK_SELECTOR, timeout=5_000) except PlaywrightTimeoutError: break + current_page_url = page.url + if page_number > 0 and current_page_url in seen_page_urls: + break + + seen_page_urls.add(current_page_url) + hrefs = page.locator(_RESULT_LINK_SELECTOR).evaluate_all( "nodes => nodes.map(node => node.href)" ) if not hrefs: break - added_any_result = False for href in hrefs: source_job_id = _extract_source_job_id(href) dedupe_key = source_job_id or href @@ -142,20 +164,15 @@ class ApecAdapter: seen_keys.add(dedupe_key) results.append(ApecSearchResult(url=href, source_job_id=source_job_id)) - added_any_result = True if len(results) >= self.max_listings: break - if added_any_result: - no_progress_pages = 0 - else: - no_progress_pages += 1 - if no_progress_pages >= 2: - break - page_number += 1 + if not usable_search_page_seen: + raise ApecSearchError("Apec search page was not reachable for any query") + return results def fetch_listing_html(self, url: str) -> str: diff --git a/tests/apec/test_adapter.py b/tests/apec/test_adapter.py index 345d239..ee5152d 100644 --- a/tests/apec/test_adapter.py +++ b/tests/apec/test_adapter.py @@ -9,6 +9,7 @@ from job_research.apec.adapter import ApecAdapter, ApecSearchFilters _RESULT_LINK_SELECTOR = "a[href*='/candidat/recherche-emploi.html/emploi/detail-offre/']" +_SEARCH_INPUT_SELECTOR = 'input[name="keywords"]' class _FakeResultButton: @@ -41,11 +42,18 @@ class _FakeLocator: class _FakeDetailPage: - def __init__(self, result_pages: dict[str, dict[int, list[str]]], *, rendered_html: str = "rendered") -> None: + def __init__( + self, + result_pages: dict[str, dict[int, list[str]]], + *, + rendered_html: str = "rendered", + search_ready: bool = True, + ) -> None: self.result_pages = result_pages self.rendered_html = rendered_html self.shell_html = "shell" self.waited_functions: list[tuple[str, int | None]] = [] + self.search_ready = search_ready self.goto_urls: list[str] = [] self.current_query = "" self.current_page = 0 @@ -75,6 +83,12 @@ class _FakeDetailPage: return None def wait_for_selector(self, selector: str, timeout: int | None = None) -> None: + if selector == _SEARCH_INPUT_SELECTOR: + if self.search_ready: + return None + + raise PlaywrightTimeoutError(f"selector not found: {selector}") + if selector == _RESULT_LINK_SELECTOR and self.current_results(): return None @@ -127,13 +141,13 @@ def test_search_continues_past_duplicate_only_pages(monkeypatch) -> None: assert any("page=1" in url for url in page.goto_urls) -def test_search_stops_after_repeated_duplicate_only_pages(monkeypatch) -> None: +def test_search_continues_past_duplicate_only_pages_until_a_later_hit(monkeypatch) -> None: first_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=0&selectedIndex=0" - second_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222?motsCles=beta&page=2&selectedIndex=0" + second_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222?motsCles=beta&page=3&selectedIndex=0" page = _FakeDetailPage( { - "alpha": {0: [first_result], 1: []}, - "beta": {0: [first_result], 1: [first_result], 2: [second_result], 3: []}, + "alpha": {0: [first_result], 1: [first_result], 2: [first_result], 3: [second_result], 4: []}, + "beta": {0: [first_result], 1: []}, } ) @@ -144,11 +158,56 @@ def test_search_stops_after_repeated_duplicate_only_pages(monkeypatch) -> None: search_filters=ApecSearchFilters(location="France", contract_type="CDI"), ) + assert [result.url for result in results] == [first_result, second_result] + assert any("page=1" in url for url in page.goto_urls) + assert any("page=3" in url for url in page.goto_urls) + + +def test_search_stops_when_result_page_url_repeats(monkeypatch) -> None: + first_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=0&selectedIndex=0" + page = _FakeDetailPage( + { + "alpha": {0: [first_result], 1: [first_result], 2: [first_result]}, + } + ) + + original_goto = page.goto + initial_result_page_url: str | None = None + + def looping_goto(url: str, wait_until: str | None = None) -> None: + nonlocal initial_result_page_url + + original_goto(url, wait_until=wait_until) + + if initial_result_page_url is None and page.current_page == 0: + initial_result_page_url = page.url + elif initial_result_page_url is not None and page.current_page > 0: + page.url = initial_result_page_url + + monkeypatch.setattr(page, "goto", looping_goto) + monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page)) + + results = ApecAdapter(max_listings=10).search( + ["alpha"], + search_filters=ApecSearchFilters(location="France", contract_type="CDI"), + ) + assert [result.url for result in results] == [first_result] assert any("page=1" in url for url in page.goto_urls) assert not any("page=2" in url for url in page.goto_urls) +def test_search_raises_when_every_query_fails_to_load_a_search_page(monkeypatch) -> None: + page = _FakeDetailPage({"alpha": {0: []}}, search_ready=False) + monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page)) + + with pytest.raises(adapter_module.ApecSearchError): + ApecAdapter(max_listings=10).search( + ["alpha", "beta"], + search_filters=ApecSearchFilters(location="France", contract_type="CDI"), + ) + + def test_fetch_listing_html_waits_for_rendered_offer_content(monkeypatch) -> None: page = _FakeDetailPage({}, rendered_html="rendered offer") monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))