爬虫代码

Posted 2020-09-22 李雪琦

tieba_spider.py:

# -*- coding: utf-8 -*-

import scrapy
import json
from tieba.items import ThreadItem, PostItem, CommentItem
import helper
import time

class TiebaSpider(scrapy.Spider):
name = "tieba"
cur_page = 1 #modified by pipelines (open_spider)
end_page = 9999
filter = None
see_lz = False

def parse(self, response): #forum parser
for sel in response.xpath(‘//li[contains(@class, "j_thread_list")]‘):
data = json.loads(sel.xpath(‘@data-field‘).extract_first())
item = ThreadItem()
item[‘id‘] = data[‘id‘]
item[‘author‘] = data[‘author_name‘]
item[‘reply_num‘] = data[‘reply_num‘]
item[‘good‘] = data[‘is_good‘]
if not item[‘good‘]:
item[‘good‘] = False
item[‘title‘] = sel.xpath(‘.//div[contains(@class, "threadlist_title")]/a/text()‘).extract_first()
if self.filter and not self.filter(item["id"], item["title"], item[‘author‘], item[‘reply_num‘], item[‘good‘]):
continue
#filter过滤掉的帖子及其回复均不存入数据库

yield item
meta = {‘thread_id‘: data[‘id‘], ‘page‘: 1}
url = ‘http://tieba.baidu.com/p/%d‘ % data[‘id‘]
if self.see_lz:
url += ‘?see_lz=1‘
yield scrapy.Request(url, callback = self.parse_post, meta = meta)
next_page = response.xpath(‘//a[@class="next pagination-item "]/@href‘)
self.cur_page += 1
if next_page:
if self.cur_page <= self.end_page:
yield self.make_requests_from_url(next_page.extract_first())

def parse_post(self, response):
meta = response.meta
has_comment = False
for floor in response.xpath("//div[contains(@class, ‘l_post‘)]"):
if not helper.is_ad(floor):
data = json.loads(floor.xpath("@data-field").extract_first())
item = PostItem()
item[‘id‘] = data[‘content‘][‘post_id‘]
item[‘author‘] = data[‘author‘][‘user_name‘]
item[‘comment_num‘] = data[‘content‘][‘comment_num‘]
if item[‘comment_num‘] > 0:
has_comment = True
content = floor.xpath(".//div[contains(@class,‘j_d_post_content‘)]").extract_first()
#以前的帖子, data-field里面没有content
item[‘content‘] = helper.parse_content(content, True)
#以前的帖子, data-field里面没有thread_id
item[‘thread_id‘] = meta[‘thread_id‘]
item[‘floor‘] = data[‘content‘][‘post_no‘]
#只有以前的帖子, data-field里面才有date
if ‘time‘ in data[‘content‘].keys():
item[‘time‘] = data[‘content‘][‘date‘]
#只有以前的帖子, data-field里面才有date
else:
item[‘time‘] = floor.xpath(".//span[@class=‘tail-info‘]")\
.re_first(r‘[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}‘)
yield item
if has_comment:
url = "http://tieba.baidu.com/p/totalComment?tid=%d&fid=1&pn=%d" % (meta[‘thread_id‘], meta[‘page‘])
if self.see_lz:
url += ‘&see_lz=1‘
yield scrapy.Request(url, callback = self.parse_comment, meta = meta)
next_page = response.xpath(u".//ul[@class=‘l_posts_num‘]//a[text()=‘下一页‘]/@href")
if next_page:
meta[‘page‘] += 1
url = response.urljoin(next_page.extract_first())
yield scrapy.Request(url, callback = self.parse_post, meta = meta)

def parse_comment(self, response):
comment_list = json.loads(response.body)[‘data‘][‘comment_list‘]
for value in comment_list.values():
comments = value[‘comment_info‘]
for comment in comments:
item = CommentItem()
item[‘id‘] = comment[‘comment_id‘]
item[‘author‘] = comment[‘username‘]
item[‘post_id‘] = comment[‘post_id‘]
item[‘content‘] = helper.parse_content(comment[‘content‘], False)
item[‘time‘] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(comment[‘now_time‘]))
yield item

 

helper.py:

# -*- coding: utf-8 -*-

import re
import urllib2
from bs4 import BeautifulSoup
import emotion

def is_ad(s): #判断楼层是否为广告
ad = s.xpath(u".//span[contains(text(), ‘广告‘)]")
# 广告楼层中间有个span含有广告俩字
return ad

def parse_content(content, is_post):
if not content or not content.strip():
return None
content = content.replace(‘\r‘, ‘\n‘) #古老的帖子会出现奇怪的\r
s = BeautifulSoup(content, ‘lxml‘)
if is_post:
s = s.div #post 外层有个div

l = list(s.children)
for i in range(len(l)):
parse_func = (is_str, is_br, is_img, is_video, other_case)
for func in parse_func:
try:
ret = func(l[i])
except:
continue
if ret is not False:
l[i] = ret
break

return strip_blank(‘‘.join(l))

def strip_blank(s): #按个人喜好去掉空白字符
s = re.sub(r‘\n[ \t]+\n‘, ‘\n‘, s)
s = re.sub(r‘ +‘, ‘ ‘, s) #去掉多余的空格
s = re.sub(r‘\n\n\n+‘, ‘\n\n‘, s) #去掉过多的连续换行
return s.strip()

def is_str(s):
if s.name:
return False
#NavigableString类型需要手动转换下
return unicode(s)

def is_br(s):
if s.name == ‘br‘:
return ‘\n‘
return False

def is_img(s):
# 处理了部分表情
if s.name == ‘img‘:
src = unicode(s.get(‘src‘))
return emotion.get_text(src)
return False

def is_video(s):
t = unicode(s.get(‘class‘))
if ‘video‘ in t:
url = s.find(‘a‘).get(‘href‘)
return ‘ ‘ + getJumpUrl(url) + ‘ ‘
return False

#bs带的get_text功能，很好很强大
#粗体红字之类的都一句话搞定了
def other_case(s):
return s.get_text()