第五天，知乎问题和回答字段提取和存入数据库-白红宇

第五天，知乎问题和回答字段提取和存入数据库

阅读量：5075 次

发布时间：2019-06-12

本文共 13082 字，大约阅读时间需要 43 分钟。

对应github地址：

摘要：

1. Scrapy的Request类支持设置cookie属性，要在爬虫请求中带上cookie，可以重载Spider的start_requests方法。start_requests()方法可以返回一个请求给爬虫的起始网站，这个返回的请求相当于start_urls，start_requests()返回的请求会替代start_urls里的请求

参考：

2. json.loads把json字符串转变为python格式的，json.dumps把python格式字符串转为json格式

一. selenium进行模拟登陆

进入项目目录后，执行下面代码

scrapy genspider zhihu_sel

1. 模拟登陆知乎并获取cookie信息

from selenium import webdriverimport timeimport pickle def start_requests(self):     browser = webdriver.Chrome()    browser.get('https://www.zhihu.com/signin')     input1 = browser.find_element_by_css_selector("input[name=username]")    input1.send_keys('xx')    input2 = browser.find_element_by_css_selector("input[name=password]")    input2.send_keys('xx')    button = browser.find_element_by_class_name('SignFlow-submitButton')    button.click()     time.sleep(10)    Cookies = browser.get_cookies()    print(Cookies)    cookie_dict = {}    for cookie in Cookies:        f = open('./cookie/' + cookie['name']+'.zhihu', 'wb')        pickle.dump(cookie, f)        f.close()        cookie_dict[cookie['name']] = cookie['value']     browser.close()    return [scrapy.Request(url=self.start_urls[0], dont_filter=True, cookies=cookie_dict, headers=self.headers)] 使用scrapy读取本地cookie文件的时候，需要在加上最后一行代码 并在zhihu_sql.py中添加如下信息，这样可以保证后续的request请求都把cookie信息自动加上去custom_settings = {        "COOKIES_ENABLED": True,        "DOWNLOAD_DELAY": 1.5,    }

注意：

1）只能使用chrome60版本和相应的驱动，否则会报grant type错误

2） www.zhihu.com/signin这个网站地址很简洁方便模拟登陆，网上搜的地址很麻烦，登不上

3) 经测试，代码中cookie文件的保存位置./cookie必须是在cmd中进入虚拟环境后，使用mkdir cookie命令建立目录，其他情况都不能保存文件到cookie目录中

2. 修改调试的main.py，把jobbole注释掉，添加知乎信息

3. 使用requests来模拟登陆知乎，自己查资料里的zhihu_login_requests.py文件

将cookie信息保存在本地，下次登陆直接读取cookie本地文件进行登陆

注意：

1）csrf会在用户名和密码的session信息中加入一段随机码并加密存储在session_value中，保存在数据库中的session包括session_key, session_value, 有效时间。

二. 知乎分析和数据表设计

1. scrapy shell中增加user_agent信息

有时候代码访问网页的时候不加user_agent信息，会访问不到页面内容，在cmd虚拟环境中执行

scrapy shell -s USER_AGENT="头信息"

然后执行如下代码，可以把网页内容写入到自定义文件中

2. 安装插件JsonView可以有序的查看json信息，需要加载解压包里的webcontent

然后双击ajax网页链接就可查看

3. 问题和回答数据表设计

ask表

注意：ask表中并没有create_time和update_time字段，所以在设计item时可以不添加这两个字段，但是可以在answer的ajax数据中找到，可以后面在ask表中加上

answer表

注意：用户回答可以是匿名的，所以author_id可为空

三. item loader方式提取question

1. scrapy默认是使用深度优先算法来提取信息

2. 在F12开发者工具中，Request Header中一般能找到如下两条信息

3. 为防止被ban，执行如下措施

1）添加头信息

headers = {    # HOST就是要访问的域名地址，https://blog.csdn.net/zhangqi_gsts/article/details/50775341    "HOST": "www.zhihu.com",    # referer表示从哪个网页跳转过来的，可防止盗链。https://blog.csdn.net/shenqueying/article/details/79426884    "Referer": "https://www.zhihu.com",    'User-Agent': "user-agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "                  "Chrome/60.0.3112.113 Safari/537.36"}

2）禁用cookie和设置下载延迟

# 重点：防止被bancustom_settings = {    "COOKIES_ENABLED": True,    "DOWNLOAD_DELAY": 1}

3. 编写parse函数

selenium模拟登陆后，会首先执行parse函数

"""

提取出html页面中的所有url 并跟踪这些url进行一步爬取

如果提取的url中格式为 /question/xxx 就下载之后直接进入解析函数

"""

def parse(self, response):    all_urls = response.css("a::attr(href)").extract()    all_urls = [parse.urljoin(response.url, url) for url in all_urls]    # 使用lambda函数对于每一个url进行过滤，如果是true放回列表，返回false去除。    all_urls = filter(lambda x: True if x.startswith("https") else False, all_urls)    for url in all_urls:        # 具体问题以及具体答案的url我们都要提取出来。用或关系实现，要用小括号括起来。因为具体答案的url没斜杠        match_obj = re.match("(.*zhihu.com/question/(\d+))(/|$).*", url)        if match_obj:            # 如果提取到question相关的页面则下载后交由提取函数进行提取            request_url = match_obj.group(1)             yield scrapy.Request(request_url, headers=self.headers, callback=self.parse_question)        else:            # 注释这里方便调试            pass            # 如果不是question页面则直接进一步跟踪            yield scrapy.Request(url, headers=self.headers, callback=self.parse)

说明：

1）拼接相对地址，并过滤出来以https开头的URL

from urllib import parseurl1 = response.css("a::attr(href)").extract()url2 = [parse.urljoin(response.url, url) for url in url1]

2） x是url2中的每一个值，如果x是以https开头的就为True，然后就可以过滤出来存放到url3中

url3 = filter(lambda x: True if x.startswith("https") else False, url2)如果觉得不好理解，可以如下写url_list = []for url in url3:    if url.startwith("https"):        url_list = url_list.append(url)

4. 补充内容，filter，map

filter() 函数用于过滤序列，过滤掉不符合条件的元素，返回一个迭代器对象，如果要转换为列表，可以使用 list() 来转换。

该接收两个参数，第一个为函数，第二个为序列，序列的每个元素作为参数传递给函数进行判，然后返回 True 或 False，最后将返回 True 的元素放到新列表中。

def is_odd(n):     return n % 2 == 1 tmplist = filter(is_odd, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) newlist = list(tmplist) print(newlist)

map()接收一个函数 f 和一个或多个list，并通过把函数 f 依次作用在 list 的每个元素上，得到一个新的 list 并返回。

# 一个列表的情况>>> map(lambda x: x ** 2, [1, 2, 3, 4, 5])[1, 4, 9, 16, 25]# 提供了两个列表，对相同位置的列表数据进行相加 >>> map(lambda x, y: x + y, [1, 3, 5, 7, 9], [2, 4, 6, 8, 10]) [3, 7, 11, 15, 19]

5. 参看定义好的数据库字段，在items.py中定义问题和答案的item

# 知乎问题的itemclass ZhihuQuestionItem(scrapy.Item):    zhihu_id = scrapy.Field()    topoics = scrapy.Field()    url = scrapy.Field()    title = scrapy.Field()    content = scrapy.Field()    answer_num = scrapy.Field()    comments_num = scrapy.Field()    watch_user_num = scrapy.Field()    click_num = scrapy.Field()    crawl_time = scrapy.Field()    crawl_update_time = scrapy.Field()  # 知乎回答的itemclass ZhihuAnswerItem(scrapy.Item):    zhihu_id = scrapy.Field()    url = scrapy.Field()    question_id = scrapy.Field()    author_id = scrapy.Field()    content = scrapy.Field()    praise_num = scrapy.Field()    comments_num = scrapy.Field()    create_time = scrapy.Field()    update_time = scrapy.Field()    crawl_time = scrapy.Field()    crawl_update_time = scrapy.Field()

6. 在zhuhu.py中编写parse_question函数，从页面中提取问题的各字段

使用下面命令可以测试知乎字段爬取是否有效

scrapy shell -s USER_AGENT="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.113 Safari/537.36"

例如：提取一个回答的方法

# response.css(".QuestionAnswers-answers .List-item:nth-child(1) .RichContent-inner span::text").extract()

def parse_question(self, response):    # 处理新版本, 新版本有唯一类QuestionHeader-title来设置标题，老版本没这个类    if "QuestionHeader-title" in response.text:         match_obj = re.match("(.*zhihu.com/question/(\d+))(/|$).*", response.url)        if match_obj:            # group(2)取到的为(\d+)中的内容            question_id = int(match_obj.group(2))         # 使用scrapy默认提供的ItemLoder使代码更简洁，首先实例化        item_loader = ItemLoader(item=ZhihuQuestionItem(), response=response)         item_loader.add_value("url_object_id", get_md5(response.url))        item_loader.add_value("zhihu_id", question_id)        item_loader.add_css("title", "h1.QuestionHeader-title::text")        # 下面一个回答内容的例子，可参考下，提取content的方法提取了所有回答内容        # response.css(".QuestionAnswers-answers .List-item:nth-child(1) .RichContent-inner span::text").extract()        item_loader.add_css("content", ".QuestionAnswers-answers")        item_loader.add_css("topics", ".QuestionHeader-topics .Tag.QuestionTopic .Popover div::text")        item_loader.add_css("answer_num", ".List-headerText span::text")        item_loader.add_css("comments_num", ".QuestionHeader-Comment button::text")        # 这里的watch_user_num 包含Watch 和 click, 在clean data中分离        item_loader.add_css("watch_user_num", ".NumberBoard-itemValue ::text")        item_loader.add_value("url", response.url)         question_item = item_loader.load_item()  # 发起向后台具体answer的接口请求yield scrapy.Request(self.start_answer_url.format(question_id, 20, 0), headers=self.headers,                     callback=self.parse_answer)yield question_item

注意

1）调试的时候可以在if "QuestionHeader-title" in response.text:处打一个断点，一路F6一步步调试，否则会报数据库的错误

2）使用xpath"或"的方式来提取字段

有时候不同页面中的标题会有2种格式，比如<a>标签里的标题和<span>span标签里的标题

此时使用css的方式response.css(".zh-question-title a:text")就不能适用了，这时需要一个或的表示方式，可用xpath来实现，如下

item_loader.add_xpath("title","//*[@id='zh-question-title']/h2/a/text()|//*[@id='zh-question-title']/h2/span/text()")

7. 编写回答的处理函数 parse_answer

回答的内容是用ajax加载的，经分析发现有api接口可调用，如下面的next和previous地址

7.1 首先在zhuhu.py中定义一个变量来发起一个关于回答的初始请求，注意里面的变量要替换为{0}，{1}，{2}

start_answer_url = "https://www.zhihu.com/api/v4/questions/{0}/answers?include=data%5B%2A%5D.is_normal%2Cadmin_closed_comment%2Creward_info%2Cis_collapsed%2Cannotation_action%2Cannotation_detail%2Ccollapse_reason%2Cis_sticky%2Ccollapsed_by%2Csuggest_edit%2Ccomment_count%2Ccan_comment%2Ccontent%2Ceditable_content%2Cvoteup_count%2Creshipment_settings%2Ccomment_permission%2Ccreated_time%2Cupdated_time%2Creview_info%2Crelevant_info%2Cquestion%2Cexcerpt%2Crelationship.is_authorized%2Cis_author%2Cvoting%2Cis_thanked%2Cis_nothelp%3Bdata%5B%2A%5D.mark_infos%5B%2A%5D.url%3Bdata%5B%2A%5D.author.follower_count%2Cbadge%5B%2A%5D.topics&limit={1}&offset={2}&sort_by=default" def parse_answer(self, response):    # json.loads把json字符串转变为python格式的    ans_json = json.loads(response.text)    # 判断是否有后续页面，以及下一个页面的URL，就是页面分析中Preview里的paging信息    is_end = ans_json["paging"]["is_end"]    next_url = ans_json["paging"]["next"]     # 提取answer的具体字段    for answer in ans_json["data"]:        answer_item = ZhihuAnswerItem()         #answer_item["url_object_id"] = get_md5(url=answer["url"])        answer_item["zhihu_id"] = answer["id"]        answer_item["question_id"] = answer["question"]["id"]        # 有时候回答是匿名的，此时author字段中没id值，那么就返回None        answer_item["author_id"] = answer["author"]["id"] if "id" in answer["author"] else None        answer_item["author_name"] = answer["author"]["name"] if "name" in answer["author"] else None        answer_item["content"] = answer["content"] if "content" in answer else None        answer_item["praise_num"] = answer["voteup_count"]        answer_item["comments_num"] = answer["comment_count"]        answer_item["url"] = "https://www.zhihu.com/question/{0}/answer/{1}".format(answer["question"]["id"], answer["id"])        answer_item["create_time"] = answer["created_time"]        answer_item["update_time"] = answer["updated_time"]        answer_item["crawl_time"] = datetime.now()         yield answer_item     # 如果不是最后一个URL，继续请求下一个页面    if not is_end:        yield scrapy.Request(next_url, headers=self.headers, callback=self.parse_answer)

四. 数据入库

方法一：根据不同的item执行不同的mysql语句

def do_insert(self, cursor, item):

if item.__class__.__name__ == "JobBoleArticleItem":

insert_sql = """..."""

上面代码可以取到当前函数所在类的名字

上面这种方法把Item名字写死了，后期如果有修改就比较麻烦

方法二，可以把不同的sql语句写在items.py中的具体项目的类中，定义一个函数来存放sql语句

1. 知乎问题类的item，ZhihuQuestionItem中添加如下代码

def get_insert_sql(self):    # 插入知乎question表的sql语句    insert_sql = """        insert into zhihu_question(zhihu_id, topics, url, title, content, answer_num, comments_num,          watch_user_num, click_num, crawl_time          )        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)        ON DUPLICATE KEY UPDATE content=VALUES(content), answer_num=VALUES(answer_num), comments_num=VALUES(comments_num),          watch_user_num=VALUES(watch_user_num), click_num=VALUES(click_num)    """# scrapy.Field返回类型为列表zhihu_id = self["zhihu_id"][0]topics = ",".join(self["topics"])url = self["url"][0]title = "".join(self["title"])content = "".join(self["content"])# extract_num就是上面定义的get_nums，只是把它重新定义为一个常用函数了answer_num = extract_num("".join(self["answer_num"]))comments_num = extract_num("".join(self["comments_num"])) # 浏览数和点击数是一起取出来的，并且用逗号分隔，需要单独取出来if len(self["watch_user_num"]) == 2:    watch_user_num_click = self["watch_user_num"]    watch_user_num = extract_num_include_dot(watch_user_num_click[0])    click_num = extract_num_include_dot(watch_user_num_click[1])else:    watch_user_num_click = self["watch_user_num"]    watch_user_num = extract_num_include_dot(watch_user_num_click[0])    click_num = 0 # 要把时间格式转为字符串格式crawl_time = datetime.datetime.now().strftime(SQL_DATETIME_FORMAT)# 顺序要和sql语句中的保持一样params = (zhihu_id, topics, url, title, content, answer_num, comments_num,          watch_user_num, click_num, crawl_time) return insert_sql, params

注意：

浏览数取出来的方式，单独定义了一个函数extract_num_include_dot，早utils.common文件中

2，改造MysqlTwistedPipeline

def do_insert(self, cursor, item):    # 根据不同的Item构建不同的sql语句并插入到mysql中    insert_sql, params = item.get_insert_sql()    cursor.execute(insert_sql, params)

3. ZhihuAnswerItem(scrapy.Item)中添加插入数据代码

def get_insert_sql(self):    # 插入知乎question表的sql语句    insert_sql = """        insert into zhihu_answer(zhihu_id, url, question_id, author_id, author_name, content, comments_num,          create_time, update_time, crawl_time          ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)          ON DUPLICATE KEY UPDATE content=VALUES(content), comments_num=VALUES(comments_num),          update_time=VALUES(update_time)    """     # int类型转为datetime类型，需要使用fromtimestamp函数；再转为字符串类型，需要strftime函数    create_time = datetime.datetime.fromtimestamp(self["create_time"]).strftime(SQL_DATETIME_FORMAT)    update_time = datetime.datetime.fromtimestamp(self["update_time"]).strftime(SQL_DATETIME_FORMAT)     params = (        self["zhihu_id"], self["url"], self["question_id"],        self["author_id"], self["author_name"], self["content"],        self["comments_num"], create_time, update_time,        self["crawl_time"].strftime(SQL_DATETIME_FORMAT),    )     return insert_sql, params