feat: add boilerplate

app/linkedin.py

@@ -0,0 +1,141 @@
+import json
+import re
+
+import httpx
+from bs4 import BeautifulSoup
+
+from app.config import settings
+from app.models import Job
+from app.ral import detect_subject, detect_work_type, extract_salary, hide_salary
+
+
+def _text(el) -> str:
+    return el.get_text(" ", strip=True) if el else ""
+
+
+def _from_ld_json(data: dict) -> dict:
+    title = data.get("title") or ""
+    description = BeautifulSoup(data.get("description") or "", "lxml").get_text(" ", strip=True)
+    org = data.get("hiringOrganization") or {}
+    company = org.get("name") if isinstance(org, dict) else ""
+    loc = data.get("jobLocation") or {}
+    if isinstance(loc, list) and loc:
+        loc = loc[0]
+    location = ""
+    if isinstance(loc, dict):
+        address = loc.get("address") or {}
+        if isinstance(address, dict):
+            location = address.get("addressLocality") or address.get("addressRegion") or ""
+    extracted = {"ral_min": None, "ral_max": None, "ral_exact": None}
+    salary = data.get("baseSalary") or {}
+    value = salary.get("value") if isinstance(salary, dict) else {}
+    if isinstance(value, dict):
+        try:
+            if value.get("value"):
+                extracted["ral_exact"] = int(float(str(value["value"]).replace(",", "")))
+            if value.get("minValue"):
+                extracted["ral_min"] = int(float(str(value["minValue"]).replace(",", "")))
+            if value.get("maxValue"):
+                extracted["ral_max"] = int(float(str(value["maxValue"]).replace(",", "")))
+        except (TypeError, ValueError):
+            pass
+    return {
+        "title": title,
+        "description": description,
+        "company": company or "",
+        "location": location or "",
+        **extracted,
+    }
+
+
+def parse_job_html(html: str, url: str) -> dict:
+    soup = BeautifulSoup(html, "lxml")
+    title = _text(soup.select_one("h1")) or _text(soup.select_one(".top-card-layout__title"))
+    company = _text(soup.select_one(".topcard__org-name-link")) or _text(
+        soup.select_one(".top-card-layout__card .topcard__flavor")
+    )
+    location = _text(soup.select_one(".topcard__flavor--bullet"))
+    description = _text(soup.select_one(".show-more-less-html__markup")) or _text(
+        soup.select_one(".description__text")
+    )
+    extracted = {"ral_min": None, "ral_max": None, "ral_exact": None}
+
+    for script in soup.select('script[type="application/ld+json"]'):
+        try:
+            data = json.loads(script.text)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(data, list):
+            data = next(
+                (item for item in data if isinstance(item, dict) and item.get("@type") == "JobPosting"),
+                {},
+            )
+        if isinstance(data, dict) and data.get("@type") == "JobPosting":
+            parsed = _from_ld_json(data)
+            title = title or parsed["title"]
+            description = description or parsed["description"]
+            company = company or parsed["company"]
+            location = location or parsed["location"]
+            extracted = {k: parsed[k] for k in ("ral_min", "ral_max", "ral_exact")}
+            break
+
+    if not title:
+        og = soup.select_one('meta[property="og:title"]')
+        title = og["content"] if og and og.get("content") else "LinkedIn job"
+
+    blob = f"{title} {company} {description}"
+    if not any(extracted.values()):
+        extracted = extract_salary(blob)
+
+    extra_bits = []
+    if location:
+        extra_bits.append(f"Location: {location}")
+    extra_bits.extend(_text(el) for el in soup.select(".description__job-criteria-item") if _text(el))
+
+    has_salary = bool(extracted.get("ral_exact") or extracted.get("ral_min"))
+    spread = 0
+    if extracted.get("ral_max") and extracted.get("ral_min"):
+        spread = extracted["ral_max"] - extracted["ral_min"]
+
+    fields = {c.name for c in Job.__table__.columns}
+    payload = {
+        "linkedin_url": url,
+        "title": (title or "LinkedIn job")[:300],
+        "company": (company or "Unknown company")[:200],
+        "location": (location or "")[:200],
+        "description": description or "",
+        "description_clean": hide_salary(description or ""),
+        "extra_info": " | ".join(extra_bits)[:500],
+        "subject": detect_subject(title or "", description or ""),
+        "work_type": detect_work_type(blob),
+        "difficulty": 2 if spread > 15000 else 1,
+        "source": "linkedin",
+        "status": "ready" if has_salary else "needs_salary",
+        "fetch_error": "" if has_salary else "Salary not found; edit in admin",
+        **extracted,
+    }
+    return {k: v for k, v in payload.items() if k in fields}
+
+
+async def fetch_linkedin_job(url: str) -> dict:
+    if "linkedin.com" not in url:
+        raise ValueError("Only LinkedIn job URLs are supported")
+    headers = {
+        "User-Agent": settings.linkedin_user_agent,
+        "Accept-Language": "en-US,en;q=0.9,it;q=0.8",
+        "Accept": "text/html,application/xhtml+xml",
+    }
+    async with httpx.AsyncClient(follow_redirects=True, timeout=25.0, headers=headers) as client:
+        response = await client.get(url)
+        response.raise_for_status()
+        if "authwall" in str(response.url):
+            raise RuntimeError("LinkedIn showed a login wall for this URL")
+        html = response.text
+        if "/jobs/view/" in url and "JobPosting" not in html:
+            guest = url.split("?")[0]
+            response = await client.get(guest)
+            html = response.text
+    parsed = parse_job_html(html, url)
+    if not parsed.get("description") and not parsed.get("title"):
+        raise RuntimeError("Could not parse job posting from LinkedIn page")
+    return parsed