鱼皮面试鸭面试题爬虫：从API接口到批量数据采集 | ZapcoMan's world

前言

在准备Java面试的过程中，拥有一个全面、结构化的面试题库是非常重要的。鱼皮开发的”面试鸭”平台聚合了大量优质面试题，涵盖Java基础、并发、Spring、MySQL、Redis等核心技术栈。然而，手动浏览网页逐题学习效率较低，如果能将这些面试题批量抓取到本地，不仅可以离线复习，还能进行关键词检索和分类整理。

本文基于真实项目源码，介绍如何通过分析面试鸭平台的API接口，使用Python编写爬虫程序，实现面试题数据的批量采集与本地化存储。我们将从最基础的版本出发，逐步迭代优化，最终实现一个具备多线程并发、随机User-Agent、自动终止判断等功能的生产级爬虫。

技术背景

本项目的核心技术栈包括：

• requests：发送HTTP POST请求，获取API返回的JSON数据
• json：解析和格式化存储API响应数据
• os：文件目录管理，确保输出路径存在
• time：请求间隔控制，避免触发反爬机制
• random：随机User-Agent选择，模拟不同浏览器环境
• concurrent.futures.ThreadPoolExecutor：多线程并发爬取，提升采集效率

面试鸭平台采用前后端分离架构，前端通过调用 api.mianshiya.com 下的RESTful API获取数据。我们通过浏览器开发者工具分析网络请求，发现面试题列表接口为 POST /api/question_bank/list_question，题库分类接口为 POST /api/questionBankCategory/list_questionBank，两者均接受JSON格式的请求体。

实现思路

整体爬虫架构分为两阶段：

1. 获取题库列表：调用题库分类API，获取所有面试题库的ID和标题（如”Java基础面试题”、”Spring面试题”等）
2. 逐库爬取题目：对每个题库ID，分页请求题目列表API，将JSON数据写入本地文件，同时提取题目标题写入TXT文件

迭代过程分为四个版本：

• V1：硬编码题库ID，基础分页请求
• V2：封装为函数，支持单个/多个ID传入
• V3：动态获取题库列表，自动遍历所有分类，增加异常处理
• V4：多线程并发 + 随机User-Agent + 超时控制 + 自动终止判断

核心代码解析

1. 基础版本：分页请求与数据写入

最初的版本直接硬编码题库ID，循环请求每一页数据：

import requests
import json

url = "https://api.mianshiya.com/api/question_bank/list_question"

# 多个题库ID
question_bank_ids = [
    "1860871861809897474",
    "1787463103423897602",
    "1789249312885223425",
    "1791375592078610434",
]

# 去重处理：将列表转为集合去重，再转回列表
question_bank_ids = list(set(question_bank_ids))

payload = {
    "current": 1,          # 当前页码
    "pageSize": 20,         # 每页条数
    "questionBankId": "",   # 题库ID（动态填充）
    "tagList": []
}

headers = {
    'authority': 'api.mianshiya.com',
    'method': 'POST',
    'path': '/api/question_bank/list_question',
    'scheme': 'https',
    'accept': '*/*',
    'content-type': 'application/json',
    'origin': 'https://www.mianshiya.com',
    'referer': 'https://www.mianshiya.com/',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}

# 遍历每个题库ID
for question_bank_id in question_bank_ids:
    payload["questionBankId"] = question_bank_id
    # 每个题库爬取20页数据
    for current in range(1, 20):
        payload['current'] = current
        response = requests.post(url, json=payload, headers=headers)
        if response.status_code == 200:
            data = response.json()
            filename = f"output_{question_bank_id}.json"
            # 以追加模式写入JSON文件，保持中文可读
            with open(filename, 'a', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=4)
                f.write('\n')  # 换行分隔每次请求的数据
            print(f"已将第 {current} 页的数据写入文件：{filename}")
        else:
            print(f"获取第 {current} 页数据失败，状态码：{response.status_code}")

这里的关键点是通过 list(set(...)) 对题库ID去重，避免重复请求。json.dump 的 ensure_ascii=False 参数确保中文字符正常显示。

2. 函数封装：支持灵活调用

将核心逻辑封装为函数，支持字符串（单个ID）或列表（多个ID）传入：

def fetch_and_save_questions(question_bank_id):
    """
    根据 question_bank_id 获取面试题数据并保存到本地 JSON 文件中

    :param question_bank_id: 题库ID，可以是字符串（单个ID）或列表（多个ID）
    """
    url = "https://api.mianshiya.com/api/question_bank/list_question"

    payload = {
        "current": 1,
        "pageSize": 20,
        "questionBankId": "",
        "tagList": []
    }

    headers = { /* 请求头配置 */ }

    # 如果是字符串，转成列表统一处理
    if isinstance(question_bank_id, str):
        question_bank_ids = [question_bank_id]
    else:
        question_bank_ids = question_bank_id

    for question_bank_id in question_bank_ids:
        payload["questionBankId"] = question_bank_id
        for current in range(1, 11):  # 每个题库爬取10页
            payload['current'] = current
            response = requests.post(url, json=payload, headers=headers)
            if response.status_code == 200:
                data = response.json()
                filename = f"output_{question_bank_id}.json"
                with open(filename, 'a', encoding='utf-8') as f:
                    json.dump(data, f, ensure_ascii=False, indent=4)
                    f.write('\n')
                print(f"已将第 {current} 页的数据写入文件：{filename}")

# 示例调用
if __name__ == "__main__":
    fetch_and_save_questions("1860871861809897474")

3. 动态获取题库列表 + 标题提取

第三版通过题库分类API动态获取所有题库信息，不再需要手动查找ID：

import os
import time

# 创建输出目录
output_dir = "output"
os.makedirs(output_dir, exist_ok=True)

def fetch_and_save_questions(question_bank_id, title):
    """
    获取面试题数据并保存到本地，同时提取题目标题写入TXT文件

    :param question_bank_id: 题库ID（字符串）
    :param title: 对应标题，用于命名输出文件
    """
    url = "https://api.mianshiya.com/api/question_bank/list_question"

    for current in range(1, 11):
        payload['current'] = current
        payload["questionBankId"] = question_bank_id
        try:
            response = requests.post(url, json=payload, headers=headers)
            if response.status_code == 200:
                data = response.json()
                filename = os.path.join(output_dir, f"{title}.json")
                txt_filename = os.path.join(output_dir, f"{title}.txt")

                # 写入原始JSON数据
                with open(filename, 'a', encoding='utf-8') as f:
                    json.dump(data, f, ensure_ascii=False, indent=4)
                    f.write('\n')

                # 提取 data.records 中的 title 字段，写入TXT
                records = data.get("data", {}).get("records", [])
                titles = [record.get("title") for record in records if record.get("title")]

                if titles:
                    with open(txt_filename, 'w', encoding='utf-8') as f:
                        for t in titles:
                            f.write(f"{t}\n")
                    print(f"已将第 {current} 页的标题写入文件：{txt_filename}")

                # 关键优化：records为空时提前终止，避免无效请求
                if current > 1 and not records:
                    print(f"第 {current} 页 records 为空，数据已请求完毕，切换到下一个题库...")
                    break
            else:
                print(f"获取第 {current} 页数据失败，状态码：{response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"请求第 {current} 页时发生异常：{e}")

        time.sleep(2)  # 每次请求后等待2秒，避免请求过快


# 第一步：调用题库分类API，获取所有题库的ID和标题
url = "https://api.mianshiya.com/api/questionBankCategory/list_questionBank"
data = {
    "current": 1,
    "pageSize": 200,   # 一次获取所有题库
    "isHasChoice": False,
    "questionBankCategoryId": "1821883312558432257"
}

response = requests.post(url, json=data, headers=headers)
response_json = response.json()

# 提取 records 中的 id 和 title
records = response_json.get("data", {}).get("records", [])
extracted_data = [{"id": record.get("id"), "title": record.get("title")} for record in records]

# 第二步：遍历所有题库，逐个爬取面试题
for item in extracted_data:
    question_bank_id = item.get("id")
    title = item.get("title")
    if question_bank_id and title:
        fetch_and_save_questions(question_bank_id, title)

这个版本的亮点在于 if current > 1 and not records: break —— 当第二页开始返回空数据时，自动跳出循环，避免对没有更多内容的题库发起无效请求。

4. 最终优化版：多线程并发 + 随机反爬

最终版本引入了线程池和多重反爬策略：

import random
from concurrent.futures import ThreadPoolExecutor

# 多组User-Agent，模拟不同浏览器和设备
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"
]

def get_session_with_headers():
    """创建带默认 headers 的 session，并随机设置 User-Agent"""
    session = requests.Session()
    session.headers.update({
        'authority': 'api.mianshiya.com',
        'accept': '*/*',
        'content-type': 'application/json',
        'origin': 'https://www.mianshiya.com',
        'referer': 'https://www.mianshiya.com/',
        'user-agent': random.choice(USER_AGENTS)  # 每次创建session随机选择
    })
    return session


def fetch_and_save_questions(session, question_bank_id, title):
    """获取面试题数据并保存到本地 JSON 和 TXT 文件中"""
    url = "https://api.mianshiya.com/api/question_bank/list_question"
    filename = os.path.join(output_dir, f"{title}.json")
    txt_filename = os.path.join(txt_output_dir, f"{title}.txt")
    all_data = []

    for current in range(1, 11):
        payload['current'] = current
        payload["questionBankId"] = question_bank_id

        try:
            # 设置连接超时3秒，读取超时10秒
            response = session.post(url, json=payload, timeout=(3, 10))
            if response.status_code == 200:
                data = response.json()
                all_data.append(data)

                records = data.get("data", {}).get("records", [])
                titles = [record.get("title") for record in records if record.get("title")]

                if titles:
                    with open(txt_filename, 'w', encoding='utf-8') as f:
                        for t in titles:
                            f.write(f"{t}\n")

                # records为空时提前终止
                if current > 1 and not records:
                    break
            else:
                break
        except requests.exceptions.RequestException as e:
            print(f"请求第 {current} 页时发生异常：{e}")
            continue  # 异常时跳过当前页，继续下一页

        # 随机延时4~15秒，模拟人类操作节奏
        time.sleep(random.uniform(4, 15))

    # 所有数据收集完成后统一写入JSON文件（避免频繁IO）
    if all_data:
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(all_data, f, ensure_ascii=False, indent=4)


def main():
    MAX_THREADS = 10  # 最大并发线程数

    with get_session_with_headers() as session:
        # 获取题库列表
        url = "https://api.mianshiya.com/api/questionBankCategory/list_questionBank"
        data = {
            "current": 1,
            "pageSize": 20,
            "isHasChoice": False,
            "questionBankCategoryId": "1821883312558432257"
        }

        response = session.post(url, json=data, timeout=(3, 10))
        response.raise_for_status()
        response_json = response.json()

        records = response_json.get("data", {}).get("records", [])
        extracted_data = [{"id": r.get("id"), "title": r.get("title")} for r in records]

        # 使用线程池并发执行爬取任务
        with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
            futures = []
            for item in extracted_data:
                question_bank_id = item.get("id")
                title = item.get("title")
                if question_bank_id and title:
                    futures.append(executor.submit(
                        fetch_and_save_questions, session, question_bank_id, title
                    ))

            # 等待所有任务完成
            for future in futures:
                try:
                    future.result()
                except Exception as e:
                    print(f"线程任务出错: {e}")

关键代码汇总

最终版本的完整执行流程如下：

if __name__ == "__main__":
    main()

核心流程：

1. get_session_with_headers() 创建带随机UA的Session
2. 调用题库分类API获取所有题库的ID和标题
3. 使用 ThreadPoolExecutor(max_workers=10) 创建10个线程
4. 每个线程负责一个题库，分页爬取面试题
5. 数据先缓存到内存 all_data，爬取完毕后统一写入文件
6. 随机延时4~15秒，有效规避频率检测

运行效果

程序运行后，控制台输出类似：

JSON 数据已写入文件: 优化强化output/response_output.json
已提取 records 中的 id 和 title，并保存到文件: 优化强化output/extracted_records.json
已将第 1 页的数据写入缓存
已将标题写入文件：优化强化txt/Java 基础面试题.txt
已将第 2 页的数据写入缓存
...
第 6 页 records 为空，判断数据已请求完毕，切换到下一个题库 ID...
已将全部数据写入文件：优化强化output/Java 基础面试题.json

运行结束后，优化强化output/ 目录下生成多个JSON文件，优化强化txt/ 目录下生成对应的TXT文件，内容为各题库的面试题标题列表。

总结与优化方向

本项目从基础的API调用出发，逐步迭代为多线程并发爬虫，核心优化包括：

• 去重处理：list(set(question_bank_ids)) 避免重复爬取
• 自动终止：检测 records 为空时 break，减少无效请求
• 随机延时：random.uniform(4, 15) 模拟人类操作节奏
• 多线程并发：ThreadPoolExecutor 提升整体采集效率约10倍
• 统一写入：数据先缓存到内存，爬取完成后一次性写入，减少IO开销
• 异常容错：try/except 捕获网络异常，continue 跳过失败页

后续优化方向：

1. 断点续传：记录已完成的题库ID，支持中断后恢复
2. 代理IP池：引入代理IP，进一步降低被封风险
3. 数据清洗：对JSON数据进行结构化提取，导入数据库
4. 增量更新：定期运行，只抓取新增的面试题