1"""
2Jobs Board Scraper: Indeed, Reed, Adzuna, RemoteOK & More - Multi-Board Aggregator
3Scrapes job listings from UK, US, EU and remote job boards with unified output.
4
5Supported boards:
6 UK: Reed, Totaljobs, CV-Library, CWJobs, Indeed UK, GOV.UK Find a Job
7 US: USAJobs, Indeed US
8 EU: Indeed DE/FR/NL, Arbeitnow
9 Global: Adzuna (multi-country API), RemoteOK
10"""
11
12import asyncio
13import os
14import random
15import re
16import statistics
17from datetime import datetime, timezone
18
19import httpx
20from apify import Actor
21from playwright.async_api import async_playwright
22
23from .utils import make_headers
24from .boards.reed import ReedScraper
25from .boards.totaljobs import TotaljobsScraper
26from .boards.cvlibrary import CVLibraryScraper
27from .boards.cwjobs import CWJobsScraper
28from .boards.indeed import IndeedUKScraper, IndeedScraper
29from .boards.findajob import FindAJobScraper
30from .boards.adzuna import AdzunaScraper
31from .boards.usajobs import USAJobsScraper
32from .boards.remoteok import RemoteOKScraper
33from .boards.arbeitnow import ArbeitnowScraper
34from .boards.themuse import TheMuseScraper
35from .boards.remotive import RemotiveScraper
36from .boards.jobicy import JobicyScraper
37from . import pipeline
38
39
40
41BOARD_REGISTRY = {
42
43 "reed": ReedScraper,
44 "totaljobs": TotaljobsScraper,
45 "cvlibrary": CVLibraryScraper,
46 "cwjobs": CWJobsScraper,
47 "indeed": IndeedUKScraper,
48 "findajob": FindAJobScraper,
49
50 "usajobs": USAJobsScraper,
51 "indeed_us": lambda client, **kw: IndeedScraper(client, base_url="https://www.indeed.com", source="indeed.com", **kw),
52
53 "indeed_de": lambda client, **kw: IndeedScraper(client, base_url="https://de.indeed.com", source="indeed.de", **kw),
54 "indeed_fr": lambda client, **kw: IndeedScraper(client, base_url="https://fr.indeed.com", source="indeed.fr", **kw),
55 "indeed_nl": lambda client, **kw: IndeedScraper(client, base_url="https://nl.indeed.com", source="indeed.nl", **kw),
56 "indeed_au": lambda client, **kw: IndeedScraper(client, base_url="https://au.indeed.com", source="indeed.au", **kw),
57
58 "remoteok": RemoteOKScraper,
59 "arbeitnow": ArbeitnowScraper,
60 "themuse": TheMuseScraper,
61 "remotive": RemotiveScraper,
62 "jobicy": JobicyScraper,
63
64}
65
66
67
68
69
70
71BROWSER_BOARDS = {"totaljobs", "cwjobs", "cvlibrary", "indeed", "reed",
72 "indeed_us", "indeed_de", "indeed_fr", "indeed_nl", "indeed_au",
73 "findajob"}
74
75
76HTTP_ONLY_BOARDS = {"usajobs", "remoteok", "arbeitnow", "adzuna",
77 "themuse", "remotive", "jobicy"}
78
79
80
81COUNTRY_DEFAULTS = {
82 "uk": ["reed", "totaljobs", "cvlibrary", "cwjobs", "indeed", "findajob", "adzuna", "themuse"],
83 "us": ["usajobs", "indeed_us", "adzuna", "remoteok", "themuse", "remotive"],
84 "de": ["indeed_de", "adzuna", "arbeitnow", "themuse"],
85 "fr": ["indeed_fr", "adzuna", "themuse"],
86 "nl": ["indeed_nl", "adzuna", "themuse"],
87 "au": ["indeed_au", "adzuna", "themuse"],
88 "remote": ["remoteok", "arbeitnow", "remotive", "jobicy", "themuse", "adzuna"],
89}
90
91
92ADZUNA_COUNTRY_MAP = {
93 "uk": "gb", "us": "us", "de": "de", "fr": "fr",
94 "nl": "nl", "au": "au", "remote": "gb",
95}
96
97
98
99
100
101
102
103FREE_API_BOARDS = {"adzuna", "usajobs", "remoteok", "arbeitnow",
104 "themuse", "remotive", "jobicy"}
105
106
107MAX_SEARCH_TERMS = 10
108
109
110
111SOURCE_CURRENCY = {
112 "usajobs.gov": "USD",
113 "remoteok.com": "USD",
114 "remotive.com": "USD",
115 "jobicy.com": "USD",
116 "themuse.com": "USD",
117 "indeed.com": "USD",
118 "arbeitnow.com": "EUR",
119 "indeed.de": "EUR",
120 "indeed.fr": "EUR",
121 "indeed.nl": "EUR",
122 "indeed.au": "AUD",
123 "adzuna.us": "USD",
124 "adzuna.de": "EUR",
125 "adzuna.fr": "EUR",
126 "adzuna.nl": "EUR",
127 "adzuna.au": "AUD",
128}
129
130
131STEALTH_ARGS = [
132 "--no-sandbox",
133 "--disable-dev-shm-usage",
134 "--disable-blink-features=AutomationControlled",
135 "--disable-features=IsolateOrigins,site-per-process",
136 "--disable-infobars",
137 "--disable-background-networking",
138 "--disable-default-apps",
139 "--disable-extensions",
140 "--disable-sync",
141 "--disable-translate",
142 "--no-first-run",
143 "--ignore-certificate-errors",
144 "--window-size=1920,1080",
145]
146
147
148async def run_board(scraper, keyword, location, max_per_board, job_type, salary_min) -> list[dict]:
149 """Run a single board scraper with error handling."""
150 try:
151 return await scraper.search(
152 keyword=keyword,
153 location=location,
154 max_results=max_per_board,
155 job_type=job_type,
156 salary_min=salary_min,
157 )
158 except Exception as e:
159 Actor.log.error(f"[{scraper.source_name}] Scraper failed: {e}")
160 return []
161
162
163def categorize_job(title: str) -> str:
164 """Infer a job category from the title using keyword matching.
165
166 Returns one of a fixed set of categories, or "Other" if no match.
167 Checked in order — first match wins, so more specific patterns come first.
168 """
169 if not title:
170 return "Other"
171 t = title.lower()
172
173
174 if any(k in t for k in ("software eng", "software dev", "full stack", "fullstack",
175 "full-stack", "frontend", "front-end", "front end",
176 "backend", "back-end", "back end", "web dev",
177 "mobile dev", "ios dev", "android dev", "flutter",
178 "react", "angular", "vue.js", "node.js", "python dev",
179 "java dev", ".net dev", "c# dev", "c++ dev", "rust dev",
180 "golang", "ruby dev", "php dev", "programmer", "coder",
181 "software architect", "principal engineer",
182 "lead developer", "lead engineer", "tech lead")):
183 return "Software Development"
184
185 if any(k in t for k in ("data scien", "data analy", "data eng", "machine learn",
186 "ml eng", "ai eng", "artificial intelligence",
187 "deep learn", "nlp", "computer vision",
188 "business intellig", "bi analyst", "bi developer",
189 "analytics eng", "data architect")):
190 return "Data & Analytics"
191
192 if any(k in t for k in ("devops", "sre", "site reliab", "platform eng",
193 "cloud eng", "cloud arch", "infrastructure",
194 "kubernetes", "docker", "terraform", "aws eng",
195 "azure eng", "gcp eng", "systems eng", "linux eng",
196 "network eng", "devsecops")):
197 return "DevOps & Infrastructure"
198
199 if any(k in t for k in ("cyber", "security eng", "security analy",
200 "penetration", "pen test", "infosec",
201 "information security", "soc analyst",
202 "security architect", "security consult")):
203 return "Cybersecurity"
204
205 if any(k in t for k in ("qa ", "qa eng", "test eng", "tester", "sdet",
206 "automation eng", "quality assurance", "test analy",
207 "test lead", "test manager")):
208 return "QA & Testing"
209
210 if any(k in t for k in ("product manager", "product owner", "product lead",
211 "head of product", "vp product", "cpo",
212 "product director")):
213 return "Product Management"
214
215 if any(k in t for k in ("project manager", "programme manager",
216 "program manager", "delivery manager",
217 "scrum master", "agile coach", "release manager",
218 "pmo", "project coordinator", "delivery lead")):
219 return "Project & Delivery Management"
220
221 if any(k in t for k in ("ux ", "ui ", "ux/ui", "ui/ux", "user experience",
222 "user interface", "product design", "interaction design",
223 "visual design", "graphic design", "web design",
224 "design lead", "creative director")):
225 return "Design & UX"
226
227 if any(k in t for k in ("it manager", "it director", "cto", "cio",
228 "head of engineering", "head of it",
229 "vp engineering", "engineering manager",
230 "development manager", "it service",
231 "service desk", "helpdesk", "help desk",
232 "it support", "desktop support", "it admin",
233 "systems admin", "it operations")):
234 return "IT Management & Support"
235
236 if any(k in t for k in ("dba", "database admin", "database eng",
237 "database dev", "sql dev", "etl dev",
238 "data warehouse", "database architect")):
239 return "Database & BI"
240
241 if any(k in t for k in ("solution architect", "enterprise architect",
242 "technical architect", "integration architect",
243 "it consult", "technology consult",
244 "technical consult", "sap consult",
245 "salesforce", "dynamics 365", "erp consult",
246 "crm consult", "business analyst")):
247 return "IT Consulting & Architecture"
248
249
250 if any(k in t for k in ("accountant", "accounting", "finance manager",
251 "financial analy", "financial controller",
252 "bookkeeper", "payroll", "tax ", "audit",
253 "treasury", "credit analy", "fund manager",
254 "investment analy", "actuary", "cfo")):
255 return "Finance & Accounting"
256
257 if any(k in t for k in ("marketing", "seo ", "ppc ", "content writer",
258 "copywriter", "social media", "digital market",
259 "brand manager", "communications", "pr manager",
260 "public relations", "cmo")):
261 return "Marketing & Communications"
262
263 if any(k in t for k in ("sales manager", "sales exec", "sales rep",
264 "account manager", "account exec",
265 "business develop", "bdr ", "sdr ",
266 "commercial manager", "sales director")):
267 return "Sales & Business Development"
268
269 if any(k in t for k in ("hr ", "human resource", "recruiter", "recruitment",
270 "talent acqui", "people manager", "people partner",
271 "learning and dev", "l&d ", "training manager",
272 "compensation", "reward", "hrbp")):
273 return "HR & Recruitment"
274
275 if any(k in t for k in ("nurse", "doctor", "clinical", "pharmacist",
276 "physiotherapist", "occupational therap",
277 "healthcare", "medical", "gp ", "surgeon",
278 "dental", "radiographer", "midwife",
279 "care assistant", "support worker")):
280 return "Healthcare & Medical"
281
282 if any(k in t for k in ("teacher", "lecturer", "professor", "tutor",
283 "teaching assistant", "education", "headteacher",
284 "head of department", "send ", "pastoral")):
285 return "Education & Training"
286
287 if any(k in t for k in ("mechanical eng", "electrical eng", "civil eng",
288 "structural eng", "chemical eng", "process eng",
289 "manufacturing eng", "maintenance eng",
290 "building services", "quantity surveyor",
291 "site manager", "construction manager")):
292 return "Engineering"
293
294 if any(k in t for k in ("solicitor", "barrister", "paralegal", "legal counsel",
295 "legal advisor", "lawyer", "conveyancer",
296 "compliance officer", "regulatory")):
297 return "Legal & Compliance"
298
299 if any(k in t for k in ("supply chain", "logistics", "procurement",
300 "warehouse", "buyer", "purchasing",
301 "operations manager", "operations director",
302 "facilities", "fleet manager")):
303 return "Operations & Logistics"
304
305 if any(k in t for k in ("customer service", "customer support",
306 "call centre", "call center", "contact centre",
307 "client service", "customer success")):
308 return "Customer Service"
309
310 if any(k in t for k in ("admin", "office manager", "receptionist",
311 "personal assistant", "executive assistant",
312 "secretary", "office coordinator")):
313 return "Administration"
314
315 return "Other"
316
317
318def normalize_job(job: dict) -> dict:
319 """Ensure all job dicts have every expected field (no undefined in output)."""
320 defaults = {
321 "title": "",
322 "company": "",
323 "location": "",
324 "salary_raw": "",
325 "salary_min": None,
326 "salary_max": None,
327 "salary_currency": "",
328 "salary_period": "",
329 "snippet": "",
330
331 "employment_type": "",
332 "date_posted": "",
333 "valid_through": "",
334 "url": "",
335 "job_id": "",
336 "source": "",
337 "category": "",
338 }
339 normalized = {**defaults, **{k: v for k, v in job.items() if v is not None and v != ""}}
340
341
342
343
344 for key in ["title", "company", "location", "salary_raw", "snippet",
345 "employment_type", "date_posted",
346 "valid_through", "url", "job_id", "source", "category",
347 "salary_period", "salary_currency"]:
348 if normalized.get(key) is None:
349 normalized[key] = defaults[key]
350
351 if not normalized["category"]:
352 normalized["category"] = categorize_job(normalized["title"])
353
354 if not normalized["salary_currency"]:
355 normalized["salary_currency"] = SOURCE_CURRENCY.get(normalized["source"], "GBP")
356 return normalized
357
358
359
360RESOLVE_HARD_CAP = 200
361
362
363async def resolve_apply_urls(jobs: list[dict], client: httpx.AsyncClient) -> None:
364 """Opt-in (resolveApplyUrl): follow redirects on aggregator links to reveal
365 the real ATS/company apply URL, then re-detect the ATS. Bounded concurrency
366 + a hard cap so it never overwhelms the pool or blows up cost. Runs over the
367 un-proxied client and only on the final, already-truncated job set."""
368 targets = [j for j in jobs if j.get("url")][:RESOLVE_HARD_CAP]
369 if not targets:
370 return
371 Actor.log.info(f"[resolveApplyUrl] Resolving apply URLs for {len(targets)} jobs...")
372 sem = asyncio.Semaphore(15)
373
374 async def resolve(job: dict) -> None:
375 url = job.get("url")
376 async with sem:
377 final = None
378 try:
379 resp = await client.head(url, follow_redirects=True, timeout=15.0)
380
381 if resp.status_code == 405:
382 resp = await client.get(url, follow_redirects=True, timeout=15.0)
383 final = str(resp.url)
384 except Exception:
385 return
386 if final and final != url:
387 job["resolved_url"] = final
388 pipeline.redetect_ats(job)
389
390 await asyncio.gather(*[resolve(j) for j in targets], return_exceptions=True)
391
392
393def compute_salary_benchmarks(jobs: list[dict]) -> list[dict]:
394 """Compute salary benchmarks grouped by title keyword and location."""
395 from collections import defaultdict
396
397
398
399
400 buckets = defaultdict(list)
401 for job in jobs:
402 ann_min = job.get("salary_annual_min")
403 ann_max = job.get("salary_annual_max")
404
405 if not ann_min and not ann_max:
406 continue
407
408 mid = ((ann_min or ann_max) + (ann_max or ann_min)) / 2
409
410 if mid < 5000 or mid > 500000:
411 continue
412
413 title = job.get("title", "").lower().strip()
414 location = job.get("location", "").lower().strip()
415
416 loc_key = location.split(",")[0].strip() if location else "unknown"
417 buckets[(title, loc_key)].append(mid)
418
419 benchmarks = []
420 for (title, loc), salaries in buckets.items():
421 if len(salaries) < 2:
422 continue
423 salaries.sort()
424 benchmarks.append({
425 "benchmark_title": title,
426 "benchmark_location": loc,
427 "count": len(salaries),
428 "salary_mean": round(statistics.mean(salaries)),
429 "salary_median": round(statistics.median(salaries)),
430 "salary_p25": round(salaries[len(salaries) // 4]),
431 "salary_p75": round(salaries[(len(salaries) * 3) // 4]),
432 "salary_min": round(min(salaries)),
433 "salary_max": round(max(salaries)),
434 "_type": "salary_benchmark",
435 })
436
437 return sorted(benchmarks, key=lambda b: b["count"], reverse=True)
438
439
440async def main() -> None:
441 async with Actor:
442
443 actor_input = await Actor.get_input() or {}
444
445
446 keyword = actor_input.get("custom_keyword") or actor_input.get("keyword", "software engineer")
447 if keyword == "__custom__":
448 keyword = actor_input.get("custom_keyword", "software engineer")
449 location = actor_input.get("custom_location") or actor_input.get("location", "London")
450 if location == "__custom__":
451 location = actor_input.get("custom_location", "London")
452
453 unlimited = actor_input.get("unlimited", False)
454 max_results = 0 if unlimited else max(actor_input.get("max_results", 100), 100)
455 salary_min = actor_input.get("salary_min") or None
456 job_type = actor_input.get("job_type", "all")
457 country = actor_input.get("country", "uk")
458 salary_benchmark = actor_input.get("salary_benchmark", False)
459
460
461
462
463 search_terms = actor_input.get("searchTerms") or []
464 search_terms = [t.strip() for t in search_terms if t and t.strip()]
465 if not search_terms:
466 search_terms = [keyword]
467
468 search_terms = list(dict.fromkeys(search_terms))[:MAX_SEARCH_TERMS]
469
470 job_type_norm = actor_input.get("jobType", "any") or "any"
471 posted_within_hours = actor_input.get("postedWithinHours") or None
472 remote_only = actor_input.get("remoteOnly", False)
473 radius_miles = actor_input.get("radiusMiles") or None
474 description_format = actor_input.get("descriptionFormat", "markdown") or "markdown"
475 resolve_apply_url = actor_input.get("resolveApplyUrl", False)
476 incremental_only = actor_input.get("incrementalOnly", False)
477
478
479
480
481 if job_type == "all" and job_type_norm != "any":
482 job_type = {
483 "fulltime": "permanent", "parttime": "part-time",
484 "contract": "contract", "internship": "all",
485 }.get(job_type_norm, "all")
486
487
488 selected_boards = actor_input.get("boards") or COUNTRY_DEFAULTS.get(country, COUNTRY_DEFAULTS["uk"])
489
490
491 adzuna_app_id = actor_input.get("adzuna_app_id", "")
492 adzuna_app_key = actor_input.get("adzuna_app_key", "")
493
494
495 usajobs_api_key = actor_input.get("usajobs_api_key") or ""
496 usajobs_email = actor_input.get("usajobs_email") or ""
497
498
499 deduplicate = actor_input.get("deduplicate", True)
500
501 Actor.log.info(f"Starting multi-board scrape: {search_terms} in '{location}' (country={country})")
502 Actor.log.info(f"Boards: {selected_boards}")
503 Actor.log.info(f"Max results: {max_results or 'unlimited'} | Contract type: {job_type} | "
504 f"Job type: {job_type_norm} | Remote only: {remote_only} | "
505 f"Radius: {radius_miles or '-'} mi | Incremental: {incremental_only}")
506
507
508
509
510
511
512
513 effective_boards = list(selected_boards)
514 if "usajobs" in effective_boards and not usajobs_api_key:
515 effective_boards.remove("usajobs")
516 if "adzuna" in effective_boards and not (adzuna_app_id and adzuna_app_key):
517 effective_boards.remove("adzuna")
518 if len(effective_boards) != len(selected_boards):
519 skipped = sorted(set(selected_boards) - set(effective_boards))
520 Actor.log.info(
521 f"Budget adjustment: {skipped} will be skipped (no API key) and are "
522 f"excluded from the per-board budget — Max Results is divided across "
523 f"the {len(effective_boards)} board(s) that can actually return jobs.")
524 num_terms = len(search_terms)
525 divisor_boards = max(1, len(effective_boards))
526 if max_results == 0:
527 max_per_board = 10000
528 else:
529 max_per_board = max(10, max_results // (divisor_boards * max(1, num_terms)))
530
531
532 headers = make_headers()
533 proxy_config = None
534 http_proxy_url = None
535 is_on_apify = os.environ.get("APIFY_IS_AT_HOME", "0") == "1"
536 need_browser = any(b in BROWSER_BOARDS for b in selected_boards)
537
538
539
540
541 if is_on_apify and need_browser:
542 proxy_country = ADZUNA_COUNTRY_MAP.get(country, "GB").upper()
543 try:
544 proxy_config = await Actor.create_proxy_configuration(
545 groups=["RESIDENTIAL"],
546 country_code=proxy_country,
547 )
548 http_proxy_url = await proxy_config.new_url(
549 session_id=f"http_main_{random.randint(1000, 9999)}"
550 )
551 Actor.log.info(f"Using Apify residential proxy ({proxy_country}) for HTTP + browser")
552 except Exception as e:
553 Actor.log.warning(f"Proxy config failed: {e}")
554
555
556
557 if not need_browser:
558 try:
559 run_memory = int(os.environ.get("ACTOR_MEMORY_MBYTES")
560 or os.environ.get("APIFY_MEMORY_MBYTES") or 0)
561 except (TypeError, ValueError):
562 run_memory = 0
563 if run_memory >= 2048:
564 Actor.log.info(
565 f"COST TIP: this run uses no browser boards, so no Chromium is "
566 f"launched. It runs fine at 1024 MB — lower the memory in the "
567 f"run options to roughly halve the compute cost "
568 f"(currently {run_memory} MB).")
569
570
571
572
573
574
575 async with httpx.AsyncClient(
576 headers=headers,
577 timeout=30.0,
578 proxy=http_proxy_url,
579 ) as client, httpx.AsyncClient(
580 headers=headers,
581 timeout=30.0,
582 ) as api_client:
583
584
585 browser = None
586 playwright_instance = None
587
588 if need_browser:
589 Actor.log.info("Launching Playwright browser for JS-heavy boards...")
590 try:
591 playwright_instance = await async_playwright().start()
592
593
594
595 launch_kwargs = {
596 "headless": True,
597 "args": STEALTH_ARGS,
598 }
599 if proxy_config:
600
601 initial_proxy_url = await proxy_config.new_url(
602 session_id=f"browser_init_{random.randint(1000, 9999)}"
603 )
604 from urllib.parse import urlparse
605 parsed = urlparse(initial_proxy_url)
606 launch_kwargs["proxy"] = {
607 "server": f"{parsed.scheme}://{parsed.hostname}:{parsed.port}",
608 "username": parsed.username or "",
609 "password": parsed.password or "",
610 }
611 Actor.log.info(f"Browser launched with proxy: {parsed.hostname}:{parsed.port}")
612
613 browser = await playwright_instance.chromium.launch(**launch_kwargs)
614 Actor.log.info("Playwright browser launched successfully")
615 except Exception as e:
616 Actor.log.error(f"Failed to launch Playwright: {e}")
617 Actor.log.warning("Browser boards will fall back to httpx (may return fewer results)")
618
619 try:
620
621 browser_scrapers = []
622 api_scrapers = []
623 adzuna_country = ADZUNA_COUNTRY_MAP.get(country, "gb")
624
625 for board_name in selected_boards:
626 if board_name == "adzuna":
627 scraper = AdzunaScraper(
628 api_client,
629 delay=0.5,
630 app_id=adzuna_app_id,
631 app_key=adzuna_app_key,
632 country=adzuna_country,
633 )
634 api_scrapers.append(scraper)
635
636 elif board_name == "usajobs":
637 scraper = USAJobsScraper(
638 api_client,
639 delay=0.5,
640 api_key=usajobs_api_key,
641 email=usajobs_email,
642 )
643 api_scrapers.append(scraper)
644
645 elif board_name in BOARD_REGISTRY:
646 factory = BOARD_REGISTRY[board_name]
647
648 if board_name in BROWSER_BOARDS and browser and proxy_config:
649
650 safe_name = re.sub(r"[^a-zA-Z0-9._~]", "_", board_name)
651 proxy_url = await proxy_config.new_url(
652 session_id=f"jobs_{safe_name}_{random.randint(1000, 9999)}"
653 )
654 if callable(factory) and not isinstance(factory, type):
655 scraper = factory(client, delay=1.5, browser=browser,
656 proxy_url=proxy_url, proxy_config=proxy_config)
657 else:
658 scraper = factory(client, delay=1.5, browser=browser,
659 proxy_url=proxy_url, proxy_config=proxy_config)
660 browser_scrapers.append(scraper)
661 elif board_name in BROWSER_BOARDS and browser:
662 if callable(factory) and not isinstance(factory, type):
663 scraper = factory(client, delay=1.5, browser=browser)
664 else:
665 scraper = factory(client, delay=1.5, browser=browser)
666 browser_scrapers.append(scraper)
667 elif board_name in FREE_API_BOARDS:
668
669 scraper = factory(api_client, delay=0.5)
670 api_scrapers.append(scraper)
671 else:
672
673 extra = {}
674 if browser:
675 extra["browser"] = browser
676 if proxy_config:
677 safe_name = re.sub(r"[^a-zA-Z0-9._~]", "_", board_name)
678 purl = await proxy_config.new_url(
679 session_id=f"jobs_{safe_name}_{random.randint(1000, 9999)}"
680 )
681 extra["proxy_url"] = purl
682 extra["proxy_config"] = proxy_config
683 if callable(factory) and not isinstance(factory, type):
684 scraper = factory(client, delay=0.5, **extra)
685 else:
686 scraper = factory(client, delay=0.5, **extra)
687 api_scrapers.append(scraper)
688 else:
689 Actor.log.warning(f"Unknown board: {board_name}, skipping.")
690
691 all_scrapers = api_scrapers + browser_scrapers
692 if not all_scrapers:
693 Actor.log.error("No valid boards selected!")
694 return
695
696 all_jobs: list[dict] = []
697
698
699
700
701
702 for s in all_scrapers:
703 s.radius_miles = radius_miles
704 s.country_hint = country
705
706
707
708
709 UNLIMITED_BOARD_CAP = 2000
710 RAW_CAP = 0 if unlimited else max_results * 3
711 raw_jobs: list[dict] = []
712 board_attempts = 0
713 board_errors = 0
714 dead_browser_boards: set[str] = set()
715 unlimited_counts: dict[str, int] = {}
716
717 def board_limit(name: str) -> int:
718 if not unlimited:
719 return max_per_board
720
721 return max(0, UNLIMITED_BOARD_CAP - unlimited_counts.get(name, 0))
722
723 async def collect(scraper, term):
724 nonlocal board_attempts, board_errors
725 board_attempts += 1
726 lim = board_limit(scraper.source_name)
727 if lim <= 0:
728 return scraper.source_name, [], False
729 try:
730 jobs = await scraper.search(keyword=term, location=location,
731 max_results=lim, job_type=job_type,
732 salary_min=salary_min)
733 return scraper.source_name, jobs, False
734 except Exception as e:
735 board_errors += 1
736 Actor.log.error(f"[{scraper.source_name}] failed on '{term}': {e}")
737 return scraper.source_name, [], True
738
739
740
741
742
743 seen_source_urls: set[tuple[str, str]] = set()
744
745 def ingest(src, jobs):
746 fresh = []
747 for j in jobs:
748 url = j.get("url") or ""
749 if url:
750 key = (j.get("source") or src, url)
751 if key in seen_source_urls:
752 continue
753 seen_source_urls.add(key)
754 fresh.append(j)
755 if len(fresh) < len(jobs):
756 Actor.log.info(f"[{src}] {len(jobs) - len(fresh)} duplicate listing(s) "
757 f"already collected earlier in this run — skipped")
758 if unlimited:
759 unlimited_counts[src] = unlimited_counts.get(src, 0) + len(fresh)
760 raw_jobs.extend(normalize_job(j) for j in fresh)
761
762 for term in search_terms:
763 if RAW_CAP and len(raw_jobs) >= RAW_CAP:
764 Actor.log.info(f"Raw buffer cap ({RAW_CAP}) reached — stopping collection early.")
765 break
766 Actor.log.info(f"── Search term: '{term}' ──")
767
768 if api_scrapers:
769 results = await asyncio.gather(*[collect(s, term) for s in api_scrapers])
770 for src, jobs, _err in results:
771 Actor.log.info(f"[{src}] '{term}' → {len(jobs)} jobs")
772 ingest(src, jobs)
773
774 for scraper in browser_scrapers:
775 if scraper.source_name in dead_browser_boards:
776 continue
777 src, jobs, errored = await collect(scraper, term)
778 if errored or not jobs:
779 dead_browser_boards.add(scraper.source_name)
780 Actor.log.warning(f"[{src}] no results — skipping it for remaining terms")
781 Actor.log.info(f"[{src}] '{term}' → {len(jobs)} jobs")
782 ingest(src, jobs)
783
784 Actor.log.info(f"Collected {len(raw_jobs)} raw jobs across {len(search_terms)} term(s)")
785
786
787
788
789 if not raw_jobs and board_attempts > 0 and board_errors == board_attempts:
790 msg = (f"All {board_attempts} board attempts failed "
791 f"(network / anti-bot). No data was returned.")
792 Actor.log.error(msg)
793 if hasattr(Actor, "fail"):
794 await Actor.fail(status_message=msg)
795 else:
796 await Actor.set_status_message(msg)
797 raise RuntimeError(msg)
798 return
799
800
801 now = datetime.now(timezone.utc)
802 opts = {
803 "description_format": description_format,
804 "remote_only": remote_only,
805 "job_type_norm": job_type_norm,
806 "posted_within_hours": posted_within_hours,
807 "salary_min": salary_min,
808 }
809
810
811
812 for job in raw_jobs:
813 pipeline.enrich_job(job, opts, now)
814
815
816 drop_counts = {}
817 filtered = []
818 for j in raw_jobs:
819 reason = pipeline.filter_reason(j, opts)
820 if reason is None:
821 filtered.append(j)
822 else:
823 drop_counts[reason] = drop_counts.get(reason, 0) + 1
824 Actor.log.info(f"{len(filtered)}/{len(raw_jobs)} jobs pass filters "
825 f"(dropped: {drop_counts or 'none'})")
826 if raw_jobs and not filtered:
827 await Actor.set_status_message(
828 "Every collected job was removed by your filters — "
829 "try relaxing Remote Only / Job Type / Posted Within / Minimum Salary.")
830
831
832 merged = pipeline.merge_jobs(filtered, do_merge=deduplicate)
833 if deduplicate:
834 Actor.log.info(f"Merged {len(filtered)} → {len(merged)} records "
835 f"({len(filtered) - len(merged)} cross-board duplicates collapsed)")
836
837
838 merged.sort(key=lambda j: j.get("posted_at") or "", reverse=True)
839
840
841
842
843 if not unlimited and len(merged) > max_results:
844 merged = merged[:max_results]
845
846
847 if resolve_apply_url:
848 await resolve_apply_urls(merged, api_client)
849
850
851
852 for job in merged:
853 pipeline.derive_salary_insights(job)
854
855
856
857 store = seen_key = None
858 prior_seen = []
859 if incremental_only:
860
861
862
863 store = await Actor.open_key_value_store(name="uk-jobs-incremental")
864 seen_key = pipeline.incremental_key({
865 "search_terms": search_terms, "location": location,
866 "country": country, "boards": selected_boards,
867 })
868 prior_seen = await store.get_value(seen_key) or []
869 before = len(merged)
870 merged = pipeline.filter_unseen(merged, set(prior_seen))
871 Actor.log.info(f"Incremental: {len(merged)}/{before} jobs are new since last run")
872
873
874 all_jobs = [pipeline.finalize(j) for j in merged]
875 Actor.log.info(f"Pushing {len(all_jobs)} jobs to dataset...")
876 for i in range(0, len(all_jobs), 500):
877 await Actor.push_data(all_jobs[i:i + 500])
878
879
880
881
882 if incremental_only and store is not None:
883 new_seen = pipeline.updated_seen_list(prior_seen, all_jobs)
884 await store.set_value(seen_key, new_seen)
885 Actor.log.info(f"Incremental store '{seen_key}' now holds {len(new_seen)} fingerprints")
886
887
888 if salary_benchmark:
889 benchmarks = compute_salary_benchmarks(all_jobs)
890 Actor.log.info(f"Generated {len(benchmarks)} salary benchmarks")
891 for i in range(0, len(benchmarks), 500):
892 await Actor.push_data(benchmarks[i:i + 500])
893
894 finally:
895
896 if browser:
897 try:
898 await browser.close()
899 except Exception:
900 pass
901 if playwright_instance:
902 try:
903 await playwright_instance.stop()
904 except Exception:
905 pass
906
907
908 source_counts = {}
909 for job in all_jobs:
910 src = job.get("source", "unknown")
911 source_counts[src] = source_counts.get(src, 0) + 1
912
913 Actor.log.info(f"╔══════════════════════════════════════╗")
914 Actor.log.info(f"║ SCRAPE COMPLETE ║")
915 Actor.log.info(f"║ Total jobs: {len(all_jobs):<23}║")
916 for src, count in sorted(source_counts.items()):
917 Actor.log.info(f"║ {src}: {count:<26}║")
918 Actor.log.info(f"╚══════════════════════════════════════╝")
919
920
921if __name__ == "__main__":
922 asyncio.run(main())
