位置：首页-资讯-数据库

【微博爬虫教程&实例】基于requests、mysql爬取大数据量博主关键字下博文及评论

2023-10-21 20:56

短信预约 -IT技能 免费直播动态提醒

【关键词：手把手教程、反爬、数据库、python爬虫、微博关键词爬虫、较大数据量、数据简单过滤】

本教程适合微博相关爬虫需求者阅读，完整实例源码将放置在文末github链接中。

该实例针对微博的反爬措施进行优化，可实现较大数据量的数据爬取需求（十万量级）。

项目准备

Python & Pycharm

主要使用到的库有：

requests：requests库是Python中用来模拟浏览器发送网络请求与得到响应数据包的库，可以说是实现爬虫原理的核心

json：用于解析（如格式化为方便python处理的字典格式）得到的响应数据包，以便于后续对得到的数据进行操作处理

pymysql：是python下用于对mysql数据库进行相关操作的一个库，本项目中用于将爬取数据加入到本地构建好的数据库中，以及其他的相关操作（增删改查）

Mysql数据库 & Datagrip

为了适用本项目较大规模数据的存放处理以及后续的使用，因此使用了数据库。

如果数据量不大或没有较高的数据分析需要，你也可以考虑将数据以.csv或.xlsx的格式保存（该方式直接在python源码中即可完成，可以看看其他教程）

本项目使用本地sql数据库存放爬取数据，为方便数据库的相关使用，选用Datagrip作为IDE对数据库进行一些操作。

Datagrip界面如下：

需求分析

本实例希望实现对账号特定时间段内按关键字筛选出的所有博文的爬取，并获得博文评论与评论者个人信息。

数据库结构设计

爬取的数据分博主数据、博文数据、博文评论数据（包含评论者的信息）三种类型。因此我们需要在数据库中建立blogger_table、blog_table、comment_table三个数据表分别存储。

各数据表间通过索引相关联（如comment_table中一条评论数据可由索引对应到blog_table中其所属的博文数据行），以此实现数据间的联系。

对每一个数据表，其内容是这样的：

生成该表结构的mysql语句如下:

create table blog_base(    id_blog_base int(11) NOT NULL AUTO_INCREMENT,    blog_date char(30) NOT NULL DEFAULT 'null' COMMENT '博文发布时间',    blog_text varchar(6000) NOT NULL DEFAULT 'null' COMMENT '博文内容',    blog_id varchar(20) NOT NULL DEFAULT 'null' COMMENT '博文id',    blog_comments int NOT NULL DEFAULT -100 COMMENT '博文评论量',    blog_likes int NOT NULL DEFAULT -100 COMMENT '博文点赞',    blog_keywords varchar(100) NOT NULL DEFAULT 'null' COMMENT '博文关键词',    retweeted_blogger_name varchar(50) NOT NULL DEFAULT 'null' COMMENT '转推原博主名称',    retweeted_verified_type int NOT NULL DEFAULT -100 COMMENT '转推原博主认证',    retweeted_text varchar(3000) NOT NULL DEFAULT 'null' COMMENT '转推原文',    id_blogger varchar(20) NOT NULL DEFAULT 'null' COMMENT '所属博主id',    PRIMARY KEY (id_blog_base))ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

数据定位

阅读本内容请先搜索requests库的基本使用，了解get请求与响应的基本概念

定位博主个人信息

1、浏览器打开微博网站（pc版，请务必登录），点开任一博主的个人界面，f12打开开发者工具。

2、选中XHR

3、选中该文件

在右侧栏中找到响应，就可以得到博主的个人信息了（如screen_name、location、gender)。

稍后我们在编写过程中使用requests包模拟浏览器得到该JSON响应文件的过程，然后获得如下图所示的JSON数据，再使用json包将该数据格式化为字典，方便获得想要的数据项。

定位博主的博文信息

微博博文的响应数据包是按照一页多条blog来发送的，该响应数据包对应的文件名为：

uid这里指博主的用户id，page取不同数是不同页的博文数据包。

下方的若干Object 每一项均为一条博文

点开一条博文的评论然后进入该博文的详情页，可以得到文件名为：

鼠标滚轮往下翻评论的过程中会刷新出现若干新评论的响应数据包。但值得注意的是，与上一个博文定位不同，此时各数据包的区分并不是用page的增加来实现的

该博文的评论区，除了最前面的若干评论（对应文件列表里未刷新时的唯一一个评论数据包）以外，其余评论数据包的GET请求参数都多了一项参数"max_id" (注意下列两项的GET参数区分，其中一个含有max_id参数）。

博文的定位中，我们修改page的值然后模拟发送即可获得所有的博文内容，而此处我们无法采用同样的操作——很显然，max_id的数值不是按照page那样自增的规律。

该如何获得该博文的所有的评论呢？我们需要解决如何获得max_id值的问题，否则我们无法正确模拟发送获得后续评论的请求。

解决方法：

1、你会发现前一个评论数据包的响应JSON包含了一个max_id（如图）,这个max_id其实是下一个评论数据包GET请求中所需的max_id值

2、因此，只需要获取首个评论数据包的响应JSON中的max_id值，我们就可以获得第二个评论数据包，第二个评论数据包的响应JSON中又包含了一个max_id值，对应下一个评论数据包的GET请求参数，以此类推。最后一个评论数据包的响应JSON不含max_id，可以作为python脚本中该部分过程结束的判断条件。

3、这样我们就可以获得全部的评论信息了

评论用户个人信息定位

该部分的采集伴随评论的采集（评论数据包会同时包含评论者的大部分个人信息）。

代码实现

以下是基本功能的实现过程，下一步会对各函数进行修改以完善。

从用户信息爬起

关于requests库的基本使用在此不做叙述，请查阅其他文章。

1、需要传入的headers项中，我设置了如下四项以确保获得响应JSON。

项目中的cookie我定义为了全局变量，以方便cookie的替换，Connection项设置为close避免过多请求带来的可能异常（或许这一项并不需要？）

2、requests.get()后记得进行utf8编码

3、用json.loads()解析res，得到字典格式的json_data

4、参照之前开发者工具中显示的层级关系，在json_data中获得需要的数据

5、认证信息存在为空但用户是认证用户的情况需要特判（verified_type是200、220这些）

import requests, jsondef user_info(user_id):    params = {        'custom': user_id,    }    headers = {        'Referer': 'https://weibo.com/' + str(user_id),        'Host': 'weibo.com',        'Cookie': cookie,        'Connection': 'close'    }    res = requests.get('https://weibo.com/ajax/profile/info', headers=headers, params=params).content.decode("utf-8")    json_data = json.loads(res)    print('博主id:', json_data['data']['user']['id'])  # 博主id    print('ip归属地:', json_data['data']['user']['location'])  # ip归属地    print('博主名:', json_data['data']['user']['screen_name'])  # 博主名    print('粉丝数:', json_data['data']['user']['followers_count'])  # 粉丝数    print('关注数:', json_data['data']['user']['friends_count'])  # 关注数    print('性别:', json_data['data']['user']['gender'])  # 性别（f & m）    print('总微博数量:', json_data['data']['user']['statuses_count'])  # 总微博数量    print('认证类型:', json_data['data']['user']['verified_type'])  # 认证类型    if json_data['data']['user']['verified_type'] != -1:        print('认证信息:', json_data['data']['user']['verified_reason'])  # 认证信息（注意缺省）    else:        print("Not verified")

verified_type的字段：

-1普通用户;

0名人,

1政府,

2企业,

3媒体,

4校园,

5网站,

6应用,

7团体（机构）,

8待审企业,

200初级达人,

220中高级达人,

400已故V用户。

-1为普通用户，200和220为达人用户，0为黄V用户，其它即为蓝V用户

爬取博主博文

从前述内容可以看出，博文是按照一个个json文件由服务器发给浏览器的。我们按照月份先打包所有该博主某月的所有博文（对应的一系列JSON文件），然后再根据关键字对其中需要的博文进行筛选以及进行评论的爬取。

注意，为了加快效率，我们在程序中设置了多线程。因而有task_queue.put(json_data)入队列的过程。(该处理原理是多线程任务的生产者-消费者模型）

并且在爬取过程中有一定概率会请求服务器失败，我设置了重新请求的处理（retried_times）

# 请求单页博文页的函数def post_req(user_id, page, curmonth, retried_times):    headers = {        'Host': 'weibo.com',        'Referer': 'https://weibo.com/' + user_id + '?refer_flag=1001030103_',        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:105.0) Gecko/20100101 Firefox/105.0',        'Cookie': cookie,    }    params = {        'uid': user_id,        'page': page,        'feature': 0,        'displayYear': 2023,        'curMonth': curmonth,        'stat_date': '2023' + str(curmonth).rjust(2, '0'),    }    if retried_times < 5:        try:            res = requests.get('https://weibo.com/ajax/statuses/mymblog', headers=headers,   params=params).content.decode('UTF-8')        except:            traceback.print_exc()            random_sleep()            retried_times += 1            post_req(user_id, page, curmonth, retried_times)        else:            json_data = json.loads(res)  # 加载json文件            if json_data['data']['list']:                return json_data            else:                return None    else:        return -1# 打包所有当月博文jsondef post_page_list_producer(task_queue, user_id, curmonth):    post_page = 1    while True:        # random_sleep()        json_data = post_req(user_id, post_page, curmonth, retried_times=0)        if json_data is None:            print('全部博文包请求成功，完成打包')            break        elif json_data == -1:            return -1        else:            task_queue.put(json_data)            print('博文包请求成功', 'month:', curmonth, 'page:', post_page)            post_page += 1            random_sleep()

打包完成后，我们使用遍历对所有博文据关键字筛选，然后对博文信息进行写入数据库操作。

def post_json_data_insert(user_id, object, in_db, in_cursor, inserted_list):    # 处理关键字并匹配    pattern = re.compile('|'.join(key_words))    text = object['text_raw']    result_findall = pattern.findall(text)    if result_findall and object['comments_count'] >= 10:        if object['id'] not in inserted_list:            date_formatted = date_format(object)            if '展开' in object['text']:                mblogid = object['mblogid']                req = long_text_req(user_id, mblogid)                print(object['text_raw'])                object['text_raw'] = req['data']['longTextContent']                print(object['text_raw'])            if 'retweeted_status' in object:                print("#################################################")                print(object['retweeted_status']['user']['verified_type'])                print(object['retweeted_status']['text_raw'])                print("#################################################")                in_cursor.execute(                    "INSERT INTO blog_base("                    "blog_date,"                    "blog_text,"                    "blog_id,"                    "blog_comments,"                    "blog_likes,"                    "blog_keywords,"                    "id_blogger,"                    "retweeted_blogger_name,"                    "retweeted_verified_type,"                    "retweeted_text"                    ") VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s); ",                    (date_formatted,                     object['text_raw'],                     object['id'],                     object['comments_count'],                     object['attitudes_count'],                     str(set(result_findall)).strip('{}'),                     user_id,                     object['retweeted_status']['user']['screen_name'],                     object['retweeted_status']['user']['verified_type'],                     object['retweeted_status']['text_raw']                     )                )            else:                in_cursor.execute(                    "INSERT INTO blog_base("                    "blog_date,"                    "blog_text,"                    "blog_id,"                    "blog_comments,"                    "blog_likes,"                    "blog_keywords,"                    "id_blogger"                    ") VALUES(%s,%s,%s,%s,%s,%s,%s); ",                    (date_formatted,                     object['text_raw'],                     object['id'],                     object['comments_count'],                     object['attitudes_count'],                     str(set(result_findall)).strip('{}'),                     user_id                     )                )            in_db.commit()            print('相关博文', object['text_raw'], '\n')            counts = 0            # 调用评论抓取函数            if post_comment_data(user_id, object['id'], in_db, in_cursor, counts) is None:                print('id%s 评论获得失败，删除该条博文数据、程序终止' % object['id'])                in_cursor.execute("delete from comment_base where id_blog=%s" % object['id'])                in_cursor.execute("delete from blog_base where blog_id=%s" % object['id'])                in_cursor.commit()                return -1            inserted_list.append(object['id'])  # 记录该博文id防止重复            return 1        else:            print('id%s 已存博文，跳过该条' % object['id'])            return 1    else:        print('无关博文(comments_count:%s)' % object['comments_count'])        return 1# 消化队列中的博文jsondef post_data_get_consumer(task_queue, user_id, inserted_list):  # 多线程处理page的json_data    global thread_status    in_db = pymysql.connect(host='localhost', port=3306, user='root', passwd='root', db='dataday0224',charset='utf8mb4')    in_cursor = in_db.cursor()    while task_queue.empty() is not True:        json_data = task_queue.get()        for object in json_data['data']['list']:            date_formatted = date_format(object)            if is_that_day(date_formatted):                if post_json_data_insert(user_id, object, in_db, in_cursor, inserted_list) == -1:                    print('由于本条博文数据插入失败，该线程结束')                    thread_status = -1                    return -1            else:                print(date_formatted+'is not wanted date')        task_queue.task_done()

在post_json_data_insert()的尾部我们调用了post_comment_data函数进行博文下评论的爬取。需要注意的是，由于评论信息存在一个需要特别关注的max_id参数需要处理，post_comment_data采集完第一次评论数据后，后续的评论数据会调用post_comment_data_rest()函数来处理max_id问题：

def post_comment_data_rest(user_id, post_id, max_id, in_db, in_cursor, counts):    headers = {        'Host': 'weibo.com',        'Referer': 'https://weibo.com/' + str(user_id) + '?refer_flag=1001030103_',        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:105.0) Gecko/20100101 Firefox/105.0',        'Cookie': cookie,        # 'Connection': 'close'    }    params = {        'flow': 1,        'id': post_id,        'is_show_bulletin': 2,        'count': 20,        'max_id': max_id,    }    try:        res = requests.get('https://weibo.com/ajax/statuses/buildComments', headers=headers,                           params=params).content.decode('UTF-8')        json_data = json.loads(res)  # 加载json文件    except:        print('【' + str(post_id) + '】''post_comment_data_rest() request error, trying again...')        traceback.print_exc()        try:            post_comment_data_rest(user_id, post_id, max_id, in_db, in_cursor, counts)        except:            print("retried failed")            return None    else:        for object in json_data['data']:            if counts >= 1000:                break            date_formatted = date_format(object)            random_sleep()            print('【', str(date_formatted), '[' + str(object['floor_number']) + ']】', '正在存储该条评论, postID:',                  post_id)            user_json = object['user']            if user_json['verified_type'] != -1:                try:                    veri_info = user_json['verified_reason']  # 认证信息（注意缺省）                except:                    veri_info = ''            else:                veri_info = "Not verified"            try:                in_cursor.execute(                    "INSERT INTO comment_base("                    "comment_date,"                    "comment_text,"                    "commenter_name,"                    "commenter_region,"                    "commenter_id,"                    "commenter_fans,"                    "commenter_follow,"                    "commenter_gender,"                    "commenter_weibo,"                    "commenter_veri_type,"                    "commenter_veri_info,"                    "id_blog"                    ") VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);",                    (date_formatted,                     object['text'],                     object['user']['screen_name'],                     user_json['location'],                     object['user']['id'],                     user_json['followers_count'],                     user_json['friends_count'],                     user_json['gender'],                     user_json['statuses_count'],                     user_json['verified_type'],                     veri_info,                     post_id                     )                )                in_db.commit()                counts += 1            except:                print('An sql error on post_comment_data_rest')                print(object['text'])                traceback.print_exc()        if json_data['max_id'] != 0:            random_sleep()            return post_comment_data_rest(user_id, post_id, json_data['max_id'], in_db, in_cursor, counts)        else:            return 1

该程序的主功能集成函数为：

def all_post_data(user_id):  # user_id的全体博文内容获取    global thread_status    thread_status = 1    months = [2, ]    restartPoint = None    while True:  # 生产者队列获取页面json失败时，重复该月json获取        for curmonth in months:            if restartPoint:                if curmonth != restartPoint:                    continue            print('正在打包博主%s月的全部博文页...' % curmonth)            task_queue = Queue()            if post_page_list_producer(task_queue, user_id, curmonth) == -1:  # 队列一次性获得全部的当月博文页                restartPoint = curmonth                break            inserted_list = []            thread_list = []            for index in range(20):                consumer_thread = Thread(target=post_data_get_consumer, args=(task_queue, user_id, inserted_list))                thread_list.append(consumer_thread)            for t in thread_list:                t.start()            for t in thread_list:                t.join()            if thread_status == -1:                print('有线程未完成任务，程序终止：检查用户 %s 的 %s 月博文数据' % (user_id, curmonth))                exit()            else:                print('用户 %s 的 %s 月数据采集完成' % (user_id, curmonth))            restartPoint = None        if restartPoint is None:            break

自此，对代码的主要设计部分讲解结束。除此之外，脚本中还有一些边缘的函数（如日期格式化、处理文本内容采集不全的函数）。

完整代码请访问我的github项目：GitHub - otonashi-ayana/SpiderWeibo: Collect blog posts and comments, commentator information and build a database