添加脚本

app/spider/offerqingbaoju.py

@@ -0,0 +1,69 @@
+"""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