70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
"""offerqingbaoju.cn 招聘公告地址抓取。
|
||
|
||
接口说明见 doc/offerqingbaoju.md：
|
||
- 明文 JSON，无加密。
|
||
- 匿名访问只能取 page=1，但 per_page 可以拉很大（实测 6000+ 条一次返回）。
|
||
- 公告地址取每条数据的 `公告链接` 字段。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import httpx
|
||
|
||
from app.core.logger import log
|
||
|
||
# 招聘数据接口
|
||
_API_URL = "https://offerqingbaoju.cn/api/simple/navigation/60/data"
|
||
|
||
_HEADERS = {
|
||
"accept": "application/json",
|
||
"user-agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"
|
||
),
|
||
}
|
||
|
||
# 请求超时（秒）
|
||
_TIMEOUT = 30
|
||
|
||
|
||
def fetch_offerqingbaoju(limit: int) -> list[str]:
|
||
"""抓取 offerqingbaoju 招聘公告地址。
|
||
|
||
Args:
|
||
limit: 抓取条数（映射到接口的 per_page，匿名只取第 1 页）。
|
||
|
||
Returns:
|
||
公告地址列表（已去重、过滤空值，保持原始顺序）。
|
||
"""
|
||
if limit <= 0:
|
||
return []
|
||
|
||
try:
|
||
response = httpx.get(
|
||
_API_URL,
|
||
params={"page": 1, "per_page": limit},
|
||
headers=_HEADERS,
|
||
timeout=_TIMEOUT,
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
except (httpx.HTTPError, ValueError) as exc:
|
||
log.error(f"offerqingbaoju 抓取失败: {exc}")
|
||
return []
|
||
|
||
rows = payload.get("data") or []
|
||
|
||
urls: list[str] = []
|
||
seen: set[str] = set()
|
||
for row in rows:
|
||
url = (row.get("公告链接") or "").strip()
|
||
if not url or not url.startswith("http"):
|
||
continue
|
||
if url in seen:
|
||
continue
|
||
seen.add(url)
|
||
urls.append(url)
|
||
|
||
log.info(f"offerqingbaoju 抓取到 {len(urls)} 条公告地址（原始 {len(rows)} 条）")
|
||
return urls
|