如何获取掌上高考（招生计划）信息

爬取掌上高考（招生计划）

厦门航空没有反爬机制，较为简单（返回数据较为复杂），现在我们爬取掌上高考作为练手项目，相较于网上爬取豆瓣电影以及其他小demo,这个项目复杂的多。掌上高考有复杂的反爬机制，非常适合练手，同时可以自己整理自己的框架，这样针对下一个爬虫任务来说，只需要做小部分的改动就行。

1.抓包

请求网址：2024河南大学分数线_招生计划|掌上高考 (gaokao.cn)

通过点击下一页，分析返回的数据以及数据响应。

此时为我们需要的数据，分析请求头：

我们可以对url进行拼接，得到我们需要的数据，比如说更换学校id就可以查询不同的学校。再观察请求数据，如下：

后面发现这个data并没有什么暖用，不需要对其进行构造，就算构造以后也会被url查询字符串给覆盖，因此我们仅仅需要构造url就行

2.发起请求

配置请求头，请求数据和url之后，就可以直接发起请求了，构造请求数据

def get_config(page,id):
    # 打开你的json文件
    date = {"local_batch_id":7,"local_province_id":"41","local_type_id":"2","page":1,"school_id":"459","signsafe":"bbddec69cb7fe50f4d9ea21404d72fcf","size":10,"special_group":"","uri":"apidata/api/gkv3/plan/school","year":2023}
    herders = {"Accept":"application/json, text/plain, */*",
               "User-Agent": UserAgent().random}
    url = "https://api.eol.cn/web/api/?local_batch_id=7&local_province_id=41&local_type_id=1&page={}&school_id={}&size=10&special_group=&uri=apidata/api/gkv3/plan/school&year=2023".format(page,id)
    return date,herders,url

发起请求

def request_api(url, params, header):
    try:
        response = requests.post(url, json=params, headers=header)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"请求失败，状态码：{response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"请求出现异常：{e}")
        return None

成功获取到数据，获取数据如下

分析发现，一次请求只能获取到10行数据，因此我们需要对其进行翻页处理，分析返回的数据携带数据的总条数，因此，我们可以根据这个总条数进行翻页处理，需要再次发送请求（（可以修改size这个参数，减少请求次数，再提高效率的同时，做到反爬效果，但是仍需要动态的获取页数））

翻页处理

    result_numFound = result['numFound'] #获取总条数
    page_list = [i + 1 for i in range(int(result_numFound / 10) + 1)] #构造页码，用作参数重新构造请求头
    #循环发起请求
    for page in page_list:
        result_page = date, herders,url= get_config(page,i)

3.提取数据

同样根据自己的需要提取数据，我提取的数据如下：

data1 = {}
data1['name'] = result[i]['name']
data1['province_name'] = result[i]['province_name']
data1['num'] = result[i]['num']
data1['spname'] = result[i]['spname']
data1['spcode'] = result[i]['spcode']
data1['length'] = result[i]['length']
data1['level2_name'] = result[i]['level2_name']
data1['local_batch_name'] = result[i]['local_batch_name']
data1['local_type_name'] = result[i]['local_type_name']

此时已经完成了单页面的数据爬取，效果如下：

4.循环采集多所学校

通过构造不同的学校id对学校进行循环爬取，将学校id以参数的方式输入

 url = "https://api.eol.cn/web/api/?local_batch_id=7&local_province_id=41&local_type_id=1&page={}&school_id={}&size=10&special_group=&uri=apidata/api/gkv3/plan/school&year=2023".format(page,id)

其中page是页面，id是学校id

5.多线程爬取

将前面的爬虫封装为单个函数，同时将循环爬取学校提取到多线程函数里面，将学校id再线程池里面提交，结果如下：

def spider(i):
    #读取参数
    date, herders,url= get_config(1,i)
    proxy = get_proxy().get("proxy")
    result = request_api(url,date,herders,proxy)['data']
    result_numFound = result['numFound']
    result_item = result['item']
    page_list = [i + 1 for i in range(int(result_numFound / 10) + 1)]
    for page in page_list:
        result_page = date, herders,url= get_config(page,i)
        result = request_api(url, date, herders,proxy)['data']['item']
        get_data(temp,result)
    index = 0
 
max_workers = 10
t = []
t1 = time.time()
threadPool = ThreadPoolExecutor(max_workers)  # 创建最大线程数为

for i in range(32,34):  # 循环向线程池中提交task任务
    future = threadPool.submit(spider, i)
    t.append(future)
    # 若不需要获取返回值，则可不需要下面两行代码
    for i in t:
        print(i.result())  # 获取每个任务的返回值，result()会阻塞主线程

        threadPool.shutdown()  # 阻塞主线程，所有任务执行完后关闭线程池

此时可以实现多线程爬取，可以先把循环写小进行测试，可以很快的爬取数据，不要写大了，容易被封！！

6.反爬策略

不仅仅需要一个伪装头就行，要保证每一次请求头都尽量不同，因此我们可以网上收集一部分伪装头，然后随机的读取，或者使用工具类的方法：

from fake_useragent import UserAgent
"User-Agent": UserAgent().random

这样比较简单，也可自己写一个工具类

如果每一次请求都睡眠相同的时间（请求的时间不相同），会导致后台监控我们像爬虫程序，因此我们需要每一次请求的时间进行随机休眠。设置方法如下：

time.sleep(random.uniform(1,3))

7.IP代理技术

上面的反爬策略仅仅能缓解一部分反爬机制，最核心的还是需要使用IP代理技术，如果使用IP代理？可以再网上找一些免费的代理，但是一般免费的成功率都很低，而收费的都很贵，因此我们都不采用。

我们可以使用网上的IP代理池项目ProxyPool ，来动态的获取IP，实测效果很不错。

git clone git@github.com:jhao104/proxy_pool.git

pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/

最好加上镜像，否则很慢。其他镜像地址如下：

https://pypi.tuna.tsinghua.edu.cn/simple 清华
https://pypi.mirrors.ustc.edu.cn/simple  中科大
http://pypi.douban.com/simple/  豆瓣
https://mirrors.aliyun.com/pypi/simple/ 阿里镜像
http://pypi.hustunique.com/  华中科大
https://mirror.baidu.com/pypi/simple  百度

# setting.py 为项目配置文件

# 配置API服务

HOST = "0.0.0.0"               # IP
PORT = 5000                    # 监听端口


# 配置数据库

DB_CONN = 'redis://:pwd@127.0.0.1:8888/0'

# 配置 ProxyFetcher

PROXY_FETCHER = [
    "freeProxy01",      # 这里是启用的代理抓取方法名，所有fetch方法位于fetcher/proxyFetcher.py
    "freeProxy02",
    # ....
]

主要修改的几项配置是监听端口（PORT）、 Redis 数据库的配置（DB_CONN）和启用的代理方法名（PROXY_FETCHER）。

需要安装redis环境，存储再redis,可以使用docker随便装一个，百度一下教程！！

这个项目总体分为两个部分：爬取代理 IP 和 取用代理 IP。

如果你要启用爬取代理 IP 的服务，直接运行下面命令：

python proxyPool.py schedule

其实，作者在这个项目中运用的原来就是到一些免费的代理网站采集 IP，然后测试 IP 的可用性，可用的就存入 Redis 中，不可用就丢弃。

python proxyPool.py server

会使用flask搭建一个简单的服务，还需要自己手动封装一下

封装一个获取Ip的函数

def get_proxy():
    return requests.get("http://127.0.0.1:5010/get/").json()

可以自己测试一下，每次获取到的IP都不同！因此反爬策略做好，这个代理技术不仅仅可以做爬虫，好可以根据自己的需求干其他的

此时需要修改请求的代码，需要把ip传入

def request_api(url, params, header,proxy):
    try:
        response = requests.post(url, json=params, headers=header,proxies={"http": "http://{}".format(proxy)})
        if response.status_code == 200:
            return response.json()
        else:
            print(f"请求失败，状态码：{response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"请求出现异常：{e}")
        return None

8.保存文件

保存为两种格式

    path = "河南大学"
    with open(path + ".json", "w", encoding='utf-8') as f:
        json.dump(temp, f, ensure_ascii=False)
    pd.read_json(path + ".json", orient='records').T.to_excel(path + ".xlsx", index=False)

项目的复用性很高，比如说多线程，浏览器伪装和IP代理技术可以很快迁移到其他爬虫任务。

完整代码

import time
import pandas as pd
import requests
import json
from fake_useragent import UserAgent
import random
from concurrent.futures import ThreadPoolExecutor  # 导入ThreadPoolExecutor模块

temp = {}
index = 0

#加载配置文件,请求参数和请求头
def get_config(page,id):
    # 打开你的json文件
    date = {"local_batch_id":7,"local_province_id":"41","local_type_id":"2","page":1,"school_id":"459","signsafe":"bbddec69cb7fe50f4d9ea21404d72fcf","size":10,"special_group":"","uri":"apidata/api/gkv3/plan/school","year":2023}
    herders = {"Accept":"application/json, text/plain, */*",
               "User-Agent": UserAgent().random}
    url = "https://api.eol.cn/web/api/?local_batch_id=7&local_province_id=41&local_type_id=1&page={}&school_id={}&size=10&special_group=&uri=apidata/api/gkv3/plan/school&year=2023".format(page,id)
    return date,herders,url
#获取Ip地址
def get_proxy():
    return requests.get("http://127.0.0.1:5010/get/").json()
#发起请求
def request_api(url, params, header,proxy):
    try:
        response = requests.post(url, json=params, headers=header,proxies={"http": "http://{}".format(proxy)})
        if response.status_code == 200:
            return response.json()
        else:
            print(f"请求失败，状态码：{response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"请求出现异常：{e}")
        return None


#发起请求，需要发两次
def spider(i):
    #读取参数
    date, herders,url= get_config(1,i)
    proxy = get_proxy().get("proxy")
    result = request_api(url,date,herders,proxy)['data']
    result_numFound = result['numFound']
    result_item = result['item']
    page_list = [i + 1 for i in range(int(result_numFound / 10) + 1)]
    for page in page_list:
        result_page = date, herders,url= get_config(page,i)
        result = request_api(url, date, herders,proxy)['data']['item']
        time.sleep(random.uniform(1,3))
        get_data(temp,result)
    index = 0
#数据提取
def get_data(data,result):
    global index
    for i in range(len(result)):
        data1 = {}
        data1['name'] = result[i]['name']
        data1['province_name'] = result[i]['province_name']
        data1['num'] = result[i]['num']
        data1['spname'] = result[i]['spname']
        data1['spcode'] = result[i]['spcode']
        data1['length'] = result[i]['length']
        data1['level2_name'] = result[i]['level2_name']
        data1['local_batch_name'] = result[i]['local_batch_name']
        data1['local_type_name'] = result[i]['local_type_name']
        data[index] = data1
        print(data1)
        index += 1

#多线程调用爬虫
def main():

    max_workers = 10
    t = []
    t1 = time.time()
    threadPool = ThreadPoolExecutor(max_workers)  # 创建最大线程数为

    for i in range(32,34):  # 循环向线程池中提交task任务
        future = threadPool.submit(spider, i)
        t.append(future)
    # 若不需要获取返回值，则可不需要下面两行代码
    for i in t:
        print(i.result())  # 获取每个任务的返回值，result()会阻塞主线程

    threadPool.shutdown()  # 阻塞主线程，所有任务执行完后关闭线程池

    path = "河南大学"
    with open(path + ".json", "w", encoding='utf-8') as f:
        json.dump(temp, f, ensure_ascii=False)
    pd.read_json(path + ".json", orient='records').T.to_excel(path + ".xlsx", index=False)
    print(time.time() - t1)
if __name__ == '__main__':
    main()

作者：dxw-1997