142 lines
5.7 KiB
Python
142 lines
5.7 KiB
Python
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
|