1"""Remote Job Feed Aggregator - one search across five public remote-job feeds."""
2
3from __future__ import annotations
4
5import asyncio
6from datetime import datetime, timedelta, timezone
7from typing import Any
8
9import httpx
10from apify import Actor
11
12from . import feeds
13
14USER_AGENT = (
15 'Mozilla/5.0 (compatible; RemoteJobFeedAggregator/0.1; '
16 '+https://apify.com/store) apify-actor'
17)
18RETRY_STATUS = {429, 500, 502, 503, 504}
19MAX_ATTEMPTS = 3
20
21
22async def get(client: httpx.AsyncClient, url: str) -> tuple[httpx.Response | None, str | None]:
23 delay = 1.0
24 for attempt in range(1, MAX_ATTEMPTS + 1):
25 try:
26 response = await client.get(url)
27 except httpx.HTTPError as exc:
28 if attempt == MAX_ATTEMPTS:
29 return None, f'network error: {type(exc).__name__}'
30 await asyncio.sleep(delay)
31 delay *= 2
32 continue
33 if response.status_code in RETRY_STATUS:
34 if attempt == MAX_ATTEMPTS:
35 return None, f'HTTP {response.status_code} after {MAX_ATTEMPTS} attempts'
36 await asyncio.sleep(delay)
37 delay *= 2
38 continue
39 if response.status_code >= 400:
40 return None, f'HTTP {response.status_code}'
41 return response, None
42 return None, 'exhausted retries'
43
44
45async def collect_remotive(client, pages, delay, include_description):
46 items, errors = [], []
47 response, error = await get(client, 'https://remotive.com/api/remote-jobs')
48 if error:
49 errors.append(f'remotive: {error}')
50 return items, errors
51 try:
52 items = feeds.parse_remotive(response.json(), include_description)
53 except ValueError:
54 errors.append('remotive: invalid JSON')
55 return items, errors
56
57
58async def collect_himalayas(client, pages, delay, include_description):
59 items, errors, cursor, offset = [], [], None, 0
60 for page in range(pages):
61
62 url = 'https://himalayas.app/jobs/api?limit=20'
63 url += f'&cursor={cursor}' if cursor else f'&offset={offset}'
64 response, error = await get(client, url)
65 if error:
66 errors.append(f'himalayas: {error}')
67 break
68 try:
69 payload = response.json()
70 except ValueError:
71 errors.append('himalayas: invalid JSON')
72 break
73 batch = feeds.parse_himalayas(payload, include_description)
74 items.extend(batch)
75 cursor = payload.get('nextCursor')
76 offset += 20
77 if not batch or (not cursor and offset >= (payload.get('totalCount') or 0)):
78 break
79 if page < pages - 1:
80 await asyncio.sleep(delay)
81 return items, errors
82
83
84async def collect_arbeitnow(client, pages, delay, include_description):
85 items, errors = [], []
86 for page in range(1, pages + 1):
87 response, error = await get(
88 client, f'https://www.arbeitnow.com/api/job-board-api?page={page}'
89 )
90 if error:
91 errors.append(f'arbeitnow: {error}')
92 break
93 try:
94 payload = response.json()
95 except ValueError:
96 errors.append('arbeitnow: invalid JSON')
97 break
98 batch = feeds.parse_arbeitnow(payload, include_description)
99 items.extend(batch)
100 if not batch:
101 break
102 if page < pages:
103 await asyncio.sleep(delay)
104 return items, errors
105
106
107async def collect_wwr(client, pages, delay, include_description):
108 items, errors = [], []
109 categories = feeds.WWR_CATEGORIES[: max(1, min(pages * 2, len(feeds.WWR_CATEGORIES)))]
110 for index, category in enumerate(categories):
111 response, error = await get(
112 client, f'https://weworkremotely.com/categories/{category}.rss'
113 )
114 if error:
115 errors.append(f'weworkremotely/{category}: {error}')
116 continue
117 items.extend(feeds.parse_wwr_rss(response.text, category, include_description))
118 if index < len(categories) - 1:
119 await asyncio.sleep(delay)
120 return items, errors
121
122
123async def collect_hn(client, pages, delay, include_description):
124 """Find the newest 'Ask HN: Who is hiring?' story, then read its comments."""
125 items, errors = [], []
126 response, error = await get(
127 client,
128 'https://hn.algolia.com/api/v1/search?tags=story,author_whoishiring'
129 '&query=who%20is%20hiring&hitsPerPage=1&restrictSearchableAttributes=title',
130 )
131 if error:
132 errors.append(f'hackernews: {error}')
133 return items, errors
134 try:
135 hits = response.json().get('hits') or []
136 except ValueError:
137 errors.append('hackernews: invalid JSON')
138 return items, errors
139 if not hits:
140 errors.append('hackernews: no "Who is hiring" thread found')
141 return items, errors
142 story_id = hits[0].get('objectID')
143 await asyncio.sleep(delay)
144 for page in range(pages):
145 response, error = await get(
146 client,
147 f'https://hn.algolia.com/api/v1/search_by_date?tags=comment,story_{story_id}'
148 f'&hitsPerPage=100&page={page}',
149 )
150 if error:
151 errors.append(f'hackernews: {error}')
152 break
153 try:
154 payload = response.json()
155 except ValueError:
156 errors.append('hackernews: invalid JSON')
157 break
158 batch = feeds.parse_hackernews(payload, include_description)
159 items.extend(batch)
160 if page + 1 >= (payload.get('nbPages') or 1) or not batch:
161 break
162 if page < pages - 1:
163 await asyncio.sleep(delay)
164 return items, errors
165
166
167COLLECTORS = {
168 'remotive': collect_remotive,
169 'himalayas': collect_himalayas,
170 'arbeitnow': collect_arbeitnow,
171 'weworkremotely': collect_wwr,
172 'hackernews': collect_hn,
173}
174
175
176PUSH_CHUNK = 250
177
178
179async def safe_push(rows: list[dict[str, Any]]) -> int:
180 """Push in chunks; if a chunk is rejected, salvage it row by row.
181
182 Apify validates every pushed item against `.actor/dataset_schema.json` and
183 rejects the WHOLE request if one item fails, so a single odd row from a feed
184 used to fail the entire run and save nothing. (It did: Himalayas publishes
185 hourly pay as a decimal, and 2 rows in 947 took the run down.) Now the run
186 keeps whatever is valid and says what it dropped.
187 """
188 pushed = 0
189 dropped = 0
190 for start in range(0, len(rows), PUSH_CHUNK):
191 chunk = rows[start:start + PUSH_CHUNK]
192 try:
193 await Actor.push_data(chunk)
194 pushed += len(chunk)
195 continue
196 except Exception as exc:
197 Actor.log.warning(
198 'A batch of %d row(s) was rejected (%s); retrying them one at a time.',
199 len(chunk), type(exc).__name__,
200 )
201 for row in chunk:
202 try:
203 await Actor.push_data(row)
204 pushed += 1
205 except Exception as exc:
206 dropped += 1
207 Actor.log.warning('Dropped 1 row the dataset would not accept (%s): %s',
208 type(exc).__name__, row.get('url') or row.get('jobId'))
209 if dropped:
210 Actor.log.warning('%d row(s) dropped, %d pushed.', dropped, pushed)
211 return pushed
212
213
214def matches(item: dict[str, Any], cfg: dict[str, Any]) -> bool:
215 title = (item.get('title') or '').lower()
216 if cfg['exclude'] and any(k in title for k in cfg['exclude']):
217 return False
218 if cfg['keywords']:
219 haystack = ' '.join(
220 str(x).lower()
221 for x in (
222 item.get('title'),
223 item.get('companyName'),
224 item.get('category'),
225 item.get('seniority'),
226 item.get('_excerpt'),
227 item.get('descriptionText'),
228 ' '.join(item.get('tags') or []),
229 )
230 if x
231 )
232 if not any(k in haystack for k in cfg['keywords']):
233 return False
234 if cfg['locations']:
235 haystack = ' '.join(
236 str(x).lower()
237 for x in ([item.get('locationText')] + list(item.get('locationRestrictions') or []))
238 if x
239 )
240 if not any(k in haystack for k in cfg['locations']):
241 return False
242 if cfg['cutoff'] is not None and item.get('postedAt'):
243 try:
244 if datetime.fromisoformat(item['postedAt']) < cfg['cutoff']:
245 return False
246 except ValueError:
247 pass
248 return True
249
250
251async def main() -> None:
252 async with Actor:
253 actor_input = await Actor.get_input() or {}
254
255 sources = [s for s in (actor_input.get('sources') or list(feeds.SOURCES)) if s in COLLECTORS]
256 if not sources:
257 sources = ['remotive', 'himalayas', 'arbeitnow', 'weworkremotely']
258 max_items = int(actor_input.get('maxItems') or 200)
259 pages = int(actor_input.get('maxPagesPerSource') or 3)
260 delay = float(actor_input.get('requestDelaySeconds') or 1)
261 include_description = bool(actor_input.get('includeDescription', False))
262 deduplicate = bool(actor_input.get('deduplicate', True))
263 posted_within = int(actor_input.get('postedWithinDays') or 0)
264
265 cfg = {
266 'keywords': [k.lower() for k in (actor_input.get('keywords') or []) if k],
267 'exclude': [k.lower() for k in (actor_input.get('excludeKeywords') or []) if k],
268 'locations': [k.lower() for k in (actor_input.get('locationKeywords') or []) if k],
269 'cutoff': (
270 datetime.now(timezone.utc) - timedelta(days=posted_within)
271 if posted_within > 0
272 else None
273 ),
274 }
275
276 Actor.log.info('Querying %d feed(s): %s', len(sources), ', '.join(sources))
277
278 collected: list[dict[str, Any]] = []
279 errors: list[str] = []
280 per_source: dict[str, int] = {}
281 timeout = httpx.Timeout(60.0, connect=15.0)
282 headers = {'User-Agent': USER_AGENT, 'Accept': 'application/json, application/xml, */*'}
283
284 async with httpx.AsyncClient(
285 timeout=timeout, headers=headers, follow_redirects=True
286 ) as client:
287 for index, source in enumerate(sources):
288 if index:
289 await asyncio.sleep(delay)
290 try:
291 items, source_errors = await COLLECTORS[source](
292 client, pages, delay, include_description
293 )
294 except Exception as exc:
295 Actor.log.exception('Feed %s failed', source)
296 errors.append(f'{source}: unexpected {type(exc).__name__}')
297 continue
298 errors.extend(source_errors)
299 per_source[source] = len(items)
300 collected.extend(items)
301 Actor.log.info('%s -> %d job(s) returned', source, len(items))
302
303 kept = [item for item in collected if matches(item, cfg)]
304 Actor.log.info('%d of %d job(s) match the filters', len(kept), len(collected))
305
306 if deduplicate:
307 merged: dict[str, dict[str, Any]] = {}
308 for item in kept:
309 key = feeds.dedup_key(item)
310 if not key:
311 continue
312 if key in merged:
313 if item['source'] not in merged[key]['alsoSeenOn']:
314 merged[key]['alsoSeenOn'].append(item['source'])
315 else:
316 merged[key] = item
317 kept = list(merged.values())
318 Actor.log.info('%d job(s) after deduplication', len(kept))
319
320 kept.sort(key=lambda x: x.get('postedAt') or '', reverse=True)
321 kept = kept[:max_items]
322
323 now = datetime.now(timezone.utc).isoformat()
324 for item in kept:
325 item.pop('_excerpt', None)
326 item['scrapedAt'] = now
327
328 if kept:
329 await safe_push(kept)
330 else:
331 Actor.log.warning(
332 'No job matched. Try fewer keywords, more sources, or a larger '
333 '"Max pages per source".'
334 )
335
336 await Actor.set_value('RUN_SUMMARY', {
337 'sources': sources,
338 'returnedPerSource': per_source,
339 'matched': len(kept),
340 'errors': errors,
341 'finishedAt': now,
342 })
343 Actor.log.info('Done. Pushed %d job(s). %d feed error(s).', len(kept), len(errors))
