import json
import os
import re
from datetime import datetime, timezone
from nstdata_ai_crawl import (
Format,
NstDataClient,
ScrapeRequestDto,
)
job_urls = [
"https://careers.example.com/jobs/123",
"https://careers.example.com/jobs/456",
]
def first_heading(markdown: str) -> str | None:
for line in markdown.splitlines():
match = re.match(r"^\s*#{1,3}\s+(.+?)\s*#*\s*$", line)
if match:
return match.group(1).strip()
return None
def labeled_value(markdown: str, labels: list[str]) -> str | None:
"""Example for pages that expose fields as 'Location: ...'."""
for raw_line in markdown.splitlines():
line = re.sub(r"[*_]", "", raw_line).strip()
line = re.sub(r"^[-*]\s+", "", line)
for label in labels:
prefix = f"{label}:"
if line.casefold().startswith(prefix.casefold()):
return line[len(prefix):].strip() or None
return None
def parse_job(url: str, markdown: str) -> dict:
return {
"source_url": url,
"observed_at": datetime.now(timezone.utc).isoformat(),
# Validate this candidate against the target site's page layout.
"title_candidate": first_heading(markdown),
"employer": labeled_value(markdown, ["Company", "Employer"]),
"location": labeled_value(markdown, ["Location"]),
"employment_type": labeled_value(
markdown, ["Employment type", "Job type"]
),
"salary_text": labeled_value(markdown, ["Salary", "Compensation"]),
# Preserve the original text so downstream parsing can be corrected.
"description_markdown": markdown,
}
records = []
failures = []
with NstDataClient(os.environ["NSTDATA_API_TOKEN"]) as client:
for url in job_urls:
try:
result = client.submit_scrape_task_sync(
ScrapeRequestDto(
url=url,
formats=[Format.MARKDOWN],
timeout=60_000,
onlyMainContent=True,
)
)
markdown = result.data.get_markdown()
if not markdown or not markdown.strip():
failures.append({"source_url": url, "reason": "empty_markdown"})
continue
records.append(parse_job(url, markdown))
except Exception as exc:
failures.append({
"source_url": url,
"error_type": type(exc).__name__,
})
with open("jobs.jsonl", "w", encoding="utf-8") as output_file:
for record in records:
output_file.write(json.dumps(record, ensure_ascii=False) + "\n")
with open("job_scrape_failures.jsonl", "w", encoding="utf-8") as error_file:
for failure in failures:
error_file.write(json.dumps(failure, ensure_ascii=False) + "\n")
print(f"已保存 {len(records)} 条记录；{len(failures)} 个页面需要审核。")
