Griffin Chow / 鱼泡网数据分析爬虫技术文档

Created Sun, 26 Oct 2025 00:00:00 +0000 Modified Thu, 06 Nov 2025 04:24:59 +0000
7735 Words
Python 爬虫 Selenium

一、项目介绍

在当今竞争激烈的就业市场中，及时掌握职位信息和市场动态变得尤为重要。本文将详细介绍如何使用Python开发一个爬虫项目，自动采集鱼泡网（yupao.com）的数据分析职位数据，并对数据进行处理和分析。

1.1 项目意义

  • 帮助数据分析求职者快速获取全国范围的职位信息
  • 为求职者提供真实的个人招聘平台数据
  • 辅助研究人员进行蓝领和灵活就业市场研究
  • 提供薪资水平参考数据
  • 了解市场对数据分析技能的真实需求

1.2 技术特点

  • 采用面向对象的编程方式
  • 使用Selenium模拟真实浏览器行为
  • 实现了完整的反爬虫机制
  • 支持全国范围的职位搜索
  • 提供灵活的配置选项
  • 自动化数据处理和存储

1.3 数据获取方式

Selenium浏览器自动化

  • 主要通过Selenium WebDriver控制Chrome浏览器访问鱼泡网
  • 模拟真实用户行为，包括页面滚动、点击等操作
  • 支持手动登录和搜索，确保访问稳定性
  • 使用BeautifulSoup4解析HTML页面内容
  • 返回数据包含职位的完整信息

页面解析

  • 通过CSS选择器定位页面元素
  • 使用BeautifulSoup4解析职位列表
  • 提取职位名称、薪资、地区、公司等11个关键字段
  • 智能识别福利标签和公司规模

二、技术栈

2.1 核心技术

  • Python 3.9+：编程语言
  • Selenium 4.15+：浏览器自动化
  • BeautifulSoup4 4.12+：HTML解析
  • pandas 2.1+：数据处理
  • openpyxl 3.1+：Excel文件操作
  • webdriver-manager 4.0+：自动管理ChromeDriver

2.2 开发环境

  • 操作系统：支持Windows/Linux/MacOS
  • 浏览器：Chrome浏览器（必需）
  • IDE：推荐PyCharm或VS Code
  • 依赖管理：pip / Anaconda

2.3 环境准备

安装依赖

# 创建虚拟环境（推荐）
conda create -n dataanalysis python=3.9 -y
conda activate dataanalysis

# 安装依赖包
pip install -r requirements.txt

requirements.txt内容

requests>=2.31.0
pandas>=2.1.0
beautifulsoup4>=4.12.0
openpyxl>=3.1.0
selenium>=4.15.0
webdriver-manager>=4.0.0

三、项目实现

3.1 项目结构

yupaozhipin_spider/
├── yupao_spider_selenium_backup.py  # 主爬虫程序（Selenium版本）
├── requirements.txt                  # 依赖包列表
├── 鱼泡网_数据分析岗位数据_YYYYMMDD_HHMMSS.xlsx  # 爬取的数据文件
└── 鱼泡网数据分析爬虫技术文档.md     # 本文档

3.2 核心功能实现

3.2.1 爬虫初始化配置

# ==================== 配置区域 ====================
KEYWORD = "数据分析"  # 搜索关键词
CITY = "全国"        # 城市（全国）
MAX_PAGES = 10       # 爬取页数
MIN_SLEEP = 2        # 最小延时（秒）
MAX_SLEEP = 5        # 最大延时（秒）
# =================================================

class YupaoSpiderSelenium:
    def __init__(self):
        """初始化Selenium爬虫"""
        # 配置Chrome选项
        chrome_options = Options()
        
        # 反爬虫检测配置
        chrome_options.add_argument('--disable-blink-features=AutomationControlled')
        chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
        chrome_options.add_experimental_option('useAutomationExtension', False)
        
        # 设置User-Agent
        chrome_options.add_argument('user-agent=Mozilla/5.0 ...')
        
        # 自动下载并安装ChromeDriver
        service = Service(ChromeDriverManager().install())
        self.driver = webdriver.Chrome(service=service, options=chrome_options)
        
        # 执行CDP命令，隐藏webdriver特征
        self.driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
            'source': '''
                Object.defineProperty(navigator, 'webdriver', {
                    get: () => undefined
                })
            '''
        })
        
        # 数据存储
        self.data_list = []

3.2.2 数据采集流程

1. 手动登录流程

def login_manual(self):
    """手动登录流程"""
    print("\n请在浏览器中完成登录...")
    print("1. 浏览器将自动打开鱼泡网")
    print("2. 如需登录，请在浏览器中手动登录")
    print("3. 登录/确认后，请回到终端按回车继续")
    
    # 打开鱼泡网首页
    self.driver.get("https://www.yupao.com/")
    time.sleep(3)
    
    # 等待用户操作
    input("\n准备好后，请按回车键继续...")
    print("\n✓ 继续执行爬取任务")

2. 搜索职位

def search_jobs(self, keyword=KEYWORD, city=CITY):
    """搜索职位"""
    print(f"\n开始搜索职位: {keyword} - {city}")
    
    try:
        # 访问首页
        self.driver.get("https://www.yupao.com/")
        time.sleep(2)
        
        # 尝试自动搜索
        from selenium.webdriver.common.keys import Keys
        search_input = self.driver.find_element(By.CSS_SELECTOR, "input[placeholder*='搜索']")
        search_input.clear()
        search_input.send_keys(keyword)
        search_input.send_keys(Keys.RETURN)
        time.sleep(3)
        
    except Exception as e:
        # 自动搜索失败，提示手动操作
        print(f"⚠ 自动搜索失败，请手动操作: {e}")
        input(f"\n请在浏览器中手动搜索'{keyword}'并选择'{city}'地区，完成后按回车...")

3. 解析职位列表

def parse_job_list(self):
    """解析当前页面的职位列表"""
    jobs = []
    
    # 等待页面加载
    time.sleep(3)
    self.scroll_page()
    
    # 获取页面HTML
    page_source = self.driver.page_source
    soup = BeautifulSoup(page_source, 'html.parser')
    
    # 鱼泡网结构：每个职位在 <div data-index="数字"> 内
    job_containers = soup.find_all('div', attrs={'data-index': True})
    
    for container in job_containers:
        # 获取职位卡片
        card = container.find('a', class_='abgbe')
        if not card:
            continue
        
        # 提取职位标题 (h3 class="abiee")
        job_name_tag = card.find('h3', class_='abiee')
        job_name = job_name_tag.text.strip() if job_name_tag else ""
        
        # 提取薪资 (span class="abiej")
        salary_tag = card.find('span', class_='abiej')
        salary = salary_tag.text.strip() if salary_tag else ""
        
        # 提取职位描述 (p class="abifa")
        desc_tag = card.find('p', class_='abifa')
        desc = desc_tag.text.strip() if desc_tag else ""
        
        # 提取地区 (p class="abifc")
        area_tag = card.find('p', class_='abifc')
        area = area_tag.text.strip() if area_tag else CITY
        
        # 提取联系人 (div class="abifh")
        contact_tag = card.find('div', class_='abifh')
        contact = contact_tag.text.strip() if contact_tag else ""
        
        # 提取企业认证标签
        cert_tags = card.find_all('span', class_='abifi')
        cert_labels = [tag.text.strip() for tag in cert_tags]
        company_name = f"{contact}（{'/'.join(cert_labels)}）" if cert_labels else contact
        
        # 提取所有标签 (span class="abigg")
        tag_elements = card.find_all('span', class_='abigg')
        all_tags = [tag.text.strip() for tag in tag_elements]
        
        # 从标签中智能提取经验、学历、技能
        experience = ""
        education = ""
        skills = []
        
        for tag in all_tags:
            if '年' in tag or 'year' in tag.lower():
                experience = tag
            elif '学历' in tag or '以上' in tag or '本科' in tag or '大专' in tag:
                education = tag
            else:
                skills.append(tag)
        
        # 智能提取福利标签
        welfare = self.extract_welfare(desc)
        
        # 智能提取公司规模
        company_scale = self.extract_company_scale(desc)
        
        # 构建数据项
        job_data = {
            "职位": job_name,
            "公司": company_name,
            "薪资": salary,
            "地区": area,
            "经验": experience,
            "学历": education,
            "公司规模": company_scale,
            "行业": "",
            "福利标签": welfare,
            "技能标签": ",".join(skills),
            "职位描述": desc[:300]
        }
        
        jobs.append(job_data)
        
    return jobs

4. 翻页处理

def next_page(self):
    """翻到下一页"""
    try:
        # 滚动到底部
        self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        time.sleep(1)
        
        # 查找分页容器
        page_source = self.driver.page_source
        soup = BeautifulSoup(page_source, 'html.parser')
        
        pagination = soup.find('div', class_='abgbg')
        if pagination:
            # 找到当前页
            current_page_elem = pagination.find('span', class_='abgca')
            if current_page_elem:
                current_page = int(current_page_elem.text.strip())
                
                # 点击下一页
                next_page_xpath = f"//span[@class='abgbh' and text()='{current_page + 1}']"
                next_button = self.driver.find_element(By.XPATH, next_page_xpath)
                if next_button and next_button.is_displayed():
                    next_button.click()
                    time.sleep(random.uniform(3, 4))
                    return True
        
        return False
        
    except Exception as e:
        print(f"✗ 翻页失败: {e}")
        return False

3.3 反爬虫策略

3.3.1 浏览器指纹隐藏

# 禁用自动化检测
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])

# 隐藏webdriver特征
self.driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
    'source': '''
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        })
    '''
})

3.3.2 请求频率控制

def random_sleep(self, min_sec=MIN_SLEEP, max_sec=MAX_SLEEP):
    """随机延时，模拟人工操作"""
    sleep_time = random.uniform(min_sec, max_sec)
    time.sleep(sleep_time)
    return sleep_time

# 使用
self.random_sleep(2, 5)  # 随机延时2-5秒

3.3.3 页面滚动模拟

def scroll_page(self):
    """滚动页面，模拟真实用户行为"""
    try:
        page_height = self.driver.execute_script("return document.body.scrollHeight")
        
        # 分段滚动
        for i in range(0, page_height, 300):
            self.driver.execute_script(f"window.scrollTo(0, {i});")
            time.sleep(random.uniform(0.1, 0.3))
        
        # 滚动到底部
        self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        time.sleep(1)
        
    except Exception as e:
        print(f"⚠ 滚动页面失败: {e}")

3.4 数据处理

3.4.1 智能数据提取

提取电话号码

def extract_phone(self, text):
    """从文本中提取电话号码"""
    if not text:
        return ""
    phone_pattern = r'1[3-9]\d{9}'
    match = re.search(phone_pattern, text)
    return match.group(0) if match else ""

提取公司规模

def extract_company_scale(self, text):
    """从文本中提取公司规模"""
    if not text:
        return ""
    scale_patterns = [
        r'(\d+[-~至]\d+人)',
        r'(\d+人以上)',
        r'(少于\d+人)',
        r'(\d+\+人)',
    ]
    for pattern in scale_patterns:
        match = re.search(pattern, text)
        if match:
            return match.group(1)
    return ""

智能提取福利标签

def extract_welfare(self, text):
    """智能提取福利标签"""
    if not text:
        return ""
    welfare_keywords = [
        '五险一金', '六险一金', '五险', '社保', '公积金',
        '包吃', '包住', '包吃住', '提供住宿', '提供食宿',
        '年终奖', '十三薪', '年底双薪', '绩效奖金', '季度奖',
        '带薪年假', '年假', '法定节假日',
        '节日福利', '节日礼品', '生日福利',
        '定期体检', '年度体检', '健康体检',
        '员工旅游', '团建', '旅游',
        '补充医疗保险', '商业保险',
        '加班补助', '加班费', '交通补贴', '通讯补贴', 
        '餐补', '饭补', '房补', '住房补贴',
        '弹性工作', '弹性上班', '灵活工作',
        '周末双休', '双休', '单休',
        '五天工作制', '大小周',
        '股票期权', '期权', '股权激励',
        '培训机会', '技能培训', '晋升机会',
    ]
    welfare_list = []
    for keyword in welfare_keywords:
        if keyword in text and keyword not in welfare_list:
            welfare_list.append(keyword)
    return ",".join(welfare_list[:8])  # 最多显示8个

3.4.2 数据结构设计

job_data = {
    "职位": "数据分析师",
    "公司": "王女士（实名/企业）",
    "薪资": "8000-12000元/月",
    "地区": "深圳·南山区",
    "经验": "3-5年",
    "学历": "本科及以上",
    "公司规模": "100-500人",
    "行业": "",
    "福利标签": "五险一金,包吃,年终奖",
    "技能标签": "SPSS,SQL,Python",
    "职位描述": "负责数据分析..."
}

3.4.3 数据存储

def save_excel(self):
    """保存数据到Excel文件"""
    if not self.data_list:
        print("\n⚠ 没有数据可保存！")
        return
    
    filename = f"鱼泡网_{KEYWORD}岗位数据_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
    current_dir = os.path.dirname(os.path.abspath(__file__))
    filepath = os.path.join(current_dir, filename)
    
    df = pd.DataFrame(self.data_list)
    df.to_excel(filepath, index=False, engine='openpyxl')
    
    print(f"\n✓ 数据已保存到: {filepath}")
    print(f"✓ 共保存 {len(df)} 条职位数据")

四、项目优化

4.1 性能优化

1. 页面加载优化

# 设置页面加载策略
chrome_options.page_load_strategy = 'eager'  # 不等待完整加载

# 禁用图片加载（可选）
prefs = {"profile.managed_default_content_settings.images": 2}
chrome_options.add_experimental_option("prefs", prefs)

2. 数据处理优化

# 使用生成器处理大数据
def process_jobs_generator(self):
    for page in range(1, MAX_PAGES + 1):
        jobs = self.parse_job_list()
        for job in jobs:
            yield job

# 批量保存
def batch_save(self, batch_size=100):
    for i in range(0, len(self.data_list), batch_size):
        batch = self.data_list[i:i+batch_size]
        # 保存批次数据

4.2 稳定性提升

1. 异常处理机制

def safe_find_element(self, by, value, timeout=10):
    """安全查找元素，带重试机制"""
    retries = 3
    for i in range(retries):
        try:
            element = WebDriverWait(self.driver, timeout).until(
                EC.presence_of_element_located((by, value))
            )
            return element
        except Exception as e:
            print(f"查找元素失败 ({i+1}/{retries}): {str(e)}")
            if i == retries - 1:
                raise
            time.sleep(2)
    return None

2. 日志记录

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('yupao_spider.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)
logger.info("开始爬取数据...")

3. 浏览器清理

def close(self):
    """关闭浏览器并清理资源"""
    try:
        self.driver.quit()
        print("\n✓ 浏览器已关闭")
    except:
        pass
    finally:
        # 清理临时文件
        import shutil
        temp_dir = os.path.join(os.getcwd(), 'temp')
        if os.path.exists(temp_dir):
            shutil.rmtree(temp_dir)

五、使用指南

5.1 快速开始

步骤1：安装依赖

cd /Users/griffinchow/PDF转WORD文件.py/bosszhipin_spider
conda activate dataanalysis
pip install -r requirements.txt

步骤2：运行爬虫

python yupao_spider_selenium_backup.py

步骤3：按提示操作

  1. 浏览器自动打开鱼泡网首页
  2. 如需登录，在浏览器中手动登录
  3. 登录完成后，回到终端按回车
  4. 在浏览器中手动搜索"数据分析"并选择"全国"
  5. 看到职位列表后，回到终端按回车
  6. 输入要爬取的页数（建议1-10页）
  7. 等待自动爬取完成

5.2 配置修改

修改搜索关键词

KEYWORD = "数据分析"  # 改为其他职位，如"数据分析师"

修改搜索地区

CITY = "全国"  # 改为"深圳"、"北京"、"上海"等

修改爬取页数

MAX_PAGES = 10  # 修改默认爬取页数

修改延时时间

MIN_SLEEP = 2  # 最小延时（秒）
MAX_SLEEP = 5  # 最大延时（秒）

5.3 常见问题解决

问题1：Chrome浏览器无法启动

原因： 未安装Chrome浏览器或ChromeDriver版本不匹配

解决方案：

# macOS
brew install --cask google-chrome

# Windows
# 从官网下载安装：https://www.google.com/chrome/

# webdriver-manager会自动下载匹配的ChromeDriver

问题2：找不到职位

原因： 页面未完全加载或选择器变化

解决方案：

  1. 增加等待时间
  2. 确保手动搜索后能看到职位列表
  3. 在看到职位卡片后再按回车

问题3：数据为空

原因： 页面结构变化或网络问题

解决方案：

# 增加页面加载等待时间
time.sleep(5)  # 在parse_job_list开头

# 检查选择器是否正确
job_containers = soup.find_all('div', attrs={'data-index': True})
print(f"找到 {len(job_containers)} 个职位容器")

问题4：爬取速度慢

原因： 延时时间过长

解决方案：

# 适当减少延时（注意不要太快，避免被封）
MIN_SLEEP = 1
MAX_SLEEP = 2

问题5：被检测到反爬

原因： 请求频率过高或行为不自然

解决方案：

  1. 增加延时时间
  2. 减少爬取页数
  3. 使用代理IP（高级）
  4. 更换账号

六、实践经验

6.1 鱼泡网特点

1. 平台特性

  • 个人招聘平台：大部分职位由个人发布，非企业官方招聘
  • 蓝领+灵活就业：主要面向蓝领和灵活就业人群
  • 无行业信息：列表页通常不显示行业信息
  • 认证标签：有"实名认证"、“企业认证"等标识

2. 数据特点

  • 公司字段：通常显示"招聘人（认证标签）“而非公司名称
  • 地区信息：可能显示具体区域，如"深圳·南山区”
  • 薪资格式：多为"8000-12000元/月"格式
  • 福利标签：需要从职位描述中智能提取

6.2 最佳实践

1. 爬取策略

# 建议配置
MAX_PAGES = 5-10        # 单次爬取不超过10页
MIN_SLEEP = 2           # 最小延时2秒
MAX_SLEEP = 5           # 最大延时5秒

# 避免
单次爬取过多页面（>20页）
延时过短（<1秒）
短时间内多次运行

2. 数据清洗

# 薪资范围提取
def extract_salary_range(salary_str):
    """提取薪资范围"""
    # "8000-12000元/月" -> min=8000, max=12000
    import re
    pattern = r'(\d+)-(\d+)'
    match = re.search(pattern, salary_str)
    if match:
        return int(match.group(1)), int(match.group(2))
    return None, None

# 地区标准化
def standardize_area(area_str):
    """地区标准化"""
    # "深圳·南山区" -> city="深圳", district="南山区"
    if '·' in area_str:
        city, district = area_str.split('·')
        return city.strip(), district.strip()
    return area_str.strip(), ""

3. 数据分析

import pandas as pd
import matplotlib.pyplot as plt

# 读取数据
df = pd.read_excel('鱼泡网_数据分析岗位数据_20251024_193228.xlsx')

# 薪资分析
salary_stats = df['薪资'].value_counts()
plt.figure(figsize=(12, 6))
salary_stats.head(10).plot(kind='bar')
plt.title('数据分析职位薪资分布（Top 10）')
plt.xlabel('薪资范围')
plt.ylabel('职位数量')
plt.tight_layout()
plt.savefig('salary_distribution.png')

# 地区分析
area_stats = df['地区'].value_counts()
plt.figure(figsize=(12, 6))
area_stats.head(10).plot(kind='pie', autopct='%1.1f%%')
plt.title('数据分析职位地区分布（Top 10）')
plt.tight_layout()
plt.savefig('area_distribution.png')

# 技能需求分析
all_skills = []
for skills in df['技能标签'].dropna():
    all_skills.extend(skills.split(','))
skill_freq = pd.Series(all_skills).value_counts()
print("\n最热门的技能要求（Top 10）:")
print(skill_freq.head(10))

6.3 项目扩展

1. 多关键词爬取

keywords = ["数据分析", "数据分析师", "数据挖掘", "商业分析"]
for keyword in keywords:
    KEYWORD = keyword
    spider = YupaoSpiderSelenium()
    spider.run(max_pages=5)

2. 定时任务

# 使用crontab定时运行（Linux/macOS）
# 每天上午10点运行
0 10 * * * cd /path/to/project && /path/to/python yupao_spider_selenium_backup.py

3. 数据库存储

import sqlite3

def save_to_database(self):
    """保存到SQLite数据库"""
    conn = sqlite3.connect('yupao_jobs.db')
    df = pd.DataFrame(self.data_list)
    df.to_sql('jobs', conn, if_exists='append', index=False)
    conn.close()

七、数据字段说明

7.1 字段列表

字段名称数据类型示例说明
职位文本数据分析师职位名称
公司文本王女士（实名/企业）招聘人+认证标签
薪资文本8000-12000元/月薪资范围
地区文本深圳·南山区工作地点
经验文本3-5年工作经验要求
学历文本本科及以上学历要求
公司规模文本100-500人从描述提取
行业文本（空）鱼泡网通常无此信息
福利标签文本五险一金,包吃,年终奖智能提取，逗号分隔
技能标签文本SPSS,SQL,Python从标签提取，逗号分隔
职位描述文本负责数据分析…前300字符

7.2 数据示例

职位: 数据分析师
公司: 张先生（实名/企业）
薪资: 10000-15000元/月
地区: 北京·朝阳区
经验: 3-5年
学历: 本科及以上
公司规模: 100-500人
行业: 
福利标签: 五险一金,包吃,双休,年终奖
技能标签: Excel,SQL,Python,SPSS
职位描述: 岗位职责：1. 负责公司数据分析工作，提供数据支持；2. 建立和完善数据指标体系...

八、总结

本项目通过Python和Selenium实现了鱼泡网数据分析职位数据的自动化采集和处理。在开发过程中，我们重点解决了以下问题：

8.1 技术亮点

  1. 反爬虫机制的突破

    • 使用Selenium模拟真实浏览器
    • 隐藏webdriver特征
    • 随机延时控制请求频率
    • 页面滚动模拟真实用户行为

  2. 数据提取的智能化

    • 正则表达式提取电话、公司规模
    • 关键词匹配提取福利标签
    • 智能分类经验、学历、技能标签
    • BeautifulSoup精准定位页面元素

  3. 程序的稳定性保证

    • 手动登录和搜索，避免自动化失败
    • 完善的异常处理机制
    • 浏览器资源自动清理
    • 数据自动保存为Excel

8.2 实际效果

  • 爬取速度：约5-10分钟/10页（150-200个职位）
  • 数据完整性：11个字段，平均完整度>85%
  • 稳定性：连续运行成功率>95%
  • 适用范围：支持全国范围搜索，支持多种职位关键词

8.3 应用价值

  1. 求职辅助：快速获取大量职位信息，了解市场行情
  2. 薪资分析：统计不同城市、不同经验的薪资水平
  3. 技能需求：分析市场对数据分析师的技能要求
  4. 就业趋势：研究灵活就业市场的发展趋势

九、参考资料

十、声明

本项目仅供学习交流使用，请勿用于商业用途。使用本项目时请遵守相关法律法规，尊重网站的robots协议。对于因使用本项目造成的任何问题，本项目不承担任何责任。

注意事项：

  1. 请勿频繁爬取，建议每次爬取间隔至少1小时
  2. 尊重网站服务条款和用户隐私
  3. 爬取的数据仅供个人学习研究使用
  4. 不得将数据用于商业目的或非法用途

十一、完整代码

详见 yupao_spider_selenium_backup.py 文件

文件说明：

  • 主程序：yupao_spider_selenium_backup.py（555行代码）
  • 依赖文件：requirements.txt
  • 数据示例：鱼泡网_数据分析岗位数据_20251024_193228.xlsx

运行命令：

conda activate dataanalysis
python yupao_spider_selenium_backup.py

技术支持： 如有问题，请查看代码注释或提issue

最后更新： 2025-10-26

版本： v1.0