#!/usr/bin/env python3 """ Role Radar - TIER 1 detector. Polls each target company's ATS read-API (Greenhouse / Ashby / Lever), filters postings by role-family title patterns, diffs against last-seen state, and for each NEW matching role: saves the JD into the resume repo and records a row in the local tracker (CSV + Obsidian markdown). Writes nothing outward; this is detection only. Tier 2 (resume tailoring) is driven separately by the resume-system skill inside a Claude session. Stdlib only - no pip installs. Run under the repo .venv per AGENTS.md, or any Python 3.8+. Usage: python role_radar.py # full run (seed on first run, else incremental) python role_radar.py --dry-run # fetch + filter + print, write nothing python role_radar.py --company anthropic # limit to one company (repeatable via comma) python role_radar.py --dry-run --company ramp,tabs Statuses written to the tracker: backlog - matched on the first-ever (seeding) run; recorded, NOT auto-drafted new - matched, posted after seeding; an auto-draft candidate for Tier 2 (drafted / needs-input / applied / skipped are set later by Tier 2 / by Darsh and are never overwritten by this script) """ import argparse import csv import datetime as dt import html import json import os import re import sys import urllib.request import urllib.error from pathlib import Path HERE = Path(__file__).resolve().parent # .../jobs-resume-studio/role-radar ROOT = HERE.parent # .../jobs-resume-studio JD_ROOT = ROOT / "job-descriptions" COMPANIES_JSON = HERE / "companies.json" FILTERS_JSON = HERE / "filters.json" SEEN_JSON = HERE / "seen_jobs.json" TRACKER_CSV = HERE / "tracker.csv" TRACKER_MD = HERE / "tracker.md" INBOX_MD = HERE / "INBOX.md" TRACKER_COLUMNS = [ "date_found", "company", "role_title", "role_family", "location", "comp", "ats", "job_id", "job_url", "status", "jd_path", "resume_path", "notes", ] UA = {"User-Agent": "Mozilla/5.0 (role-radar; personal job watcher)"} TODAY = dt.date.today().isoformat() # --------------------------------------------------------------------------- # # HTTP + HTML helpers # --------------------------------------------------------------------------- # def fetch_json(url): req = urllib.request.Request(url, headers=UA) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode("utf-8", "replace")) def html_to_text(s): """Best-effort HTML -> readable plain text (for Greenhouse 'content').""" if not s: return "" s = html.unescape(s) # <p> ->

s = re.sub(r"(?i)", "\n", s) s = re.sub(r"(?i)", "\n\n", s) s = re.sub(r"(?i)]*>", "\n- ", s) s = re.sub(r"(?i)", "\n", s) s = re.sub(r"<[^>]+>", "", s) # strip remaining tags s = html.unescape(s) # any entities revealed s = re.sub(r"[ \t]{2,}", " ", s) s = re.sub(r"\n{3,}", "\n\n", s) return s.strip() # --------------------------------------------------------------------------- # # ATS adapters -> normalized job dict # {id, title, location, posted_at, url, apply_url, comp, department, jd_text} # --------------------------------------------------------------------------- # def adapter_greenhouse(slug): data = fetch_json( f"https://boards-api.greenhouse.io/v1/boards/{slug}/jobs?content=true" ) out = [] for j in data.get("jobs", []): depts = j.get("departments") or [] out.append({ "id": str(j.get("id")), "title": (j.get("title") or "").strip(), "location": ((j.get("location") or {}).get("name") or "").strip(), "posted_at": (j.get("first_published") or j.get("updated_at") or "")[:10], "url": j.get("absolute_url") or "", "apply_url": j.get("absolute_url") or "", "comp": "", "department": (depts[0].get("name") if depts else "") or "", "jd_text": html_to_text(j.get("content") or ""), }) return out def adapter_ashby(slug): data = fetch_json( f"https://api.ashbyhq.com/posting-api/job-board/{slug}?includeCompensation=true" ) out = [] for j in data.get("jobs", []): if j.get("isListed") is False: continue comp = "" c = j.get("compensation") or {} comp = (c.get("compensationTierSummary") or c.get("scrapeableCompensationSalarySummary") or "") out.append({ "id": str(j.get("id")), "title": (j.get("title") or "").strip(), "location": (j.get("location") or "").strip(), "posted_at": (j.get("publishedAt") or "")[:10], "url": j.get("jobUrl") or j.get("applyUrl") or "", "apply_url": j.get("applyUrl") or j.get("jobUrl") or "", "comp": comp, "department": (j.get("department") or j.get("team") or "") or "", "jd_text": (j.get("descriptionPlain") or html_to_text(j.get("descriptionHtml") or "")), }) return out def adapter_lever(slug): data = fetch_json(f"https://api.lever.co/v0/postings/{slug}?mode=json") out = [] for j in data: cats = j.get("categories") or {} created = j.get("createdAt") posted = "" if isinstance(created, (int, float)): posted = dt.datetime.utcfromtimestamp(created / 1000).date().isoformat() out.append({ "id": str(j.get("id")), "title": (j.get("text") or "").strip(), "location": (cats.get("location") or "").strip(), "posted_at": posted, "url": j.get("hostedUrl") or "", "apply_url": (j.get("applyUrl") or j.get("hostedUrl") or ""), "comp": (cats.get("commitment") or ""), "department": (cats.get("team") or cats.get("department") or "") or "", "jd_text": (j.get("descriptionPlain") or html_to_text(j.get("description") or "")), }) return out ADAPTERS = { "greenhouse": adapter_greenhouse, "ashby": adapter_ashby, "lever": adapter_lever, } # --------------------------------------------------------------------------- # # Filtering # --------------------------------------------------------------------------- # class Filters: def __init__(self, cfg): self.exclude = [re.compile(p, re.I) for p in cfg.get("exclude", [])] enabled = cfg.get("enabled_families") or list(cfg.get("families", {})) self.families = [] for fam in enabled: pats = cfg.get("families", {}).get(fam, []) self.families.append((fam, [re.compile(p, re.I) for p in pats])) def classify(self, title): """Return matching family name, or None if excluded / no match.""" if any(p.search(title) for p in self.exclude): return None for fam, pats in self.families: if any(p.search(title) for p in pats): return fam return None # --------------------------------------------------------------------------- # # Slugs, paths, tracker IO # --------------------------------------------------------------------------- # def slugify(text, maxlen=60): s = re.sub(r"[^a-z0-9]+", "_", (text or "").lower()).strip("_") return s[:maxlen].strip("_") or "role" def role_folder(company, title): return JD_ROOT / f"{slugify(company, 24)}_{slugify(title, 48)}" # folder/date -> job_id of the posting that owns plain "jd.md" this run, so two # distinct postings with the same title don't overwrite each other's JD. _JD_OWNER = {} def write_jd(company, comp_protected, job, family): folder = role_folder(company, job["title"]) / TODAY folder.mkdir(parents=True, exist_ok=True) owner = _JD_OWNER.setdefault(str(folder), job["id"]) fname = "jd.md" if owner == job["id"] else f"jd_{job['id']}.md" jd_path = folder / fname header = [ f"# JD - {job['title']} @ {company}", "", f"- **Company:** {company}" + (" *(PROTECTED - hold for review)*" if comp_protected else ""), f"- **Role:** {job['title']}", f"- **Role family:** {family}", f"- **Location:** {job['location'] or 'n/a'}", f"- **Comp:** {job['comp'] or 'n/a'}", f"- **Posted:** {job['posted_at'] or 'n/a'}", f"- **ATS job id:** {job['id']}", f"- **Listing:** {job['url']}", f"- **Apply:** {job['apply_url']}", f"- **Detected:** {TODAY}", "", "---", "", "## Job description", "", job["jd_text"] or "*(no description text returned by the ATS)*", "", ] jd_path.write_text("\n".join(header), encoding="utf-8") # store path relative to repo root for portability in the tracker return jd_path.relative_to(ROOT).as_posix() def load_tracker(): """Return (rows_by_jobkey, ordered_keys). Key = company::job_id.""" rows, order = {}, [] if TRACKER_CSV.exists(): with TRACKER_CSV.open(newline="", encoding="utf-8") as f: for row in csv.DictReader(f): key = f"{row.get('company')}::{row.get('job_id')}" rows[key] = row order.append(key) return rows, order def write_tracker(rows, order): with TRACKER_CSV.open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=TRACKER_COLUMNS, extrasaction="ignore") w.writeheader() for key in order: w.writerow(rows[key]) render_tracker_md(rows, order) def render_tracker_md(rows, order): status_rank = {"new": 0, "needs-input": 1, "backlog": 2, "drafted": 3, "applied": 4, "skipped": 5} keys = sorted(order, key=lambda k: (status_rank.get(rows[k].get("status"), 9), rows[k].get("date_found", "")), reverse=False) lines = [ "# Role Radar - Tracker", "", f"_Regenerated {TODAY}. Source of truth: `tracker.csv`. " "Set a row's status to `applied` or `skipped` yourself once you act on it — " "the detector never re-surfaces a job it has already seen._", "", "| Status | Company | Role | Family | Location | Comp | Found | Resume | Apply |", "|---|---|---|---|---|---|---|---|---|", ] for k in keys: r = rows[k] resume = f"`{r['resume_path']}`" if r.get("resume_path") else "—" apply = f"[link]({r['job_url']})" if r.get("job_url") else "—" title = r.get("role_title", "").replace("|", "\\|") lines.append( f"| {r.get('status','')} | {r.get('company','')} | {title} | " f"{r.get('role_family','')} | {r.get('location','') or '—'} | " f"{r.get('comp','') or '—'} | {r.get('date_found','')} | {resume} | {apply} |" ) lines.append("") TRACKER_MD.write_text("\n".join(lines), encoding="utf-8") def write_inbox(run_summary, rows, order): new_rows = [rows[k] for k in order if rows[k].get("status") == "new"] backlog = [rows[k] for k in order if rows[k].get("status") == "backlog"] needs = [rows[k] for k in order if rows[k].get("status") == "needs-input"] drafted = [rows[k] for k in order if rows[k].get("status") == "drafted"] def block(title, items): out = [f"### {title} ({len(items)})", ""] if not items: out.append("_none_") for r in items: out.append( f"- **{r['company']} — {r['role_title']}** " f"({r['role_family']}, {r.get('location') or 'n/a'}, {r.get('comp') or 'n/a'}) " f"\n JD: `{r.get('jd_path','')}` · [apply]({r.get('job_url','')})" + (f" · resume: `{r['resume_path']}`" if r.get("resume_path") else "") ) out.append("") return out lines = [ "# Role Radar - INBOX", "", f"_Last run: {dt.datetime.now().strftime('%Y-%m-%d %H:%M')}_", "", "**Per-company this run:** " + run_summary, "", "**What to do:** review `new` roles below; for each you want, ask Claude to " "*\"draft the resume for via resume-system\"* (Tier 2). " "Then apply yourself and set the tracker status to `applied`.", "", ] lines += block("🆕 New (auto-draft candidates)", new_rows) lines += block("⏳ Needs input (Tier 2 paused — JD needs evidence not in your repo)", needs) lines += block("✅ Drafted (resume ready to review & apply)", drafted) lines += block("📋 Backlog (already-open matches from seeding — draft on demand)", backlog) INBOX_MD.write_text("\n".join(lines), encoding="utf-8") # --------------------------------------------------------------------------- # # Main # --------------------------------------------------------------------------- # VALID_STATUSES = {"backlog", "new", "drafted", "needs-input", "applied", "skipped"} def do_mark(key, status, resume_path): """Safely set a tracker row's status (+ optional resume_path) and re-render. key = 'company::job_id' or 'company::' (first match).""" if status not in VALID_STATUSES: print(f"status must be one of {sorted(VALID_STATUSES)}", file=sys.stderr) return 2 rows, order = load_tracker() match = None if key in rows: match = key elif "::" in key: co, frag = key.split("::", 1) frag = frag.lower() for k in order: r = rows[k] if r["company"] == co and (frag == r["job_id"] or frag in r["role_title"].lower()): match = k break if not match: print(f"no tracker row matched '{key}'", file=sys.stderr) return 1 rows[match]["status"] = status if resume_path: rows[match]["resume_path"] = resume_path write_tracker(rows, order) write_inbox("(status update)", rows, order) r = rows[match] print(f"marked {r['company']} / {r['role_title']} -> {status}" + (f" resume={resume_path}" if resume_path else "")) return 0 def main(): ap = argparse.ArgumentParser(description="Role Radar Tier-1 detector") ap.add_argument("--dry-run", action="store_true", help="fetch + filter + print, write nothing") ap.add_argument("--company", default="", help="limit to these company names (comma-separated)") ap.add_argument("--mark", default="", help="set a row's status: --mark 'company::job_id_or_title' " "--status drafted [--resume PATH]") ap.add_argument("--status", default="", help="status for --mark") ap.add_argument("--resume", default="", help="resume_path for --mark") args = ap.parse_args() if args.mark: return do_mark(args.mark, args.status, args.resume) companies = json.loads(COMPANIES_JSON.read_text(encoding="utf-8"))["companies"] if args.company: want = {c.strip().lower() for c in args.company.split(",")} companies = [c for c in companies if c["name"].lower() in want] if not companies: print(f"No companies matched --company={args.company}", file=sys.stderr) return 2 filters = Filters(json.loads(FILTERS_JSON.read_text(encoding="utf-8"))) first_run = not SEEN_JSON.exists() seen = {} if first_run else json.loads(SEEN_JSON.read_text(encoding="utf-8")) rows, order = load_tracker() if first_run and not args.dry_run: print("FIRST RUN: seeding seen-state; current matches recorded as " "'backlog' (not auto-drafted).") summary_bits, total_new, total_backlog = [], 0, 0 for c in companies: name, platform, slug = c["name"], c["platform"], c["slug"] protected = bool(c.get("protected")) adapter = ADAPTERS.get(platform) if not adapter: print(f" ! {name}: unknown platform '{platform}'", file=sys.stderr) continue try: jobs = adapter(slug) except urllib.error.HTTPError as e: print(f" ! {name} ({platform}/{slug}): HTTP {e.code} — check slug", file=sys.stderr) continue except Exception as e: # noqa: BLE001 - keep the loop alive per company print(f" ! {name} ({platform}/{slug}): {e}", file=sys.stderr) continue seen_co = seen.setdefault(name, {}) n_match = n_new = n_back = 0 for j in jobs: fam = filters.classify(j["title"]) is_new_id = j["id"] not in seen_co if fam: n_match += 1 if args.dry_run: if fam: flag = "NEW" if is_new_id else "seen" print(f" [{name}] {flag:4s} {fam:20s} {j['title']}") continue if not is_new_id: continue # record the id as seen (matched or not) so 'new' = truly newly posted seen_co[j["id"]] = {"title": j["title"], "first_seen": TODAY} if not fam: continue status = "backlog" if first_run else "new" jd_rel = write_jd(name, protected, j, fam) note = "PROTECTED — hold for review" if protected else "" key = f"{name}::{j['id']}" rows[key] = { "date_found": TODAY, "company": name, "role_title": j["title"], "role_family": fam, "location": j["location"], "comp": j["comp"], "ats": platform, "job_id": j["id"], "job_url": j["url"], "status": status, "jd_path": jd_rel, "resume_path": "", "notes": note, } if key not in order: order.append(key) if status == "new": n_new += 1 else: n_back += 1 total_new += n_new total_backlog += n_back tag = (f"{n_match} match" + (f", {n_new} new" if n_new else "") + (f", {n_back} backlog" if n_back else "")) summary_bits.append(f"{name}: {tag}") print(f" {name:11s} {len(jobs):4d} jobs | {tag}") if args.dry_run: print("\n(dry-run: nothing written)") return 0 SEEN_JSON.write_text(json.dumps(seen, indent=2, ensure_ascii=False), encoding="utf-8") write_tracker(rows, order) write_inbox("; ".join(summary_bits), rows, order) print(f"\nWrote: seen_jobs.json, tracker.csv, tracker.md, INBOX.md") print(f"Totals this run: {total_new} new, {total_backlog} backlog.") if total_new: print("-> NEW roles found. Run Tier 2 (resume-system) to draft them.") return 0 if __name__ == "__main__": sys.exit(main())