From 6d2c9d95b997e6ba05b0fd8600a7e3480a9e0cd4 Mon Sep 17 00:00:00 2001 From: freedakgmail Date: Wed, 1 Jul 2026 23:37:47 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20add=20Boss=E7=9B=B4=E8=81=98=20job=20sp?= =?UTF-8?q?ider=20test=20script?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/spiders/test_boss_zhipin.py | 196 ++++++++++++++++++++++++++++++ 1 file changed, 196 insertions(+) create mode 100644 tests/spiders/test_boss_zhipin.py diff --git a/tests/spiders/test_boss_zhipin.py b/tests/spiders/test_boss_zhipin.py new file mode 100644 index 0000000..262e6ea --- /dev/null +++ b/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()