Scrapling/tests/spiders/test_boss_zhipin.py

"""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()