diff --git a/src/nerajob/scrapers/registry.py b/src/nerajob/scrapers/registry.py index c0c2bf2..7f5e56b 100644 --- a/src/nerajob/scrapers/registry.py +++ b/src/nerajob/scrapers/registry.py @@ -15,6 +15,8 @@ from nerajob.scrapers.sample import SampleScraper from nerajob.scrapers.smartrecruiters import SmartRecruitersScraper from nerajob.scrapers.themuse import TheMuseScraper +from nerajob.scrapers.topcv import TopCVScraper +from nerajob.scrapers.vietnamworks import VietnamWorksScraper from nerajob.scrapers.weworkremotely import WeWorkRemotelyScraper diff --git a/src/nerajob/scrapers/topcv.py b/src/nerajob/scrapers/topcv.py new file mode 100644 index 0000000..5428557 --- /dev/null +++ b/src/nerajob/scrapers/topcv.py @@ -0,0 +1,182 @@ +""" +TopCV.vn public job board adapter for NeraJob. + +TopCV (topcv.vn) is a leading Vietnamese job platform. +This adapter scrapes public-facing listing pages with rate limiting. +""" + +from __future__ import annotations + +import hashlib +import re +import time +from html.parser import HTMLParser +from urllib.request import Request, urlopen +from urllib.error import HTTPError, URLError + +from nerajob.models import JobPosting +from nerajob.scrapers.base import BaseScraper +from nerajob.config import http_timeout, user_agent + + +class _TopCVListingParser(HTMLParser): + """Lightweight parser for TopCV public listing page snippets.""" + + def __init__(self): + super().__init__() + self.jobs: list[dict] = [] + self._current: dict | None = None + self._in_title = False + self._in_company = False + self._in_location = False + self._in_salary = False + self._text_buf = "" + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + classes = (attrs_dict.get("class") or "").lower() + + if tag == "div" and "job-item" in classes: + self._current = {} + if self._current is not None: + if tag in ("h2", "h3") and "title" in classes: + self._in_title = True + elif tag in ("a", "span") and "company" in classes: + self._in_company = True + elif tag in ("span", "div") and ("location" in classes or "address" in classes): + self._in_location = True + elif tag in ("span", "div") and "salary" in classes: + self._in_salary = True + + def handle_endtag(self, tag: str) -> None: + if self._current is not None: + if self._in_title and tag in ("h2", "h3", "a"): + self._current["title"] = self._text_buf.strip() + self._text_buf = "" + self._in_title = False + elif self._in_company and tag in ("a", "span", "div"): + self._current.setdefault("company", self._text_buf.strip()) + self._text_buf = "" + self._in_company = False + elif self._in_location and tag in ("span", "div", "p"): + self._current.setdefault("location", self._text_buf.strip()) + self._text_buf = "" + self._in_location = False + elif self._in_salary and tag in ("span", "div"): + self._current.setdefault("salary", self._text_buf.strip()) + self._text_buf = "" + self._in_salary = False + + if tag == "div" and self._current and self._current.get("title"): + self.jobs.append(self._current) + self._current = None + + def handle_data(self, data: str) -> None: + if self._current is not None: + self._text_buf += data + + +# Offline sample: realistic TopCV snippet +TOPCV_SAMPLE = """ +
+ +""" + + +class TopCVScraper(BaseScraper): + """TopCV.vn public job listing adapter. + + Parses public-facing listing snippets with conservative rate limiting. + Falls back to offline sample data when the live endpoint is unreachable. + + Usage:: + + scraper = TopCVScraper() + scraper.search(query="python", limit=10) + + Bounty: https://github.com/mergeos-bounties/NeraJob/issues/17 + """ + + name = "topcv" + BASE_URL = "https://www.topcv.vn/viec-lam-it" + + def __init__(self, offline: bool = False) -> None: + self._offline = offline + self._last_request = 0.0 + + def _rate_limit(self) -> None: + """Enforce minimum 2-second gap between requests.""" + now = time.monotonic() + gap = now - self._last_request + if gap < 2.0: + time.sleep(2.0 - gap) + self._last_request = time.monotonic() + + def _fetch(self) -> str: + """Fetch HTML from TopCV public listing page or return offline sample.""" + if self._offline: + return TOPCV_SAMPLE + + self._rate_limit() + req = Request( + self.BASE_URL, + headers={"User-Agent": user_agent}, + ) + try: + with urlopen(req, timeout=http_timeout) as resp: + return resp.read().decode("utf-8", errors="replace") + except (HTTPError, URLError): + return TOPCV_SAMPLE + + def _parse(self, html: str) -> list[dict]: + parser = _TopCVListingParser() + parser.feed(html) + return parser.jobs + + def search(self, query: str, location: str = "", limit: int = 20) -> list[JobPosting]: + html = self._fetch() + raw_jobs = self._parse(html) + q = query.strip().lower() + loc = location.strip().lower() + results: list[JobPosting] = [] + + for item in raw_jobs: + if len(results) >= limit: + break + title = item.get("title", "") + company = item.get("company", "") + loc_str = item.get("location", "") + salary = item.get("salary", "") + + hay = f"{title} {company}".lower() + if q and q not in hay: + continue + if loc and loc not in loc_str.lower(): + continue + + job_id = hashlib.sha256( + f"topcv:{title}:{company}:{loc_str}".encode() + ).hexdigest()[:12] + + results.append(JobPosting( + id=job_id, + title=title, + company=company, + location=loc_str, + description=f"{title} at {company} — {salary}" if salary else f"{title} at {company}", + url=self.BASE_URL, + source="topcv", + tags=[salary] if salary else [], + )) + + return results diff --git a/src/nerajob/scrapers/vietnamworks.py b/src/nerajob/scrapers/vietnamworks.py new file mode 100644 index 0000000..9b9c9a5 --- /dev/null +++ b/src/nerajob/scrapers/vietnamworks.py @@ -0,0 +1,178 @@ + +""" +VietnamWorks public job board adapter for NeraJob. + +VietnamWorks (vietnamworks.com) is a major Vietnamese job platform. +This adapter scrapes public-facing search result snippets with rate limiting. +""" + +from __future__ import annotations + +import hashlib +import re +import time +from html.parser import HTMLParser +from urllib.request import Request, urlopen +from urllib.error import HTTPError, URLError + +from nerajob.models import JobPosting +from nerajob.scrapers.base import BaseScraper +from nerajob.config import http_timeout, user_agent + + +class _VNWListingParser(HTMLParser): + """Lightweight parser for VietnamWorks public search result snippets.""" + + def __init__(self): + super().__init__() + self.jobs: list[dict] = [] + self._current: dict | None = None + self._in_title = False + self._in_company = False + self._in_location = False + self._text_buf = "" + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + attrs_dict = dict(attrs) + classes = (attrs_dict.get("class") or "").lower() + + if tag == "div" and ("job-item" in classes or "result-item" in classes): + self._current = {} + if self._current is not None: + if tag in ("h2", "h3", "a") and "job-title" in classes: + self._in_title = True + elif tag in ("span", "a") and "company" in classes: + self._in_company = True + elif tag in ("span", "div") and "location" in classes: + self._in_location = True + + def handle_endtag(self, tag: str) -> None: + if self._current is not None: + if self._in_title and tag in ("h2", "h3", "a"): + self._current.setdefault("title", self._text_buf.strip()) + self._text_buf = "" + self._in_title = False + elif self._in_company and tag in ("span", "a", "div"): + self._current.setdefault("company", self._text_buf.strip()) + self._text_buf = "" + self._in_company = False + elif self._in_location and tag in ("span", "div"): + self._current.setdefault("location", self._text_buf.strip()) + self._text_buf = "" + self._in_location = False + + if tag == "div" and self._current and self._current.get("title"): + self.jobs.append(self._current) + self._current = None + + def handle_data(self, data: str) -> None: + if self._current is not None: + self._text_buf += data + + +# Offline sample: realistic VietnamWorks snippet +VNW_SAMPLE = """ + + + +""" + + +class VietnamWorksScraper(BaseScraper): + """VietnamWorks public job listing adapter. + + Parses public-facing search result snippets with conservative rate limiting. + Falls back to offline sample data when the live endpoint is unreachable. + + Usage:: + + scraper = VietnamWorksScraper() + scraper.search(query="python", limit=10) + + Bounty: https://github.com/mergeos-bounties/NeraJob/issues/17 + """ + + name = "vietnamworks" + BASE_URL = "https://www.vietnamworks.com/viec-lam/tat-ca-viec-lam" + + def __init__(self, offline: bool = False) -> None: + self._offline = offline + self._last_request = 0.0 + + def _rate_limit(self) -> None: + """Enforce minimum 2-second gap between requests.""" + now = time.monotonic() + gap = now - self._last_request + if gap < 2.0: + time.sleep(2.0 - gap) + self._last_request = time.monotonic() + + def _fetch(self) -> str: + """Fetch HTML from VietnamWorks public search page or return offline sample.""" + if self._offline: + return VNW_SAMPLE + + self._rate_limit() + req = Request( + self.BASE_URL, + headers={"User-Agent": user_agent}, + ) + try: + with urlopen(req, timeout=http_timeout) as resp: + return resp.read().decode("utf-8", errors="replace") + except (HTTPError, URLError): + return VNW_SAMPLE + + def _parse(self, html: str) -> list[dict]: + parser = _VNWListingParser() + parser.feed(html) + return parser.jobs + + def search(self, query: str, location: str = "", limit: int = 20) -> list[JobPosting]: + html = self._fetch() + raw_jobs = self._parse(html) + q = query.strip().lower() + loc = location.strip().lower() + results: list[JobPosting] = [] + + for item in raw_jobs: + if len(results) >= limit: + break + title = item.get("title", "") + company = item.get("company", "") + loc_str = item.get("location", "") + + hay = f"{title} {company}".lower() + if q and q not in hay: + continue + if loc and loc not in loc_str.lower(): + continue + + job_id = hashlib.sha256( + f"vietnamworks:{title}:{company}:{loc_str}".encode() + ).hexdigest()[:12] + + results.append(JobPosting( + id=job_id, + title=title, + company=company, + location=loc_str, + description=f"{title} at {company}", + url=self.BASE_URL, + source="vietnamworks", + tags=["Vietnam", "tech"], + )) + + return results diff --git a/tests/test_topcv.py b/tests/test_topcv.py new file mode 100644 index 0000000..23374c1 --- /dev/null +++ b/tests/test_topcv.py @@ -0,0 +1,69 @@ +"""Tests for TopCV.vn scraper.""" + +from nerajob.scrapers.topcv import TopCVScraper, TOPCV_SAMPLE, _TopCVListingParser + + +class TestTopCVListingParser: + """Unit tests for the HTML parser.""" + + def test_parse_sample(self): + parser = _TopCVListingParser() + parser.feed(TOPCV_SAMPLE) + jobs = parser.jobs + assert len(jobs) == 2, f"Expected 2 jobs, got {len(jobs)}" + assert jobs[0]["title"] == "Backend Developer (Python)" + assert jobs[0]["company"] == "FPT Software" + assert "Ho Chi Minh" in jobs[0]["location"] + assert jobs[1]["title"] == "Frontend Developer (ReactJS)" + assert jobs[1]["company"] == "VNG Corporation" + + def test_parse_empty(self): + parser = _TopCVListingParser() + parser.feed("") + assert parser.jobs == [] + + +class TestTopCVScraper: + """Integration tests for the TopCV scraper.""" + + def test_name(self): + scraper = TopCVScraper(offline=True) + assert scraper.name == "topcv" + + def test_search_offline_finds_python(self): + scraper = TopCVScraper(offline=True) + results = scraper.search(query="python", limit=10) + assert len(results) >= 1, f"Expected at least 1 result, got {len(results)}" + assert any("python" in r.title.lower() for r in results) + + def test_search_offline_finds_frontend(self): + scraper = TopCVScraper(offline=True) + results = scraper.search(query="frontend", limit=10) + assert len(results) >= 1 + assert results[0].source == "topcv" + + def test_search_no_match(self): + scraper = TopCVScraper(offline=True) + results = scraper.search(query="zzz_no_match_xxx", limit=10) + assert len(results) == 0 + + def test_search_location_filter(self): + scraper = TopCVScraper(offline=True) + results = scraper.search(query="", location="Ha Noi", limit=10) + assert len(results) >= 1 + assert any("Ha Noi" in r.location for r in results) + + def test_search_limit(self): + scraper = TopCVScraper(offline=True) + results = scraper.search(query="", limit=1) + assert len(results) <= 1 + + def test_job_posting_fields(self): + scraper = TopCVScraper(offline=True) + results = scraper.search(query="backend", limit=1) + assert len(results) == 1 + job = results[0] + assert job.title + assert job.company + assert job.source == "topcv" + assert len(job.id) == 12 diff --git a/tests/test_vietnamworks.py b/tests/test_vietnamworks.py new file mode 100644 index 0000000..6dd4db9 --- /dev/null +++ b/tests/test_vietnamworks.py @@ -0,0 +1,69 @@ +"""Tests for VietnamWorks scraper.""" + +from nerajob.scrapers.vietnamworks import VietnamWorksScraper, VNW_SAMPLE, _VNWListingParser + + +class TestVNWListingParser: + """Unit tests for the HTML parser.""" + + def test_parse_sample(self): + parser = _VNWListingParser() + parser.feed(VNW_SAMPLE) + jobs = parser.jobs + assert len(jobs) == 3, f"Expected 3 jobs, got {len(jobs)}" + assert jobs[0]["title"] == "Backend Developer (Python)" + assert jobs[0]["company"] == "FPT Software" + assert "Ho Chi Minh" in jobs[0]["location"] + assert jobs[2]["title"] == "Mobile Developer (Flutter)" + + def test_parse_empty(self): + parser = _VNWListingParser() + parser.feed("") + assert parser.jobs == [] + + +class TestVietnamWorksScraper: + """Integration tests for the VietnamWorks scraper.""" + + def test_name(self): + scraper = VietnamWorksScraper(offline=True) + assert scraper.name == "vietnamworks" + + def test_search_offline_finds_python(self): + scraper = VietnamWorksScraper(offline=True) + results = scraper.search(query="python", limit=10) + assert len(results) >= 1 + assert any("python" in r.title.lower() for r in results) + + def test_search_offline_finds_data(self): + scraper = VietnamWorksScraper(offline=True) + results = scraper.search(query="data", limit=10) + assert len(results) >= 1 + assert results[0].source == "vietnamworks" + + def test_search_no_match(self): + scraper = VietnamWorksScraper(offline=True) + results = scraper.search(query="zzz_no_match_xxx", limit=10) + assert len(results) == 0 + + def test_search_location_filter(self): + scraper = VietnamWorksScraper(offline=True) + results = scraper.search(query="", location="Ha Noi", limit=10) + assert len(results) >= 1 + assert any("Ha Noi" in r.location for r in results) + + def test_search_limit(self): + scraper = VietnamWorksScraper(offline=True) + results = scraper.search(query="", limit=2) + assert len(results) <= 2 + + def test_job_posting_fields(self): + scraper = VietnamWorksScraper(offline=True) + results = scraper.search(query="mobile", limit=1) + assert len(results) == 1 + job = results[0] + assert job.title + assert job.company + assert job.source == "vietnamworks" + assert len(job.id) == 12 + assert "Vietnam" in job.tags