feat: add Boss直聘 job spider test script

tests/spiders/test_boss_zhipin.py

@@ -0,0 +1,196 @@
+"""Boss直聘岗位爬虫 - headful模式，支持手动登录后爬取
+
+使用方法:
+  1. 第一次运行时，浏览器会弹出Boss直聘登录页，手动扫码/登录
+  2. 登录成功后按回车键继续爬取
+  3. 登录状态会保存到 user_data_dir，下次运行可跳过登录
+"""
+
+import json
+import os
+import time
+
+from scrapling.fetchers import StealthyFetcher
+from scrapling.core._types import Any, Dict
+
+
+# ===== 配置 =====
+SEARCH_KEYWORD = "Python"
+SEARCH_CITY = "100010000"  # 全国
+USER_DATA_DIR = os.path.join(os.path.dirname(__file__), "boss_browser_data")
+OUTPUT_FILE = os.path.join(os.path.dirname(__file__), "boss_zhipin_jobs.json")
+MAX_PAGES = 5
+
+
+def crawl_boss_zhipin():
+    """使用StealthyFetcher爬取Boss直聘岗位信息"""
+
+    # 构建搜索URL
+    url = f"https://www.zhipin.com/web/geek/job?query={SEARCH_KEYWORD}&city={SEARCH_CITY}"
+    print(f"目标URL: {url}")
+    print(f"关键词: {SEARCH_KEYWORD} | 城市: {SEARCH_CITY}")
+    print(f"最大页数: {MAX_PAGES}")
+    print("-" * 60)
+
+    all_jobs = []
+
+    for page_num in range(1, MAX_PAGES + 1):
+        page_url = f"{url}&page={page_num}"
+        print(f"\n正在爬取第 {page_num}/{MAX_PAGES} 页...")
+
+        try:
+            response = StealthyFetcher.fetch(
+                page_url,
+                headless=False,  # headful模式，方便登录
+                network_idle=True,
+                wait=3000,
+                timeout=90000,
+                block_ads=True,
+                disable_resources=True,
+                locale="zh-CN",
+                timezone_id="Asia/Shanghai",
+                user_data_dir=USER_DATA_DIR,  # 保存浏览器会话
+                real_chrome=True,  # 使用本机Chrome
+            )
+
+            print(f"  状态码: {response.status}")
+            print(f"  最终URL: {response.url}")
+
+            # 检查是否被重定向到登录页
+            if "/web/user/" in str(response.url) or "注册登录" in (response.css("title::text").get() or ""):
+                print(f"\n⚠️  页面被重定向到登录页！")
+                print(f"    浏览器窗口已打开，请在浏览器中完成登录（扫码/账号登录）。")
+                print(f"    登录成功后，请在此处按回车键继续...")
+                input("    >>> 按回车继续 <<<")
+
+                # 重新抓取
+                print(f"\n  重新爬取第 {page_num} 页...")
+                response = StealthyFetcher.fetch(
+                    page_url,
+                    headless=False,
+                    network_idle=True,
+                    wait=5000,
+                    timeout=90000,
+                    block_ads=True,
+                    disable_resources=True,
+                    locale="zh-CN",
+                    timezone_id="Asia/Shanghai",
+                    user_data_dir=USER_DATA_DIR,
+                    real_chrome=True,
+                )
+                print(f"  状态码: {response.status}")
+                print(f"  最终URL: {response.url}")
+
+            # 解析岗位卡片
+            job_cards = response.css(".job-card-wrapper")
+            if not job_cards:
+                job_cards = response.css('[class*="job-card"]')
+            if not job_cards:
+                # 尝试更通用的选择器
+                job_cards = response.css('[data-type="job"]')
+
+            print(f"  找到 {len(job_cards)} 个岗位")
+
+            if not job_cards:
+                # 调试: 打印页面信息
+                title = response.css("title::text").get() or ""
+                print(f"  页面标题: {title}")
+                # 保存当前页面HTML供调试
+                debug_path = os.path.join(os.path.dirname(__file__), f"boss_page_{page_num}_debug.html")
+                with open(debug_path, "w", encoding="utf-8") as f:
+                    f.write(response.body.decode("utf-8", errors="ignore"))
+                print(f"  页面HTML已保存到: {debug_path}")
+
+                if page_num == 1:
+                    print("\n  第一页未找到岗位，可能需要登录。")
+                    print("  请在浏览器中完成登录后按回车重试...")
+                    input("  >>> 按回车继续 <<<")
+                    continue
+                else:
+                    break
+
+            page_jobs = 0
+            for card in job_cards:
+                try:
+                    job_name = card.css(".job-name::text").get() or card.css('[class*="job-name"]::text').get() or ""
+                    salary = card.css(".salary::text").get() or card.css('[class*="salary"]::text').get() or ""
+                    company_name = (
+                        card.css(".company-name::text").get()
+                        or card.css('[class*="company-name"] a::text').get()
+                        or card.css(".company-name a::text").get()
+                        or ""
+                    )
+                    area = card.css(".job-area::text").get() or card.css('[class*="job-area"]::text').get() or ""
+                    tag_list = (
+                        card.css(".tag-list li::text").getall()
+                        or card.css('[class*="tag-list"]::text').getall()
+                        or []
+                    )
+                    job_link = card.css(".job-card-left::attr(href)").get() or card.css("a::attr(href)").get() or ""
+
+                    # 获取经验/学历要求
+                    info_items = card.css(".job-info .tag-list li::text").getall() or []
+                    if not info_items:
+                        info_items = card.css('[class*="info"] li::text').getall() or []
+
+                    item = {
+                        "job_name": job_name.strip() if job_name else "",
+                        "salary": salary.strip() if salary else "",
+                        "company_name": company_name.strip() if company_name else "",
+                        "area": area.strip() if area else "",
+                        "tags": [t.strip() for t in tag_list if t.strip()],
+                        "requirements": [t.strip() for t in info_items if t.strip()],
+                        "job_url": response.urljoin(job_link) if job_link else "",
+                        "page": page_num,
+                    }
+
+                    if item["job_name"]:
+                        all_jobs.append(item)
+                        page_jobs += 1
+                        print(f"    [{page_jobs}] {item['job_name']} | {item['salary']} | {item['company_name']}")
+                except Exception as e:
+                    print(f"    解析岗位出错: {e}")
+
+            print(f"  第 {page_num} 页成功解析 {page_jobs} 个岗位")
+
+            if page_jobs == 0 and page_num > 1:
+                print("  没有更多岗位，停止翻页")
+                break
+
+            # 翻页延迟
+            if page_num < MAX_PAGES:
+                delay = 3
+                print(f"  等待 {delay}s 后翻页...")
+                time.sleep(delay)
+
+        except Exception as e:
+            print(f"  爬取出错: {e}")
+            import traceback
+            traceback.print_exc()
+            break
+
+    # 输出结果
+    print("\n" + "=" * 60)
+    print(f"爬取完成!")
+    print(f"  总岗位数: {len(all_jobs)}")
+    print("=" * 60)
+
+    for i, item in enumerate(all_jobs[:30]):
+        print(f"\n[{i+1}] {item['job_name']}")
+        print(f"    薪资: {item['salary']}")
+        print(f"    公司: {item['company_name']}")
+        print(f"    地区: {item['area']}")
+        print(f"    标签: {', '.join(item.get('tags', []))}")
+        print(f"    要求: {', '.join(item.get('requirements', []))}")
+        print(f"    链接: {item['job_url']}")
+
+    # 保存到JSON
+    with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
+        json.dump(all_jobs, f, ensure_ascii=False, indent=2)
+    print(f"\n结果已保存到: {OUTPUT_FILE}")
+
+    return all_jobs
+
+
+if __name__ == "__main__":
+    jobs = crawl_boss_zhipin()