---

前言

为了更好的掌握数据处理的能力，因而开启Python网络爬虫系列小项目文章。

1. 小项目小需求驱动，每篇文章会使用两种以上的方式（Xpath、Bs4、PyQuery、正则等）获取想要的数据。
2. 博客系列完结后，将会总结各种方式。

一、需求

• 获取高校数据

  • 高校排行（软科综合、校友会综合、武书连）
  • 基本信息（博士点、硕士点、创建时间、占地面积、学校地址）

• 多种数据存储方式

  • 本地文件存储

  • redis

  • MongoDB

二、分析

F12 打开抓包工具，刷新https://www.gaokao.cn/school/search接口页面

name.json 接口包含所有的高校id

进入学校详情页，获取到info.json接口信息

数据可以直接通过请求接口即可返回
已获取学校id（一共2820个数据，可以用上协程异步访问提高获取的数据的速度）
已获取学校信息的接口https://static-gkcx.gaokao.cn/www/2.0/json/live/v2/school/102/info.json（只需要修改下学校id即可访问其它学校信息）

三、处理

编写获取大学id的接口

import requests
import json
HEADERS = {"user-agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36"}

# 获取大学school_id
def get_school_id():
    url= "https://static-data.gaokao.cn/www/2.0/school/name.json"
    res = requests.request("GET",url=url,headers=HEADERS)
    res.encoding = "utf-8"
    # 转为json格式
    data = json.loads(res.text)
    return data

获取一次以后，本地可以进行处理提取学校id

import re
def get_id():
    f = open("name.txt", "r", encoding="utf-8")
    f1 = f.read()
    deal = re.compile(r"'school_id': '(?P<id>.*?)'",re.S)
    result = deal.findall(f1)
    # print(len(result))
    f.close()
    return  result

编写访问学校信息的接口请求

def get_school_info_1():
    url = "https://static-data.gaokao.cn/www/2.0/school/3419/info.json"
    res = requests.request("GET",url=url,headers=HEADERS)
    res.encoding = "utf-8"
    # 转为json格式
    data = json.loads(res.text)
    return data

协程异步访问接口请求（本地文件存储）

import asyncio
import aiohttp
import aiofiles


# 获取学校详细信息
async def get_school_info(url,sem):
    try:
        async with sem:
            async with aiohttp.ClientSession() as session:
                # 发送请求
                async with session.get(url) as res:
                    # 读取数据
                    content = await res.text(encoding="utf-8")
                    data = json.loads(content)
                    # 高校排行- 软科综合/校友会排名/武书连/US世界
                    school_info = {}
                    school_info["学校名称"] = data["data"]["name"]
                    school_info["软科综合"] = data["data"]["ruanke_rank"]
                    school_info["武书连"] = data["data"]["wsl_rank"]
                    school_info["校友会"] = data["data"]["xyh_rank"]
                    school_info["US世界"] = data["data"]["us_rank"]
                    # 博士点/硕士点/国家重点学科
                    school_info["博士点"] = data["data"]["num_doctor"]
                    school_info["硕士点"] = data["data"]["num_master"]
                    school_info["国家重点学科"] = data["data"]["num_subject"]
                    # 创建时间/占地面积/学校地址
                    school_info["创建时间"] = data["data"]["create_date"]
                    school_info["占地面积"] = data["data"]["area"]
                    school_info["学校地址"] = data["data"]["address"]
                    # 写入文件
                    async with aiofiles.open("school_info.txt",mode="a") as f:
                        await f.write(f"{str(school_info)}\n")
                        print("一条数据下载完毕！")
    except Exception as e:
        print(e)
 
 async def main(list_id):
    # 添加信号量控制速度
    sem = asyncio.Semaphore(100)
    tasks = []
    for i in list_id:
        # https://static-data.gaokao.cn/www/2.0/school/3419/info.json
        url = f"https://static-data.gaokao.cn/www/2.0/school/{i}/info.json"
        # 创建任务
        task = asyncio.create_task(get_school_info(url,sem))
        tasks.append(task)
    await asyncio.wait(tasks)



if __name__ == '__main__':
    # 本地获取id
    id_list = get_id()
  	asyncio.run(main(id_list))

写入Redis数据库，
如有报错DataError: Invalid input of type: ‘NoneType’. Convert to a byte, string or number first.
解决方法，降低redis版本 pip install redis==2.10.6

# 获取学校详细信息
async def get_school_info(url,sem):
    try:
        async with sem:
            async with aiohttp.ClientSession() as session:
                # 发送请求
                async with session.get(url) as res:
                    # 读取数据
                    content = await res.text(encoding="utf-8")
                    data = json.loads(content,encoding="utf-8")
                    # 高校排行- 软科综合/校友会排名/武书连/US世界
                    school_info = {}
                    school_name = data["data"]["name"]
                    school_info["软科综合"] = data["data"]["ruanke_rank"]
                    school_info["武书连"] = data["data"]["wsl_rank"]
                    school_info["校友会"] = data["data"]["xyh_rank"]
                    school_info["US世界"] = data["data"]["us_rank"]
                    # 博士点/硕士点/国家重点学科
                    school_info["博士点"] = data["data"]["num_doctor"]
                    school_info["硕士点"] = data["data"]["num_master"]
                    school_info["国家重点学科"] = data["data"]["num_subject"]
                    # 创建时间/占地面积/学校地址
                    school_info["创建时间"] = data["data"]["create_date"]
                    school_info["占地面积"] = data["data"]["area"]
                    school_info["学校地址"] = data["data"]["address"]
                    # 写入redis文件中
                    r = connect_redis()
                    r.hset("school",school_name,json.dumps(school_info))
                    print("一条数据写入成功")
    except Exception as e:
        print(e)
  
  # 连接Redis数据库
def connect_redis():
    pool  = redis.ConnectionPool(host="127.0.0.1", port=6379,db=3)
    r = redis.Redis(connection_pool=pool)
    return r

写入MongoDB数据库

import pymongo
from pymongo import MongoClient
# MongoDB数据库初始化
def get_db(database,user=None,pwd=None):
    client = MongoClient(host="localhost",port=27017)
    db = client[database]
    # 有账号密码即验证
    if user:
        db.authenticate(user,pwd)
    return db

# 增加多条数据
def insert_data_many(collection, data):
    db = get_db("school")
    result = db[collection].insert_many(data)
    return result

async def get_s
chool_info(url,sem):
    try:
        async with sem:
            async with aiohttp.ClientSession() as session:
                # 发送请求
                async with session.get(url) as res:
                    # 读取数据
                    content = await res.text(encoding="utf-8")
                    data = json.loads(content,encoding="utf-8")
                    # 高校排行- 软科综合/校友会排名/武书连/US世界
                    school_info = {}
                    school_info["学校名称"] = data["data"]["name"]
                    school_info["软科综合"] = data["data"]["ruanke_rank"]
                    school_info["武书连"] = data["data"]["wsl_rank"]
                    school_info["校友会"] = data["data"]["xyh_rank"]
                    school_info["US世界"] = data["data"]["us_rank"]
                    # 博士点/硕士点/国家重点学科
                    school_info["博士点"] = data["data"]["num_doctor"]
                    school_info["硕士点"] = data["data"]["num_master"]
                    school_info["国家重点学科"] = data["data"]["num_subject"]
                    # 创建时间/占地面积/学校地址
                    school_info["创建时间"] = data["data"]["create_date"]
                    school_info["占地面积"] = data["data"]["area"]
                    school_info["学校地址"] = data["data"]["address"]
                    insert_data_many("school",[school_info])
                    print("一条数据写入成功")
    except Exception as e:
        print(e)
        

四、运行效果

通过Redis可视化工具查看写入的2821条数据

MongoDB可视化工具查询数据

本地读取数据

完整源码文件附在在知识星球
我正在「Python^self-study」和朋友们讨论有趣的话题，你⼀起来吧？
https://t.zsxq.com/076uG3kOn