#!/usr/bin/env python3 """Apify job-discovery source for the daily job sweep. Runs the Apify actor curious_coder/linkedin-jobs-scraper (pay-per-result, $1.00 / 1,000 results) against LinkedIn public job search, one actor run per lane defined in .apify_config.json, and writes normalized results to .apify_results_YYYY-MM-DD.json in this folder. DISCOVERY ONLY. This script never touches the tracker. Every candidate here is UNVERIFIED until the sweep confirms it on the company's own careers site (the daily-job-sweep verification bar stays authoritative). Usage: python3 apify_jobs.py # real run python3 apify_jobs.py --dry-run # print planned queries + cost, no API calls python3 apify_jobs.py --count N # override per-lane result count (min 10) Stdlib only. Exits non-zero with a clear message on auth/credit/network errors. """ import json import ssl import sys import time import urllib.error import urllib.parse import urllib.request from datetime import date, datetime, timedelta, timezone from pathlib import Path def _jobdash_data_dir(_default): """Pi deployment: use the Nextcloud data-dir copy of the Job Search folder when jobdash_pi.json (written by deploy_to_pi.py) sits next to this script.""" try: import json as _json return Path(_json.loads( (_default / "jobdash_pi.json").read_text())["data_dir"]) except Exception: return _default HERE = _jobdash_data_dir(Path(__file__).resolve().parent) CONFIG_PATH = HERE / ".apify_config.json" LOG_PATH = HERE / ".apify_log.txt" API = "https://api.apify.com/v2" ACTOR_MIN_COUNT = 10 # actor input schema minimum SSL_REMEDY = ("no usable CA bundle for SSL verification — " "run: /Applications/Python*/Install Certificates.command " "or pip3 install certifi") def die(msg, code=1): print(f"ERROR: {msg}", file=sys.stderr) sys.exit(code) def _log(msg): try: with open(LOG_PATH, "a") as f: f.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} [scrape] {msg}\n") except Exception: pass _SSL_CTX = None def get_ssl_context(): """Verified SSL context, tried in order: (1) certifi bundle if importable, (2) macOS system bundle /etc/ssl/cert.pem, (3) stdlib default context. NEVER falls back to unverified SSL — these requests carry the API token. If nothing works, logs a one-line remedy to .apify_log.txt and exits.""" global _SSL_CTX if _SSL_CTX is not None: return _SSL_CTX try: import certifi _SSL_CTX = ssl.create_default_context(cafile=certifi.where()) return _SSL_CTX except Exception: pass try: if Path("/etc/ssl/cert.pem").exists(): _SSL_CTX = ssl.create_default_context(cafile="/etc/ssl/cert.pem") return _SSL_CTX except Exception: pass try: _SSL_CTX = ssl.create_default_context() return _SSL_CTX except Exception: _log(SSL_REMEDY) die(SSL_REMEDY) def api_call(path, token, method="GET", body=None, timeout=60): url = f"{API}{path}{'&' if '?' in path else '?'}token={token}" data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout, context=get_ssl_context()) as r: return json.loads(r.read().decode()) except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace")[:500] if e.code == 401: die("Apify token rejected (401). Fix: apify.com > Settings > API & Integrations, " "generate a new token and update .apify_config.json.") if e.code == 402 or "usage-limit" in detail or "not enough" in detail.lower(): die("Apify credit exhausted (402). Fix: wait for the free $5 monthly credit to reset " "or add a payment method at apify.com > Billing.") die(f"Apify API HTTP {e.code} on {path}: {detail}") except urllib.error.URLError as e: if isinstance(getattr(e, "reason", None), ssl.SSLCertVerificationError): _log(SSL_REMEDY) die(f"SSL certificate verification failed reaching api.apify.com " f"({e.reason}). {SSL_REMEDY}") die(f"Cannot reach api.apify.com ({e.reason}). If this is a sandboxed/proxied " "environment, api.apify.com may be blocked; run this script directly on the Mac.") def build_search_url(q, posted_within_seconds): params = {"keywords": q["keywords"], "location": q["location"], "f_TPR": f"r{posted_within_seconds}"} if q.get("remote"): params["f_WT"] = "2" if q.get("company_ids"): params["f_C"] = ",".join(q["company_ids"]) return "https://www.linkedin.com/jobs/search/?" + urllib.parse.urlencode(params) def first(*vals): for v in vals: if v: return v return "" def normalize(item, lane): salary = item.get("salary") or item.get("salaryInfo") or "" if isinstance(salary, list): salary = " - ".join(str(s) for s in salary if s) return { "lane": lane, "title": first(item.get("title")), "company": first(item.get("companyName"), item.get("company")), "location": first(item.get("location")), "salary": salary, "url": first(item.get("link"), item.get("url"), item.get("jobUrl"), item.get("applyUrl")), "posted_date": first(item.get("postedAt"), item.get("postedDate"), item.get("publishedAt")), "source": "apify:linkedin (UNVERIFIED - verify on company site before tracker entry)", } def excluded(job, keywords): t = (job["title"] or "").lower() return any(k in t for k in keywords) def main(): dry = "--dry-run" in sys.argv count_override = None if "--count" in sys.argv: count_override = max(ACTOR_MIN_COUNT, int(sys.argv[sys.argv.index("--count") + 1])) if not CONFIG_PATH.exists(): die(f"missing {CONFIG_PATH.name} next to this script") cfg = json.loads(CONFIG_PATH.read_text()) token = cfg["token"] actor = cfg.get("actor_id") or cfg["actors"][0].replace("/", "~") price = cfg.get("price_per_1000_results_usd", 1.0) posted_within = cfg.get("posted_within_seconds", 259200) cap = cfg.get("max_results_per_query", 50) excl = [k.lower() for k in cfg.get("exclude_title_keywords", [])] # Plan runs: one actor run per lane, all lane query URLs in a single run. plans = [] for lane in cfg["lanes"]: count = count_override or lane.get("count", 40) count = max(ACTOR_MIN_COUNT, min(count, cap)) # budget clamp urls = [build_search_url(q, posted_within) for q in lane["queries"]] plans.append({"lane": lane["name"], "label": lane.get("label", lane["name"]), "count": count, "urls": urls}) total_requested = sum(p["count"] for p in plans) est_cost = total_requested * price / 1000.0 max_cost = cfg.get("max_cost_per_run_usd", 0.15) print(f"Planned: {len(plans)} actor run(s), <= {total_requested} results, " f"est. max cost ${est_cost:.3f} (guard: ${max_cost:.2f})") if est_cost > max_cost: die(f"budget guard tripped: estimated ${est_cost:.3f} > max_cost_per_run_usd " f"${max_cost:.2f}. Lower lane counts in .apify_config.json.") if dry: for p in plans: print(f"\n[{p['lane']}] {p['label']} (count {p['count']})") for u in p["urls"]: print(" " + u) return # Start all lane runs. for p in plans: body = {"urls": p["urls"], "count": p["count"], "scrapeCompany": cfg.get("scrape_company_details", False)} run = api_call(f"/acts/{actor}/runs", token, "POST", body)["data"] p["run_id"], p["dataset_id"], p["status"] = run["id"], run["defaultDatasetId"], run["status"] print(f"[{p['lane']}] started run {p['run_id']}") # Poll to completion. deadline = time.time() + cfg.get("run_timeout_seconds", 600) pending = {p["run_id"] for p in plans} while pending and time.time() < deadline: time.sleep(10) for p in plans: if p["run_id"] not in pending: continue p["status"] = api_call(f"/actor-runs/{p['run_id']}", token)["data"]["status"] if p["status"] not in ("READY", "RUNNING"): pending.discard(p["run_id"]) print(f"[{p['lane']}] {p['status']}") for p in plans: if p["status"] in ("READY", "RUNNING"): print(f"WARNING: [{p['lane']}] run {p['run_id']} still running at timeout; " f"fetching whatever is in the dataset so far.") # Collect + normalize. jobs, seen = [], set() for p in plans: items = api_call(f"/datasets/{p['dataset_id']}/items?clean=true&format=json", token) or [] p["raw_results"] = len(items) kept = 0 for it in items: j = normalize(it, p["lane"]) key = j["url"].split("?")[0] or (j["title"] + "|" + j["company"]) if not j["title"] or key in seen or excluded(j, excl): continue seen.add(key) jobs.append(j) kept += 1 p["kept_results"] = kept total_raw = sum(p["raw_results"] for p in plans) actual_cost = total_raw * price / 1000.0 out = { "generated_at": datetime.now(timezone.utc).isoformat(), "actor": cfg["actors"][0], "note": "DISCOVERY ONLY - unverified. Sweep must verify each job on the company's own careers site.", "runs": [{k: p[k] for k in ("lane", "label", "run_id", "status", "raw_results", "kept_results")} for p in plans], "total_raw_results": total_raw, "estimated_cost_usd": round(actual_cost, 4), "jobs": jobs, } out_path = HERE / f".apify_results_{date.today().isoformat()}.json" out_path.write_text(json.dumps(out, indent=2)) print(f"\nWrote {len(jobs)} jobs ({total_raw} raw) to {out_path.name}; " f"cost ~= ${actual_cost:.3f}") for p in plans: print(f" {p['lane']}: {p['kept_results']} kept / {p['raw_results']} raw ({p['status']})") prune_old_results() def prune_old_results(days=7): """Delete .apify_results_YYYY-MM-DD.json older than `days` days, so the folder doesn't accumulate one file per day forever. Best-effort.""" cutoff = (date.today() - timedelta(days=days)).isoformat() for p in HERE.glob(".apify_results_*.json"): stamp = p.name[len(".apify_results_"):-len(".json")] if stamp < cutoff: try: p.unlink() _log(f"pruned old results file {p.name}") except OSError: pass if __name__ == "__main__": main()