高考录取分数线爬虫

# -*- coding: utf-8 -*-
'''
作者 : dy
开发时间 : 2021/6/15 17:15
'''
import aiohttp
import asyncio
import pandas as pd
from pathlib import Path
from tqdm import tqdm
import time

current_path = Path.cwd()

def get_url_list(max_id):
    url = 'https://static-data.eol.cn/www/2.0/school/%d/info.json'
    not_crawled = set(range(max_id))
    if Path.exists(Path(current_path, 'college_info.csv')):
        df = pd.read_csv(Path(current_path, 'college_info.csv'))
        not_crawled -= set(df['学校id'].unique())
    return [url%id for id in not_crawled]


async def get_json_data(url, semaphore):
    async with semaphore:
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36',
        }
        async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False), trust_env=True) as session:
            try:
                async with session.get(url=url, headers=headers, timeout=6) as response:
                    # 更改相应数据的编码格式
                    response.encoding = 'utf-8'
                    # 遇到IO请求挂起当前任务，等IO操作完成执行之后的代码，当协程挂起时，事件循环可以去执行其他任务。
                    json_data = await response.json()
                    if json_data != '':
                        # print(f"{url} collection succeeded!")
                        return save_to_csv(json_data['data'])
            except:
                return None


def save_to_csv(json_info):
    save_info = {}
    save_info['学校id'] = json_info['school_id']              # 学校id
    save_info['学校名称'] = json_info['name']                  # 学校名字
    level = ""
    if json_info['f985'] == '1' and json_info['f211'] == '1':
        level += "985 211"
    elif json_info['f211'] == '1':
        level += "211"
    else:
        level += json_info['level_name']
    save_info['学校层次'] = level                               # 学校层次
    save_info['软科排名'] = json_info['rank']['ruanke_rank']    # 软科排名
    save_info['校友会排名'] = json_info['rank']['xyh_rank']     # 校友会排名
    save_info['武书连排名'] = json_info['rank']['wsl_rank']     # 武书连排名
    save_info['QS世界排名'] = json_info['rank']['qs_world']     # QS世界排名
    save_info['US世界排名'] = json_info['rank']['us_rank']      # US世界排名
    save_info['学校类型'] = json_info['type_name']              # 学校类型
    save_info['省份'] = json_info['province_name']              # 省份
    save_info['城市'] = json_info['city_name']                  # 城市名称
    save_info['所处地区'] = json_info['town_name']              # 所处地区
    save_info['招生办电话'] = json_info['phone']                # 招生办电话
    save_info['招生办官网'] = json_info['site']                 # 招生办官网


    df = pd.DataFrame(save_info, index=[0])

    header = False if Path.exists(Path(current_path, 'college_info.csv')) else True
    df.to_csv(Path(current_path, 'college_info.csv'), index=False, mode='a', header=header)


async def main(loop):
    # 获取url列表
    url_list =  get_url_list(5000)
    # 限制并发量
    semaphore = asyncio.Semaphore(500)
    # 创建任务对象并添加到任务列表中
    tasks = [loop.create_task(get_json_data(url, semaphore)) for url in url_list]
    # 挂起任务列表
    for t in tqdm(asyncio.as_completed(tasks), total=len(tasks)):
        await t


if __name__ == '__main__':
    start = time.time()
    # 修改事件循环的策略
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    # 创建事件循环对象
    loop = asyncio.get_event_loop()
    # 将任务添加到事件循环中并运行循环直至完成
    loop.run_until_complete(main(loop))
    # 关闭事件循环对象
    loop.close()
    df = pd.read_csv(Path(current_path, 'college_info.csv'))
    df.drop_duplicates(keep='first', inplace=True)
    df.reset_index(drop=True, inplace=True)
    df.sort_values('学校id', inplace=True)
    df.loc[df['软科排名'] == 0, '软科排名'] = 999
    df.to_csv(Path(current_path, 'college_info.csv'), index=False)
    print(f'采集完成，共耗时：{round(time.time() - start, 2) } 秒')+680

转自：1分钟爬取全国高校信息，制成大屏可视化！_俊红的数据分析之路的博客-CSDN博客