Upwork - Public Job Posts (title, budget, experience, skills, client country, proposals)

import json, time
from urllib.parse import quote, urlsplit
 
BASE = "https://www.upwork.com"
WRAPPERS = ("ShallowReactive", "Reactive", "Ref", "ShallowRef")
STATE = ("JSON.stringify([location.pathname, document.readyState, document.title,"
         " (performance.getEntriesByType('navigation')[0] || {}).responseStatus || 0,"
         " !!document.getElementById('__NUXT_DATA__')])")
 
class Blocked(Exception):
    """Challenge page, 429, or repeated 403s: stop the run and back off. Never retried."""
 
def text(v):
    return v if isinstance(v, str) and v else None
 
def num(v):
    return v if isinstance(v, (int, float)) and not isinstance(v, bool) else None
 
def dget(d, *keys):
    for k in keys:
        d = d.get(k) if isinstance(d, dict) else None
    return d
 
def decode_nuxt(raw):
    """Decode Nuxt 3 __NUXT_DATA__ (devalue format: one flat array, dict/list values are indexes into it)."""
    flat, memo = json.loads(raw), {}
    def ref(i):
        if not isinstance(i, int) or isinstance(i, bool) or not 0 <= i < len(flat):   # -1 undefined, -2 null
            return None
        if i not in memo:
            v = flat[i]
            if isinstance(v, list):
                memo[i] = ref(v[1]) if len(v) == 2 and v[0] in WRAPPERS else [ref(x) for x in v]
            elif isinstance(v, dict):
                memo[i] = {k: ref(x) for k, x in v.items()}
            else:
                memo[i] = v
        return memo[i]
    return ref(0)
 
_tab = {"open": False}
 
def open_page(url, timeout=30):
    """Navigate, wait for THIS url's complete document, return (http_status, payload or None)."""
    if _tab["open"]:
        goto_url(url)
    else:
        new_tab(url)
        _tab["open"] = True
    want = urlsplit(url).path.rstrip("/")
    deadline = time.time() + timeout
    while time.time() < deadline:
        time.sleep(0.5)
        try:
            path, ready, title, status, has_data = json.loads(js(STATE))
        except (TimeoutError, RuntimeError) as e:   # js() can time out while a large page parses
            if isinstance(e, TimeoutError) or "timed out" in str(e):
                continue
            raise
        if path.rstrip("/") != want or ready != "complete":
            continue
        if status == 429 or "Just a moment" in title or (status == 403 and not has_data):
            raise Blocked(f"HTTP {status} {title!r} at {url}")   # challenge or rate limit: stop
        raw = js("(document.getElementById('__NUXT_DATA__') || {}).textContent || ''") if has_data else ""
        return status, decode_nuxt(raw) if isinstance(raw, str) and raw else None
    raise RuntimeError(f"{url} did not load in {timeout} s; tab is at {js('location.href')}")

TIER = {1: "Entry level", 2: "Intermediate", 3: "Expert"}   # 0 = not specified
JOB_TYPE = {1: "Fixed-price", 2: "Hourly"}
 
def skill_page(slug):
    """One /freelance-jobs/<slug>/ page: its 30 jobs + the category's open-job total. Raises on failure."""
    url = f"{BASE}/freelance-jobs/{quote(slug, safe='')}/"
    status, payload = open_page(url)
    if status != 200:
        raise RuntimeError(f"HTTP {status} on {url}")
    for v in (dget(payload, "data") or {}).values():
        if isinstance(v, dict) and isinstance(v.get("jobs"), list) and "totalJobs" in v:
            return v["jobs"], num(v["totalJobs"])
    raise RuntimeError(f"no job list in __NUXT_DATA__ on {url}")
 
def job_detail(path):
    """Returns (jobDetails dict, error string or None). A miss is explicit, never silent."""
    try:
        status, payload = open_page(BASE + quote(path, safe="/%"))
    except RuntimeError as e:   # page did not finish loading; Blocked is not caught here
        return {}, f"did not load: {e}"
    if status != 200:
        return {}, f"HTTP {status}"
    jd = dget(payload, "vuex", "jobDetails")
    return (jd, None) if isinstance(dget(jd, "job"), dict) else ({}, "no jobDetails in page")
 
def iso_utc(v):
    s = text(v)   # detail uses "...Z"; the card fallback uses "+0000"
    return s[:-5] + "Z" if s and s.endswith("+0000") else s
 
def to_record(card, jd, err):
    job, buyer = dget(jd, "job") or {}, dget(jd, "buyer") or {}
    jtype = num(job.get("type")) or num(card.get("type"))
    tier = num(job.get("contractorTier"))
    tier = num(card.get("contractorTier")) if tier is None else tier
    fixed = num(dget(job, "budget", "amount"))
    fixed = num(dget(card, "amount", "amount")) if fixed is None else fixed
    skills = [dget(s, "prefLabel") for s in card.get("ontologySkills") or []]
    skills += [dget(s, "name") for s in dget(jd, "sands", "additionalSkills") or []]
    return {
        "id": text(card.get("ciphertext")),
        "title": text(job.get("title")) or text(card.get("title")),
        "posted_on": iso_utc(job.get("postedOn")) or iso_utc(card.get("createdOn")),
        "posted_ago": text(card.get("ago")),
        "job_type": JOB_TYPE.get(jtype),
        "fixed_budget": fixed if jtype == 1 else None,
        "hourly_min": num(dget(job, "extendedBudgetInfo", "hourlyBudgetMin")) if jtype == 2 else None,
        "hourly_max": num(dget(job, "extendedBudgetInfo", "hourlyBudgetMax")) if jtype == 2 else None,
        "experience_level": TIER.get(tier),
        "skills": list(dict.fromkeys(s for s in skills if text(s))),
        "client_country": text(dget(buyer, "location", "country")),
        "proposals": num(dget(job, "clientActivity", "totalApplicants")),
        "workload": text(job.get("workload")) or text(card.get("engagement")),
        "duration": text(job.get("durationLabel")) or text(card.get("durationLabel")),
        "url": BASE + card["url"],
        "detail_error": err,
    }
 
SKILLS = ["web-scraping", "python"]      # each slug = 1 listing page + up to 30 detail pages
rows, seen, streak, stopped = [], set(), 0, None
try:
    for slug in SKILLS:
        cards, total = skill_page(slug)  # raises if the listing cannot be read
        print(f"{slug}: {len(cards)} cards of {total} open jobs (no page 2 exists; the rest are not reachable logged out)")
        time.sleep(2)
        for card in cards:
            if not isinstance(card, dict) or not text(card.get("url")):
                print("  card without url, skipped:", card)
                continue
            key = text(card.get("ciphertext")) or card["url"]   # never drop a card just because its id is missing
            if key in seen:
                continue
            seen.add(key)
            jd, err = job_detail(card["url"])
            if err:
                print("  detail failed:", err, card["url"])
            rows.append(to_record(card, jd, err))
            streak = streak + 1 if err == "HTTP 403" else 0
            if streak >= 2:              # one 403 = private listing; two in a row = treat as a block
                raise Blocked(f"2 detail pages in a row returned 403 (last: {card['url']})")
            time.sleep(2)                # polite pacing: ~1 page per 2-4 s
except Blocked as e:
    stopped = e
    print("STOPPED, back off before retrying:", e)
 
OUT = "upwork_jobs.partial.jsonl" if stopped else "upwork_jobs.jsonl"   # a stopped run never writes the full-run name
with open(OUT, "w") as f:
    for r in rows:
        f.write(json.dumps(r) + "\n")
print(len(rows), "jobs saved to", OUT + ";", sum(r["detail_error"] is not None for r in rows), "without detail data")
print(json.dumps(rows[:2], indent=1))
print("missing country:", sum(r["client_country"] is None for r in rows),
      "missing proposals:", sum(r["proposals"] is None for r in rows))
if stopped:
    print(f"PARTIAL: the run stopped early; {len(rows)} jobs in {OUT} are not all of {SKILLS}")
    raise stopped

web-scraping: 30 cards of 291 open jobs (no page 2 exists; the rest are not reachable logged out)
python: 30 cards of 1631 open jobs (no page 2 exists; the rest are not reachable logged out)
  detail failed: HTTP 403 /freelance-jobs/apply/Port-Viva-com-payment-module-Odoo-and-add-ISV-key_~022113260749740632003/
57 jobs saved to upwork_jobs.jsonl; 1 without detail data
[
 {
  "id": "~022112614550818987209",
  "title": "E-commerce Product Monitoring System",
  "posted_on": "2026-10-20T18:38:54.416Z",
  "posted_ago": "3 days ago",
  "job_type": "Fixed-price",
  "fixed_budget": 100,
  "hourly_min": null,
  "hourly_max": null,
  "experience_level": "Intermediate",
  "skills": ["Selenium", "Data Scraping", "HTML", "Python", "Web Scraping", "Web Scraping Framework", "Playwright", "Data Extraction", "AI Agents", "pandas"],
  "client_country": "IND",
  "proposals": 26,
  "workload": null,
  "duration": "1 to 3 months",
  "url": "https://www.upwork.com/freelance-jobs/apply/commerce-Product-Monitoring-System_~022112614550818987209/",
  "detail_error": null
 },
 ...
]
missing country: 1 missing proposals: 1

ROBOTS_DISALLOWED = {"fiction-writing"}   # robots.txt: Disallow: /freelance-jobs/fiction-writing/
 
status, payload = open_page(f"{BASE}/freelance-jobs/")
skill_pages = None
for v in (dget(payload, "data") or {}).values():
    if isinstance(v, dict) and "categories" in v and isinstance(v.get("jobs"), list):
        skill_pages = [p for p in v["jobs"] if isinstance(p, dict) and text(p.get("id"))]
if not skill_pages:
    raise RuntimeError(f"HTTP {status}: skill list not found in __NUXT_DATA__ on {BASE}/freelance-jobs/")
print(len(skill_pages), "skill listing pages")
 
def find_skill_slugs(keyword, max_slugs=5):
    """Capped: each slug costs block 1 up to 31 page loads (~2 min at 2 s pacing)."""
    kw = keyword.lower().strip()
    return [p["id"] for p in skill_pages
            if p["id"] not in ROBOTS_DISALLOWED
            and (kw in p["id"].replace("-", " ") or kw in (text(p.get("title")) or "").lower())][:max_slugs]
 
for kw in ["scraping", "react", "data entry", "python", "design"]:
    print(kw, "->", find_skill_slugs(kw))

482 skill listing pages
scraping -> ['data-scraping', 'web-scraping']
react -> ['react-js', 'react-native']
data entry -> ['data-entry']
python -> ['python-script', 'python']
design -> ['visual-design', 'packaging-design', 'social-media-design', 'book-design', 'adobe-indesign']